import csv import contextlib import hashlib import importlib.util import io import json import tempfile import unittest from pathlib import Path def load_compare_extracted_csv(): module_path = Path(__file__).resolve().parent / "compare_extracted_csv.py" spec = importlib.util.spec_from_file_location("compare_extracted_csv", module_path) module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) return module def write_json(path: Path, payload: dict): path.parent.mkdir(parents=True, exist_ok=True) path.write_text(json.dumps(payload, indent=2), encoding="utf-8") def write_csv(path: Path, rows: list[dict[str, str]]): path.parent.mkdir(parents=True, exist_ok=True) with path.open("w", newline="", encoding="utf-8") as handle: writer = csv.DictWriter(handle, fieldnames=list(rows[0])) writer.writeheader() writer.writerows(rows) def write_text(path: Path, text: str): path.parent.mkdir(parents=True, exist_ok=True) path.write_text(text, encoding="utf-8") def metadata_payload() -> dict: return { "schema_version": "abaqus-user-subroutine-artifact-v1", "feature_id": "umat", "model_id": "single-element", "artifact_status": "ready-for-comparison", "abaqus": {"version": "2024", "precision": "double"}, "compiler": {"vendor": "Intel oneAPI", "name": "ifx", "version": "2024"}, "subroutine": {"entry_points": ["UMAT"], "source_files": []}, "input_file": "model.inp", "outputs": { "tails": { "msg": "job.msg.tail.txt", "dat": "job.dat.tail.txt", "log": "job.log.tail.txt", "sta": "job.sta.tail.txt", }, "csv": {"stresses": "extracted/stresses.csv"}, }, "extraction": { "source_odb": "job.odb", "tool": "Abaqus Python", "extracted_at": "2026-06-10T00:00:00+09:00", "csv_directory": "extracted", }, "comparisons": { "stresses": { "reference_csv": "extracted/stresses.csv", "actual_csv": "extracted/stresses.csv", "required_columns": [ "step", "frame", "instance", "element_label", "integration_point", "section_point", "output_position", "component", "coordinate_system", "unit", "value", ], "key_columns": [ "step", "frame", "instance", "element_label", "integration_point", "section_point", "output_position", "component", ], "value_column": "value", "unit_column": "unit", "coordinate_system_column": "coordinate_system", "tolerance": {"absolute": 1.0e-8, "relative": 1.0e-6, "relative_floor": 1.0e-12}, } }, } def stress_rows(value: str = "100.0") -> list[dict[str, str]]: return [ { "step": "Step-1", "frame": "1", "instance": "PART-1-1", "element_label": "1", "integration_point": "1", "section_point": "", "output_position": "INTEGRATION_POINT", "component": "S11", "coordinate_system": "GLOBAL", "unit": "MPa", "value": value, } ] def make_metadata_valid_for_artifact_validation(root: Path, model_dir: Path, payload: dict) -> dict: source = root / "src" / "fortran" / "abaqus" / "UMAT.for" source.parent.mkdir(parents=True, exist_ok=True) source.write_text(" subroutine umat()\n end\n", encoding="utf-8") source_hash = hashlib.sha256(source.read_bytes()).hexdigest() payload["subroutine"]["source_files"] = [ { "path": "src/fortran/abaqus/UMAT.for", "language": "Fortran", "sha256": source_hash, } ] for name in [ "model.inp", "job.msg.tail.txt", "job.dat.tail.txt", "job.log.tail.txt", "job.sta.tail.txt", ]: path = model_dir / name path.parent.mkdir(parents=True, exist_ok=True) path.write_text("ok\n", encoding="utf-8") return payload def compare_stresses(reference_rows: list[dict[str, str]], actual_rows: list[dict[str, str]]) -> dict: compare = load_compare_extracted_csv() with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) reference = root / "references" / "umat" / "single-element" actual = root / "external-results" / "umat" / "single-element" write_json(reference / "metadata.json", metadata_payload()) write_csv(reference / "extracted" / "stresses.csv", reference_rows) write_csv(actual / "extracted" / "stresses.csv", actual_rows) return compare.compare_metadata( reference / "metadata.json", actual, quantities=["stresses"], validate_artifacts=False, ) def call_main_silently(compare, argv: list[str]) -> int: with contextlib.redirect_stdout(io.StringIO()), contextlib.redirect_stderr(io.StringIO()): return compare.main(argv) class CompareExtractedCsvTests(unittest.TestCase): def test_quantity_passes_when_schema_keys_units_and_values_match_within_tolerance(self): compare = load_compare_extracted_csv() with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) reference = root / "references" / "umat" / "single-element" actual = root / "external-results" / "umat" / "single-element" write_json(reference / "metadata.json", metadata_payload()) write_csv(reference / "extracted" / "stresses.csv", stress_rows("100.0")) write_csv(actual / "extracted" / "stresses.csv", stress_rows("100.00000001")) report = compare.compare_metadata( reference / "metadata.json", actual, quantities=["stresses"], validate_artifacts=False, ) self.assertEqual(report["overall_result"], "pass") self.assertEqual(report["quantities"][0]["result"], "pass") self.assertEqual(report["quantities"][0]["classification"], "N/A") self.assertEqual(report["quantities"][0]["compared_rows"], 1) def test_missing_actual_csv_is_missing_generated_output(self): compare = load_compare_extracted_csv() with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) reference = root / "references" / "umat" / "single-element" actual = root / "external-results" / "umat" / "single-element" write_json(reference / "metadata.json", metadata_payload()) write_csv(reference / "extracted" / "stresses.csv", stress_rows("100.0")) report = compare.compare_metadata( reference / "metadata.json", actual, quantities=["stresses"], validate_artifacts=False, ) self.assertEqual(report["overall_result"], "fail") self.assertEqual(report["quantities"][0]["classification"], "missing-generated-output") def test_missing_required_column_is_schema_mismatch(self): compare = load_compare_extracted_csv() with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) reference = root / "references" / "umat" / "single-element" actual = root / "external-results" / "umat" / "single-element" write_json(reference / "metadata.json", metadata_payload()) row = stress_rows("100.0")[0] write_csv(reference / "extracted" / "stresses.csv", [row]) actual_row = dict(row) actual_row.pop("coordinate_system") write_csv(actual / "extracted" / "stresses.csv", [actual_row]) report = compare.compare_metadata( reference / "metadata.json", actual, quantities=["stresses"], validate_artifacts=False, ) self.assertEqual(report["overall_result"], "fail") self.assertEqual(report["quantities"][0]["classification"], "schema-mismatch") def test_missing_quantity_contract_is_upstream_contract(self): compare = load_compare_extracted_csv() with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) reference = root / "references" / "umat" / "single-element" actual = root / "external-results" / "umat" / "single-element" payload = metadata_payload() payload["comparisons"].pop("stresses") write_json(reference / "metadata.json", payload) report = compare.compare_metadata( reference / "metadata.json", actual, quantities=["stresses"], validate_artifacts=False, ) self.assertEqual(report["overall_result"], "fail") self.assertEqual(report["quantities"][0]["classification"], "upstream-contract") def test_missing_comparisons_block_is_upstream_contract(self): compare = load_compare_extracted_csv() with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) reference = root / "references" / "umat" / "single-element" actual = root / "external-results" / "umat" / "single-element" payload = metadata_payload() payload.pop("comparisons") write_json(reference / "metadata.json", payload) report = compare.compare_metadata(reference / "metadata.json", actual, validate_artifacts=False) self.assertEqual(report["overall_result"], "fail") self.assertEqual(report["quantities"][0]["classification"], "upstream-contract") def test_incomplete_quantity_contract_is_upstream_contract(self): compare = load_compare_extracted_csv() with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) reference = root / "references" / "umat" / "single-element" actual = root / "external-results" / "umat" / "single-element" payload = metadata_payload() payload["comparisons"]["stresses"].pop("value_column") write_json(reference / "metadata.json", payload) report = compare.compare_metadata( reference / "metadata.json", actual, quantities=["stresses"], validate_artifacts=False, ) self.assertEqual(report["overall_result"], "fail") self.assertEqual(report["quantities"][0]["classification"], "upstream-contract") def test_duplicate_header_is_schema_mismatch(self): compare = load_compare_extracted_csv() with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) reference = root / "references" / "umat" / "single-element" actual = root / "external-results" / "umat" / "single-element" write_json(reference / "metadata.json", metadata_payload()) write_csv(reference / "extracted" / "stresses.csv", stress_rows("100.0")) header = ",".join(list(stress_rows("100.0")[0]) + ["value"]) row = ",".join(stress_rows("100.0")[0].values()) + ",100.0" write_text(actual / "extracted" / "stresses.csv", f"{header}\n{row}\n") report = compare.compare_metadata( reference / "metadata.json", actual, quantities=["stresses"], validate_artifacts=False, ) self.assertEqual(report["overall_result"], "fail") self.assertEqual(report["quantities"][0]["classification"], "schema-mismatch") def test_changed_row_key_is_id_mismatch(self): actual_row = dict(stress_rows("100.0")[0]) actual_row["element_label"] = "2" report = compare_stresses(stress_rows("100.0"), [actual_row]) self.assertEqual(report["overall_result"], "fail") self.assertEqual(report["quantities"][0]["classification"], "id-mismatch") self.assertEqual(report["quantities"][0]["missing_rows"], 1) self.assertEqual(report["quantities"][0]["extra_rows"], 1) def test_unit_mismatch_is_unit_or_coordinate_mismatch(self): actual_row = dict(stress_rows("100.0")[0]) actual_row["unit"] = "Pa" report = compare_stresses(stress_rows("100.0"), [actual_row]) self.assertEqual(report["overall_result"], "fail") self.assertEqual(report["quantities"][0]["classification"], "unit-or-coordinate-mismatch") def test_coordinate_system_mismatch_is_unit_or_coordinate_mismatch(self): actual_row = dict(stress_rows("100.0")[0]) actual_row["coordinate_system"] = "LOCAL-1" report = compare_stresses(stress_rows("100.0"), [actual_row]) self.assertEqual(report["overall_result"], "fail") self.assertEqual(report["quantities"][0]["classification"], "unit-or-coordinate-mismatch") def test_nonfinite_value_is_nonfinite_result(self): report = compare_stresses(stress_rows("100.0"), stress_rows("nan")) self.assertEqual(report["overall_result"], "fail") self.assertEqual(report["quantities"][0]["classification"], "nonfinite-result") def test_value_outside_tolerance_is_tolerance_failure(self): report = compare_stresses(stress_rows("100.0"), stress_rows("101.0")) self.assertEqual(report["overall_result"], "fail") self.assertEqual(report["quantities"][0]["classification"], "tolerance-failure") self.assertEqual(report["quantities"][0]["max_abs_error"], 1.0) self.assertEqual(report["quantities"][0]["result"], "fail") def test_cli_writes_json_report_and_returns_zero_when_all_quantities_pass(self): compare = load_compare_extracted_csv() with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) reference = root / "references" / "umat" / "single-element" actual = root / "external-results" / "umat" / "single-element" report_json = root / "build" / "reference-verification" / "umat-single-element.json" payload = make_metadata_valid_for_artifact_validation(root, reference, metadata_payload()) write_json(reference / "metadata.json", payload) write_csv(reference / "extracted" / "stresses.csv", stress_rows("100.0")) write_csv(actual / "extracted" / "stresses.csv", stress_rows("100.00000001")) exit_code = call_main_silently( compare, [ "--metadata", str(reference / "metadata.json"), "--actual-root", str(actual), "--quantity", "stresses", "--report-json", str(report_json), ], ) report = json.loads(report_json.read_text(encoding="utf-8")) self.assertEqual(exit_code, 0) self.assertEqual(report["overall_result"], "pass") def test_cli_writes_json_report_and_returns_one_when_a_quantity_fails(self): compare = load_compare_extracted_csv() with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) reference = root / "references" / "umat" / "single-element" actual = root / "external-results" / "umat" / "single-element" report_json = root / "build" / "reference-verification" / "umat-single-element.json" payload = make_metadata_valid_for_artifact_validation(root, reference, metadata_payload()) write_json(reference / "metadata.json", payload) write_csv(reference / "extracted" / "stresses.csv", stress_rows("100.0")) write_csv(actual / "extracted" / "stresses.csv", stress_rows("101.0")) exit_code = call_main_silently( compare, [ "--metadata", str(reference / "metadata.json"), "--actual-root", str(actual), "--quantity", "stresses", "--report-json", str(report_json), ], ) report = json.loads(report_json.read_text(encoding="utf-8")) self.assertEqual(exit_code, 1) self.assertEqual(report["overall_result"], "fail") def test_cli_returns_two_for_invalid_arguments(self): compare = load_compare_extracted_csv() self.assertEqual(call_main_silently(compare, []), 2) def test_default_comparison_validates_reference_artifact_metadata(self): compare = load_compare_extracted_csv() with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) reference = root / "references" / "umat" / "single-element" actual = root / "external-results" / "umat" / "single-element" write_json(reference / "metadata.json", metadata_payload()) write_csv(reference / "extracted" / "stresses.csv", stress_rows("100.0")) write_csv(actual / "extracted" / "stresses.csv", stress_rows("100.0")) report = compare.compare_metadata(reference / "metadata.json", actual) self.assertEqual(report["overall_result"], "fail") self.assertEqual(report["quantities"][0]["classification"], "missing-reference-artifact") if __name__ == "__main__": unittest.main()