e07c8e76e9
- 76 new tests covering CLI commands, API routes, auth, runner, load modules — coverage 99% (169 remaining lines are infrastructure) - Enhanced /health endpoint with DuckDB, bib, pipeline service checks - Added `stack health` CLI command with graceful degradation - SnapshotManager for DuckDB data versioning and rollback - .woodpecker.yml CI pipeline: lint, test (99% gate), marimo check - Error handling audit: no bare except clauses, all exceptions logged Closes #31, #32, #33, #34, #35.
162 lines
4.2 KiB
Python
162 lines
4.2 KiB
Python
"""Coverage tests for aco.pipe.runner edge cases."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from unittest.mock import MagicMock
|
|
|
|
import pytest
|
|
|
|
from aco.pipe.runner import (
|
|
SchemaError,
|
|
_fingerprint_df,
|
|
_fingerprint_inputs,
|
|
_validate_output,
|
|
load_fingerprints,
|
|
run_pipeline,
|
|
save_fingerprints,
|
|
)
|
|
|
|
|
|
class TestFingerprint:
|
|
def test_fingerprint_df(self) -> None:
|
|
df = MagicMock()
|
|
df.columns = ["a", "b"]
|
|
df.__len__ = lambda s: 10
|
|
fp = _fingerprint_df(df)
|
|
assert isinstance(fp, str)
|
|
assert len(fp) == 12
|
|
|
|
def test_fingerprint_inputs(self) -> None:
|
|
df = MagicMock()
|
|
df.columns = ["x"]
|
|
df.__len__ = lambda s: 5
|
|
|
|
def my_fn(core__patient):
|
|
pass
|
|
|
|
fp = _fingerprint_inputs({"core__patient": df}, my_fn)
|
|
assert isinstance(fp, str)
|
|
assert len(fp) == 16
|
|
|
|
|
|
class TestValidateOutput:
|
|
def test_missing_columns(self) -> None:
|
|
class FakeModel:
|
|
__module__ = "test"
|
|
__qualname__ = "FakeModel"
|
|
|
|
@classmethod
|
|
def column_names(cls):
|
|
return ["a", "b", "c"]
|
|
|
|
df = MagicMock()
|
|
df.columns = ["a", "b"]
|
|
|
|
with pytest.raises(SchemaError, match="missing columns"):
|
|
_validate_output(
|
|
"test.output", df, FakeModel, fn=lambda: None, input_tables=[]
|
|
)
|
|
|
|
def test_extra_columns(self) -> None:
|
|
class FakeModel:
|
|
__module__ = "test"
|
|
__qualname__ = "FakeModel"
|
|
|
|
@classmethod
|
|
def column_names(cls):
|
|
return ["a"]
|
|
|
|
df = MagicMock()
|
|
df.columns = ["a", "b", "c"]
|
|
|
|
with pytest.raises(SchemaError, match="extra columns"):
|
|
_validate_output(
|
|
"test.output", df, FakeModel, fn=lambda: None, input_tables=[]
|
|
)
|
|
|
|
|
|
class TestRunPipeline:
|
|
def test_fingerprint_skip(self) -> None:
|
|
"""Step is skipped when fingerprint matches."""
|
|
df = MagicMock()
|
|
df.columns = ["x"]
|
|
df.__len__ = lambda s: 3
|
|
|
|
def step(core__patient):
|
|
return df
|
|
|
|
# First run to get fingerprints
|
|
cache = run_pipeline(
|
|
[("out", step, None)],
|
|
load=lambda ref: df,
|
|
fingerprints={},
|
|
)
|
|
fp = cache["__fingerprints__"]["out"]
|
|
|
|
# Second run with matching fingerprint — should load from cache
|
|
load_called = []
|
|
|
|
def tracking_load(ref):
|
|
load_called.append(ref)
|
|
return df
|
|
|
|
cache2 = run_pipeline(
|
|
[("out", step, None)],
|
|
load=tracking_load,
|
|
fingerprints={"out": fp},
|
|
)
|
|
# "out" should have been loaded from cache (via load) rather than re-run
|
|
assert "out" in cache2
|
|
|
|
def test_fingerprint_cache_miss(self) -> None:
|
|
"""Step re-executes when cached load fails."""
|
|
df = MagicMock()
|
|
df.columns = ["x"]
|
|
df.__len__ = lambda s: 3
|
|
|
|
call_count = [0]
|
|
|
|
def step(core__patient):
|
|
call_count[0] += 1
|
|
return df
|
|
|
|
# First run
|
|
cache = run_pipeline(
|
|
[("out", step, None)],
|
|
load=lambda ref: df,
|
|
fingerprints={},
|
|
)
|
|
fp = cache["__fingerprints__"]["out"]
|
|
|
|
# Second run — load fails for "out" so it re-executes
|
|
def failing_load(ref):
|
|
if ref == "out":
|
|
raise KeyError("cache miss")
|
|
return df
|
|
|
|
call_count[0] = 0
|
|
run_pipeline(
|
|
[("out", step, None)],
|
|
load=failing_load,
|
|
fingerprints={"out": fp},
|
|
)
|
|
assert call_count[0] == 1
|
|
|
|
|
|
class TestFingerprintIO:
|
|
def test_load_missing_file(self, tmp_path) -> None:
|
|
path = str(tmp_path / "missing.json")
|
|
assert load_fingerprints(path) == {}
|
|
|
|
def test_load_invalid_json(self, tmp_path) -> None:
|
|
path = tmp_path / "bad.json"
|
|
path.write_text("not json")
|
|
assert load_fingerprints(str(path)) == {}
|
|
|
|
def test_save_and_load(self, tmp_path) -> None:
|
|
path = str(tmp_path / "fp.json")
|
|
data = {"step1": "abc123"}
|
|
save_fingerprints(path, data)
|
|
loaded = load_fingerprints(path)
|
|
assert loaded == data
|