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()