Testing
The dissmodel.executor.testing module provides ExecutorTestHarness, a
contract validator for ModelExecutor subclasses. It is designed to run
in Jupyter before opening a PR — the same checks are reused in CI via pytest,
so a passing notebook guarantees a passing pipeline.
Contract tests
run_contract_tests() runs structural checks without loading any data:
from dissmodel.executor.testing import ExecutorTestHarness
from my_package.my_executor import MyExecutor
harness = ExecutorTestHarness(MyExecutor)
harness.run_contract_tests()
ExecutorTestHarness — MyExecutor
────────────────────────────────────────────────────
✅ name attribute exists
✅ name is a non-empty string
✅ name has no whitespace
✅ load() is implemented
✅ run() is implemented
✅ save() is implemented
✅ run() signature is correct
✅ save() signature is correct
────────────────────────────────────────────────────
All 8 checks passed ✅
The run() signature is correct check validates that the executor uses the
0.4.0 signature run(self, data, record). Executors still using the old
run(self, record) form will fail here with a clear message:
❌ run() signature is correct: run() must accept exactly two parameters
(data, record), got ['record']. Did you update to the 0.4.0 signature?
Full lifecycle test
run_with_sample_data() executes validate → load → run(data, record) → save
with a real record. Use it for smoke testing after migrating a model:
from dissmodel.executor.schemas import DataSource, ExperimentRecord
record = ExperimentRecord(
model_name = "my_model",
model_commit = "local-test",
code_version = "dev",
source = DataSource(type="local", uri="data/sample.gpkg"),
)
harness.run_with_sample_data(record)
▶ Running my_model...
validate()...
load()...
run()...
save()...
✅ Cycle OK — status=completed sha256=deadbeef123...
If no record is provided, the harness creates a minimal synthetic one:
harness.run_with_sample_data() # uses _minimal_record()
Using in pytest
# tests/test_contract.py
from dissmodel.executor.testing import ExecutorTestHarness
from my_package.my_executor import MyExecutor
def test_contract():
assert ExecutorTestHarness(MyExecutor).run_contract_tests() is True
API Reference
dissmodel.executor.testing.ExecutorTestHarness
Validates that an executor fulfills the ModelExecutor contract.
Designed to run in Jupyter before opening a PR — the same checks
are reused in CI via pytest parametrize, so a passing notebook
guarantees a passing pipeline.
Usage
harness = ExecutorTestHarness(MyExecutor)
harness.run_contract_tests() # structural — no data needed
harness.run_with_sample_data(record) # full cycle with real data
Source code in dissmodel/executor/testing.py
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174 | class ExecutorTestHarness:
"""
Validates that an executor fulfills the ModelExecutor contract.
Designed to run in Jupyter before opening a PR — the same checks
are reused in CI via pytest parametrize, so a passing notebook
guarantees a passing pipeline.
Usage
-----
harness = ExecutorTestHarness(MyExecutor)
harness.run_contract_tests() # structural — no data needed
harness.run_with_sample_data(record) # full cycle with real data
"""
def __init__(self, executor_cls: type[ModelExecutor]) -> None:
self.executor_cls = executor_cls
self.executor = executor_cls()
self._passed: list[str] = []
self._failed: list[str] = []
# ── Public interface ──────────────────────────────────────────────────────
def run_contract_tests(self) -> bool:
"""
Run structural checks — no data required.
Returns True if all checks pass.
"""
self._passed.clear()
self._failed.clear()
self._check("name attribute exists", self._check_name_exists)
self._check("name is a non-empty string", self._check_name_type)
self._check("name has no whitespace", self._check_name_format)
self._check("load() is implemented", self._check_load)
self._check("run() is implemented", self._check_run)
self._check("save() is implemented", self._check_save)
self._check("run() signature is correct", self._check_run_signature)
self._check("save() signature is correct", self._check_save_signature)
self._print_report()
return len(self._failed) == 0
def run_with_sample_data(self, record: ExperimentRecord | None = None) -> bool:
"""
Run the full executor lifecycle with real or synthetic data.
Calls validate → load → run(data, record) → save in sequence.
Returns True if the cycle completes without error.
"""
if record is None:
record = self._minimal_record()
print(" No record provided — using minimal synthetic record")
print(f"\n▶ Running {self.executor_cls.name}...")
try:
print(" validate()...")
self.executor.validate(record)
print(" load()...")
data = self.executor.load(record)
print(" run()...")
result = self.executor.run(data, record)
print(" save()...")
completed = self.executor.save(result, record)
if completed.status != "completed":
print(f" ⚠ save() returned status='{completed.status}' — expected 'completed'")
return False
if not completed.output_sha256:
print(" ⚠ save() did not set output_sha256")
return False
print(f" ✅ Cycle OK — status={completed.status} sha256={completed.output_sha256[:12]}...")
return True
except NotImplementedError:
print(" ⚠ Some methods are not yet implemented")
return False
except Exception:
print(f" ❌ Error during execution:\n{traceback.format_exc()}")
return False
# ── Individual checks ─────────────────────────────────────────────────────
def _check_name_exists(self) -> None:
assert hasattr(self.executor_cls, "name"), \
"Class must define a 'name' class attribute"
def _check_name_type(self) -> None:
assert isinstance(self.executor_cls.name, str) and self.executor_cls.name, \
f"'name' must be a non-empty string, got {self.executor_cls.name!r}"
def _check_name_format(self) -> None:
assert " " not in self.executor_cls.name, \
f"'name' must not contain whitespace: {self.executor_cls.name!r}"
def _check_load(self) -> None:
assert not _is_abstract(self.executor, "load"), \
"load() must be implemented"
def _check_run(self) -> None:
assert not _is_abstract(self.executor, "run"), \
"run() must be implemented"
def _check_save(self) -> None:
assert not _is_abstract(self.executor, "save"), \
"save() must be implemented"
def _check_run_signature(self) -> None:
sig = inspect.signature(self.executor.run)
params = [p for p in sig.parameters.values() if p.name != "self"]
assert len(params) == 2, (
f"run() must accept exactly two parameters (data, record), "
f"got {[p.name for p in params]}. "
f"Did you update to the 0.4.0 signature?"
)
def _check_save_signature(self) -> None:
sig = inspect.signature(self.executor.save)
params = [p for p in sig.parameters.values() if p.name != "self"]
assert len(params) == 2, \
f"save() must accept exactly two parameters (result, record), got {[p.name for p in params]}"
# ── Helpers ───────────────────────────────────────────────────────────────
def _check(self, label: str, fn) -> None:
try:
fn()
self._passed.append(label)
except AssertionError as exc:
self._failed.append(f"{label}: {exc}")
except Exception as exc:
self._failed.append(f"{label}: unexpected error — {exc}")
def _print_report(self) -> None:
print(f"\nExecutorTestHarness — {self.executor_cls.__name__}")
print("─" * 52)
for label in self._passed:
print(f" ✅ {label}")
for label in self._failed:
print(f" ❌ {label}")
print("─" * 52)
if self._failed:
print(f" {len(self._passed)} passed, {len(self._failed)} failed\n")
else:
print(f" All {len(self._passed)} checks passed ✅\n")
def _minimal_record(self) -> ExperimentRecord:
"""Synthetic record for contract testing without real data."""
return ExperimentRecord(
model_name = getattr(self.executor_cls, "name", "unknown"),
model_commit = "local-test",
code_version = "dev",
resolved_spec = {"model": {"class": getattr(self.executor_cls, "name", ""), "parameters": {}}},
source = DataSource(type="local", uri=""),
)
|
run_contract_tests()
Run structural checks — no data required.
Returns True if all checks pass.
Source code in dissmodel/executor/testing.py
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54 | def run_contract_tests(self) -> bool:
"""
Run structural checks — no data required.
Returns True if all checks pass.
"""
self._passed.clear()
self._failed.clear()
self._check("name attribute exists", self._check_name_exists)
self._check("name is a non-empty string", self._check_name_type)
self._check("name has no whitespace", self._check_name_format)
self._check("load() is implemented", self._check_load)
self._check("run() is implemented", self._check_run)
self._check("save() is implemented", self._check_save)
self._check("run() signature is correct", self._check_run_signature)
self._check("save() signature is correct", self._check_save_signature)
self._print_report()
return len(self._failed) == 0
|
run_with_sample_data(record=None)
Run the full executor lifecycle with real or synthetic data.
Calls validate → load → run(data, record) → save in sequence.
Returns True if the cycle completes without error.
Source code in dissmodel/executor/testing.py
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99 | def run_with_sample_data(self, record: ExperimentRecord | None = None) -> bool:
"""
Run the full executor lifecycle with real or synthetic data.
Calls validate → load → run(data, record) → save in sequence.
Returns True if the cycle completes without error.
"""
if record is None:
record = self._minimal_record()
print(" No record provided — using minimal synthetic record")
print(f"\n▶ Running {self.executor_cls.name}...")
try:
print(" validate()...")
self.executor.validate(record)
print(" load()...")
data = self.executor.load(record)
print(" run()...")
result = self.executor.run(data, record)
print(" save()...")
completed = self.executor.save(result, record)
if completed.status != "completed":
print(f" ⚠ save() returned status='{completed.status}' — expected 'completed'")
return False
if not completed.output_sha256:
print(" ⚠ save() did not set output_sha256")
return False
print(f" ✅ Cycle OK — status={completed.status} sha256={completed.output_sha256[:12]}...")
return True
except NotImplementedError:
print(" ⚠ Some methods are not yet implemented")
return False
except Exception:
print(f" ❌ Error during execution:\n{traceback.format_exc()}")
return False
|