from __future__ import annotations import importlib.util import json import struct import tempfile import unittest from pathlib import Path from unittest.mock import patch from core.software_contracts import get_contract ROOT = Path(__file__).resolve().parents[1] ACCEPTANCE_PATH = ( ROOT / "windows-node" / "adapters" / "origin.plot@v2" / "acceptance.py" ) SPEC = importlib.util.spec_from_file_location("zcbot_origin_acceptance", ACCEPTANCE_PATH) assert SPEC and SPEC.loader acceptance = importlib.util.module_from_spec(SPEC) SPEC.loader.exec_module(acceptance) class OriginAcceptanceTests(unittest.TestCase): def test_fixed_cases_are_valid_contract_requests(self) -> None: contract = get_contract("origin.plot@v2") cases = acceptance._cases() self.assertEqual( list(cases), [ "annotations", "recipe_2x2", "heatmap", "surface_3d", "stacked", "stacked_line", "violin", "band", ], ) for case in cases.values(): normalized, digest = contract.normalize_request(case["request"]) self.assertEqual(normalized, case["request"]) self.assertEqual(len(digest), 64) self.assertEqual( {item["format"] for item in case["request"]["outputs"]}, {"opju", "png", "svg", "pdf"}, ) def test_scientific_oracles_use_known_values(self) -> None: cases = acceptance._cases() self.assertEqual(cases["stacked"]["oracle"]["stacked_totals"], [15, 18, 17, 22]) self.assertEqual(cases["heatmap"]["oracle"]["grid_shape"], [5, 7]) self.assertTrue(cases["band"]["oracle"]["lower_not_above_center"]) self.assertTrue(cases["band"]["oracle"]["center_not_above_upper"]) self.assertEqual( cases["stacked_line"]["oracle"]["stacked_line_baselines"], [0, 58.3, 116.6], ) self.assertEqual(cases["violin"]["oracle"]["violin_medians"], [50, 56, 65]) def test_stage_uses_stable_digest_and_keyed_csv_inputs(self) -> None: with tempfile.TemporaryDirectory() as directory: root = Path(directory) case = acceptance._cases()["annotations"] job_dir = acceptance._stage(root, "annotations", 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.assertEqual(record["request"], case["request"]) self.assertTrue((job_dir / "input" / "sample" / "sample.csv").is_file()) def test_project_data_validation_recomputes_stacked_oracle(self) -> None: class Sheet: def __init__(self, name, labels, rows): self.lname = name self.labels = labels self.rows = rows def get_labels(self, _kind): return self.labels def to_list2(self, **_arguments): return [list(column) for column in zip(*self.rows)] case = acceptance._cases()["stacked"] headers, rows = acceptance._csv_fixture(case["files"]["sample"]) staging = [[row[0], row[1], row[2], row[3]] for row in rows] workbook = [ Sheet("sample", ["age", "相 A", "相 B", "相 C"], rows), Sheet("stacked_plot_data", ["age", "相 A", "相 B", "相 C"], staging), ] result = acceptance._validate_project_data([workbook], [], case) self.assertTrue(result["oracle_validation"]["input_sheets_match"]) self.assertEqual( result["oracle_validation"]["stacked_totals"], [15.0, 18.0, 17.0, 22.0], ) def test_project_data_validation_recomputes_stacked_line_offsets(self) -> None: class Sheet: def __init__(self, name, labels, rows): self.lname = name self.labels = labels self.rows = rows def get_labels(self, _kind): return self.labels def to_list2(self, **_arguments): return [list(column) for column in zip(*self.rows)] case = acceptance._cases()["stacked_line"] headers, rows = acceptance._csv_fixture(case["files"]["spectrum"]) display = [ [row[0], row[1] - 12, row[0], row[2] - 15 + 58.3, row[0], row[3] - 8 + 116.6] for row in rows ] workbook = [ Sheet("spectrum", ["two_theta", "试样 A", "试样 B", "试样 C"], rows), Sheet( "stacked_line_plot_data", ["two_theta", "试样 A", "two_theta", "试样 B", "two_theta", "试样 C"], display, ), ] result = acceptance._validate_project_data([workbook], [], case) self.assertEqual( result["oracle_validation"]["stacked_line_baselines"], [0.0, 58.3, 116.6], ) def test_project_data_validation_recomputes_violin_medians(self) -> None: class Sheet: def __init__(self, name, labels, rows): self.lname = name self.labels = labels self.rows = rows def get_labels(self, _kind): return self.labels def to_list2(self, **_arguments): return [list(column) for column in zip(*self.rows)] case = acceptance._cases()["violin"] _, rows = acceptance._csv_fixture(case["files"]["strength"]) workbook = [[Sheet("strength", ["基准组", "掺量 5%", "掺量 10%"], rows)]] result = acceptance._validate_project_data(workbook, [], case) self.assertEqual( result["oracle_validation"]["violin_medians"], [50.0, 56.0, 65.0], ) def test_manifest_validation_checks_all_outputs_and_png_size(self) -> None: with tempfile.TemporaryDirectory() as directory: job_dir = Path(directory) output = job_dir / "output" output.mkdir() case = acceptance._cases()["annotations"] width = round(300 * 160 / 25.4) files = { "project.opju": b"O" * 64, "figure.png": b"\x89PNG\r\n\x1a\n" + b"\x00\x00\x00\rIHDR" + struct.pack(">II", width, 1000), "figure.svg": b"", "figure.pdf": b"%PDF-1.7\n%%EOF", } for name, content in files.items(): (output / name).write_bytes(content) acceptance.worker._atomic_json(output / "plot-spec.json", case["request"]) acceptance.worker._atomic_json( output / "provenance.json", {"adapter_version": acceptance.worker.ADAPTER_VERSION}, ) paths = [ output / "project.opju", output / "figure.png", output / "figure.svg", output / "figure.pdf", output / "plot-spec.json", output / "provenance.json", ] terminal = { "status": "succeeded", "artifact_manifest": [ acceptance.worker._manifest(path, "application/octet-stream") for path in paths ], } result = acceptance._validate_manifest(job_dir, case["request"], terminal) self.assertEqual(result["png_pixels"], [width, 1000]) self.assertEqual(set(result["artifacts"]), acceptance.EXPECTED_ARTIFACTS) def test_suite_always_writes_atomic_failure_report(self) -> None: with tempfile.TemporaryDirectory() as directory: root = Path(directory) / "acceptance" with ( patch.object(acceptance, "_environment_fingerprint", return_value={"test": True}), patch.object(acceptance, "_origin_processes", return_value={}), patch.object(acceptance, "_run_case", side_effect=RuntimeError("boom")), self.assertRaisesRegex(RuntimeError, "boom"), ): acceptance.run_suite(root, ["annotations"]) report = json.loads((root / "acceptance-report.json").read_text(encoding="utf-8")) self.assertFalse(report["passed"]) self.assertEqual(report["failure"]["type"], "RuntimeError") self.assertEqual(report["failure"]["completed_cases"], 0) def test_suite_records_selected_case_success(self) -> None: with tempfile.TemporaryDirectory() as directory: root = Path(directory) / "acceptance" result = {"case": "annotations", "elapsed_seconds": 1.0} with ( patch.object(acceptance, "_environment_fingerprint", return_value={"test": True}), patch.object(acceptance, "_origin_processes", return_value={}), patch.object(acceptance, "_run_case", return_value=result), ): report = acceptance.run_suite(root, ["annotations"]) self.assertTrue(report["passed"]) self.assertEqual(report["cases"], [result]) persisted = json.loads( (root / "acceptance-report.json").read_text(encoding="utf-8") ) self.assertTrue(persisted["passed"]) if __name__ == "__main__": unittest.main()