Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
20 changes: 11 additions & 9 deletions sdks/python/src/opik/runner/snapshot.py
Original file line number Diff line number Diff line change
Expand Up @@ -35,14 +35,18 @@
re.compile(r"entrypoint:\s*true"),
]

_CONFIGURATION_PATTERNS = [
re.compile(r"get_or_create_config\("),
re.compile(r"create_config\("),
re.compile(r"getOrCreateConfig\("),
re.compile(r"createConfig\("),
_PROMPT_PATTERNS = [
re.compile(r"\.get_prompt\("),
re.compile(r"\.create_prompt\("),
re.compile(r"\.get_chat_prompt\("),
re.compile(r"\.create_chat_prompt\("),
re.compile(r"\.getPrompt\("),
re.compile(r"\.createPrompt\("),
Comment thread
petrotiurin marked this conversation as resolved.
re.compile(r"\.getChatPrompt\("),
re.compile(r"\.createChatPrompt\("),
]

_ALL_PATTERNS = _TRACING_PATTERNS + _ENTRYPOINT_PATTERNS + _CONFIGURATION_PATTERNS
_ALL_PATTERNS = _TRACING_PATTERNS + _ENTRYPOINT_PATTERNS + _PROMPT_PATTERNS


def _git_files(repo_root: Path) -> Optional[Set[str]]:
Expand Down Expand Up @@ -79,9 +83,7 @@ def build_checklist(
"entrypoint": any(
_matches_any(line, _ENTRYPOINT_PATTERNS) for line in matches
),
"configuration": any(
_matches_any(line, _CONFIGURATION_PATTERNS) for line in matches
),
"prompts": any(_matches_any(line, _PROMPT_PATTERNS) for line in matches),
Comment thread
petrotiurin marked this conversation as resolved.
},
"instrumentation_matches": matches,
}
Expand Down
92 changes: 92 additions & 0 deletions sdks/python/tests/unit/runner/test_snapshot.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,9 @@
import platform
import sys
from pathlib import Path
from typing import Dict

import pytest

from opik.runner.snapshot import build_checklist

Expand Down Expand Up @@ -60,3 +63,92 @@ def test_build_checklist__command_none__serialized_as_none(
result = build_checklist(tmp_path, command=None)

assert result["command"] is None

@pytest.mark.parametrize(
"files,expected",
[
pytest.param(
{"app.py": "x = 1\n"},
{"tracing": False, "entrypoint": False, "prompts": False},
id="empty_repo",
),
pytest.param(
{"app.py": "import opik\n@opik.track\ndef f():\n pass\n"},
{"tracing": True, "entrypoint": False, "prompts": False},
id="tracing_only",
),
pytest.param(
{"agent.py": "entrypoint = True\n"},
{"tracing": False, "entrypoint": True, "prompts": False},
id="entrypoint_python",
),
pytest.param(
{"agent.ts": "export const config = { entrypoint: true };\n"},
{"tracing": False, "entrypoint": True, "prompts": False},
id="entrypoint_ts",
),
pytest.param(
{"app.py": "client.get_prompt('hello')\n"},
{"tracing": False, "entrypoint": False, "prompts": True},
id="get_prompt_python",
),
pytest.param(
{"app.py": "client.create_prompt('hello', 'tmpl')\n"},
{"tracing": False, "entrypoint": False, "prompts": True},
id="create_prompt_python",
),
pytest.param(
{"app.py": "client.get_chat_prompt('hello')\n"},
{"tracing": False, "entrypoint": False, "prompts": True},
id="get_chat_prompt_python",
),
pytest.param(
{"app.py": "client.create_chat_prompt('hello', [])\n"},
{"tracing": False, "entrypoint": False, "prompts": True},
id="create_chat_prompt_python",
),
pytest.param(
{"main.ts": "await client.getPrompt('hello');\n"},
{"tracing": False, "entrypoint": False, "prompts": True},
id="get_prompt_ts",
),
pytest.param(
{"main.ts": "await client.prompts.createPrompt({});\n"},
{"tracing": False, "entrypoint": False, "prompts": True},
id="create_prompt_ts",
),
pytest.param(
{"main.ts": "await client.getChatPrompt('hello');\n"},
{"tracing": False, "entrypoint": False, "prompts": True},
id="get_chat_prompt_ts",
),
pytest.param(
{"main.ts": "await client.createChatPrompt({});\n"},
{"tracing": False, "entrypoint": False, "prompts": True},
id="create_chat_prompt_ts",
),
pytest.param(
{
"app.py": (
"import opik\n"
"entrypoint = True\n"
"client.get_chat_prompt('hello')\n"
),
},
{"tracing": True, "entrypoint": True, "prompts": True},
id="all_flags_set",
),
],
)
def test_build_checklist__instrumentation_flags(
self,
tmp_path: Path,
files: Dict[str, str],
expected: Dict[str, bool],
) -> None:
for name, content in files.items():
(tmp_path / name).write_text(content)

result = build_checklist(tmp_path, command=None)

assert result["instrumentation"] == expected
Loading