diff --git a/lamindb/.agents b/lamindb/.agents index d109acd4d..2c252f7fe 160000 --- a/lamindb/.agents +++ b/lamindb/.agents @@ -1 +1 @@ -Subproject commit d109acd4d939cbcf362aee56e599f9897706cc67 +Subproject commit 2c252f7fee6daa1fe22e2df4c64fe8897ca872e3 diff --git a/lamindb/core/__init__.py b/lamindb/core/__init__.py index 00a1f303c..8faef61eb 100644 --- a/lamindb/core/__init__.py +++ b/lamindb/core/__init__.py @@ -42,7 +42,7 @@ from . import subsettings from ._context import Context from ._settings import Settings - +from ._verify_lineage import verify_lineage def __getattr__(name: str): # need to lazy import a few auxliary modules to maintain backward compatibility diff --git a/lamindb/core/_context.py b/lamindb/core/_context.py index 4441c2ab5..95c03d0fb 100644 --- a/lamindb/core/_context.py +++ b/lamindb/core/_context.py @@ -578,13 +578,6 @@ def _track( lamin save /path/to/.cursor/plans/curate-dataset-x.plan.md lamin save /path/to/.claude/plans/curate-dataset-x.md - .. dropdown:: `lamin track copilot` says it can't find the active session? - - In VS Code, make sure **"Copilot"** is selected — not **"Local"** — in the mode picker below the chat input box. `lamin track copilot` can only see sessions that go through the "Copilot"; sessions run via "Local" aren't visible to it. - - .. image:: https://lamin-site-assets.s3.amazonaws.com/.lamindb/f7Nw4RNYkvlw966d0000.png - :width: 800px - To sync code with a git repo, see: :ref:`sync-code-with-git`. To track parameters and features, see: :ref:`track-run-parameters`. diff --git a/lamindb/core/_verify_lineage.py b/lamindb/core/_verify_lineage.py new file mode 100644 index 000000000..66592cdb7 --- /dev/null +++ b/lamindb/core/_verify_lineage.py @@ -0,0 +1,436 @@ +import ast +import re +from dataclasses import dataclass +from pathlib import Path + +# Path matching regex for URIs, absolute/relative paths, and extensions +PATH_PATTERN = re.compile( + r"^(?:" + r"[a-zA-Z0-9]+://" # URIs (s3://, gs://, https://) + r"|[a-zA-Z]:[\\/]" # Windows drive prefix (C:\, D:/) + r"|~?[/\\]|\.[\./][/\\]" # Path prefixes (/, ./, ../, ~/) + r")[\w\-\.\s/]+$" # Path body + r"|^[\w\-\.\s]+[/\\][\w\-\.\s/]*$" # Directory relative paths with slashes + r"|^[\w\-\.\s]+\.[a-zA-Z0-9]{1,10}$" # Filenames with extensions +) + +# Calls in these categories can carry path strings but are considered +# environment/path setup (not data lineage I/O), so they are ignored. +NON_LINEAGE_EXACT_CALLS = frozenset( + { + "Path", + "PurePath", + "PosixPath", + "WindowsPath", + "makedirs", + "os.makedirs", + } +) +NON_LINEAGE_CALL_PREFIXES = ( + "sys.path.", + "os.path.", + "importlib.", +) +NON_LINEAGE_ATTR_CALLS = frozenset( + { + "Path", + "PurePath", + "PosixPath", + "WindowsPath", + "mkdir", + "exists", + "is_file", + "is_dir", + "resolve", + "expanduser", + "glob", + "rglob", + } +) + + +@dataclass(frozen=True) +class VerifyLineageResult: + """Public result object returned by `verify_lineage()`.""" + is_fully_tracked: bool + missing_lineage: tuple[str, ...] + + +class LaminLineageChecker(ast.NodeVisitor): + def __init__(self): + # Paths observed in LaminDB API calls, keyed by path string -> line numbers. + self.tracked_paths: dict[str, list[int]] = {} + # Paths observed in non-LaminDB calls, keyed by path string -> line numbers. + self.untracked_paths: dict[str, list[int]] = {} + # Best-effort binding of variable names to path-like strings as code is traversed. + self.var_map: dict[str, set[str]] = {} # var_name -> set of path strings + # Variable names that currently refer to a lamindb Artifact(...) instance. + self.artifact_vars: set[str] = set() # vars assigned from lamindb.Artifact(...) + # Session-level lifecycle flags required for a "fully tracked" script. + self.has_track_call: bool = False + self.has_finish_call: bool = False + # Module aliases bound to `lamindb` via `import lamindb as `. + self.lamindb_module_aliases: set[str] = {"lamindb"} + # Local symbols imported from `lamindb`, keyed by local alias -> imported name. + self.imported_lamindb_symbols: dict[str, str] = {} + # Function definitions are cached so later calls can be re-visited with bound args. + self.function_defs: dict[str, ast.FunctionDef] = {} + # Prevent recursive re-entry while tracing user-defined function calls. + self._active_function_calls: set[str] = set() + + def visit_ImportFrom(self, node: ast.ImportFrom): + if node.module != "lamindb": + return + + for imported in node.names: + local_name = imported.asname or imported.name + self.imported_lamindb_symbols[local_name] = imported.name + + def visit_Import(self, node: ast.Import): + for imported in node.names: + if imported.name == "lamindb": + local_name = imported.asname or imported.name + self.lamindb_module_aliases.add(local_name) + + def visit_FunctionDef(self, node: ast.FunctionDef): + for decorator in node.decorator_list: + decorator_name = ( + self._get_func_name(decorator.func) + if isinstance(decorator, ast.Call) + else self._get_func_name(decorator) + ) + if self._is_lamindb_flow_name(decorator_name): + # `@lamindb.flow` provides run lifecycle management. + self.has_track_call = True + self.has_finish_call = True + # Save function bodies for later path-flow tracing at call sites. + self.function_defs[node.name] = node + self.generic_visit(node) + + def visit_Assign(self, node: ast.Assign): + """Trace variables assigned to path strings and LaminDB Artifact instances.""" + if self._is_env_var_setup_assignment(node): + # Environment variable assignment is runtime setup and not lineage I/O. + # Skip traversing nested helper calls (e.g. str(path)) in this subtree. + return + + paths = self._extract_paths_from_node(node.value) + is_artifact_ctor = ( + isinstance(node.value, ast.Call) + and self._is_lamindb_artifact_constructor_call(node.value) + ) + + for target in node.targets: + if isinstance(target, ast.Name): + if paths: + self.var_map[target.id] = paths + else: + self.var_map.pop(target.id, None) + if is_artifact_ctor: + self.artifact_vars.add(target.id) + else: + self.artifact_vars.discard(target.id) + elif isinstance(target, (ast.Tuple, ast.List)): + for elt in target.elts: + if isinstance(elt, ast.Name): + if paths: + self.var_map[elt.id] = paths + else: + self.var_map.pop(elt.id, None) + if is_artifact_ctor: + self.artifact_vars.add(elt.id) + else: + self.artifact_vars.discard(elt.id) + + self.generic_visit(node) + + def _is_env_var_setup_assignment(self, node: ast.Assign) -> bool: + for target in node.targets: + if not isinstance(target, ast.Subscript): + continue + if self._get_func_name(target.value) == "os.environ": + return True + return False + + def visit_Call(self, node: ast.Call): + func_name = self._get_func_name(node.func) + lineno = getattr(node, "lineno", 0) + + # Record explicit session lifecycle calls used to open/close lineage tracking. + if self._is_lamindb_module_call(func_name, "track") or self._is_imported_lamindb_symbol( + func_name, "track" + ): + self.has_track_call = True + if self._is_lamindb_module_call(func_name, "finish") or self._is_imported_lamindb_symbol( + func_name, "finish" + ): + self.has_finish_call = True + if self._is_lamindb_flow_name(func_name): + self.has_track_call = True + self.has_finish_call = True + + # Distinguish LaminDB API calls from ordinary Python calls. + # Any path referenced inside LaminDB calls is considered lineage-tracked. + is_lamin_call = self._is_lamindb_call(node, func_name) + is_non_lineage_setup_call = self._is_non_lineage_setup_call(node, func_name) + if is_non_lineage_setup_call: + # Ignore full setup-call subtrees (e.g. `sys.path.insert(..., str(path))`) + # so nested helper calls are not misclassified as lineage-relevant I/O. + return + + paths = self._extract_paths_from_node(node) + + if is_lamin_call: + for path in paths: + self.tracked_paths.setdefault(path, []).append(lineno) + + elif not is_non_lineage_setup_call: + for path in paths: + self.untracked_paths.setdefault(path, []).append(lineno) + + self._trace_user_function_call(node) + self.generic_visit(node) + + def _get_func_name(self, node: ast.AST) -> str: + """Recursively resolves AST attribute names (e.g. lamindb.Artifact.get).""" + if isinstance(node, ast.Name): + return node.id + elif isinstance(node, ast.Attribute): + return f"{self._get_func_name(node.value)}.{node.attr}" + return "" + + def _is_lamindb_artifact_constructor_call(self, node: ast.Call) -> bool: + func_name = self._get_func_name(node.func) + return self._is_lamindb_module_call(func_name, "Artifact") or self._is_imported_lamindb_symbol( + func_name, "Artifact" + ) + + def _is_lamindb_module_call(self, func_name: str, call_name: str) -> bool: + root_name, _, remainder = func_name.partition(".") + return root_name in self.lamindb_module_aliases and remainder == call_name + + def _is_imported_lamindb_symbol(self, func_name: str, symbol_name: str) -> bool: + root_name = func_name.split(".", 1)[0] + return self.imported_lamindb_symbols.get(root_name) == symbol_name + + def _is_lamindb_flow_name(self, func_name: str) -> bool: + return self._is_lamindb_module_call(func_name, "flow") or self._is_imported_lamindb_symbol( + func_name, "flow" + ) + + def _is_lamindb_artifact_save_call(self, node: ast.Call) -> bool: + # Treat both `lamindb.Artifact(...).save()` and `artifact.save()` + # (where artifact was created from `lamindb.Artifact(...)`) + # as LaminDB lineage-tracked operations. + if not (isinstance(node.func, ast.Attribute) and node.func.attr == "save"): + return False + if isinstance(node.func.value, ast.Call): + return self._is_lamindb_artifact_constructor_call(node.func.value) + if isinstance(node.func.value, ast.Name): + return node.func.value.id in self.artifact_vars + return False + + def _is_lamindb_call(self, node: ast.Call, func_name: str) -> bool: + # All calls rooted at `lamindb` (or any imported alias like `ln`) are lineage-aware. + # We also include artifact save calls recognized above. + root_name = func_name.split(".", 1)[0] + if root_name in self.lamindb_module_aliases: + return True + if root_name in self.imported_lamindb_symbols: + return True + return self._is_lamindb_artifact_save_call(node) + + def _is_non_lineage_setup_call(self, node: ast.Call, func_name: str) -> bool: + # Centralized allowlist for path/setup helpers that should not be + # interpreted as lineage-relevant file I/O. + if func_name in NON_LINEAGE_EXACT_CALLS: + return True + if func_name.startswith(NON_LINEAGE_CALL_PREFIXES): + return True + return isinstance(node.func, ast.Attribute) and node.func.attr in NON_LINEAGE_ATTR_CALLS + + def _extract_paths_from_node(self, node: ast.AST) -> set[str]: + """Extracts paths from string constants, variables, and nested function arguments.""" + paths = set() + + if isinstance(node, ast.Constant) and isinstance(node.value, str): + if self._is_path_like(node.value): + paths.add(node.value) + + elif isinstance(node, ast.Name): + if node.id in self.var_map: + paths.update(self.var_map[node.id]) + + elif isinstance(node, ast.Call): + for arg in node.args: + paths.update(self._extract_paths_from_node(arg)) + for kw in node.keywords: + paths.update(self._extract_paths_from_node(kw.value)) + + elif isinstance(node, ast.BinOp) and isinstance(node.op, (ast.Add, ast.Div)): + # Handle both string concatenation and pathlib-style joins. + left_paths = self._extract_paths_from_node(node.left) + right_paths = self._extract_paths_from_node(node.right) + for left_path in left_paths: + for right_path in right_paths: + combined = self._join_paths(left_path, right_path) + if self._is_path_like(combined): + paths.add(combined) + + return paths + + def _is_path_like(self, s: str) -> bool: + s = s.strip() + return bool(s and PATH_PATTERN.match(s)) + + def _join_paths(self, left: str, right: str) -> str: + left = left.strip() + right = right.strip() + if left.endswith(("/", "\\")) or right.startswith(("/", "\\")): + return f"{left}{right}" + return f"{left}/{right}" + + def _trace_user_function_call(self, node: ast.Call): + """Propagate path bindings into user-defined functions at call sites. + + This gives the checker a lightweight inter-procedural view, so path usage + hidden behind helper functions can still be classified as tracked/untracked. + """ + if not isinstance(node.func, ast.Name): + return + + function_name = node.func.id + function_def = self.function_defs.get(function_name) + if function_def is None or function_name in self._active_function_calls: + return + + bindings = self._get_param_path_bindings(node, function_def) + if not bindings: + return + + self._active_function_calls.add(function_name) + # Temporarily augment variable scope with call argument bindings, then + # restore state after visiting the function body. + original_var_map = self.var_map.copy() + original_artifact_vars = self.artifact_vars.copy() + try: + self.var_map.update(bindings) + for stmt in function_def.body: + self.visit(stmt) + finally: + self.var_map = original_var_map + self.artifact_vars = original_artifact_vars + self._active_function_calls.remove(function_name) + + def _get_param_path_bindings( + self, call: ast.Call, function_def: ast.FunctionDef + ) -> dict[str, set[str]]: + bindings: dict[str, set[str]] = {} + + positional_params = function_def.args.posonlyargs + function_def.args.args + for param, arg in zip(positional_params, call.args): + paths = self._extract_paths_from_node(arg) + if paths: + bindings[param.arg] = paths + + keyword_args = { + kw.arg: self._extract_paths_from_node(kw.value) + for kw in call.keywords + if kw.arg is not None + } + + for param in positional_params: + if param.arg in bindings: + continue + paths = keyword_args.get(param.arg, set()) + if paths: + bindings[param.arg] = paths + + for param in function_def.args.kwonlyargs: + paths = keyword_args.get(param.arg, set()) + if paths: + bindings[param.arg] = paths + + return bindings + + +def verify_lineage(script_path: str) -> VerifyLineageResult: + """Statically analyze one script and report missing LaminDB lineage hooks. + + Coverage (static AST checks): + - Confirms the script contains explicit session lifecycle calls from LaminDB: + `track()` and `finish()` (or `flow()`). + - Extracts path-like strings from literals, variables, call arguments, + and simple path joins (string `+` or path `/` operations). + - Classifies path usage as tracked when it appears in LaminDB calls + (`ln.*`, `lamindb.*`, or any alias like `import lamindb as alias`, + including `x=lamindb.Artifact(...)->x.save()` patterns and + `from lamindb import Artifact`), + and flags paths that only appear in non-LaminDB calls. + - Propagates path-like argument bindings into user-defined helper + functions to catch indirect path usage. + + Important exclusions: + - Path construction/inspection/setup helpers are allowlisted and ignored + for lineage (for example: `Path(...)`, `os.path.*`, `.mkdir()`, + `.exists()`, `.is_file()`, `.resolve()`, and glob helpers). + - `importlib.*` calls are treated as import/module setup and ignored for + lineage purposes. + - `os.environ[...] = ...` assignments are treated as environment setup and + ignored for lineage purposes. + - `sys.path.*` calls (for example `sys.path.insert(...)`) are treated + as import-path setup and ignored for lineage purposes. + """ + path = Path(script_path) + if not path.is_file(): + return VerifyLineageResult( + is_fully_tracked=False, + missing_lineage=(f"File not found: {script_path}",), + ) + + with open(path, encoding="utf-8") as f: + source_code = f.read() + + try: + tree = ast.parse(source_code, filename=script_path) + except SyntaxError as e: + return VerifyLineageResult( + is_fully_tracked=False, + missing_lineage=(f"Syntax error while parsing {script_path}: {e}",), + ) + + checker = LaminLineageChecker() + checker.visit(tree) + + # Keep only paths that are seen in non-LaminDB calls and never seen in LaminDB calls. + # These are candidates for missing lineage registration. + tracked_path_keys = set(checker.tracked_paths) + truly_untracked_paths = { + path: lines + for path, lines in checker.untracked_paths.items() + if path not in tracked_path_keys + } + + missing_lineage: list[str] = [] + + # A script is fully tracked only if it opens and closes tracking + # (or uses `flow()` lifecycle management). + if not checker.has_track_call: + missing_lineage.append("Missing ln.track() call in script.") + + if not checker.has_finish_call: + missing_lineage.append("Missing ln.finish() call in script.") + + if truly_untracked_paths: + # Point to concrete source lines where paths are used. + for fname, lines in sorted(truly_untracked_paths.items()): + lines_str = ", ".join(f"line {l}" for l in lines) + missing_lineage.append(f"File not tracked in lamindb: {fname} ({lines_str})") + + + is_fully_tracked = not missing_lineage + return VerifyLineageResult( + is_fully_tracked=is_fully_tracked, + missing_lineage=tuple(missing_lineage), + ) + diff --git a/sub/lamin-cli b/sub/lamin-cli index b4e0aff3e..9041d0ef7 160000 --- a/sub/lamin-cli +++ b/sub/lamin-cli @@ -1 +1 @@ -Subproject commit b4e0aff3ea6dcc78c845bba2da83c8d26f84f9f5 +Subproject commit 9041d0ef732d6b50ca041f5d70560a4e1847a329 diff --git a/tests/no_instance/test_verify_lineage.py b/tests/no_instance/test_verify_lineage.py new file mode 100644 index 000000000..d3ad92651 --- /dev/null +++ b/tests/no_instance/test_verify_lineage.py @@ -0,0 +1,436 @@ +from __future__ import annotations + +import textwrap +from typing import TYPE_CHECKING + +import pytest +from lamindb.core import verify_lineage + +if TYPE_CHECKING: + from pathlib import Path + + +def _write_script(tmp_path: Path, name: str, source_code: str) -> Path: + script_path = tmp_path / name + script_path.write_text(source_code, encoding="utf-8") + return script_path + + +def extract_cases(cls): + return [ + pytest.param(textwrap.dedent(v).strip(), id=k) + for k, v in vars(cls).items() + if not k.startswith("_") + ] + + +def extract_untracked_cases(cls): + cases = [] + for k, v in vars(cls).items(): + if not k.startswith("_"): + source_code = v["source_code"] + untracked_path = v["untracked_path"] + cases.append( + pytest.param(textwrap.dedent(source_code).strip(), untracked_path, id=k) + ) + return cases + + +class PositiveCases: + lamindb_alias = """ + import lamindb as ln + + ln.track() + ln.Artifact.get(uid="abcDEF1234567890") + ln.Artifact("./out.csv").save() + ln.finish() + """ + + lamindb_custom_alias = """ + import lamindb as ldb + + ldb.track() + ldb.Artifact.get(uid="abcDEF1234567890") + ldb.Artifact("./out.csv").save() + ldb.finish() + """ + + lamindb_without_alias = """ + import lamindb + + lamindb.track() + lamindb.Artifact.get(uid="abcDEF1234567890") + lamindb.Artifact("./out.csv").save() + lamindb.finish() + """ + + artifact_from_dataframe = """ + import lamindb as ln + + ln.track() + ln.Artifact.from_dataframe(df, key="out.parquet") + ln.finish() + """ + + imported_symbols = """ + from lamindb import Artifact, finish, track + + track() + Artifact.get(uid="abcDEF1234567890") + Artifact("./out.csv").save() + finish() + """ + + flow_decorator_with_alias = """ + import lamindb as ln + + @ln.flow() + def main(): + ln.Artifact.get(uid="abcDEF1234567890") + ln.Artifact("./out.csv").save() + + main() + """ + + flow_decorator_no_alias = """ + import lamindb as ln + + @lamindb.flow() + def main(): + ln.Artifact.get(uid="abcDEF1234567890") + ln.Artifact("./out.csv").save() + + main() + """ + + imported_flow_step_decorators = """ + from lamindb import Artifact, flow, step + + @step() + def build_output(): + Artifact("./out.csv").save() + + @flow() + def main(): + Artifact.get(uid="abcDEF1234567890") + build_output() + + main() + """ + + zero_io = """ + import lamindb as ln + + ln.track() + fibonacci = [0, 1] + for _ in range(2, 10): + fibonacci.append(fibonacci[-1] + fibonacci[-2]) + ln.finish() + """ + + local_write_then_save = """ + from pathlib import Path + import json + import lamindb as ln + + ln.track() + result = {"ok": True} + output_path = Path("./result.json") + output_path.write_text(json.dumps(result, indent=2) + "\\n", encoding="utf-8") + ln.Artifact(output_path).save() + ln.finish() + """ + + np_save_then_save = """ + from pathlib import Path + import lamindb as ln + import numpy as np + + ln.track() + output_path = Path("./array.npy") + np.save(output_path, np.array([1, 2, 3])) + ln.Artifact(output_path).save() + ln.finish() + """ + + open_write_then_save = """ + import lamindb as ln + + ln.track() + with open("out.txt", "w") as f: + f.write("external output test\\n") + ln.Artifact("out.txt", key="tests/external-output/out.txt").save() + ln.finish() + """ + + artifact_variable_save = """ + from pathlib import Path + import lamindb as ln + + ln.track() + out_path = Path("out.txt") + out_path.write_text("external output test\\n") + artifact = ln.Artifact(out_path, key="tests/external-output/out.txt") + artifact.save() + ln.finish() + """ + + helper_function_param = """ + import lamindb as ln + + def load_dataset(dataset_ref: str): + return ln.Artifact.get(key=dataset_ref) + + def main() -> None: + ln.track() + _ = load_dataset("datasets/rnaseq/synthetic_rnaseq_from_age_disease.csv") + ln.finish() + """ + + concatenated_output_path = """ + from pathlib import Path + import lamindb as ln + + ln.track() + output_dir = Path("outputs/proteomics") + output_dir.mkdir(parents=True, exist_ok=True) + output_path = output_dir / "uniprot_human_reviewed.tsv" + ln.Artifact( + output_path, + key="datasets/proteomics/uniprot_human_reviewed.tsv", + description="UniProt reviewed human proteomics dataset with disulfide annotations", + ).save() + ln.finish() + """ + + sys_path_insert_ignored = """ + import sys + from pathlib import Path + import lamindb as ln + + ln.track() + plugins = Path("./plugins") + scanpy_plugin = plugins / "scanpy" + sys.path.insert(0, str(scanpy_plugin)) + ln.finish() + """ + + os_path_helpers_ignored = """ + import os + import lamindb as ln + + ln.track() + base = os.path.abspath("./data") + out = os.path.join(base, "output.txt") + parent = os.path.dirname(out) + name = os.path.basename(out) + _ = os.path.splitext(name) + ln.finish() + """ + + env_var_setup_ignored = """ + import os + from pathlib import Path + import lamindb as ln + + ln.track() + plugins = Path("./plugins") + os.environ["PYTHONPATH"] = str(plugins) + ln.finish() + """ + + importlib_setup_ignored = """ + import importlib.util + from pathlib import Path + import lamindb as ln + + ln.track() + plugin_path = Path("./plugins") / "scanpy_plugin.py" + _ = importlib.util.spec_from_file_location("scanpy_plugin", str(plugin_path)) + ln.finish() + """ + + +class MissingLifecycleCases: + step_without_flow_or_track = """ + import lamindb as ln + + @ln.step() + def build_output(): + ln.Artifact("./out.csv").save() + + build_output() + """ + + no_track_finish = """ + import lamindb as ln + + ln.Artifact.get(uid="abcDEF1234567890") + ln.Artifact("./out.csv").save() + """ + + missing_track_only = """ + import lamindb as ln + + ln.Artifact("./out.csv").save() + ln.finish() + """ + + missing_finish_only = """ + import lamindb as ln + + ln.track() + ln.Artifact("./out.csv").save() + """ + + +class UntrackedPathCases: + external_input_read_csv = { + "untracked_path": "./local_input.csv", + "source_code": """ + import lamindb as ln + import pandas as pd + + ln.track() + pd.read_csv("./local_input.csv") + ln.Artifact.get(uid="abcDEF1234567890") + ln.Artifact("./out.csv").save() + ln.finish() + """, + } + + external_output_open_write = { + "untracked_path": "./local_output.txt", + "source_code": """ + import lamindb as ln + + ln.track() + with open("./local_output.txt", "w") as f: + f.write("hello") + ln.finish() + """, + } + + external_input_open_read = { + "untracked_path": "./local_input.txt", + "source_code": """ + import lamindb as ln + + ln.track() + with open("./local_input.txt", "r") as f: + _ = f.read() + ln.finish() + """, + } + + np_save_without_artifact_save = { + "untracked_path": "./array.npy", + "source_code": """ + from pathlib import Path + import lamindb as ln + import numpy as np + + ln.track() + output_path = Path("./array.npy") + np.save(output_path, np.array([1, 2, 3])) + ln.finish() + """, + } + + non_lamindb_save_call = { + "untracked_path": "./weights.bin", + "source_code": """ + import lamindb as ln + + class DummyModel: + def save(self, path): + with open(path, "w", encoding="utf-8") as f: + f.write("weights") + + ln.track() + model = DummyModel() + model.save("./weights.bin") + ln.finish() + """, + } + + untracked_directory = { + "untracked_path": "outputs/proteomics", + "source_code": """ + from pathlib import Path + import lamindb as ln + + def get_path(path: str): + print(path) + + ln.track() + output_dir = Path("outputs/proteomics") + get_path("outputs/proteomics") + output_path = output_dir / "uniprot_human_reviewed.tsv" + ln.Artifact(output_path, key="datasets/proteomics/uniprot_human_reviewed.tsv").save() + ln.finish() + """, + } + + untracked_matplotlib_savefig = { + "untracked_path": "./my_plot.png", + "source_code": """ + import lamindb as ln + import matplotlib.pyplot as plt + + ln.track() + plt.plot([1, 2]) + plt.savefig("./my_plot.png") + ln.finish() + """, + } + + untracked_pandas_to_parquet = { + "untracked_path": "./data.parquet", + "source_code": """ + import lamindb as ln + import pandas as pd + + ln.track() + df = pd.DataFrame({"a": [1, 2]}) + df.to_parquet("./data.parquet") + ln.finish() + """, + } + + +@pytest.mark.parametrize("source_code", extract_cases(PositiveCases)) +def test_verify_lineage_positive_cases(tmp_path: Path, source_code: str): + script_path = _write_script(tmp_path, "script.py", source_code) + result = verify_lineage(script_path) + assert result.is_fully_tracked is True + assert result.missing_lineage == () + + +@pytest.mark.parametrize("source_code", extract_cases(MissingLifecycleCases)) +def test_verify_lineage_negative_missing_lifecycle(tmp_path: Path, source_code: str): + script_path = _write_script(tmp_path, "script.py", source_code) + result = verify_lineage(script_path) + assert result.is_fully_tracked is False + assert ( + "Missing ln.track() call in script." in result.missing_lineage + or "Missing ln.finish() call in script." in result.missing_lineage + ) + + +@pytest.mark.parametrize( + ("source_code", "untracked_path"), extract_untracked_cases(UntrackedPathCases) +) +def test_verify_lineage_negative_untracked_paths( + tmp_path: Path, source_code: str, untracked_path: str +): + script_path = _write_script(tmp_path, "script.py", source_code) + result = verify_lineage(script_path) + assert result.is_fully_tracked is False + assert any(untracked_path in item for item in result.missing_lineage) + + +def test_verify_lineage_missing_file(): + result = verify_lineage("does-not-exist.py") + assert result.is_fully_tracked is False + assert result.missing_lineage == ("File not found: does-not-exist.py",)