310 lines
12 KiB
Python
310 lines
12 KiB
Python
from __future__ import annotations
|
|
|
|
import importlib.util
|
|
import json
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
import numpy as np
|
|
|
|
from core.software_contracts import get_contract
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
ADAPTER_ROOT = ROOT / "windows-node" / "adapters" / "origin.analysis@v1"
|
|
WORKER_PATH = ADAPTER_ROOT / "worker.py"
|
|
SPEC = importlib.util.spec_from_file_location("zcbot_origin_analysis_worker", WORKER_PATH)
|
|
assert SPEC and SPEC.loader
|
|
worker = importlib.util.module_from_spec(SPEC)
|
|
SPEC.loader.exec_module(worker)
|
|
|
|
ACCEPTANCE_PATH = ADAPTER_ROOT / "acceptance.py"
|
|
ACCEPTANCE_SPEC = importlib.util.spec_from_file_location(
|
|
"zcbot_origin_analysis_acceptance", ACCEPTANCE_PATH
|
|
)
|
|
assert ACCEPTANCE_SPEC and ACCEPTANCE_SPEC.loader
|
|
acceptance = importlib.util.module_from_spec(ACCEPTANCE_SPEC)
|
|
ACCEPTANCE_SPEC.loader.exec_module(acceptance)
|
|
|
|
|
|
def _request(operation: str, parameters: dict, *, data: dict | None = None) -> dict:
|
|
return {
|
|
"schema_version": 1,
|
|
"inputs": [
|
|
{
|
|
"key": "sample",
|
|
"artifact_id": "f4186347-65cc-4f07-9c26-bf11992beef8",
|
|
}
|
|
],
|
|
"operation": {
|
|
"analysis": {
|
|
"type": operation,
|
|
"title": f"Test {operation}",
|
|
"data": data or {"input": "sample", "x": "x", "y": "y"},
|
|
"parameters": parameters,
|
|
}
|
|
},
|
|
"outputs": [],
|
|
}
|
|
|
|
|
|
class OriginAnalysisContractTests(unittest.TestCase):
|
|
def test_contract_exposes_six_guarded_operations_and_fixed_outputs(self) -> None:
|
|
contract = get_contract("origin.analysis@v1")
|
|
self.assertFalse(contract.default_enrollment)
|
|
self.assertEqual(
|
|
set(contract.features),
|
|
{
|
|
"data_check",
|
|
"normalize",
|
|
"smooth",
|
|
"differentiate",
|
|
"integrate",
|
|
"linear_fit",
|
|
},
|
|
)
|
|
request = _request("normalize", {"method": "max_abs"})
|
|
normalized, digest = contract.normalize_request(request)
|
|
self.assertEqual(normalized, request)
|
|
self.assertEqual(len(digest), 64)
|
|
self.assertEqual(
|
|
set(contract.expected_outputs(request)),
|
|
{"project", "result_table", "diagnostics", "analysis_spec", "provenance"},
|
|
)
|
|
|
|
workbook_request = json.loads(json.dumps(request))
|
|
workbook_request["outputs"] = [
|
|
{"key": "result_workbook", "type": "table", "format": "xlsx"}
|
|
]
|
|
normalized, _ = contract.normalize_request(workbook_request)
|
|
self.assertEqual(normalized, workbook_request)
|
|
self.assertIn("result_workbook", contract.expected_outputs(workbook_request))
|
|
|
|
def test_contract_rejects_scripts_and_operation_specific_parameter_drift(self) -> None:
|
|
contract = get_contract("origin.analysis@v1")
|
|
request = _request("smooth", {
|
|
"method": "savitzky_golay",
|
|
"window": 7,
|
|
"polynomial_order": 3,
|
|
})
|
|
contract.normalize_request(request)
|
|
|
|
request["operation"]["analysis"]["parameters"]["script"] = "anything"
|
|
with self.assertRaises(ValueError):
|
|
contract.normalize_request(request)
|
|
|
|
reference = _request("normalize", {"method": "reference"})
|
|
with self.assertRaises(ValueError):
|
|
contract.normalize_request(reference)
|
|
|
|
def test_worker_and_manifest_versions_match(self) -> None:
|
|
manifest = json.loads((ADAPTER_ROOT / "adapter.json").read_text(encoding="utf-8"))
|
|
self.assertEqual(worker.ADAPTER_VERSION, manifest["adapter_version"])
|
|
|
|
|
|
class OriginAnalysisNumericalTests(unittest.TestCase):
|
|
def test_data_check_profiles_columns_and_reports_x_issues(self) -> None:
|
|
headers = ["x", "y", "label"]
|
|
rows = [[1, 2, "a"], [1, None, "b"], [3.5, float("inf"), "c"]]
|
|
spec = _request(
|
|
"data_check",
|
|
{"fail_on": ["missing", "non_finite", "duplicate_x", "uneven_spacing"]},
|
|
)["operation"]["analysis"]
|
|
result_headers, result_rows, diagnostics = worker.analyze(headers, rows, spec)
|
|
self.assertEqual(result_headers[0], "column")
|
|
self.assertEqual(len(result_rows), 3)
|
|
self.assertFalse(diagnostics["passed"])
|
|
self.assertEqual(
|
|
set(diagnostics["failed_checks"]),
|
|
{"missing", "non_finite", "duplicate_x"},
|
|
)
|
|
|
|
def test_normalize_supports_max_area_and_interpolated_reference(self) -> None:
|
|
x = np.array([0.0, 1.0, 2.0])
|
|
y = np.array([0.0, 2.0, 4.0])
|
|
_, rows, diagnostics = worker._normalize(x, y, {"method": "max_abs"})
|
|
self.assertEqual([item[2] for item in rows], [0.0, 0.5, 1.0])
|
|
self.assertEqual(diagnostics["divisor"], 4.0)
|
|
|
|
_, _, diagnostics = worker._normalize(x, y, {"method": "area"})
|
|
self.assertEqual(diagnostics["divisor"], 4.0)
|
|
|
|
_, rows, diagnostics = worker._normalize(
|
|
x, y, {"method": "reference", "reference_x": 1.5}
|
|
)
|
|
self.assertEqual(diagnostics["divisor"], 3.0)
|
|
self.assertAlmostEqual(rows[-1][2], 4 / 3)
|
|
|
|
def test_smooth_reproduces_quadratic_and_rejects_even_window(self) -> None:
|
|
x = np.arange(7, dtype=float)
|
|
y = x**2 + 2 * x + 3
|
|
_, rows, diagnostics = worker._smooth(
|
|
x,
|
|
y,
|
|
{"method": "savitzky_golay", "window": 5, "polynomial_order": 2},
|
|
)
|
|
np.testing.assert_allclose([item[2] for item in rows], y, atol=1e-10)
|
|
self.assertAlmostEqual(diagnostics["rmse"], 0.0, places=10)
|
|
with self.assertRaisesRegex(ValueError, "SMOOTH_WINDOW_MUST_BE_ODD"):
|
|
worker._smooth(
|
|
x,
|
|
y,
|
|
{"method": "savitzky_golay", "window": 6, "polynomial_order": 2},
|
|
)
|
|
|
|
def test_differentiate_and_integrate_match_polynomial_oracles(self) -> None:
|
|
x = np.arange(5, dtype=float)
|
|
y = x**2
|
|
_, derivative_rows, _ = worker._differentiate(x, y, {"order": 1})
|
|
np.testing.assert_allclose([item[2] for item in derivative_rows], 2 * x)
|
|
|
|
_, integral_rows, diagnostics = worker._integrate(x, 2 * x, {})
|
|
np.testing.assert_allclose([item[2] for item in integral_rows], x**2)
|
|
self.assertEqual(diagnostics["full_integral"], 16.0)
|
|
_, _, diagnostics = worker._integrate(x, 2 * x, {"from": 1.0, "to": 3.0})
|
|
self.assertEqual(diagnostics["integral"], 8.0)
|
|
|
|
def test_linear_fit_returns_parameters_intervals_and_residuals(self) -> None:
|
|
x = np.arange(1, 7, dtype=float)
|
|
y = 2.5 * x - 1.25
|
|
headers, rows, diagnostics = worker._linear_fit(
|
|
x,
|
|
y,
|
|
{"include_intercept": True, "confidence_level": 0.95},
|
|
)
|
|
self.assertEqual(headers, ["x", "y", "fitted_y", "residual"])
|
|
self.assertAlmostEqual(diagnostics["slope"], 2.5)
|
|
self.assertAlmostEqual(diagnostics["intercept"], -1.25)
|
|
self.assertAlmostEqual(diagnostics["r_squared"], 1.0)
|
|
np.testing.assert_allclose([item[3] for item in rows], 0, atol=1e-12)
|
|
self.assertLessEqual(
|
|
diagnostics["slope_confidence_interval"][0], diagnostics["slope"]
|
|
)
|
|
self.assertGreaterEqual(
|
|
diagnostics["slope_confidence_interval"][1], diagnostics["slope"]
|
|
)
|
|
|
|
def test_xy_values_require_finite_strictly_increasing_x(self) -> None:
|
|
binding = {"input": "sample", "x": "x", "y": "y"}
|
|
with self.assertRaisesRegex(ValueError, "X_VALUES_DUPLICATED"):
|
|
worker._xy_values(["x", "y"], [[1, 2], [1, 3]], binding)
|
|
with self.assertRaisesRegex(ValueError, "Y_VALUE_NOT_FINITE"):
|
|
worker._xy_values(["x", "y"], [[1, 2], [2, float("nan")]], binding)
|
|
|
|
def test_csv_json_and_xlsx_outputs_are_deterministically_readable(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
csv_path = root / "input.csv"
|
|
csv_path.write_text("x,y\n1,2\n3,4\n", encoding="utf-8")
|
|
self.assertEqual(
|
|
worker._read_rows(csv_path, None),
|
|
(["x", "y"], [["1", "2"], ["3", "4"]]),
|
|
)
|
|
json_path = root / "input.json"
|
|
json_path.write_text(
|
|
json.dumps([{"x": 1, "y": 2}, {"x": 3, "y": 4}]),
|
|
encoding="utf-8",
|
|
)
|
|
self.assertEqual(
|
|
worker._read_rows(json_path, None), (["x", "y"], [[1, 2], [3, 4]])
|
|
)
|
|
|
|
result_path = root / "result-table.csv"
|
|
worker._atomic_csv(result_path, ["x", "y"], [[1, 2], [3, 4]])
|
|
self.assertTrue(result_path.read_bytes().startswith(b"\xef\xbb\xbf"))
|
|
workbook_path = root / "result-table.xlsx"
|
|
worker._atomic_xlsx(workbook_path, ["x", "y"], [[1, 2], [3, 4]])
|
|
self.assertTrue(workbook_path.is_file())
|
|
|
|
def test_run_builds_fixed_manifest_without_origin_side_effects_in_request(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
job_dir = Path(directory)
|
|
(job_dir / "request").mkdir()
|
|
(job_dir / "input" / "sample").mkdir(parents=True)
|
|
(job_dir / "input" / "sample" / "sample.csv").write_text(
|
|
"x,y\n0,0\n1,2\n2,4\n", encoding="utf-8"
|
|
)
|
|
request = _request("normalize", {"method": "max_abs"})
|
|
record = {
|
|
"job_id": "job",
|
|
"lease_id": "lease",
|
|
"request_digest": "a" * 64,
|
|
"request": request,
|
|
}
|
|
worker._atomic_json(job_dir / "request" / "request.json", record)
|
|
|
|
def fake_project(path, *_args):
|
|
path.write_bytes(b"origin-project" * 16)
|
|
|
|
with patch.object(worker, "_write_origin_project", side_effect=fake_project):
|
|
artifacts = worker.run(job_dir)
|
|
|
|
self.assertEqual(
|
|
{item["artifact_id"] for item in artifacts},
|
|
{"project", "result_table", "diagnostics", "analysis_spec", "provenance"},
|
|
)
|
|
diagnostics = json.loads(
|
|
(job_dir / "output" / "diagnostics.json").read_text(encoding="utf-8")
|
|
)
|
|
self.assertEqual(diagnostics["divisor"], 4.0)
|
|
provenance = json.loads(
|
|
(job_dir / "output" / "provenance.json").read_text(encoding="utf-8")
|
|
)
|
|
self.assertEqual(provenance["request_digest"], "a" * 64)
|
|
|
|
|
|
class OriginAnalysisAcceptanceTests(unittest.TestCase):
|
|
def test_fixed_acceptance_cases_are_contract_valid(self) -> None:
|
|
contract = get_contract("origin.analysis@v1")
|
|
cases = acceptance._cases()
|
|
self.assertEqual(
|
|
list(cases),
|
|
[
|
|
"data_check",
|
|
"normalize",
|
|
"smooth",
|
|
"differentiate",
|
|
"integrate",
|
|
"linear_fit",
|
|
],
|
|
)
|
|
for name, case in cases.items():
|
|
normalized, digest = contract.normalize_request(case["request"])
|
|
self.assertEqual(normalized, case["request"], name)
|
|
self.assertEqual(len(digest), 64)
|
|
|
|
def test_acceptance_oracles_are_independently_recomputed(self) -> None:
|
|
for name, case in acceptance._cases().items():
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
path = Path(directory) / "sample.csv"
|
|
path.write_text(case["csv"], encoding="utf-8", newline="")
|
|
headers, rows = worker._read_rows(path, None)
|
|
result_headers, result_rows, diagnostics = worker.analyze(
|
|
headers, rows, case["request"]["operation"]["analysis"]
|
|
)
|
|
validation = acceptance._validate_oracle(
|
|
name,
|
|
diagnostics,
|
|
result_headers,
|
|
result_rows,
|
|
case["oracle"],
|
|
)
|
|
self.assertTrue(validation["passed"], name)
|
|
|
|
def test_stage_writes_keyed_input_and_stable_request_digest(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
case = acceptance._cases()["linear_fit"]
|
|
job_dir = acceptance._stage(root, "linear_fit", case)
|
|
record = json.loads(
|
|
(job_dir / "request" / "request.json").read_text(encoding="utf-8")
|
|
)
|
|
_, digest = acceptance._canonical_request(case["request"])
|
|
self.assertEqual(record["request_digest"], digest)
|
|
self.assertTrue((job_dir / "input" / "sample" / "sample.csv").is_file())
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|