220 lines
9.2 KiB
Python
220 lines
9.2 KiB
Python
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"<svg xmlns='http://www.w3.org/2000/svg'></svg>",
|
|
"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()
|