chore: remove retired workflow and historical material

This commit is contained in:
KOKO\Mimi
2026-10-17 22:39:47 +09:00
parent 4bd19f5bf0
commit 774b80099c
768 changed files with 0 additions and 54164 deletions
-422
View File
@@ -1,422 +0,0 @@
import json
from pathlib import Path
import re
import tomllib
import pytest
from test_execute import step_runtime
ROOT = Path(__file__).resolve().parents[1]
AGENT_DIR = ROOT / ".codex" / "agents"
SKILL_DIR = ROOT / ".codex" / "skills"
EXPECTED_AGENTS = {
"coordinator-agent",
"correction-agent",
"formulation-agent",
"implementation-agent",
"implementation-planning-agent",
"io-definition-agent",
"numerical-review-agent",
"physics-evaluation-agent",
"release-agent",
"requirement-agent",
"research-agent",
}
EXPECTED_SKILLS = {
"fem-theory-query",
"fem-implementation-tdd",
"fem-formulation-spec",
"fem-io-contract",
"fem-numerical-review",
"fem-physics-review",
"fem-release-readiness",
"fem-requirements-baseline",
"fem-research-evidence",
}
LEGACY_OUTPUT_DIRS = {
"coordination",
"requirements",
"research",
"formulations",
"numerical-reviews",
"io-definitions",
"reference-models",
"implementation-plans",
"build-test-reports",
"corrections",
"reference-verifications",
"physics-evaluations",
"releases",
}
FEATURE_FILES = {
"linear-static-3d-euler-beam": {
"coordination.md",
"requirements.md",
"research.md",
"formulation.md",
"numerical-review.md",
"reference-model.md",
"io.md",
"implementation-plan.md",
"implementation-report.md",
"build-test.md",
"reference-comparison.md",
"physics-evaluation.md",
"release.md",
},
"linear-static-mitc4-shell": {
"coordination.md",
"requirements.md",
"research.md",
"formulation.md",
"numerical-review.md",
"reference-model.md",
"io.md",
"implementation-plan.md",
"build-test.md",
"reference-comparison.md",
"physics-evaluation.md",
"release.md",
},
}
LIVE_CONTRACT_FILES = (
ROOT / "AGENTS.md",
ROOT / "docs" / "SOLVER_AGENT_DESIGN.md",
ROOT / "docs" / "SOLVER_SKILL_DESIGN.md",
)
AGENT_STAGE_GUIDANCE_FILES = (
ROOT / "AGENTS.md",
ROOT / "docs" / "SOLVER_AGENT_DESIGN.md",
)
EXPECTED_AGENT_STAGE_MAPPING = (
("requirements", "requirement-agent", ("requirements.md",)),
("research", "research-agent", ("research.md",)),
("formulation", "formulation-agent", ("formulation.md",)),
(
"numerical-review",
"numerical-review-agent",
("numerical-review.md", "reference-model.md"),
),
("io-definition", "io-definition-agent", ("io.md",)),
(
"implementation-planning",
"implementation-planning-agent",
("implementation-plan.md",),
),
(
"implementation",
"implementation-agent",
("implementation-report.md", "build-test.md", "reference-comparison.md"),
),
("physics-evaluation", "physics-evaluation-agent", ("physics-evaluation.md",)),
("release-readiness", "release-agent", ("release.md",)),
)
CANONICAL_FEATURE_DIRS = (
ROOT / "docs" / "linear-static-3d-euler-beam",
ROOT / "docs" / "linear-static-mitc4-shell",
)
# Plans and specs intentionally retain deleted source paths as migration history.
HISTORICAL_PATH_EXEMPT_DIRS = (
ROOT / "docs" / "superpowers" / "plans",
ROOT / "docs" / "superpowers" / "specs",
)
LEGACY_README_PATTERN = re.compile(
r"docs/(?:coordination|requirements|research|formulations|numerical-reviews|"
r"io-definitions|reference-models|implementation-plans|build-test-reports|"
r"corrections|reference-verifications|physics-evaluations|releases)/README\.md",
re.IGNORECASE,
)
EXPECTED_SKILL_FRONTMATTER_KEYS = ("name", "description")
EXPECTED_SKILL_INTERFACE_KEYS = (
"display_name",
"short_description",
"default_prompt",
)
RETIRED_TOKENS = (
"reference-model-agent",
"reference model agent",
"build-test-executor-agent",
"build/test executor agent",
"reference-verification-agent",
"reference verification agent",
"fesa-reference-models",
"fesa-reference-comparison",
"docs/coordination/",
"docs/requirements/",
"docs/research/",
"docs/formulations/",
"docs/numerical-reviews/",
"docs/io-definitions/",
"docs/reference-models/",
"docs/implementation-plans/",
"docs/build-test-reports/",
"docs/corrections/",
"docs/reference-verifications/",
"docs/physics-evaluations/",
"docs/releases/",
)
def read(path: Path) -> str:
return path.read_text(encoding="utf-8")
def live_contract_text() -> str:
paths = list(LIVE_CONTRACT_FILES)
paths.extend(sorted(AGENT_DIR.glob("*.toml")))
paths.extend(sorted(SKILL_DIR.glob("*/SKILL.md")))
return "\n".join(read(path) for path in paths)
def markdown_stage_mapping(path: Path) -> tuple[tuple[str, str, tuple[str, ...]], ...]:
heading = (
"## Agent/Skill Workflow"
if path.name == "AGENTS.md"
else "## 4. 단계와 소유권"
)
section = read(path).split(heading, maxsplit=1)[1].split("\n## ", maxsplit=1)[0]
rows = []
for line in section.splitlines():
if not line.lstrip().startswith("|"):
continue
columns = [column.strip().replace("`", "") for column in line.strip().split("|")[1:-1]]
if len(columns) < 3:
continue
stage = re.sub(r"^\d+\.\s+", "", columns[0])
if re.fullmatch(r"[a-z]+(?:-[a-z]+)*", stage):
reports = tuple(re.findall(r"\b[\w-]+\.md\b", columns[-1]))
rows.append((stage, columns[1], reports))
return tuple(rows)
def operational_markdown_paths() -> tuple[Path, ...]:
paths = [
path
for feature_dir in CANONICAL_FEATURE_DIRS
for path in sorted(feature_dir.glob("*.md"))
]
paths.extend(sorted(ROOT.glob("phases/**/step*.md")))
return tuple(paths)
def parse_skill_frontmatter(path: Path) -> dict[str, str]:
lines = read(path).splitlines()
assert lines and lines[0] == "---", path
assert "---" in lines[1:], path
closing_index = lines.index("---", 1)
field_lines = lines[1:closing_index]
assert len(field_lines) == len(EXPECTED_SKILL_FRONTMATTER_KEYS), path
parsed = {}
for line in field_lines:
match = re.fullmatch(r"([a-z_]+): ([^\s].*)", line)
assert match is not None, (path, line)
key, value = match.groups()
assert key not in parsed, (path, key)
parsed[key] = value
assert tuple(parsed) == EXPECTED_SKILL_FRONTMATTER_KEYS, path
assert re.fullmatch(r"[a-z0-9]+(?:-[a-z0-9]+)*", parsed["name"]), path
assert parsed["description"].startswith("Use when "), path
return parsed
def parse_skill_interface(path: Path) -> dict[str, str]:
lines = [line for line in read(path).splitlines() if line]
assert lines and lines[0] == "interface:", path
assert len(lines) == len(EXPECTED_SKILL_INTERFACE_KEYS) + 1, path
parsed = {}
for line in lines[1:]:
match = re.fullmatch(r' ([a-z_]+): ("(?:[^"\\]|\\.)*")', line)
assert match is not None, (path, line)
key, quoted_value = match.groups()
assert key not in parsed, (path, key)
parsed[key] = json.loads(quoted_value)
assert tuple(parsed) == EXPECTED_SKILL_INTERFACE_KEYS, path
assert all(parsed.values()), path
return parsed
def test_agent_inventory_is_consolidated():
actual = {path.stem for path in AGENT_DIR.glob("*.toml")}
assert actual == EXPECTED_AGENTS
def test_project_skill_inventory_is_consolidated():
actual = {
path.name
for path in SKILL_DIR.iterdir()
if path.is_dir() and (path / "SKILL.md").is_file()
}
assert actual == EXPECTED_SKILLS
def test_existing_feature_artifacts_are_bundled():
docs_dir = ROOT / "docs"
for legacy_dir in LEGACY_OUTPUT_DIRS:
assert not (docs_dir / legacy_dir).exists(), legacy_dir
for feature_id, expected_files in FEATURE_FILES.items():
feature_dir = docs_dir / feature_id
actual_files = {path.name for path in feature_dir.glob("*.md")}
# Preserve historical artifacts while allowing Feature management and linked details.
assert expected_files <= actual_files
def test_numerical_review_absorbs_reference_model_contract():
assert not (AGENT_DIR / "reference-model-agent.toml").exists()
assert not (SKILL_DIR / "fesa-reference-models").exists()
text = read(SKILL_DIR / "fem-numerical-review" / "SKILL.md")
outputs = set(re.findall(r"docs/<feature-id>/([\w-]+\.md)", text))
assert {"numerical-review.md", "reference-model.md"} <= outputs
def test_implementation_absorbs_build_test_and_reference_comparison():
assert not (AGENT_DIR / "build-test-executor-agent.toml").exists()
assert not (AGENT_DIR / "reference-verification-agent.toml").exists()
assert not (SKILL_DIR / "fesa-reference-comparison").exists()
text = read(SKILL_DIR / "fem-implementation-tdd" / "SKILL.md")
outputs = set(re.findall(r"docs/<feature-id>/([\w-]+\.md)", text))
assert {
"implementation-plan.md",
"implementation-report.md",
"build-test.md",
"reference-comparison.md",
"corrections.md",
} <= outputs
def test_implementation_verification_reference_is_resolvable():
skill = SKILL_DIR / "fem-implementation-tdd" / "SKILL.md"
links = re.findall(r"\[[^\]]+\]\(([^)]+)\)", read(skill))
assert "references/toolchain-and-verification.md" in links
for link in links:
assert (skill.parent / link).is_file(), link
def test_p_agent_001_implementation_guidance_resolves_coding_style_contract():
parsed = tomllib.loads(read(AGENT_DIR / "implementation-agent.toml"))
instructions = parsed["developer_instructions"]
assert "AGENTS.md" in instructions
project_links = re.findall(r"\[[^\]]+\]\(([^)]+)\)", read(ROOT / "AGENTS.md"))
assert "docs/CODINGSTYLE.md" in project_links
assert (ROOT / "docs/CODINGSTYLE.md").is_file()
def test_agent_profiles_reference_shared_authority_and_feature_context():
for path in sorted(AGENT_DIR.glob("*.toml")):
instructions = tomllib.loads(read(path))["developer_instructions"]
for document in (
"AGENTS.md",
"docs/SOLVER_AGENT_DESIGN.md",
"docs/FEATURE_DOCUMENTATION.md",
"FEATURED_AGENTS.md",
"PRD.md",
"ADR.md",
):
assert document in instructions, (path.name, document)
for skill_name in re.findall(r"\$([a-z0-9-]+)", instructions):
candidates = (
SKILL_DIR / skill_name / "SKILL.md",
ROOT / ".agents/skills" / skill_name / "SKILL.md",
)
assert any(candidate.is_file() for candidate in candidates), (path, skill_name)
def test_live_contracts_have_no_retired_names_or_paths():
text = live_contract_text().lower()
assert "docs/<feature-id>/" in text
for token in RETIRED_TOKENS:
assert token not in text, token
def test_live_guidance_declares_consistent_ordered_nine_stage_workflow():
for path in LIVE_CONTRACT_FILES:
text = read(path)
assert "docs/<feature-id>/" in text, path
for path in AGENT_STAGE_GUIDANCE_FILES:
assert markdown_stage_mapping(path) == EXPECTED_AGENT_STAGE_MAPPING, path
def test_operational_documents_do_not_use_deleted_stage_readmes():
paths = operational_markdown_paths()
assert paths
for path in paths:
assert not any(path.is_relative_to(directory) for directory in HISTORICAL_PATH_EXEMPT_DIRS)
match = LEGACY_README_PATTERN.search(read(path))
assert match is None, (path, match.group(0) if match else None)
def test_mitc4_live_authority_links_use_feature_bundle():
text = read(ROOT / "docs" / "MITC4_SUPP.md")
authority_section = text.split("라인 참조", maxsplit=1)[0]
for path in (
"docs/linear-static-mitc4-shell/requirements.md",
"docs/linear-static-mitc4-shell/formulation.md",
"docs/linear-static-mitc4-shell/io.md",
"docs/linear-static-mitc4-shell/reference-model.md",
"docs/linear-static-mitc4-shell/numerical-review.md",
"docs/linear-static-mitc4-shell/release.md",
):
assert path in authority_section
for legacy_path in (
"docs/requirements/linear-static-mitc4-shell.md",
"docs/formulations/mitc4-shell-formulation.md",
"docs/io-definitions/linear-static-mitc4-shell-io.md",
"docs/reference-models/linear-static-mitc4-shell-reference-models.md",
"docs/numerical-reviews/linear-static-mitc4-shell-review.md",
"docs/releases/linear-static-mitc4-shell-release.md",
):
assert legacy_path not in authority_section
def test_agent_toml_and_limited_skill_yaml_schemas_are_parseable():
for path in sorted(AGENT_DIR.glob("*.toml")):
parsed = tomllib.loads(read(path))
assert parsed["name"] == path.stem
assert parsed["model_reasoning_effort"] == "xhigh"
assert parsed["sandbox_mode"] == "workspace-write"
assert parsed["developer_instructions"].strip(), path
for skill_name in EXPECTED_SKILLS:
skill_dir = SKILL_DIR / skill_name
frontmatter = parse_skill_frontmatter(skill_dir / "SKILL.md")
interface = parse_skill_interface(skill_dir / "agents" / "openai.yaml")
assert frontmatter["name"] == skill_name
assert f"${skill_name}" in interface["default_prompt"]
@pytest.mark.parametrize("document", [".agents/skills/harness/SKILL.md", "docs/HARNESS_WORKFLOW.md"])
def test_documented_phase_example_passes_executor_preflight(step_runtime, document):
"""Exercise the documented schema with its real consumer, not prose matching."""
from scripts import execute
examples = [json.loads(block) for block in re.findall(r"```json\s*\n(.*?)\n```", read(ROOT / document), re.DOTALL)]
phase = next(example for example in examples if "steps" in example)
phase["project"] = "Harness Tests"
phase["phase"] = "runtime"
phase["steps"][0]["name"] = "runtime"
runtime = step_runtime
runtime.executor._write_json(runtime.index_file, phase)
executor = execute.StepExecutor("runtime")
executor._preflight_repository()
assert runtime.calls == []
-856
View File
@@ -1,856 +0,0 @@
"""Independent comparison contracts; generated result fixtures are not source evidence."""
import csv
import importlib.util
import json
import math
import os
from pathlib import Path
import shutil
import subprocess
import sys
import h5py
import numpy as np
import pytest
ROOT = Path(__file__).resolve().parents[1]
SPEC = importlib.util.spec_from_file_location(
"compare_hex8", ROOT / "scripts/verification/compare_hex8.py"
)
COMPARE = importlib.util.module_from_spec(SPEC)
SPEC.loader.exec_module(COMPARE)
def family(actual, reference):
identities = [{"node_label": i + 1, "component": "U1"} for i in range(len(actual))]
return COMPARE.evaluate_family(
{"case": "hexa_linearstatic", "step": "Step-1", "increment": 10,
"quantity": "U", "unit_dimensions": "length", "rows": identities},
actual, reference,
)
def test_zero_reference_scale_requires_exact_zero():
report = family([0.0, 1e-300], [0.0, 0.0])
assert not report["passed"]
assert report["diagnostic_code"] == "zero-reference-scale-nonzero-error"
def test_row_pass_does_not_override_one_percent_rms():
report = family([1.02] * 10, [1.0] * 10)
assert report["rows_passed"]
assert not report["rms_passed"]
assert not report["passed"]
def test_near_zero_band_uses_reference_scale_only():
report = family([100.0, 1.0], [100.0, 0.0])
assert report["passed"]
assert report["rows"][1]["branch"] == "near-zero"
assert not family([100.0, 1.0001], [100.0, 0.0])["passed"]
def test_nonfinite_family_fails_before_tolerance():
with pytest.raises(COMPARE.ComparisonError, match="finite"):
family([float("nan")], [1.0])
def test_header_strip_collision_rejected_before_bad_numeric(tmp_path):
path = tmp_path / "collision.csv"
path.write_text("Node Label, Node Label\n1,NaN\n", encoding="utf-8")
with pytest.raises(COMPARE.ComparisonError) as error:
COMPARE.read_reference(path, "hexa_linearstatic", "U", {})
assert error.value.code == "duplicate-reference-header"
def test_modal_quantity_and_units_are_explicit():
rows = [{"step": "Step-1", "mode": i, "quantity": "eigenvalue",
"unit_dimensions": "1/time^2", "value": float(i)} for i in range(1, 11)]
with pytest.raises(COMPARE.ComparisonError) as error:
COMPARE.compare_modal(rows, rows, "frequency", "1/time")
assert error.value.code == "wrong-modal-quantity"
def test_modal_singletons_use_common_one_percent_rms():
reference = [{"step": "Step-1", "mode": i, "quantity": "eigenvalue",
"unit_dimensions": "1/time^2", "value": float(i)} for i in range(1, 11)]
actual = [dict(row, value=row["value"] * 1.02) for row in reference]
report = COMPARE.compare_modal(reference, actual, "eigenvalue", "1/time^2")
assert not report["passed"]
assert len(report["families"]) == 10
assert all(row["rows_passed"] and not row["rms_passed"] for row in report["families"])
# These schema declarations are independent of the comparator's declarations.
TEXT = h5py.string_dtype(encoding="utf-8")
REFERENCE = ROOT / "reference/hexa"
def write_strings(obj, attributes):
for name, value in attributes.items():
obj.attrs.create(name, value, dtype=TEXT)
def compound(group, name, schema, rows):
data = np.empty(len(rows), dtype=np.dtype(schema))
for i, row in enumerate(rows):
data[i] = row
return group.create_dataset(name, data=data)
@pytest.fixture(scope="module")
def source_models():
return {case: COMPARE.input_inventory(REFERENCE / f"{case}.inp", case)
for case in ("hexa_linearstatic", "hexa_nonlinearstatic", "hexa_dynlinear", "hexa_dynnonlinear", "hexa_freq")}
def source_rows(case, quantity):
with (REFERENCE / f"{case}_{quantity}.csv").open(newline="", encoding="utf-8") as stream:
reader = csv.reader(stream)
headers = [value.strip() for value in next(reader)]
return [dict(zip(headers, [value.strip() for value in row])) for row in reader]
def make_result(path, model):
"""Build an original-sized schema fixture, not a validated physical solution."""
case = model["case"]
dynamic = case in ("hexa_dynlinear", "hexa_dynnonlinear")
nonlinear = case in ("hexa_nonlinearstatic", "hexa_dynnonlinear")
frequency = case == "hexa_freq"
reference_values = {}
if not frequency:
for quantity in (["U"] if dynamic else ["U", "RF", "S"]):
table_rows = {}
for row in source_rows(case, quantity):
increment = int(row["Frame"].split(":")[0].split()[-1])
entity = int(row["Element Label" if quantity == "S" else "Node Label"])
point = int(row["IntPt"]) if quantity == "S" else 0
names = ("S11", "S22", "S33", "S12", "S13", "S23") if quantity == "S" else tuple(f"{quantity}{c}" for c in (1, 2, 3))
table_rows[increment, entity, point] = [float(row[f"{quantity}-{name}"]) for name in names]
reference_values[quantity] = table_rows
with h5py.File(path, "w") as file:
metadata = file.create_group("metadata")
metadata.attrs.create("schema_version", np.uint64(1), dtype="<u8")
write_strings(metadata, {
"feature_id": "hex8-structural-analysis", "analysis_procedure": {
"hexa_linearstatic": "linear-static", "hexa_nonlinearstatic": "nonlinear-static-full-newton",
"hexa_dynlinear": "hht-linear-dynamics", "hexa_dynnonlinear": "hht-nonlinear-dynamics",
"hexa_freq": "linear-frequency"}[case],
"element_formulation": "FESA-Hex8", "integration_rule": "gauss-2x2x2",
"mass_formulation": "lumped-reference", "source_step_name": "Step-1",
"source_input_identity": f"path={model['source_path']};content_identity={model['content_identity']}",
"unit_system_label": "user-consistent-unspecified",
"coordinate_convention": "global-cartesian; reference-strain; current-cauchy-stress",
"component_order": "node-major; UX,UY,UZ", "solver_version": "fixture-not-source-evidence",
})
group = file.create_group("model")
compound(group, "nodes", [("internal_node_id", "<u8"), ("instance_name", TEXT),
("source_label", TEXT), ("coordinates", "<f8", (3,))],
[(i, "Part-1-1", str(label), coords) for i, (label, coords) in enumerate(model["nodes"].items())])
compound(group, "elements", [("internal_element_id", "<u8"), ("instance_name", TEXT),
("source_label", TEXT), ("source_element_type", TEXT),
("internal_formulation", TEXT), ("node_internal_ids", "<u8", (8,)),
("solid_section_internal_id", "<u8"), ("material_internal_id", "<u8")],
[(i, "Part-1-1", str(label), "C3D8", "FESA-Hex8", tuple(n - 1 for n in nodes), 0, 0)
for i, (label, nodes) in enumerate(model["elements"].items())])
solids = group.create_group("solid")
compound(solids, "materials", [("internal_material_id", "<u8"), ("name", TEXT),
("youngs_modulus", "<f8"), ("poisson_ratio", "<f8"),
("density_present", "u1"), ("density", "<f8")],
[(0, "Material-1", 2.1e11, .3, 1, 7850.)])
compound(solids, "sections", [("internal_section_id", "<u8"), ("source_file", TEXT),
("source_line", "<u8"), ("source_elset", TEXT), ("material_internal_id", "<u8")],
[(0, model["source_path"], model["section_line"], "Set-1", 0)])
point_signs = ((-1, -1, -1), (1, -1, -1), (-1, 1, -1), (1, 1, -1),
(-1, -1, 1), (1, -1, 1), (-1, 1, 1), (1, 1, 1))
node_signs = ((-1, -1, -1), (1, -1, -1), (1, 1, -1), (-1, 1, -1),
(-1, -1, 1), (1, -1, 1), (1, 1, 1), (-1, 1, 1))
points = []
for i, (_, connectivity) in enumerate(model["elements"].items()):
coords = np.array([model["nodes"][n] for n in connectivity])
for p, signs in enumerate(point_signs, 1):
natural = np.array(signs) / math.sqrt(3)
shape = np.array([math.prod(1 + sign * x for sign, x in zip(node, natural)) / 8
for node in node_signs])
points.append((i, p, natural, np.sum(shape[:, None] * coords, axis=0)))
compound(group, "solid_integration_points", [("internal_element_id", "<u8"), ("point_id", "u1"),
("natural_coordinates", "<f8", (3,)),
("reference_coordinates", "<f8", (3,))], points)
mask = group.create_dataset("nodal_constraint_mask", data=np.array(
[[(node, c) in model["constrained"] for c in (1, 2, 3)] for node in model["nodes"]], dtype="u1"))
write_strings(mask, {"component_names": "UX,UY,UZ"})
step = file.create_group("steps/Step-1")
if frequency:
eigen = step.create_group("eigen")
modes = compound(eigen, "modes", [("mode_id", "<u8"), ("eigenvalue", "<f8"),
("classification", "u1"), ("cluster_id", "<u8")],
[(m, float(m), 1, m) for m in range(1, 11)])
frequencies = compound(eigen, "frequencies", [("mode_id", "<u8"), ("frequency", "<f8")],
[(m, math.sqrt(m) / (2 * math.pi)) for m in range(1, 11)])
for obj, name, units in ((modes, "eigenvalue", "1/time^2"), (frequencies, "frequency", "1/time")):
write_strings(obj, {"component_names": name, "unit_dimensions": units,
"location": "global", "coordinate_system": "global-cartesian"})
vectors = eigen.create_dataset("vectors", data=np.zeros((10, 442, 3), dtype="<f8"))
write_strings(vectors, {"component_names": "UX,UY,UZ", "unit_dimensions": "mass-normalized",
"location": "nodal", "coordinate_system": "global-cartesian",
"normalization": "phi^T*M*phi=1"})
verification = eigen.create_group("verification")
compound(verification, "pairs", [("sorted_rank", "<u8"), ("solver_column", "<u8"),
("eigenvalue", "<f8"), ("modal_mass", "<f8"),
("rayleigh_quotient", "<f8"), ("eta_r", "<f8"),
("eta_norm", "<f8"), ("eta_energy", "<f8")],
[(m, m - 1, float(m), 1., float(m), 0., 0., 0.) for m in range(1, 11)])
verification.create_dataset("eta_gram", data=0., dtype="<f8")
verification.create_dataset("max_eta_orth", data=0., dtype="<f8")
return
count = 101 if dynamic else (10 if nonlinear else 1)
if dynamic or nonlinear:
compound(step, "accepted_frames", [("frame_index", "<u8"), ("increment_index", "<u8"),
("time" if dynamic else "load_factor", "<f8"),
("iteration_count", "<u8"),
("is_initial" if dynamic else "is_factor_one", "u1")],
[(frame, frame if dynamic else frame + 1, (frame if dynamic else frame + 1) * .1,
int(not (dynamic and frame == 0)), int(frame == (0 if dynamic else 9))) for frame in range(count)])
for frame in range(count):
increment = frame if dynamic else (frame + 1 if nonlinear else 10)
coordinate = increment * .1 if dynamic or nonlinear else 1.
frame_group = step.create_group(f"frames/{frame}")
definitions = {
"nodal/displacement": ((442, 3), "U1,U2,U3", "length", {}),
"nodal/reaction": ((442, 3), "RF1,RF2,RF3", "force", {}),
"nodal/residual": ((442, 3), "R1,R2,R3", "force", {}),
"solid/stress": ((300, 8, 6), "S11,S22,S33,S12,S13,S23", "force/length^2", {"stress_measure": "cauchy"}),
"solid/strain": ((300, 8, 6), "E11,E22,E33,E12,E13,E23", "dimensionless",
{"strain_measure": "green-lagrange" if nonlinear else "infinitesimal", "shear_convention": "tensor"}),
"global/strain_energy": ((), "scalar", "force*length", {"definition": "sum(W*detJ0)" if nonlinear else "0.5*u^T*K0*u"}),
}
if dynamic:
definitions.update({
"nodal/velocity": ((442, 3), "V1,V2,V3", "length/time", {}),
"nodal/acceleration": ((442, 3), "A1,A2,A3", "length/time^2", {"acceleration_kind": "physical"}),
"nodal/algorithmic_acceleration": ((442, 3), "A1,A2,A3", "length/time^2", {"acceleration_kind": "algorithmic"}),
"global/kinetic_energy": ((), "scalar", "force*length", {"definition": "0.5*v^T*M*v"}),
"global/external_work": ((), "scalar", "force*length", {"definition": "sum(0.5*(F_old+F_new)*delta_u)"}),
"global/energy_balance_error": ((), "scalar", "force*length", {"definition": "T+U-T0-U0-Wext"}),
"global/linear_momentum": ((3,), "P1,P2,P3", "mass*length/time", {}),
"global/angular_momentum": ((3,), "L1,L2,L3", "mass*length^2/time", {"origin": "global0"}),
})
for key, (shape, names, units, extra) in definitions.items():
values = np.zeros(shape, dtype="<f8")
quantity = {"nodal/displacement": "U", "nodal/reaction": "RF", "solid/stress": "S"}.get(key)
if quantity in reference_values and increment in (range(101) if dynamic else (10,)):
for (n, entity, point), components in reference_values[quantity].items():
if n == increment:
values[(entity - 1, point - 1) if quantity == "S" else entity - 1] = components
obj = frame_group.create_dataset(key, data=values)
write_strings(obj, {"component_names": names, "unit_dimensions": units, "step_name": "Step-1",
"coordinate_system": "global-cartesian", "location": {
"nodal": "nodal", "solid": "integration-point", "global": "global"}[key.split('/')[0]], **extra})
obj.attrs.create("frame_index", np.uint64(frame), dtype="<u8")
obj.attrs.create("increment_index", np.uint64(increment), dtype="<u8")
obj.attrs.create("time" if dynamic else "load_factor", coordinate, dtype="<f8")
@pytest.fixture(scope="module")
def result_fixtures(tmp_path_factory, source_models):
directory = tmp_path_factory.mktemp("hex8-comparison-fixtures")
paths = {}
for case, model in source_models.items():
path = directory / f"{case}.h5"
make_result(path, model)
paths[case] = path
return paths
def copy_result(tmp_path, result_fixtures, case="hexa_linearstatic"):
path = tmp_path / "mutated.h5"
shutil.copyfile(result_fixtures[case], path)
return path
def compare_fixture(path, case="hexa_linearstatic", execution=None):
return COMPARE.compare(case, REFERENCE / f"{case}.inp", path, REFERENCE, execution)
def mutate_csv(tmp_path, case, quantity, mutation):
rows = source_rows(case, quantity)
headers = list(rows[0])
mutation(headers, rows)
path = tmp_path / "mutated.csv"
with path.open("w", newline="", encoding="utf-8") as stream:
writer = csv.DictWriter(stream, fieldnames=headers)
writer.writeheader()
writer.writerows(rows)
return path
@pytest.mark.parametrize("case", ["hexa_linearstatic", "hexa_nonlinearstatic", "hexa_dynlinear", "hexa_dynnonlinear"])
def test_all_original_sized_schema_fixtures_compare_by_identity(case, result_fixtures):
report = compare_fixture(result_fixtures[case], case)
assert report["precheck"]["passed"], report["precheck"]
assert report["verdict"] == "passed"
assert report["execution"]["verified"] is False
assert len(report["families"]) == (101 if case.startswith("hexa_dyn") else 3)
if not case.startswith("hexa_dyn"):
assert [family["row_count"] for family in report["families"]] == [1326, 51, 14400]
assert report["precheck"]["reference_row_counts"]["RF"] == 442
else:
assert [family["increment"] for family in report["families"]] == list(range(101))
assert all(family["row_count"] == 1326 for family in report["families"])
assert report["families"][0]["reference_scale"] == 0.
def test_static_final_factor_maps_to_increment10_not_storage_frame(result_fixtures):
report = compare_fixture(result_fixtures["hexa_nonlinearstatic"], "hexa_nonlinearstatic")
assert {family["increment"] for family in report["families"]} == {10}
assert all(family["passed"] for family in report["families"])
@pytest.mark.parametrize("quantity", ["U", "RF", "S"])
def test_csv_order_does_not_supply_identity(tmp_path, source_models, quantity):
path = mutate_csv(tmp_path, "hexa_linearstatic", quantity, lambda _, rows: rows.reverse())
rows = COMPARE.read_reference(path, "hexa_linearstatic", quantity, source_models["hexa_linearstatic"])
original = COMPARE.read_reference(REFERENCE / f"hexa_linearstatic_{quantity}.csv", "hexa_linearstatic", quantity, source_models["hexa_linearstatic"])
assert COMPARE.reference_values(rows, quantity, source_models["hexa_linearstatic"]) == COMPARE.reference_values(original, quantity, source_models["hexa_linearstatic"])
@pytest.mark.parametrize("field,value,code", [
("Node Label", "443", "extra-reference-entity"), ("Part Instance Name", "OTHER", "wrong-reference-instance"),
("Step", "Step-2", "wrong-reference-step"), ("ODB Name", "hexa_nonlinearstatic.odb", "wrong-reference-case"),
("Frame", "Increment 9: Step Time = 1.000", "wrong-reference-frame-time"),
("Frame", "Increment 10: Step Time = 0.999", "wrong-reference-frame-time"),
])
def test_reference_identity_errors_precede_nonfinite(tmp_path, source_models, field, value, code):
def mutation(_, rows):
rows[0]["U-U1"] = "NaN"
rows[-1][field] = value
path = mutate_csv(tmp_path, "hexa_linearstatic", "U", mutation)
with pytest.raises(COMPARE.ComparisonError) as error:
COMPARE.read_reference(path, "hexa_linearstatic", "U", source_models["hexa_linearstatic"])
assert error.value.code == code
def test_rf_entire_inventory_precedes_constraint_projection(tmp_path, source_models):
def mutation(_, rows):
assert int(rows[0]["Node Label"]) == 1 # A free node, absent from projected RF family.
rows.pop(0)
path = mutate_csv(tmp_path, "hexa_linearstatic", "RF", mutation)
with pytest.raises(COMPARE.ComparisonError) as error:
COMPARE.read_reference(path, "hexa_linearstatic", "RF", source_models["hexa_linearstatic"])
assert error.value.code == "reference-inventory"
def test_duplicate_reference_identity_precedes_bad_numeric(tmp_path, source_models):
def mutation(_, rows):
rows[0]["RF-RF1"] = "NaN"
rows[-1]["Node Label"] = rows[0]["Node Label"]
path = mutate_csv(tmp_path, "hexa_linearstatic", "RF", mutation)
with pytest.raises(COMPARE.ComparisonError) as error:
COMPARE.read_reference(path, "hexa_linearstatic", "RF", source_models["hexa_linearstatic"])
assert error.value.code == "duplicate-reference-identity"
@pytest.mark.parametrize("point", ["0", "9", "bad"])
def test_invalid_stress_point_is_never_rematched(tmp_path, source_models, point):
path = mutate_csv(tmp_path, "hexa_linearstatic", "S", lambda _, rows: rows[0].update(IntPt=point))
with pytest.raises(COMPARE.ComparisonError):
COMPARE.read_reference(path, "hexa_linearstatic", "S", source_models["hexa_linearstatic"])
def test_stress_point_order_coordinates_are_auxiliary_not_identity(tmp_path, source_models):
path = mutate_csv(tmp_path, "hexa_linearstatic", "S", lambda _, rows: rows[2].update(
X=rows[3]["X"], Y=rows[3]["Y"], Z=rows[3]["Z"]))
rows = COMPARE.read_reference(path, "hexa_linearstatic", "S", source_models["hexa_linearstatic"])
with pytest.raises(COMPARE.ComparisonError) as error:
COMPARE.reference_values(rows, "S", source_models["hexa_linearstatic"])
assert error.value.code == "reference-coordinate-convention"
@pytest.mark.parametrize("header_mutation", ["missing", "extra"])
def test_wrong_component_columns_rejected(tmp_path, source_models, header_mutation):
def mutation(headers, rows):
if header_mutation == "missing":
headers.remove("U-U3")
for row in rows:
del row["U-U3"]
else:
headers.append("U-UR1")
for row in rows:
row["U-UR1"] = "0"
path = mutate_csv(tmp_path, "hexa_linearstatic", "U", mutation)
with pytest.raises(COMPARE.ComparisonError) as error:
COMPARE.read_reference(path, "hexa_linearstatic", "U", source_models["hexa_linearstatic"])
assert error.value.code == "wrong-reference-columns"
def test_dynamic_decimal_time_requires_exact_integer_frame(tmp_path, source_models):
path = mutate_csv(tmp_path, "hexa_dynlinear", "U", lambda _, rows: rows[442].update(
Frame="Increment 1: Step Time = 0.1000000001"))
with pytest.raises(COMPARE.ComparisonError) as error:
COMPARE.read_reference(path, "hexa_dynlinear", "U", source_models["hexa_dynlinear"])
assert error.value.code == "wrong-reference-frame-time"
def test_nonfinite_free_reference_rf_is_not_discarded(tmp_path, source_models):
path = mutate_csv(tmp_path, "hexa_linearstatic", "RF", lambda _, rows: rows[0].update({"RF-RF1": "NaN"}))
rows = COMPARE.read_reference(path, "hexa_linearstatic", "RF", source_models["hexa_linearstatic"])
with pytest.raises(COMPARE.ComparisonError) as error:
COMPARE.reference_values(rows, "RF", source_models["hexa_linearstatic"])
assert error.value.code == "nonfinite-reference-value"
def test_inclusive_row_and_rms_boundaries_are_independent():
report = family([21.] + [20.] * 99, [20.] * 100)
assert report["passed"] and report["rows"][0]["relative_error"] == .05
assert not family([21.00001] + [20.] * 99, [20.] * 100)["rows_passed"]
assert family([101.], [100.])["passed"]
assert not family([101.00001], [100.])["rms_passed"]
assert family([100., 2.], [100., 1.])["rows"][1]["branch"] == "near-zero"
@pytest.mark.parametrize("actual,reference", [([1e308], [-1e308]), ([float('inf')], [1.])])
def test_overflow_cannot_create_passing_metric(actual, reference):
with pytest.raises(COMPARE.ComparisonError):
family(actual, reference)
def test_wrong_result_node_beats_nonfinite_displacement(tmp_path, result_fixtures):
path = copy_result(tmp_path, result_fixtures)
with h5py.File(path, "r+") as file:
nodes = file["model/nodes"][:]
nodes[-1]["source_label"] = b"1"
file["model/nodes"][:] = nodes
file["steps/Step-1/frames/0/nodal/displacement"][0, 0] = np.nan
report = compare_fixture(path)
assert report["precheck"]["code"] == "result-node-inventory"
assert report["families"] == []
@pytest.mark.parametrize("attribute,value,code", [
("source_input_identity", "path=wrong;content_identity=fnv1a64:0000000000000000", "wrong-result-attribute"),
("analysis_procedure", "hht-linear-dynamics", "wrong-result-attribute"),
("mass_formulation", "consistent-reference", "wrong-result-attribute"),
("component_order", "UX,UY,UZ,URX,URY,URZ", "wrong-result-attribute"),
])
def test_metadata_identity_rejects_other_model(tmp_path, result_fixtures, attribute, value, code):
path = copy_result(tmp_path, result_fixtures)
with h5py.File(path, "r+") as file:
file["metadata"].attrs.modify(attribute, value)
assert compare_fixture(path)["precheck"]["code"] == code
@pytest.mark.parametrize("attribute,value", [("component_names", "U1,U3,U2"), ("unit_dimensions", "force"),
("location", "integration-point"), ("coordinate_system", "local")])
def test_result_quantity_semantics_must_match(tmp_path, result_fixtures, attribute, value):
path = copy_result(tmp_path, result_fixtures)
with h5py.File(path, "r+") as file:
file["steps/Step-1/frames/0/nodal/displacement"].attrs.modify(attribute, value)
assert compare_fixture(path)["precheck"]["code"] == "wrong-result-attribute"
@pytest.mark.parametrize("shape,dtype,code", [((442, 6), "<f8", "wrong-result-shape"),
((442, 3), "<f4", "wrong-result-value-type"),
((442, 3), ">f8", "wrong-result-value-type")])
def test_wrong_shape_or_dtype_is_not_sliced_into_supported_components(tmp_path, result_fixtures, shape, dtype, code):
path = copy_result(tmp_path, result_fixtures)
with h5py.File(path, "r+") as file:
key = "steps/Step-1/frames/0/nodal/displacement"
del file[key]
file.create_dataset(key, shape=shape, dtype=dtype)
assert compare_fixture(path)["precheck"]["code"] == code
def test_signed_ids_and_fixed_ascii_are_not_official_compound_schema(tmp_path, result_fixtures):
path = copy_result(tmp_path, result_fixtures)
with h5py.File(path, "r+") as file:
old = file["model/nodes"][:]
del file["model/nodes"]
compound(file["model"], "nodes", [("internal_node_id", "<i8"), ("instance_name", "S16"),
("source_label", "S16"), ("coordinates", "<f8", (3,))], list(old))
assert compare_fixture(path)["precheck"]["code"] == "wrong-result-table-schema"
@pytest.mark.parametrize("target", ["missing", "extra", "noncanonical"])
def test_dynamic_frame_inventory_is_exact(tmp_path, result_fixtures, target):
path = copy_result(tmp_path, result_fixtures, "hexa_dynlinear")
with h5py.File(path, "r+") as file:
root = file["steps/Step-1/frames"]
if target == "missing":
del root["100"]
elif target == "extra":
root.create_group("101")
else:
root.move("1", "01")
assert compare_fixture(path, "hexa_dynlinear")["precheck"]["code"] == "result-frame-inventory"
def test_dynamic_binary_schedule_not_nearest_time(tmp_path, result_fixtures):
path = copy_result(tmp_path, result_fixtures, "hexa_dynlinear")
with h5py.File(path, "r+") as file:
rows = file["steps/Step-1/accepted_frames"][:]
rows[3]["time"] = .3 # Required binary64 3*.1 differs from decimal-literal .3.
file["steps/Step-1/accepted_frames"][:] = rows
assert compare_fixture(path, "hexa_dynlinear")["precheck"]["code"] == "wrong-result-frame-schedule"
def test_result_point_id_mismatch_rejected_before_nan_stress(tmp_path, result_fixtures):
path = copy_result(tmp_path, result_fixtures)
with h5py.File(path, "r+") as file:
rows = file["model/solid_integration_points"][:]
rows[2]["point_id"] = 4
file["model/solid_integration_points"][:] = rows
file["steps/Step-1/frames/0/solid/stress"][0, 0, 0] = np.nan
assert compare_fixture(path)["precheck"]["code"] == "wrong-result-point-identity"
def test_result_free_reaction_is_not_free_residual(tmp_path, result_fixtures):
path = copy_result(tmp_path, result_fixtures)
with h5py.File(path, "r+") as file:
file["steps/Step-1/frames/0/nodal/residual"][0, 0] = 1234.
assert compare_fixture(path)["verdict"] == "passed"
with h5py.File(path, "r+") as file:
file["steps/Step-1/frames/0/nodal/reaction"][0, 0] = 1234.
assert compare_fixture(path)["precheck"]["code"] == "nonzero-free-result-reaction"
def test_nonfinite_uncompared_quantity_is_schema_failure(tmp_path, result_fixtures):
path = copy_result(tmp_path, result_fixtures)
with h5py.File(path, "r+") as file:
file["steps/Step-1/frames/0/global/strain_energy"][()] = np.inf
assert compare_fixture(path)["precheck"]["code"] == "nonfinite-result-value"
def test_source_copy_must_be_byte_identical(tmp_path, result_fixtures):
source = tmp_path / "hexa_linearstatic.inp"
source.write_bytes((REFERENCE / "hexa_linearstatic.inp").read_bytes().replace(b"\r\n", b"\n"))
report = COMPARE.compare("hexa_linearstatic", source, result_fixtures["hexa_linearstatic"])
assert report["precheck"]["code"] == "source-input-hash-mismatch"
def test_report_is_deterministic_and_keeps_family_boundaries(result_fixtures):
first = compare_fixture(result_fixtures["hexa_linearstatic"])
second = compare_fixture(result_fixtures["hexa_linearstatic"])
assert json.dumps(first, sort_keys=True, allow_nan=False) == json.dumps(second, sort_keys=True, allow_nan=False)
assert [row["quantity"] for row in first["families"]] == ["U", "RF", "S"]
assert len({row["reference_sha256"] for row in first["families"]}) == 3
def test_modal_original_absence_is_not_pass(tmp_path, result_fixtures):
report = COMPARE.compare("hexa_freq", REFERENCE / "hexa_freq.inp",
result_fixtures["hexa_freq"], tmp_path)
assert report["precheck"]["passed"], report["precheck"]
assert report["verdict"] == "not-run" and not report["blocking_passed"]
assert report["families"] == []
def test_supplied_modal_values_are_compared_not_reported_missing(result_fixtures):
report = compare_fixture(result_fixtures["hexa_freq"], "hexa_freq")
assert report["precheck"]["passed"], report["precheck"]
# The artificial eigenvalues 1..10 must fail against genuine source Values.
assert report["verdict"] == "failed"
assert len(report["blocking_failures"]) == 10
assert report["precheck"]["reference_row_counts"]["Value"] == 10
assert [f["mode"] for f in report["families"]] == list(range(1, 11))
assert all(f["quantity"] == "eigenvalue" for f in report["families"])
def test_modal_eigenvalue_matches_pass_with_ten_independent_families(tmp_path, result_fixtures):
path = copy_result(tmp_path, result_fixtures, "hexa_freq")
reference = COMPARE.read_modal_reference(REFERENCE / "hexa_freq.txt")
with h5py.File(path, "r+") as file:
eigen = file["steps/Step-1/eigen"]
for name, field in (("modes", "eigenvalue"), ("verification/pairs", "eigenvalue"),
("frequencies", "frequency")):
rows = eigen[name][:]
rows[field] = [math.sqrt(row["value"]) / (2 * math.pi) if field == "frequency"
else row["value"] for row in reference]
eigen[name][:] = rows
first = compare_fixture(path, "hexa_freq")
assert first["verdict"] == "passed", first
assert first["blocking_passed"] and first["blocking_failures"] == []
assert len({f["identity"] for f in first["families"]}) == 10
assert all(f["relative_rms"] == 0 and f["row_count"] == 1 for f in first["families"])
assert first == compare_fixture(path, "hexa_freq")
def test_modal_text_reads_value_not_cycles_per_time(tmp_path):
original = REFERENCE / "hexa_freq.txt"
rows = COMPARE.read_modal_reference(original)
assert rows[0] == {"step": "Step-1", "mode": 1, "quantity": "eigenvalue",
"unit_dimensions": "1/time^2", "value": 496.28}
assert rows[-1]["value"] == 449260.0
reversed_path = tmp_path / "reversed.txt"
reversed_path.write_text("\n".join(reversed(original.read_text().splitlines())), encoding="utf-8")
assert COMPARE.read_modal_reference(reversed_path) == rows
@pytest.mark.parametrize("mutation,code", [
("missing", "modal-mode-inventory"), ("duplicate", "duplicate-modal-mode"),
("extra", "wrong-modal-mode"), ("unit", "wrong-modal-reference-format"),
("quantity", "wrong-modal-reference-format"), ("nan", "nonfinite-modal-reference"),
("negative", "invalid-modal-reference-value"), ("order", "wrong-modal-order"),
])
def test_modal_text_rejects_invalid_source(tmp_path, mutation, code):
lines = [line for line in (REFERENCE / "hexa_freq.txt").read_text().splitlines() if line.strip()]
if mutation == "missing":
lines.pop()
elif mutation == "duplicate":
lines.append(lines[0])
elif mutation == "extra":
lines.append("Mode 11: Value = 500000 Freq = 112.54 (cycles/time)")
elif mutation == "unit":
lines[0] = lines[0].replace("cycles/time", "Hz")
elif mutation == "quantity":
lines[0] = lines[0].replace("Value", "Frequency")
elif mutation == "nan":
lines[0] = lines[0].replace("496.28", "NaN")
elif mutation == "negative":
lines[0] = lines[0].replace("3.5456", "-3.5456")
else:
lines[0] = lines[0].replace("496.28", "900000")
path = tmp_path / "invalid.txt"
path.write_text("\n".join(lines), encoding="utf-8")
with pytest.raises(COMPARE.ComparisonError) as error:
COMPARE.read_modal_reference(path)
assert error.value.code == code
def test_modified_modal_reference_is_precheck_failure(tmp_path, result_fixtures):
(tmp_path / "hexa_freq.txt").write_bytes((REFERENCE / "hexa_freq.txt").read_bytes() + b"\n")
report = COMPARE.compare("hexa_freq", REFERENCE / "hexa_freq.inp",
result_fixtures["hexa_freq"], tmp_path)
assert report["verdict"] == "failed"
assert report["precheck"]["code"] == "reference-hash-mismatch"
@pytest.mark.parametrize("mutation,code", [("duplicate", "duplicate-modal-mode"), ("missing", "modal-mode-inventory"),
("extra", "wrong-modal-mode"), ("unit", "wrong-modal-unit"),
("step", "wrong-modal-step")])
def test_modal_identity_before_nonfinite(mutation, code):
rows = [{"step": "Step-1", "mode": i, "quantity": "frequency", "unit_dimensions": "1/time",
"value": float(i)} for i in range(1, 11)]
actual = [dict(row) for row in rows]
actual[0]["value"] = np.nan
if mutation == "duplicate":
actual[-1]["mode"] = 1
elif mutation == "missing":
actual.pop()
elif mutation == "extra":
actual.append(dict(rows[-1], mode=11))
elif mutation == "unit":
actual[-1]["unit_dimensions"] = "Hz_assumed_SI"
else:
actual[-1]["step"] = "Step-2"
with pytest.raises(COMPARE.ComparisonError) as error:
COMPARE.compare_modal(rows, actual, "frequency", "1/time")
assert error.value.code == code
def test_closed_execution_record_hashes_are_checked(tmp_path, result_fixtures):
path = result_fixtures["hexa_linearstatic"]
record = tmp_path / "native.json"
data = {"case": "hexa_linearstatic", "native_exit": 0, "source_unchanged": True,
"scratch_unchanged": True, "result_exists": True, "command": ["fesa.exe", str(REFERENCE / "hexa_linearstatic.inp")],
"original_sha256": COMPARE.sha256(REFERENCE / "hexa_linearstatic.inp"),
"scratch_sha256": COMPARE.sha256(REFERENCE / "hexa_linearstatic.inp"),
"result_sha256": COMPARE.sha256(path)}
record.write_text(json.dumps(data), encoding="utf-8")
assert compare_fixture(path, execution=record)["execution"]["verified"]
data["native_exit"] = 1
record.write_text(json.dumps(data), encoding="utf-8")
assert compare_fixture(path, execution=record)["precheck"]["code"] == "invalid-execution-evidence"
data["native_exit"], data["result_sha256"] = 0, "0" * 64
record.write_text(json.dumps(data), encoding="utf-8")
assert compare_fixture(path, execution=record)["precheck"]["code"] == "invalid-execution-evidence"
def test_cli_protects_reference_and_has_deterministic_exit(tmp_path, result_fixtures):
original_hash = COMPARE.sha256(REFERENCE / "hexa_linearstatic_U.csv")
command = [sys.executable, str(ROOT / "scripts/verification/compare_hex8.py"),
"--case", "hexa_linearstatic", "--input", str(REFERENCE / "hexa_linearstatic.inp"),
"--results", str(result_fixtures["hexa_linearstatic"])]
denied = subprocess.run(command + ["--report", str(REFERENCE / "hexa_linearstatic_U.csv")], capture_output=True, text=True)
assert denied.returncode == 2
assert COMPARE.sha256(REFERENCE / "hexa_linearstatic_U.csv") == original_hash
report_path = tmp_path / "report.json"
completed = subprocess.run(command + ["--report", str(report_path)], capture_output=True, text=True)
assert completed.returncode == 0, completed.stderr
assert json.loads(report_path.read_text())["verdict"] == "passed"
def test_result_projection_joins_source_identities_not_storage_order(tmp_path, result_fixtures):
path = copy_result(tmp_path, result_fixtures)
with h5py.File(path, "r+") as file:
nodes = file["model/nodes"][:][::-1]
nodes["internal_node_id"] = np.arange(442, dtype="<u8")
file["model/nodes"][:] = nodes
for name in ("displacement", "reaction", "residual"):
obj = file[f"steps/Step-1/frames/0/nodal/{name}"]
obj[:] = obj[:][::-1]
mask = file["model/nodal_constraint_mask"]
mask[:] = mask[:][::-1]
elements = file["model/elements"][:][::-1]
elements["internal_element_id"] = np.arange(300, dtype="<u8")
elements["node_internal_ids"] = 441 - elements["node_internal_ids"]
file["model/elements"][:] = elements
points = file["model/solid_integration_points"][:].reshape(300, 8)[::-1].reshape(2400)
points["internal_element_id"] = np.repeat(np.arange(300, dtype="<u8"), 8)
file["model/solid_integration_points"][:] = points
for name in ("stress", "strain"):
obj = file[f"steps/Step-1/frames/0/solid/{name}"]
obj[:] = obj[:][::-1]
report = compare_fixture(path)
assert report["verdict"] == "passed", report["precheck"]
@pytest.mark.parametrize("mutation,code", [
("point-natural", "result-coordinate-convention"), ("point-reference", "result-coordinate-convention"),
("connectivity", "wrong-result-element-connectivity"), ("mask", "wrong-result-constraint-identity"),
("node-coordinate", "wrong-result-coordinate"), ("material", "wrong-result-material-value"),
("section", "wrong-result-section-identity"),
])
def test_model_geometry_and_properties_match_frozen_input(tmp_path, result_fixtures, mutation, code):
path = copy_result(tmp_path, result_fixtures)
with h5py.File(path, "r+") as file:
if mutation.startswith("point-"):
obj = file["model/solid_integration_points"]
rows = obj[:]
field = "natural_coordinates" if mutation == "point-natural" else "reference_coordinates"
rows[0][field][0] += .001
obj[:] = rows
elif mutation == "connectivity":
obj = file["model/elements"]
rows = obj[:]
rows[0]["node_internal_ids"][[0, 1]] = rows[0]["node_internal_ids"][[1, 0]]
obj[:] = rows
elif mutation == "mask":
file["model/nodal_constraint_mask"][0, 0] = 1
elif mutation == "node-coordinate":
obj = file["model/nodes"]
rows = obj[:]
rows[0]["coordinates"][0] = .001
obj[:] = rows
elif mutation == "material":
obj = file["model/solid/materials"]
rows = obj[:]
rows[0]["youngs_modulus"] *= 2
obj[:] = rows
else:
obj = file["model/solid/sections"]
rows = obj[:]
rows[0]["source_line"] += 1
obj[:] = rows
assert compare_fixture(path)["precheck"]["code"] == code
@pytest.mark.parametrize("mutation,code", [
("metadata-width", "wrong-result-attribute-type"), ("attribute-ascii", "wrong-result-attribute-type"),
("missing-dataset", "wrong-result-quantity-inventory"), ("extra-component-dataset", "wrong-result-quantity-inventory"),
("compound-member-order", "wrong-result-table-schema"),
])
def test_required_types_and_complete_dataset_inventory(tmp_path, result_fixtures, mutation, code):
path = copy_result(tmp_path, result_fixtures)
with h5py.File(path, "r+") as file:
if mutation == "metadata-width":
obj = file["metadata"]
del obj.attrs["schema_version"]
obj.attrs.create("schema_version", 1, dtype="<u4")
elif mutation == "attribute-ascii":
obj = file["steps/Step-1/frames/0/nodal/displacement"]
del obj.attrs["unit_dimensions"]
obj.attrs.create("unit_dimensions", b"length", dtype="S6")
elif mutation == "missing-dataset":
del file["steps/Step-1/frames/0/nodal/reaction"]
elif mutation == "extra-component-dataset":
file.create_dataset("steps/Step-1/frames/0/nodal/rotation", data=np.zeros((442, 3)))
else:
obj = file["model/solid_integration_points"]
rows = obj[:]
del file[obj.name]
compound(file["model"], "solid_integration_points", [
("point_id", "u1"), ("internal_element_id", "<u8"),
("natural_coordinates", "<f8", (3,)), ("reference_coordinates", "<f8", (3,))],
[(row["point_id"], row["internal_element_id"], row["natural_coordinates"], row["reference_coordinates"]) for row in rows])
assert compare_fixture(path)["precheck"]["code"] == code
def test_reference_raw_hash_rejects_modified_original_copy(tmp_path, result_fixtures):
for quantity in ("U", "RF", "S"):
shutil.copyfile(REFERENCE / f"hexa_linearstatic_{quantity}.csv", tmp_path / f"hexa_linearstatic_{quantity}.csv")
path = tmp_path / "hexa_linearstatic_U.csv"
path.write_bytes(path.read_bytes() + b"\r\n")
report = COMPARE.compare("hexa_linearstatic", REFERENCE / "hexa_linearstatic.inp", result_fixtures["hexa_linearstatic"], tmp_path)
assert report["precheck"]["code"] == "reference-hash-mismatch"
def test_case_families_cannot_share_scale_or_cancel_sign_errors():
assert family([100.], [100.])["passed"]
assert not family([.001], [0.])["passed"]
report = family([21., 19.], [20., 20.])
assert report["rows_passed"]
assert report["absolute_rms"] == 1.
assert report["relative_rms"] == .05
assert not report["rms_passed"]
def test_alternate_reference_directory_cannot_make_original_report_writable(tmp_path, monkeypatch):
original = REFERENCE / "hexa_linearstatic_U.csv"
original_sha = COMPARE.sha256(original)
monkeypatch.setattr(sys, "argv", ["compare_hex8.py", "--case", "hexa_linearstatic",
"--input", str(REFERENCE / "hexa_linearstatic.inp"), "--results", str(tmp_path / "fixture.h5"),
"--references", str(tmp_path / "copied-references"), "--report", str(original)])
monkeypatch.setattr(COMPARE, "compare", lambda *args: {"precheck": {"passed": True},
"case": "hexa_linearstatic", "verdict": "passed", "blocking_failures": [], "blocking_passed": True})
# This tripwire proves the old branch without ever allowing an original write.
write_text = Path.write_text
def protected_write(path, *args, **kwargs):
if path.resolve() == original.resolve():
pytest.fail("Comparator attempted to write an authoritative original reference")
return write_text(path, *args, **kwargs)
monkeypatch.setattr(Path, "write_text", protected_write)
with pytest.raises(SystemExit) as error:
COMPARE.main()
assert error.value.code == 2
assert COMPARE.sha256(original) == original_sha
@pytest.mark.parametrize("shape", [(), (10, 1)])
def test_modal_pair_rank_errors_produce_failed_precheck(tmp_path, result_fixtures, shape):
path = copy_result(tmp_path, result_fixtures, "hexa_freq")
with h5py.File(path, "r+") as file:
key = "steps/Step-1/eigen/verification/pairs"
del file[key]
file.create_dataset(key, shape=shape, dtype="<f8")
report = compare_fixture(path, "hexa_freq")
assert not report["precheck"]["passed"]
assert report["precheck"]["code"] == "wrong-modal-pair-shape"
assert report["families"] == []
def test_hardlink_report_alias_cannot_overwrite_known_source(tmp_path, monkeypatch):
protected = tmp_path / "originals"
protected.mkdir()
original = protected / "reference.csv"
original.write_bytes(b"immutable source fixture\n")
report_alias = tmp_path / "report-alias.json"
os.link(original, report_alias)
original_sha = COMPARE.sha256(original)
monkeypatch.setattr(COMPARE, "REFERENCE_DIRECTORY", protected)
monkeypatch.setattr(sys, "argv", ["compare_hex8.py", "--case", "hexa_linearstatic",
"--input", str(tmp_path / "input.inp"), "--results", str(tmp_path / "fixture.h5"),
"--references", str(tmp_path / "other-references"), "--report", str(report_alias)])
monkeypatch.setattr(COMPARE, "compare", lambda *args: {"precheck": {"passed": True},
"case": "hexa_linearstatic", "verdict": "passed", "blocking_failures": [], "blocking_passed": True})
# Neither the fixture nor its alias is written during the failing old path.
monkeypatch.setattr(Path, "write_text", lambda *args, **kwargs: pytest.fail("Unsafe alias write attempted"))
with pytest.raises(SystemExit) as error:
COMPARE.main()
assert error.value.code == 2
assert COMPARE.sha256(original) == original_sha
def test_lexically_equivalent_input_path_preserves_writer_identity(result_fixtures):
source = REFERENCE / "../hexa/hexa_linearstatic.inp"
report = COMPARE.compare("hexa_linearstatic", source, result_fixtures["hexa_linearstatic"])
assert report["verdict"] == "passed", report["precheck"]
-351
View File
@@ -1,351 +0,0 @@
"""HHT reference identity, time, projection and common-tolerance regressions."""
import csv
import json
import math
from pathlib import Path
import shutil
import subprocess
import sys
import pytest
h5py = pytest.importorskip("h5py", reason="Run with uv --with h5py for HHT comparison tests.")
np = pytest.importorskip("numpy")
from scripts.verification import compare_hht_dynamic as comparator
@pytest.fixture(scope="module")
def artifacts(tmp_path_factory):
directory = tmp_path_factory.mktemp("hht-comparison")
input_path = comparator.REFERENCE_DIRECTORY / f"{comparator.CASE}.inp"
reference_path = comparator.REFERENCE_DIRECTORY / f"{comparator.CASE} displacements.csv"
nodes, content_identity = comparator.input_inventory(input_path)
reference = comparator.read_reference(reference_path)
result_path = directory / "matching_results.h5"
string_type = h5py.string_dtype("utf-8")
node_type = np.dtype([("internal_node_id", "<u8"), ("instance_name", string_type),
("source_label", string_type), ("coordinates", "<f8", (3,))])
node_rows = np.empty(729, dtype=node_type)
labels = list(nodes)
for index, label in enumerate(labels):
node_rows[index] = (index, "PART-1-1", str(label), nodes[label])
frame_type = np.dtype([("frame_index", "<u8"), ("increment_index", "<u8"),
("time", "<f8"), ("iteration_count", "<u8"), ("is_initial", "u1")])
frames = np.empty(101, dtype=frame_type)
with h5py.File(result_path, "w") as file:
metadata = file.create_group("metadata")
metadata.attrs.update({
"feature_id": "hht-implicit-dynamics", "analysis_procedure": "hht-implicit-dynamics",
"mass_formulation": "nodal-lumped-kinetic", "internal_formulation": "FESA-MITC4-total-lagrangian",
"initial_state": "zero-q-v-step-load-0+",
"acceleration_convention": "physical-endpoint-and-auxiliary-separated",
"source_step_name": "Step-1", "step_name": "Step-1", "unit_system_label": "consistent-input-units",
"solver_version": "test-fixture", "source_input_identity": f"path={input_path};content_identity={content_identity}",
"schema_version": np.uint64(1), "increment_count": np.uint64(100), "maximum_increments": np.uint64(1000),
"alpha": -0.05, "beta": 0.275625, "gamma": 0.55, "initial_increment": 0.1, "time_period": 10.0,
})
file.create_dataset("model/nodes", data=node_rows)
for increment in range(101):
time = float(increment) * 0.1
frames[increment] = (increment, increment, time, 0 if increment == 0 else 2, int(increment == 0))
values = np.zeros((729, 6), dtype="<f8")
for label in (2, 21, 22, 69, 70, 165, 309, 442, 586):
values[labels.index(label)] = reference[(increment, label)]
dataset = file.create_dataset(f"steps/Step-1/frames/{increment}/nodal/displacement", data=values)
dataset.attrs.update({
"component_names": "UX,UY,UZ,URX,URY,URZ", "component_unit_dimensions": "length,length,length,1,1,1",
"coordinate_system": "global-cartesian", "location": "nodal", "step_name": "Step-1", "source_step_name": "Step-1",
"rotation_representation": "total-rotation-vector-chart",
"quantity_semantics": "global-translation-and-total-rotation-vector",
"schema_version": np.uint64(1), "frame_index": np.uint64(increment), "increment_index": np.uint64(increment),
"time": time, "is_initial": np.uint8(increment == 0),
})
file.create_dataset("steps/Step-1/accepted_frames", data=frames)
return input_path, reference_path, result_path
def changed_result(tmp_path, artifacts):
target = tmp_path / "changed.h5"
shutil.copyfile(artifacts[2], target)
return target
def reduced_result(tmp_path, artifacts):
path = changed_result(tmp_path, artifacts)
with h5py.File(path, "r+") as file:
file["metadata"].attrs.modify("schema_version", np.uint64(2))
file["metadata"].attrs["time_integration_formulation"] = (
"reduced-coordinate-HHT-with-physical-rate-reconstruction")
for frame in range(101):
root = f"steps/Step-1/frames/{frame}/nodal"
displacement = file[f"{root}/displacement"]
displacement.attrs.modify("schema_version", np.uint64(2))
auxiliary = file.create_dataset(f"{root}/algorithmic_acceleration",
data=np.zeros((729, 6), dtype="<f8"))
for name, value in displacement.attrs.items():
auxiliary.attrs[name] = value
auxiliary.attrs["component_names"] = "AUX1,AUX2,AUX3,AUX4,AUX5,AUX6"
auxiliary.attrs["coordinate_system"] = "fixed-initial-global-chart"
auxiliary.attrs["component_unit_dimensions"] = (
"length/time^2,length/time^2,length/time^2,1/time^2,1/time^2,1/time^2")
auxiliary.attrs["quantity_semantics"] = (
"full-chart-HHT-filter-combination-not-predictor-driver")
return path
def test_reduced_schema_preserves_displacement_comparison(tmp_path, artifacts):
result = comparator.compare(artifacts[0], artifacts[1], reduced_result(tmp_path, artifacts))
assert result["precheck"]["passed"], result["precheck"]
assert result["verdict"] == "passed"
@pytest.mark.parametrize("mutation", ["missing_A", "A_semantics", "A_version", "A_nan", "method"])
def test_reduced_schema_rejects_inconsistent_auxiliary_history(tmp_path, artifacts, mutation):
path = reduced_result(tmp_path, artifacts)
with h5py.File(path, "r+") as file:
auxiliary = file["steps/Step-1/frames/99/nodal/algorithmic_acceleration"]
if mutation == "missing_A":
del file[auxiliary.name]
elif mutation == "A_semantics":
auxiliary.attrs["quantity_semantics"] = "auxiliary-HHT-chart-acceleration"
elif mutation == "A_version":
auxiliary.attrs.modify("schema_version", np.uint64(1))
elif mutation == "A_nan":
auxiliary[0, 0] = float("nan")
else:
file["metadata"].attrs["time_integration_formulation"] = "full-coordinate-HHT"
result = comparator.compare(artifacts[0], artifacts[1], path)
assert not result["precheck"]["passed"]
assert result["families"] == []
def test_original_csv_exact_matching_history_and_deterministic_report(artifacts):
first = comparator.compare(*artifacts)
second = comparator.compare(*artifacts)
assert first == second
assert first["precheck"]["passed"], first["precheck"]
assert first["precheck"]["reference_row_count"] == 909
assert first["precheck"]["checked_result_scalars"] == 441774
assert first["verdict"] == "passed"
assert first["warnings"] == []
assert len(first["families"]) == 202
assert sum(family["row_count"] for family in first["families"]) == 5454
assert [family["quantity"] for family in first["families"][:4]] == ["U", "UR", "U", "UR"]
assert all(family["passed"] for family in first["families"])
assert first["families"][0]["reference_scale"] == 0.0
assert first["families"][6]["hdf5_time"] == 3 * 0.1
assert first["families"][6]["reference_decimal_time"] == "0.3"
assert 3 * 0.1 != float("0.3") # Both legitimate representations were validated independently.
json.dumps(first, allow_nan=False)
@pytest.mark.parametrize("mutation,code", [
("procedure", "wrong-hdf5-attribute-value"),
("mass", "wrong-hdf5-attribute-value"),
("coefficient", "wrong-hdf5-attribute-value"),
("initial_convention", "wrong-hdf5-attribute-value"),
("input_content", "wrong-input-content-identity"),
("component", "wrong-hdf5-attribute-value"),
("rotation", "wrong-hdf5-attribute-value"),
("missing_component_attribute", "missing-hdf5-attribute"),
("nonprojected_nan", "nonfinite-result-value"),
("rotation_inf", "nonfinite-result-value"),
("initial_motion", "nonzero-initial-displacement"),
("duplicate_node", "duplicate-result-node"),
("extra_node_label", "extra-result-node"),
("wrong_instance", "wrong-result-instance"),
("coordinate", "wrong-result-coordinate"),
("wrong_shape", "wrong-hdf5-dataset-layout"),
("float32", "wrong-displacement-type"),
("missing_frame", "result-frame-inventory"),
("extra_frame", "result-frame-inventory"),
("leading_zero_frame", "result-frame-inventory"),
("duplicate_increment", "wrong-result-frame-time"),
("decimal_binary_time", "wrong-result-frame-time"),
("frame_attribute_time", "wrong-hdf5-attribute-value"),
("trial_iteration", "wrong-result-frame-time"),
("initial_iteration", "wrong-result-frame-time"),
])
def test_hdf5_precheck_rejects_before_tolerance(tmp_path, artifacts, mutation, code):
path = changed_result(tmp_path, artifacts)
with h5py.File(path, "r+") as file:
metadata = file["metadata"]
displacement = file["steps/Step-1/frames/3/nodal/displacement"]
nodes = file["model/nodes"]
table = file["steps/Step-1/accepted_frames"]
if mutation in {"procedure", "mass", "coefficient", "initial_convention", "input_content"}:
key, value = {
"procedure": ("analysis_procedure", "linear-static"),
"mass": ("mass_formulation", "consistent-kinetic"),
"coefficient": ("alpha", 0.0),
"initial_convention": ("initial_state", "unloaded"),
"input_content": ("source_input_identity", "wrong-content"),
}[mutation]
metadata.attrs.modify(key, value)
elif mutation in {"component", "rotation", "frame_attribute_time"}:
key, value = {"component": ("component_names", "U1,U2,U3,UR1,UR2,UR3"),
"rotation": ("rotation_representation", "incremental-rotation"),
"frame_attribute_time": ("time", 0.3)}[mutation]
displacement.attrs.modify(key, value)
elif mutation == "missing_component_attribute":
del displacement.attrs["component_names"]
elif mutation in {"nonprojected_nan", "rotation_inf"}:
displacement[0, 0 if mutation == "nonprojected_nan" else 3] = float("nan") if mutation == "nonprojected_nan" else float("inf")
elif mutation == "initial_motion":
file["steps/Step-1/frames/0/nodal/displacement"][0, 0] = 1.0e-100
elif mutation in {"duplicate_node", "extra_node_label", "wrong_instance", "coordinate"}:
rows = nodes[:]
if mutation == "duplicate_node":
rows[1]["source_label"] = rows[0]["source_label"]
elif mutation == "extra_node_label":
rows[0]["source_label"] = "99999"
elif mutation == "wrong_instance":
rows[0]["instance_name"] = "Other"
else:
rows[0]["coordinates"] = [1.0, 2.0, 3.0]
nodes[:] = rows
elif mutation in {"wrong_shape", "float32"}:
name = displacement.name
values = displacement[:]
del file[name]
file.create_dataset(name, data=values[:-1] if mutation == "wrong_shape" else values.astype("<f4"))
elif mutation == "missing_frame":
del file["steps/Step-1/frames/100"]
elif mutation == "extra_frame":
file.create_group("steps/Step-1/frames/101")
elif mutation == "leading_zero_frame":
file.move("steps/Step-1/frames/1", "steps/Step-1/frames/01")
else:
rows = table[:]
if mutation == "duplicate_increment":
rows[3]["increment_index"] = 2
elif mutation == "decimal_binary_time":
rows[3]["time"] = 0.3
elif mutation == "trial_iteration":
rows[3]["iteration_count"] = 101
elif mutation == "initial_iteration":
rows[0]["iteration_count"] = 1
table[:] = rows
report = comparator.compare(artifacts[0], artifacts[1], path)
assert report["precheck"]["passed"] is False
assert report["precheck"]["code"] == code, report["precheck"]
assert report["families"] == []
assert report["verdict"] == "failed"
@pytest.mark.parametrize("mutation,code", [
("duplicate_header", "duplicate-reference-column"),
("missing_header", "missing-reference-column"),
("duplicate_row", "duplicate-reference-row"),
("missing_row", "reference-row-inventory"),
("extra_node", "reference-row-inventory"),
("extra_frame", "reference-row-inventory"),
("time", "wrong-reference-time"),
("instance", "wrong-reference-instance"),
("label", "invalid-node-label"),
("nonfinite_rotation", "nonfinite-reference-value"),
])
def test_reference_precheck_rejects_before_tolerance(tmp_path, artifacts, mutation, code):
with artifacts[1].open(newline="", encoding="utf-8") as stream:
rows = list(csv.reader(stream))
if mutation == "duplicate_header":
rows[0][-1] = rows[0][-2]
elif mutation == "missing_header":
rows[0][-1] = "Not a required component"
elif mutation == "duplicate_row":
rows.append(rows[1])
elif mutation == "missing_row":
rows.pop()
elif mutation in {"extra_node", "extra_frame"}:
row = rows[1].copy()
if mutation == "extra_node":
row[2] = "1"
else:
row[0] = "Increment 101: Step Time = 10.1"
rows.append(row)
elif mutation == "time":
rows[1][0] = "Increment 0: Step Time = 0.0000000001"
elif mutation == "instance":
rows[1][1] = "WRONG-INSTANCE"
elif mutation == "label":
rows[1][2] = "2.0"
else:
rows[1][-1] = "nan"
reference = tmp_path / "changed.csv"
with reference.open("w", newline="", encoding="utf-8") as stream:
csv.writer(stream).writerows(rows)
report = comparator.compare(artifacts[0], reference, artifacts[2])
assert not report["precheck"]["passed"]
assert report["precheck"]["code"] == code, report["precheck"]
assert report["families"] == []
def test_projection_joins_source_identity_and_csv_order_is_irrelevant(tmp_path, artifacts):
path = changed_result(tmp_path, artifacts)
with h5py.File(path, "r+") as file:
nodes = file["model/nodes"][:][::-1]
nodes["internal_node_id"] = np.arange(729, dtype="<u8")
file["model/nodes"][:] = nodes
for frame in range(101):
values = file[f"steps/Step-1/frames/{frame}/nodal/displacement"]
values[:] = values[:][::-1]
with artifacts[1].open(newline="", encoding="utf-8") as stream:
rows = list(csv.reader(stream))
reference = tmp_path / "reordered.csv"
with reference.open("w", newline="", encoding="utf-8") as stream:
csv.writer(stream).writerows([rows[0], *reversed(rows[1:])])
report = comparator.compare(artifacts[0], reference, path)
assert report["precheck"]["passed"], report["precheck"]
assert report["verdict"] == "passed"
@pytest.mark.parametrize("component,passed", [(0, False), (3, True)])
def test_translation_blocks_rotation_warns(tmp_path, artifacts, component, passed):
path = changed_result(tmp_path, artifacts)
with h5py.File(path, "r+") as file:
values = file["steps/Step-1/frames/1/nodal/displacement"]
rows = values[:]
rows[:, component:component + 3] *= 2.0
values[:] = rows
report = comparator.compare(artifacts[0], artifacts[1], path)
assert report["precheck"]["passed"]
assert report["blocking_passed"] is passed
assert len(report["warnings"]) == (2 if passed else 0)
if passed:
assert [warning["code"] for warning in report["warnings"]] == ["row-tolerance-failure", "relative-rms-failure"]
else:
assert report["blocking_failures"] == ["U-global:Step-1:increment-1"]
def test_common_policy_boundaries_and_zero_scale():
near = comparator.evaluate_family(1, "U", [100.0, 2.0, 1.0], [100.0, 1.0, 0.0])
assert near["passed"]
assert [row["branch"] for row in near["rows"]] == ["relative", "near-zero", "near-zero"]
relative = comparator.evaluate_family(1, "U", [105.0], [100.0])
assert relative["rows_passed"] and not relative["rms_passed"]
boundary = comparator.evaluate_family(1, "U", [101.0], [100.0])
assert boundary["passed"]
assert boundary["relative_rms"] == 0.01
outside = comparator.evaluate_family(1, "U", [math.nextafter(101.0, math.inf)], [100.0])
assert outside["rows_passed"] and not outside["rms_passed"]
zero = comparator.evaluate_family(0, "U", [0.0], [0.0])
assert zero["passed"] and zero["relative_rms"] is None
nonzero = comparator.evaluate_family(0, "U", [1.0e-300], [0.0])
assert not nonzero["passed"]
assert nonzero["diagnostic_code"] == "zero-reference-scale-nonzero-error"
json.dumps(nonzero, allow_nan=False)
with pytest.raises(comparator.ComparisonError, match="U"):
comparator.evaluate_family(1, "U", [1.0e308], [-1.0e308])
def test_cli_writes_json_and_exit_two_for_missing_results(tmp_path, artifacts):
report = tmp_path / "evidence.json"
completed = subprocess.run([sys.executable, "-B", str(Path(comparator.__file__)),
"--input", str(artifacts[0]), "--reference", str(artifacts[1]),
"--results", str(tmp_path / "absent.h5"), "--report", str(report)],
check=False, capture_output=True, text=True)
assert completed.returncode == 2, completed.stderr
evidence = json.loads(report.read_text())
assert not evidence["precheck"]["passed"]
assert evidence["families"] == []
-693
View File
@@ -1,693 +0,0 @@
import json
import os
import subprocess
from types import SimpleNamespace
import pytest
from scripts import execute
@pytest.fixture(autouse=True)
def clear_sandbox_override(monkeypatch):
monkeypatch.delenv("FEM_HARNESS_CODEX_SANDBOX", raising=False)
def write_top_index(phase_dir):
(phase_dir.parent / "index.json").write_text(
json.dumps({"phases": [{"dir": phase_dir.name, "status": "pending"}]}),
encoding="utf-8",
)
def write_implementation_agent_profile(
root, *, model="gpt-5.6-terra", reasoning="xhigh"
):
profile = root / ".codex" / "agents" / "implementation-agent.toml"
profile.parent.mkdir(parents=True)
profile.write_text(
f'model = "{model}"\nmodel_reasoning_effort = "{reasoning}"\n'
'developer_instructions = """Implementation Agent test role."""\n',
encoding="utf-8",
)
def test_invoke_codex_uses_utf8_for_unicode_prompt(tmp_path, monkeypatch):
phase_dir = tmp_path / "phases" / "unicode-phase"
phase_dir.mkdir(parents=True)
write_top_index(phase_dir)
(phase_dir / "index.json").write_text(
json.dumps(
{
"project": "FESA Structural Solver",
"phase": "unicode-phase",
"steps": [{"step": 0, "name": "unicode", "status": "pending"}],
}
),
encoding="utf-8",
)
(phase_dir / "step0.md").write_text("Euler–Bernoulli 보", encoding="utf-8")
write_implementation_agent_profile(tmp_path)
invocation = {}
def capture_run(command, **kwargs):
invocation["command"] = command
invocation["kwargs"] = kwargs
return subprocess.CompletedProcess(command, 0, stdout="", stderr="")
monkeypatch.setattr(execute, "ROOT", tmp_path)
monkeypatch.setattr(execute.subprocess, "run", capture_run)
executor = execute.StepExecutor("unicode-phase")
executor._invoke_codex({"step": 0, "name": "unicode"}, "승인된 설계 – ")
assert "Euler–Bernoulli 보" in invocation["kwargs"]["input"]
assert invocation["kwargs"]["encoding"] == "utf-8"
sandbox_index = invocation["command"].index("--sandbox")
assert invocation["command"][sandbox_index + 1] == "workspace-write"
def test_invoke_codex_uses_implementation_agent_command_contract(
tmp_path, monkeypatch
):
phase_dir = tmp_path / "phases" / "agent-phase"
phase_dir.mkdir(parents=True)
write_top_index(phase_dir)
(phase_dir / "index.json").write_text(
json.dumps(
{
"project": "FESA Structural Solver",
"phase": "agent-phase",
"steps": [{"step": 0, "name": "agent", "status": "pending"}],
}
),
encoding="utf-8",
)
(phase_dir / "step0.md").write_text("agent routing", encoding="utf-8")
write_implementation_agent_profile(tmp_path)
invocation = {}
def capture_run(command, **kwargs):
invocation["command"] = command
return subprocess.CompletedProcess(command, 0, stdout="", stderr="")
monkeypatch.setattr(execute, "ROOT", tmp_path)
monkeypatch.setattr(execute.subprocess, "run", capture_run)
executor = execute.StepExecutor("agent-phase")
executor._invoke_codex({"step": 0, "name": "agent"}, "")
model_index = invocation["command"].index("--model")
config_index = invocation["command"].index("--config")
assert invocation["command"][model_index + 1] == "gpt-5.6-terra"
assert invocation["command"][config_index + 1] == 'model_reasoning_effort="xhigh"'
adjacent_arguments = list(zip(invocation["command"], invocation["command"][1:]))
assert ("--disable", "unified_exec") in adjacent_arguments
def test_invoke_codex_accepts_explicit_danger_full_access_override(
tmp_path, monkeypatch
):
phase_dir = tmp_path / "phases" / "sandbox-phase"
phase_dir.mkdir(parents=True)
write_top_index(phase_dir)
(phase_dir / "index.json").write_text(
json.dumps(
{
"project": "FESA Structural Solver",
"phase": "sandbox-phase",
"steps": [{"step": 0, "name": "sandbox", "status": "pending"}],
}
),
encoding="utf-8",
)
(phase_dir / "step0.md").write_text("sandbox override", encoding="utf-8")
write_implementation_agent_profile(tmp_path)
invocation = {}
def capture_run(command, **kwargs):
if command[0] == "git":
responses = {
("rev-parse", "--absolute-git-dir"): str(tmp_path / ".git/worktrees/sandbox"),
("rev-parse", "--git-common-dir"): str(tmp_path / ".git"),
("status", "--porcelain=v1", "-z", "--untracked-files=all"): "",
}
return subprocess.CompletedProcess(
command, 0, stdout=responses[tuple(command[1:])], stderr=""
)
invocation["command"] = command
return subprocess.CompletedProcess(command, 0, stdout="", stderr="")
monkeypatch.setattr(execute, "ROOT", tmp_path)
monkeypatch.setattr(execute.subprocess, "run", capture_run)
monkeypatch.setenv("FEM_HARNESS_CODEX_SANDBOX", "danger-full-access")
executor = execute.StepExecutor("sandbox-phase", danger_full_access_approved=True)
executor._invoke_codex({"step": 0, "name": "sandbox"}, "")
sandbox_index = invocation["command"].index("--sandbox")
assert invocation["command"][sandbox_index + 1] == "danger-full-access"
def test_invoke_codex_timeout_writes_structured_failure(tmp_path, monkeypatch):
phase_dir = tmp_path / "phases" / "timeout-phase"
phase_dir.mkdir(parents=True)
write_top_index(phase_dir)
(phase_dir / "index.json").write_text(
json.dumps(
{
"project": "FESA Structural Solver",
"phase": "timeout-phase",
"steps": [{"step": 0, "name": "timeout", "status": "pending"}],
}
),
encoding="utf-8",
)
(phase_dir / "step0.md").write_text("timeout", encoding="utf-8")
write_implementation_agent_profile(tmp_path)
def timeout_run(command, **kwargs):
raise subprocess.TimeoutExpired(
command, kwargs["timeout"], output="partial stdout", stderr="partial stderr"
)
monkeypatch.setattr(execute, "ROOT", tmp_path)
monkeypatch.setattr(execute.subprocess, "run", timeout_run)
executor = execute.StepExecutor("timeout-phase")
output = executor._invoke_codex({"step": 0, "name": "timeout"}, "")
assert output["exitCode"] == 124
assert output["transportFailure"] == "timeout"
assert output["stdout"] == "partial stdout"
assert "Codex timed out after 14400 seconds." in output["stderr"]
stored = json.loads((phase_dir / "step0-output.json").read_text(encoding="utf-8"))
assert stored == output
def test_execute_single_step_retries_structured_codex_timeout(tmp_path, monkeypatch):
phase_dir = tmp_path / "phases" / "retry-phase"
phase_dir.mkdir(parents=True)
write_top_index(phase_dir)
(phase_dir / "index.json").write_text(
json.dumps(
{
"project": "FESA Structural Solver",
"phase": "retry-phase",
"steps": [{"step": 0, "name": "retry", "status": "pending", "owned_paths": []}],
}
),
encoding="utf-8",
)
(phase_dir / "step0.md").write_text("retry", encoding="utf-8")
write_implementation_agent_profile(tmp_path)
attempts = []
def timeout_run(command, **kwargs):
if command[0] == "git":
responses = {
("symbolic-ref", "--quiet", "HEAD"): "refs/heads/feat-retry-phase",
("rev-parse", "--verify", "HEAD"): "a" * 40,
("status", "--porcelain=v1", "-z", "--untracked-files=all"): "",
}
return subprocess.CompletedProcess(
command, 0, stdout=responses[tuple(command[1:])], stderr=""
)
attempts.append(command)
raise subprocess.TimeoutExpired(command, kwargs["timeout"])
monkeypatch.setattr(execute, "ROOT", tmp_path)
monkeypatch.setattr(execute.subprocess, "run", timeout_run)
executor = execute.StepExecutor("retry-phase")
with pytest.raises(SystemExit, match="1"):
executor._execute_single_step(
{"step": 0, "name": "retry", "status": "pending"}, ""
)
assert len(attempts) == 2
index = json.loads((phase_dir / "index.json").read_text(encoding="utf-8"))
assert index["steps"][0]["status"] == "error"
assert "Codex timed out after 14400 seconds." in index["steps"][0]["error_message"]
assert index["steps"][0]["attempt_history"][-1]["reason"] == index["steps"][0]["error_message"]
top = json.loads((phase_dir.parent / "index.json").read_text(encoding="utf-8"))
assert top["phases"][0]["status"] == "error"
def test_invoke_codex_rejects_unknown_sandbox_override(tmp_path, monkeypatch):
phase_dir = tmp_path / "phases" / "sandbox-phase"
phase_dir.mkdir(parents=True)
write_top_index(phase_dir)
(phase_dir / "index.json").write_text(
json.dumps(
{
"project": "FESA Structural Solver",
"phase": "sandbox-phase",
"steps": [{"step": 0, "name": "sandbox", "status": "pending"}],
}
),
encoding="utf-8",
)
(phase_dir / "step0.md").write_text("sandbox override", encoding="utf-8")
monkeypatch.setattr(execute, "ROOT", tmp_path)
monkeypatch.setenv("FEM_HARNESS_CODEX_SANDBOX", "read-only")
executor = execute.StepExecutor("sandbox-phase")
with pytest.raises(execute.CodexEnvironmentError, match="FEM_HARNESS_CODEX_SANDBOX"):
executor._invoke_codex({"step": 0, "name": "sandbox"}, "")
@pytest.mark.parametrize(
("approved", "isolated", "porcelain", "expected_error"),
[
(False, True, "", "--approve-danger-full-access"),
(True, False, "", "isolated linked worktree"),
(True, True, " M unrelated.md\0", "clean worktree"),
],
)
def test_sandbox_elevation_rejects_missing_safety_conditions(
tmp_path, monkeypatch, approved, isolated, porcelain, expected_error
):
phase_dir = tmp_path / "phases" / "sandbox-phase"
phase_dir.mkdir(parents=True)
write_top_index(phase_dir)
(phase_dir / "step0.md").write_text("sandbox guard", encoding="utf-8")
(phase_dir / "index.json").write_text(
json.dumps({
"project": "FESA Structural Solver",
"phase": "sandbox-phase",
"steps": [{"step": 0, "name": "sandbox", "status": "pending"}],
}),
encoding="utf-8",
)
def git_probe(command, **kwargs):
assert command[0] == "git", "Rejected elevation must not launch Codex"
common_dir = tmp_path / ".git"
git_dir = common_dir / "worktrees/sandbox" if isolated else common_dir
responses = {
("rev-parse", "--absolute-git-dir"): str(git_dir),
("rev-parse", "--git-common-dir"): str(common_dir),
("status", "--porcelain=v1", "-z", "--untracked-files=all"): porcelain,
}
return subprocess.CompletedProcess(
command, 0, stdout=responses[tuple(command[1:])], stderr=""
)
monkeypatch.setattr(execute, "ROOT", tmp_path)
monkeypatch.setattr(execute.subprocess, "run", git_probe)
monkeypatch.setenv("FEM_HARNESS_CODEX_SANDBOX", "danger-full-access")
executor = execute.StepExecutor("sandbox-phase", danger_full_access_approved=approved)
with pytest.raises(execute.CodexEnvironmentError, match=expected_error):
executor._codex_sandbox_mode()
def test_codex_environment_failure_detects_missing_sandbox_helper_on_zero_exit():
output = {
"exitCode": 0,
"stdout": "",
"stderr": (
"windows sandbox: orchestrator_helper_launch_failed: "
"helper=codex-windows-sandbox-setup.exe, error=program not found"
),
}
assert execute.StepExecutor._codex_environment_failure(output) == (
"Codex Windows sandbox helper를 실행할 수 없습니다."
)
def source_echo_event(text):
return json.dumps({
"type": "item.completed",
"item": {
"type": "command_execution", "command": "Get-Content scripts/execute.py",
"aggregated_output": text, "exit_code": 0, "status": "completed",
},
})
@pytest.mark.parametrize("stdout", [
'Source excerpt: if "orchestrator_helper_launch_failed" in diagnostic:',
source_echo_event('if "orchestrator_helper_launch_failed" in diagnostic:'),
source_echo_event("windows sandbox: orchestrator_helper_launch_failed: helper not found"),
source_echo_event(json.dumps({"type": "error", "message": "orchestrator_helper_launch_failed"})),
json.dumps({"type": "item.completed", "item": {
"type": "command_execution", "exit_code": 0,
"command": "rg orchestrator_helper_launch_failed scripts/execute.py",
"aggregated_output": "", "status": "completed",
}}),
json.dumps({"type": "item.completed", "item": {
"type": "agent_message", "text": "Source contains orchestrator_helper_launch_failed",
}}),
], ids=["plain-excerpt", "source-code", "quoted-runtime-line", "quoted-runtime-event",
"command-argument", "agent-source-description"])
def test_environment_classifier_ignores_successful_source_echo(stdout):
output = {"exitCode": 0, "stdout": stdout, "stderr": ""}
assert execute.StepExecutor._codex_environment_failure(output) is None
@pytest.mark.parametrize("process_exit", [0, 1])
@pytest.mark.parametrize("route", [
"stderr", "stdout", "timestamped-stdout", "error-event", "turn-failed",
"failed-command", "mixed-stderr", "mixed-stdout",
])
def test_environment_classifier_retains_runtime_helper_failure(process_exit, route):
diagnostic = "windows sandbox: orchestrator_helper_launch_failed: helper not found"
output = {"exitCode": process_exit, "stdout": "", "stderr": ""}
if route in {"stderr", "mixed-stderr"}:
output["stderr"] = diagnostic
elif route == "error-event":
output["stdout"] = json.dumps({"type": "error", "message": diagnostic})
elif route == "turn-failed":
output["stdout"] = json.dumps({"type": "turn.failed", "error": {"message": diagnostic}})
elif route == "failed-command":
output["stdout"] = json.dumps({"type": "item.completed", "item": {
"type": "command_execution", "status": "failed", "exit_code": 1,
"aggregated_output": diagnostic,
}})
else:
output["stdout"] = diagnostic
if route == "timestamped-stdout":
output["stdout"] = "2026-09-06T01:09:28Z ERROR codex_core::tools: " + diagnostic
if route.startswith("mixed-"):
output["stdout"] = source_echo_event(diagnostic) + "\n" + output["stdout"]
reason = "Codex Windows sandbox helper를 실행할 수 없습니다."
assert execute.StepExecutor._codex_environment_failure(output) == reason
executor = object.__new__(execute.StepExecutor)
assert executor._classify_attempt({"status": "pending"}, output, None) == (
"environment", "blocked", reason,
)
def test_run_git_decodes_output_as_utf8(tmp_path, monkeypatch):
invocation = {}
def capture_run(command, **kwargs):
invocation["command"] = command
invocation["kwargs"] = kwargs
return subprocess.CompletedProcess(command, 0, stdout="", stderr="")
monkeypatch.setattr(execute.subprocess, "run", capture_run)
executor = object.__new__(execute.StepExecutor)
executor._root = str(tmp_path)
executor._run_git("status", "--short")
assert invocation["kwargs"]["encoding"] == "utf-8"
def test_feature_commit_message_is_windows_console_safe():
message = execute.StepExecutor.FEAT_MSG.format(
phase="linear-static-3d-euler-beam", num=0, name="requirements-baseline"
)
message.encode("cp949")
def test_configure_utf8_stdio_reconfigures_both_streams(monkeypatch):
class ReconfigurableStream:
def __init__(self):
self.encodings = []
def reconfigure(self, *, encoding):
self.encodings.append(encoding)
stdout = ReconfigurableStream()
stderr = ReconfigurableStream()
monkeypatch.setattr(execute.sys, "stdout", stdout)
monkeypatch.setattr(execute.sys, "stderr", stderr)
execute._configure_utf8_stdio()
assert stdout.encodings == ["utf-8"]
assert stderr.encodings == ["utf-8"]
@pytest.fixture
def step_runtime(tmp_path, monkeypatch):
"""Run the real Step loop and Git against a disposable repo; replace only Codex."""
monkeypatch.setenv("GIT_CONFIG_GLOBAL", os.devnull)
monkeypatch.setenv("GIT_CONFIG_NOSYSTEM", "1")
monkeypatch.delenv("FEM_HARNESS_CODEX_SANDBOX", raising=False)
real_run = subprocess.run
def git(*args):
return real_run(
["git", *args], cwd=tmp_path, capture_output=True,
text=True, encoding="utf-8", check=True,
).stdout
git("init", "--quiet")
git("config", "user.name", "Harness Tests")
git("config", "user.email", "harness@example.invalid")
git("config", "commit.gpgsign", "false")
empty_hooks = tmp_path / ".git" / "empty-hooks"
empty_hooks.mkdir()
git("config", "core.hooksPath", str(empty_hooks))
phase_dir = tmp_path / "phases" / "runtime"
phase_dir.mkdir(parents=True)
write_top_index(phase_dir)
index_file = phase_dir / "index.json"
index_file.write_text(json.dumps({
"project": "Harness Tests", "phase": "runtime",
"steps": [{"step": 0, "name": "runtime", "status": "pending", "owned_paths": []}],
}), encoding="utf-8")
(phase_dir / "step0.md").write_text("Run the assigned test Step.", encoding="utf-8")
(tmp_path / ".gitignore").write_text("phases/**/step*-output.json\n", encoding="utf-8")
write_implementation_agent_profile(tmp_path)
git("add", "--", ".")
git("commit", "--quiet", "-m", "test: initialize disposable repo")
monkeypatch.setattr(execute, "ROOT", tmp_path)
runtime = SimpleNamespace(
executor=execute.StepExecutor("runtime"), root=tmp_path,
index_file=index_file, git=git, real_run=real_run, responses=[], calls=[],
)
def invoke(command, **kwargs):
if command[0] == "git":
return real_run(command, **kwargs)
assert command[0] == "codex", "Only the simulated Codex process is allowed"
runtime.calls.append(kwargs["input"])
response = runtime.responses.pop(0)
if callable(response):
response = response(runtime)
if isinstance(response, Exception):
raise response
index = json.loads(index_file.read_text(encoding="utf-8"))
current = index["steps"][0]
current["status"] = response.get("status", "pending")
if current["status"] != "pending":
field = {"completed": "summary", "error": "error_message", "blocked": "blocked_reason"}[current["status"]]
current[field] = response.get("reason", "reported reason")
index_file.write_text(json.dumps(index), encoding="utf-8")
return subprocess.CompletedProcess(
command, response.get("exit", 0),
stdout=response.get("stdout", ""), stderr=response.get("stderr", ""),
)
monkeypatch.setattr(execute.subprocess, "run", invoke)
return runtime
def run_step(runtime):
index = runtime.executor._read_phase_index()
return runtime.executor._execute_single_step(index["steps"][0], "")
@pytest.mark.parametrize("status,process_exit,timeouts", [
("completed", 0, 0), ("completed", 1, 0),
("pending", 0, 0), ("pending", 1, 0), ("pending", 124, 0),
("error", 0, 0), ("error", 1, 0), ("error", 124, 0),
("blocked", 0, 0), ("blocked", 1, 0), ("blocked", 124, 0),
("completed", 0, 1), ("pending", 0, 2),
("error", 124, 1), ("blocked", 124, 1),
])
def test_successful_source_echo_preserves_terminal_and_retry(
step_runtime, status, process_exit, timeouts
):
runtime = step_runtime
echo = source_echo_event("windows sandbox: orchestrator_helper_launch_failed: helper not found")
response = {"status": status, "exit": process_exit, "reason": "original result", "stdout": echo}
timeout = subprocess.TimeoutExpired("codex", 1800, output=echo)
terminal_timeout = timeouts and status in {"error", "blocked"}
def report_then_timeout(runtime):
index = runtime.executor._read_phase_index()
current = index["steps"][0]
current["status"] = status
current["error_message" if status == "error" else "blocked_reason"] = "original result"
runtime.executor._write_json(runtime.index_file, index)
return timeout
runtime.responses = [report_then_timeout] if terminal_timeout else [timeout] * timeouts + [response]
if status == "completed" and process_exit == 0:
assert run_step(runtime)
expected = "completed"
else:
with pytest.raises(SystemExit) as stopped:
run_step(runtime)
assert stopped.value.code == (2 if status == "blocked" else 1)
expected = "blocked" if status == "blocked" else "error"
current = runtime.executor._read_phase_index()["steps"][0]
history = current["attempt_history"]
calls = 1 if terminal_timeout else min(timeouts + 1, 2)
assert len(runtime.calls) == len(history) == calls
assert [record["attempt"] for record in history] == list(range(1, calls + 1))
assert current["status"] == expected
classification = (
"transport-timeout" if timeouts == 2 else
"reported-" + status if status in {"error", "blocked"} else
"completed" if expected == "completed" else
"inconsistent-result" if status == "completed" else
"process-error" if process_exit else "missing-result"
)
assert history[-1]["classification"] == classification
assert history[-1]["exit_code"] == (124 if timeouts == 2 or terminal_timeout else process_exit)
if timeouts and not terminal_timeout:
assert history[0]["classification"] == "transport-timeout"
assert "Codex timed out after 1800 seconds." in history[0]["diagnostic"]
if status in {"error", "blocked"}:
field = "error_message" if status == "error" else "blocked_reason"
assert current[field] == history[-1]["reason"] == history[-1]["reported_reason"] == "original result"
if not timeouts:
assert history[-1]["diagnostic"] == echo
@pytest.mark.parametrize("status,expected_exit", [("error", 1), ("blocked", 2)])
@pytest.mark.parametrize("process_exit", [0, 1, 124])
def test_explicit_failure_preserves_reason_without_retry(step_runtime, status, expected_exit, process_exit):
runtime = step_runtime
runtime.responses = [
{"status": status, "reason": "original failure", "exit": process_exit, "stderr": "process diagnostic"},
{"status": "completed"},
]
with pytest.raises(SystemExit) as stopped:
run_step(runtime)
assert stopped.value.code == expected_exit
assert len(runtime.calls) == 1
current = runtime.executor._read_phase_index()["steps"][0]
assert current["status"] == status
assert current["error_message" if status == "error" else "blocked_reason"] == "original failure"
assert current["attempt_history"][0]["reported_reason"] == "original failure"
assert current["attempt_history"][0]["reason"] == "original failure"
assert current["attempt_history"][0]["exit_code"] == process_exit
assert current["attempt_history"][0]["diagnostic"] == "process diagnostic"
assert runtime.executor._read_top_index()["phases"][0]["status"] == status
assert runtime.git("rev-list", "--count", "HEAD").strip() == "1"
@pytest.mark.parametrize("process_exit", [0, 1])
@pytest.mark.parametrize("status", ["pending", "completed", "error"])
def test_environment_failure_blocks_independent_of_exit_and_report(step_runtime, process_exit, status):
runtime = step_runtime
diagnostic = "windows sandbox: orchestrator_helper_launch_failed: helper not found"
runtime.responses = [{"status": status, "exit": process_exit, "stderr": diagnostic}]
with pytest.raises(SystemExit) as stopped:
run_step(runtime)
assert stopped.value.code == 2
assert len(runtime.calls) == 1
current = runtime.executor._read_phase_index()["steps"][0]
assert current["status"] == "blocked"
assert "sandbox helper" in current["blocked_reason"]
record = current["attempt_history"][0]
assert record["reported_status"] == status
assert record["classification"] == "environment"
assert record["diagnostic"] == diagnostic
assert runtime.executor._read_top_index()["phases"][0]["status"] == "blocked"
assert runtime.git("rev-list", "--count", "HEAD").strip() == "1"
def test_missing_cli_is_persisted_as_environment_block(step_runtime):
runtime = step_runtime
runtime.responses = [FileNotFoundError("codex")]
with pytest.raises(SystemExit) as stopped:
run_step(runtime)
assert stopped.value.code == 2
current = runtime.executor._read_phase_index()["steps"][0]
assert current["status"] == "blocked"
assert current["attempt_history"][0]["exit_code"] is None
assert runtime.executor._read_top_index()["phases"][0]["status"] == "blocked"
assert len(runtime.calls) == 1
@pytest.mark.parametrize("status,process_exit", [("pending", 0), ("pending", 1), ("pending", 124), ("completed", 1)])
def test_unclassified_or_inconsistent_result_never_retries(step_runtime, status, process_exit):
runtime = step_runtime
runtime.responses = [{"status": status, "exit": process_exit}, {"status": "completed"}]
with pytest.raises(SystemExit) as stopped:
run_step(runtime)
assert stopped.value.code == 1
assert len(runtime.calls) == 1
assert runtime.executor._read_phase_index()["steps"][0]["status"] == "error"
assert runtime.git("rev-list", "--count", "HEAD").strip() == "1"
def test_timeout_retry_preserves_history_and_can_complete(step_runtime):
runtime = step_runtime
runtime.responses = [subprocess.TimeoutExpired("codex", 1800), {"status": "completed"}]
assert run_step(runtime)
current = runtime.executor._read_phase_index()["steps"][0]
assert current["status"] == "completed"
assert [record["classification"] for record in current["attempt_history"]] == ["transport-timeout", "completed"]
assert [record["attempt"] for record in current["attempt_history"]] == [1, 2]
assert "timed out" in runtime.calls[1]
assert runtime.git("status", "--porcelain").strip() == ""
def test_history_survives_assigned_pending_reset(step_runtime):
runtime = step_runtime
runtime.responses = [{"status": "error", "reason": "original failure"}]
with pytest.raises(SystemExit):
run_step(runtime)
index = runtime.executor._read_phase_index()
history = index["steps"][0]["attempt_history"]
index["steps"][0]["status"] = "pending"
runtime.executor._write_json(runtime.index_file, index)
runtime.executor._ensure_created_at()
current = runtime.executor._read_phase_index()["steps"][0]
assert "error_message" not in current
assert current["attempt_history"] == history
runtime.responses = [{"status": "completed"}]
assert run_step(runtime)
assert [record["attempt"] for record in runtime.executor._read_phase_index()["steps"][0]["attempt_history"]] == [1, 2]
def test_full_runner_completes_metadata_only_phase(step_runtime):
runtime = step_runtime
runtime.responses = [{"status": "completed"}]
runtime.executor.run()
assert len(runtime.calls) == 1
assert runtime.executor._read_top_index()["phases"][0]["status"] == "completed"
assert "completed_at" in runtime.executor._read_phase_index()
assert runtime.git("symbolic-ref", "HEAD").strip() == "refs/heads/feat-runtime"
assert runtime.git("status", "--porcelain").strip() == ""
@pytest.mark.parametrize("status", ["error", "blocked", "completed"])
def test_terminal_report_prevents_retry_even_when_wrapper_times_out(step_runtime, status):
runtime = step_runtime
def report_then_timeout(runtime):
index = runtime.executor._read_phase_index()
current = index["steps"][0]
current["status"] = status
field = {"error": "error_message", "blocked": "blocked_reason", "completed": "summary"}[status]
current[field] = "original terminal report"
runtime.executor._write_json(runtime.index_file, index)
return subprocess.TimeoutExpired("codex", 1800)
runtime.responses = [report_then_timeout]
with pytest.raises(SystemExit) as stopped:
run_step(runtime)
assert stopped.value.code == (2 if status == "blocked" else 1)
assert len(runtime.calls) == 1
assert runtime.executor._read_phase_index()["steps"][0]["attempt_history"][0]["reported_reason"] == "original terminal report"
-348
View File
@@ -1,348 +0,0 @@
import os
import subprocess
import pytest
from scripts import execute
from test_execute import run_step, step_runtime
def set_owned_paths(runtime, paths):
index = runtime.executor._read_phase_index()
index["steps"][0]["owned_paths"] = paths
runtime.executor._write_json(runtime.index_file, index)
def link_directory(runtime, link, target):
if os.name == "nt":
# Junctions exercise Windows link resolution without symlink privileges.
runtime.real_run(
["cmd", "/c", "mklink", "/J", str(link), str(target)],
cwd=runtime.root, capture_output=True, check=True,
)
else:
link.symlink_to(target, target_is_directory=True)
@pytest.mark.parametrize("paths", [
None, "src/a.py", [1], [""], ["."], ["src/"], ["src/**"],
["../a.py"], ["src/../a.py"], ["/a.py"], ["C:/a.py"], ["C:a.py"],
["\\\\server\\share\\a.py"], ["src/a.py:stream"], [".git/config"],
["phases/runtime/index.json"], ["phases/other/step0.md"],
["src/a.py", "src\\a.py"], ["src/a.py\x00"], ["src/a.py."],
])
def test_owned_paths_rejects_ambiguous_or_unsafe_entries(step_runtime, paths):
runtime = step_runtime
set_owned_paths(runtime, paths)
with pytest.raises(execute.HarnessError, match="owned_paths"):
runtime.executor._read_phase_index()
@pytest.mark.skipif(os.name != "nt", reason="Windows path identity")
def test_owned_paths_rejects_case_alias_duplicates(step_runtime):
set_owned_paths(step_runtime, ["src/a.py", "SRC/A.PY"])
with pytest.raises(execute.HarnessError, match="owned_paths"):
step_runtime.executor._read_phase_index()
def test_missing_pending_ownership_stops_before_branch_or_state_changes(step_runtime):
runtime = step_runtime
index = runtime.executor._read_phase_index()
del index["steps"][0]["owned_paths"]
runtime.executor._write_json(runtime.index_file, index)
original = runtime.index_file.read_bytes()
branch = runtime.git("symbolic-ref", "HEAD")
with pytest.raises(execute.HarnessError, match="owned_paths"):
runtime.executor.run()
assert runtime.calls == []
assert runtime.index_file.read_bytes() == original
assert runtime.git("symbolic-ref", "HEAD") == branch
def test_completed_history_without_ownership_remains_readable(step_runtime):
runtime = step_runtime
index = runtime.executor._read_phase_index()
index["steps"][0] = {"step": 0, "name": "runtime", "status": "completed", "summary": "historical"}
runtime.executor._write_json(runtime.index_file, index)
assert execute.StepExecutor("runtime")._read_phase_index()["steps"][0]["summary"] == "historical"
@pytest.mark.parametrize("path", ["src", "escape/file.py"])
def test_pending_ownership_rejects_directory_or_link_escape(step_runtime, path):
runtime = step_runtime
if path == "src":
(runtime.root / "src").mkdir()
else:
outside = runtime.root.parent / (runtime.root.name + "-outside")
outside.mkdir()
link_directory(runtime, runtime.root / "escape", outside)
set_owned_paths(runtime, [path])
with pytest.raises(execute.HarnessError, match="owned_paths"):
runtime.executor._preflight_repository()
assert runtime.calls == []
@pytest.mark.parametrize("history", [None, {}, ["bad"], [{"attempt": 1}]])
def test_malformed_attempt_history_is_rejected_without_erasure(step_runtime, history):
runtime = step_runtime
index = runtime.executor._read_phase_index()
index["steps"][0]["attempt_history"] = history
runtime.executor._write_json(runtime.index_file, index)
original = runtime.index_file.read_bytes()
with pytest.raises(execute.HarnessError, match="attempt_history"):
runtime.executor._read_phase_index()
assert runtime.index_file.read_bytes() == original
@pytest.mark.parametrize("field,value", [("owned_paths", ["unassigned.md"]), ("attempt_history", [])])
def test_agent_cannot_change_executor_or_planning_metadata(step_runtime, field, value):
runtime = step_runtime
def tamper(runtime):
index = runtime.executor._read_phase_index()
index["steps"][0][field] = value
runtime.executor._write_json(runtime.index_file, index)
return {"status": "completed"}
runtime.responses = [tamper]
with pytest.raises(execute.HarnessError, match="immutable metadata"):
run_step(runtime)
current = runtime.executor._read_phase_index()["steps"][0]
assert current["owned_paths"] == []
assert "attempt_history" not in current
assert current["status"] == "error"
assert runtime.git("rev-list", "--count", "HEAD").strip() == "1"
@pytest.mark.parametrize("field,value", [("reported_status", []), ("classification", []), ("exit_code", True), ("attempt", True)])
def test_history_record_types_fail_closed(step_runtime, field, value):
runtime = step_runtime
runtime.responses = [{"status": "error"}]
with pytest.raises(SystemExit):
run_step(runtime)
index = runtime.executor._read_phase_index()
index["steps"][0]["attempt_history"][0][field] = value
runtime.executor._write_json(runtime.index_file, index)
with pytest.raises(execute.HarnessError, match="attempt_history"):
runtime.executor._read_phase_index()
def test_all_pending_steps_need_ownership_before_execution(step_runtime):
runtime = step_runtime
index = runtime.executor._read_phase_index()
index["steps"].append({"step": 1, "name": "later", "status": "pending"})
(runtime.index_file.parent / "step1.md").write_text("Later Step", encoding="utf-8")
runtime.executor._write_json(runtime.index_file, index)
executor = execute.StepExecutor("runtime")
with pytest.raises(execute.HarnessError, match="Step 1 requires an owned_paths"):
executor.run()
assert runtime.calls == []
assert "created_at" not in executor._read_phase_index()
def test_metadata_commit_failure_preserves_implementation_commit_and_reason(step_runtime, monkeypatch):
runtime = step_runtime
set_owned_paths(runtime, ["approved.py"])
def change(runtime):
(runtime.root / "approved.py").write_text("approved\n", encoding="utf-8")
return {"status": "completed"}
original_run_git = runtime.executor._run_git
def fail_metadata_commit(*args):
if args[:2] == ("commit", "-m") and args[-1].startswith("chore("):
return subprocess.CompletedProcess(args, 1, stdout="", stderr="metadata commit failed")
return original_run_git(*args)
monkeypatch.setattr(runtime.executor, "_run_git", fail_metadata_commit)
runtime.responses = [change]
with pytest.raises(SystemExit) as stopped:
run_step(runtime)
assert stopped.value.code == 1
current = runtime.executor._read_phase_index()["steps"][0]
assert current["status"] == "error"
assert current["attempt_history"][-1]["reason"] == current["error_message"]
assert current["attempt_history"][-1]["classification"] == "commit-error"
assert runtime.git("rev-list", "--count", "HEAD").strip() == "2"
assert runtime.git("show", "HEAD:approved.py") == "approved\n"
assert runtime.executor._read_top_index()["phases"][0]["status"] == "error"
@pytest.mark.parametrize("staged", [False, True])
@pytest.mark.parametrize("result", ["completed", "timeout"])
def test_unassigned_changes_stop_commit_and_retry_without_reverting_files(step_runtime, staged, result):
runtime = step_runtime
set_owned_paths(runtime, ["approved.py"])
def change(runtime):
(runtime.root / "approved.py").write_text("approved change\n", encoding="utf-8")
(runtime.root / "unassigned.md").write_text("user change\n", encoding="utf-8")
if staged:
runtime.git("add", "--", "unassigned.md")
if result == "timeout":
return subprocess.TimeoutExpired("codex", 1800)
return {"status": "completed"}
runtime.responses = [change, {"status": "completed"}]
with pytest.raises(SystemExit) as stopped:
run_step(runtime)
assert stopped.value.code == 1
assert len(runtime.calls) == 1
current = runtime.executor._read_phase_index()["steps"][0]
assert current["status"] == "error"
assert "unassigned.md" in current["error_message"]
assert current["attempt_history"][0]["classification"] == "ownership-violation"
assert "unassigned.md" in current["attempt_history"][0]["reason"]
assert (runtime.root / "unassigned.md").read_text(encoding="utf-8") == "user change\n"
assert (runtime.root / "approved.py").read_text(encoding="utf-8") == "approved change\n"
assert runtime.git("rev-list", "--count", "HEAD").strip() == "1"
assert runtime.git("diff", "--cached", "--name-only").strip() == ("unassigned.md" if staged else "")
def test_empty_ownership_allows_only_metadata_commit(step_runtime):
runtime = step_runtime
runtime.responses = [{"status": "completed"}]
assert run_step(runtime)
assert runtime.git("log", "-1", "--format=%s").strip() == "chore(runtime): step 0 output"
assert runtime.git("diff-tree", "--no-commit-id", "--name-only", "-r", "HEAD").strip() == "phases/runtime/index.json"
def test_owned_creation_modification_deletion_and_rename_commit_separately(step_runtime):
runtime = step_runtime
for name in ("modify.py", "delete.py", "old.py"):
(runtime.root / name).write_text("original content\n", encoding="utf-8")
runtime.git("add", "--", "modify.py", "delete.py", "old.py")
runtime.git("commit", "--quiet", "-m", "test: seed tracked files")
set_owned_paths(runtime, ["created.py", "modify.py", "delete.py", "old.py", "new.py"])
def change(runtime):
(runtime.root / "created.py").write_text("created\n", encoding="utf-8")
(runtime.root / "modify.py").write_text("modified\n", encoding="utf-8")
(runtime.root / "delete.py").unlink()
runtime.git("mv", "--", "old.py", "new.py")
return {"status": "completed"}
runtime.responses = [change]
assert run_step(runtime)
assert set(runtime.git("diff-tree", "--no-commit-id", "--no-renames", "--name-only", "-r", "HEAD~1").splitlines()) == {"created.py", "modify.py", "delete.py", "old.py", "new.py"}
assert runtime.git("diff-tree", "--no-commit-id", "--name-only", "-r", "HEAD").strip() == "phases/runtime/index.json"
assert runtime.git("status", "--porcelain").strip() == ""
@pytest.mark.parametrize("owned", [["old.py"], ["new.py"]])
def test_rename_requires_both_endpoints(step_runtime, owned):
runtime = step_runtime
(runtime.root / "old.py").write_text("content\n", encoding="utf-8")
runtime.git("add", "--", "old.py")
runtime.git("commit", "--quiet", "-m", "test: seed rename")
set_owned_paths(runtime, owned)
def rename(runtime):
runtime.git("mv", "--", "old.py", "new.py")
return {"status": "completed"}
runtime.responses = [rename]
with pytest.raises(SystemExit) as stopped:
run_step(runtime)
assert stopped.value.code == 1
assert runtime.git("rev-list", "--count", "HEAD").strip() == "2"
assert (runtime.root / "new.py").is_file()
@pytest.mark.parametrize("injected_path", ["unassigned.md", "phases/runtime/index.json"])
def test_new_unassigned_staged_path_is_rejected_before_commit(step_runtime, monkeypatch, injected_path):
runtime = step_runtime
set_owned_paths(runtime, ["approved.py"])
def change(runtime):
(runtime.root / "approved.py").write_text("approved\n", encoding="utf-8")
return {"status": "completed"}
original_run_git = runtime.executor._run_git
def inject_during_add(*args):
result = original_run_git(*args)
if "add" in args:
if injected_path == "unassigned.md":
(runtime.root / injected_path).write_text("late change\n", encoding="utf-8")
runtime.git("add", "--", injected_path)
return result
monkeypatch.setattr(runtime.executor, "_run_git", inject_during_add)
runtime.responses = [change]
with pytest.raises(SystemExit) as stopped:
run_step(runtime)
assert stopped.value.code == 1
assert injected_path in runtime.executor._read_phase_index()["steps"][0]["error_message"]
assert runtime.git("rev-list", "--count", "HEAD").strip() == "1"
assert (runtime.root / injected_path).is_file()
def test_owned_path_replaced_by_external_junction_is_rejected(step_runtime):
runtime = step_runtime
set_owned_paths(runtime, ["src/approved.py"])
outside = runtime.root.parent / (runtime.root.name + "-outside")
outside.mkdir()
(outside / "approved.py").write_text("outside data\n", encoding="utf-8")
def substitute(runtime):
link_directory(runtime, runtime.root / "src", outside)
return {"status": "completed"}
runtime.responses = [substitute]
with pytest.raises(SystemExit) as stopped:
run_step(runtime)
assert stopped.value.code == 1
assert runtime.git("rev-list", "--count", "HEAD").strip() == "1"
assert (outside / "approved.py").read_text(encoding="utf-8") == "outside data\n"
def test_late_unassigned_staging_survives_index_reset(step_runtime, monkeypatch):
runtime = step_runtime
set_owned_paths(runtime, ["approved.py"])
def change(runtime):
(runtime.root / "approved.py").write_text("approved\n", encoding="utf-8")
return {"status": "completed"}
original_run_git = runtime.executor._run_git
def inject_before_reset(*args):
if "reset" in args:
(runtime.root / "unassigned.md").write_text("staged user change\n", encoding="utf-8")
runtime.git("add", "--", "unassigned.md")
return original_run_git(*args)
monkeypatch.setattr(runtime.executor, "_run_git", inject_before_reset)
runtime.responses = [change]
with pytest.raises(SystemExit) as stopped:
run_step(runtime)
assert stopped.value.code == 1
assert "unassigned.md" in runtime.git("diff", "--cached", "--name-only").splitlines()
assert runtime.git("rev-list", "--count", "HEAD").strip() == "1"
@pytest.mark.parametrize("when", ["preflight", "after-attempt"])
def test_owned_hard_link_is_rejected(step_runtime, when):
runtime = step_runtime
outside = runtime.root.parent / (runtime.root.name + "-outside.py")
outside.write_text("external data\n", encoding="utf-8")
set_owned_paths(runtime, ["approved.py"])
def substitute(runtime):
(runtime.root / "approved.py").hardlink_to(outside)
return {"status": "completed"}
if when == "preflight":
substitute(runtime)
with pytest.raises(execute.HarnessError, match="owned_paths"):
runtime.executor._preflight_repository()
assert runtime.calls == []
else:
runtime.responses = [substitute]
with pytest.raises(SystemExit) as stopped:
run_step(runtime)
assert stopped.value.code == 1
assert outside.read_text(encoding="utf-8") == "external data\n"
assert runtime.git("rev-list", "--count", "HEAD").strip() == "1"
-80
View File
@@ -1,80 +0,0 @@
import json
from pathlib import Path
import sys
ROOT = Path(__file__).resolve().parents[1]
SCRIPTS = ROOT / "scripts"
sys.path.insert(0, str(SCRIPTS))
from msvc_harness.adapters.cmake import CMakeAdapter
from msvc_harness.config import load_config
from msvc_harness.models import ProjectKind, ProjectSelection, Toolchain
PRESET_NAME = "windows-msvc-ninja-debug"
BINARY_DIR = ".harness/build-ninja-msvc-debug"
def test_harness_uses_tracked_ninja_msvc_presets(tmp_path):
presets = json.loads((ROOT / "CMakePresets.json").read_text(encoding="utf-8"))
configure = presets["configurePresets"][0]
assert configure == {
"name": PRESET_NAME,
"displayName": "Windows MSVC x64 Debug (Ninja)",
"generator": "Ninja",
"binaryDir": f"${{sourceDir}}/{BINARY_DIR}",
"cacheVariables": {
"CMAKE_BUILD_TYPE": "Debug",
"FESA_GTEST_SOURCE_DIR": "$env{FESA_GTEST_SOURCE_DIR}",
"MKL_DIR": "$env{MKL_DIR}",
"TBB_DIR": "$env{TBB_DIR}",
"HDF5_DIR": "$env{HDF5_DIR}",
},
}
assert presets["buildPresets"] == [
{"name": PRESET_NAME, "configurePreset": PRESET_NAME}
]
assert presets["testPresets"] == [
{"name": PRESET_NAME, "configurePreset": PRESET_NAME}
]
example = json.loads(
(ROOT / ".harness" / "config.example.json").read_text(encoding="utf-8")
)
assert example == {
"version": 1,
"projectType": "cmake",
"cmake": {
"sourceDir": ".",
"binaryDir": BINARY_DIR,
"configurePreset": PRESET_NAME,
"buildPreset": PRESET_NAME,
"testPreset": PRESET_NAME,
},
}
assert ".harness/build-ninja-msvc-debug/" in (
ROOT / ".gitignore"
).read_text(encoding="utf-8").splitlines()
harness_dir = tmp_path / ".harness"
harness_dir.mkdir()
(harness_dir / "config.json").write_text(
json.dumps(example), encoding="utf-8"
)
config = load_config(tmp_path)
tools = Toolchain(
installation=tmp_path / "vs",
msbuild=tmp_path / "msbuild.exe",
cmake=tmp_path / "cmake.exe",
ctest=tmp_path / "ctest.exe",
)
selection = ProjectSelection(ProjectKind.CMAKE, ROOT / "CMakePresets.json")
plan = CMakeAdapter().create_plan(tmp_path, selection, config, tools)
assert [step.command.argv[1:] for step in plan.steps] == [
("--preset", PRESET_NAME),
("--build", "--preset", PRESET_NAME),
("--preset", PRESET_NAME, "--show-only=json-v1"),
("--preset", PRESET_NAME, "--output-on-failure"),
]
-200
View File
@@ -1,200 +0,0 @@
import io
import json
import subprocess
from pathlib import Path
from types import SimpleNamespace
import pytest
from scripts import execute
from scripts.hooks import stop_validation
from scripts.msvc_harness.models import (
CommandSpec,
ProjectKind,
ValidationPlan,
ValidationStep,
)
from scripts.msvc_harness.process import ValidationFailure, execute_plan
def WriteExecutor(tmp_path, monkeypatch):
phase_dir = tmp_path / "phases" / "timeout-phase"
phase_dir.mkdir(parents=True)
(phase_dir.parent / "index.json").write_text(
json.dumps({"phases": [{"dir": "timeout-phase", "status": "pending"}]}),
encoding="utf-8",
)
(phase_dir / "index.json").write_text(
json.dumps(
{
"project": "Harness Tests",
"phase": "timeout-phase",
"steps": [{"step": 0, "name": "timeout", "status": "pending"}],
}
),
encoding="utf-8",
)
(phase_dir / "step0.md").write_text("timeout", encoding="utf-8")
profile = tmp_path / ".codex" / "agents" / "implementation-agent.toml"
profile.parent.mkdir(parents=True)
profile.write_text(
'model = "gpt-5.6-terra"\n'
'model_reasoning_effort = "xhigh"\n'
'developer_instructions = """test"""\n',
encoding="utf-8",
)
monkeypatch.setattr(execute, "ROOT", tmp_path)
return execute.StepExecutor("timeout-phase")
def TestClock(values):
iterator = iter(values)
return lambda: next(iterator)
def test_executor_uses_finite_default_for_actual_timeout(tmp_path, monkeypatch):
executor = WriteExecutor(tmp_path, monkeypatch)
timeouts = []
def timeout_run(command, **kwargs):
timeouts.append(kwargs["timeout"])
raise subprocess.TimeoutExpired(command, kwargs["timeout"])
monkeypatch.setattr(execute.subprocess, "run", timeout_run)
output = executor._invoke_codex({"step": 0, "name": "timeout"}, "")
assert execute.StepExecutor.CODEX_TIMEOUT_SECONDS == 14400
assert timeouts == [14400]
assert output["transportFailure"] == "timeout"
assert "Codex timed out after 14400 seconds." in output["stderr"]
def test_stop_passes_one_shared_deadline_to_root_toolchain_and_plan(
tmp_path, monkeypatch
):
root = tmp_path
selection = SimpleNamespace(kind=stop_validation.ProjectKind.CMAKE)
calls = {}
monkeypatch.setattr(stop_validation, "load_config", lambda _: object())
monkeypatch.setattr(
stop_validation,
"discover_project",
lambda _, __: SimpleNamespace(selection=selection),
)
def discover_toolchain(kind, *, deadline, clock):
calls["toolchain"] = (kind, deadline, clock())
return object()
class Adapter:
def create_plan(self, *args):
return object()
def capture_plan(plan, plan_root, *, env, deadline, clock):
calls["plan"] = (plan, plan_root, env, deadline, clock())
monkeypatch.setattr(stop_validation, "discover_toolchain", discover_toolchain)
monkeypatch.setattr(stop_validation, "CMakeAdapter", Adapter)
monkeypatch.setattr(stop_validation, "MSBuildAdapter", Adapter)
monkeypatch.setattr(stop_validation, "execute_plan", capture_plan)
clock = lambda: 100.0
assert stop_validation.run_validations(root, clock=clock) == (True, "")
assert stop_validation.TOTAL_TIMEOUT_SECONDS == 7200
assert calls["toolchain"][0].value == ProjectKind.CMAKE.value
assert calls["toolchain"][1] == 7300.0
assert calls["plan"][1] == root
assert calls["plan"][3] == 7300.0
assert calls["plan"][2][stop_validation.REENTRY_ENV] == "1"
observed = {}
payload = {"cwd": str(root), "stop_hook_active": False}
monkeypatch.setattr(stop_validation.sys, "stdin", io.StringIO(json.dumps(payload)))
monkeypatch.setattr(stop_validation, "is_document_only_turn", lambda *_: False)
monkeypatch.setattr(
stop_validation,
"_project_root",
lambda *args, **kwargs: observed.setdefault("root", kwargs["deadline"]) and root,
)
monkeypatch.setattr(
stop_validation,
"run_validations",
lambda *args, **kwargs: observed.setdefault("validation", kwargs["deadline"])
and (True, ""),
)
assert stop_validation.main(clock=clock) == 0
assert observed == {"root": 7300.0, "validation": 7300.0}
def test_command_defaults_and_shared_remaining_cap_are_preserved(tmp_path):
default_command = CommandSpec(("default",), tmp_path, "default")
assert default_command.timeout_seconds == 7200
plan = ValidationPlan(
ProjectKind.CMAKE,
(
ValidationStep(default_command),
ValidationStep(CommandSpec(("remaining",), tmp_path, "remaining")),
ValidationStep(CommandSpec(("explicit",), tmp_path, "explicit", 5)),
),
)
observed = []
def run(command, **kwargs):
observed.append(kwargs["timeout"])
return subprocess.CompletedProcess(command, 0, stdout="", stderr="")
execute_plan(
plan,
tmp_path,
run=run,
clock=TestClock((0.0, 0.0, 2000.0, 2000.0)),
)
assert observed == [7200, 5200.0, 5]
expired_plan = ValidationPlan(
ProjectKind.CMAKE,
(ValidationStep(default_command),),
)
with pytest.raises(ValidationFailure, match="timed out before it could start"):
execute_plan(expired_plan, tmp_path, deadline=0.0, clock=lambda: 0.0)
def nonzero(command, **kwargs):
return subprocess.CompletedProcess(command, 1, stdout="", stderr="failed")
with pytest.raises(ValidationFailure, match="failed with exit code 1"):
execute_plan(expired_plan, tmp_path, run=nonzero, clock=TestClock((0.0, 0.0)))
def test_registered_hook_and_one_time_invocation_contract(monkeypatch):
root = Path(__file__).resolve().parents[1]
hooks = json.loads((root / ".codex" / "hooks.json").read_text(encoding="utf-8"))
pre_tool = hooks["hooks"]["PreToolUse"][0]["hooks"][0]
stop = hooks["hooks"]["Stop"][0]["hooks"][0]
assert stop["timeout"] == 7500
assert pre_tool["timeout"] == 30
assert execute.StepExecutor.CODEX_TIMEOUT_SECONDS > stop["timeout"]
assert stop["timeout"] > stop_validation.TOTAL_TIMEOUT_SECONDS
assert all(
timeout > 0
for timeout in (
execute.StepExecutor.CODEX_TIMEOUT_SECONDS,
stop["timeout"],
stop_validation.TOTAL_TIMEOUT_SECONDS,
CommandSpec(("test",), root, "test").timeout_seconds,
)
)
main_calls = []
monkeypatch.setattr(execute.StepExecutor, "CODEX_TIMEOUT_SECONDS", 1800)
monkeypatch.setattr(
execute,
"main",
lambda: main_calls.append(execute.StepExecutor.CODEX_TIMEOUT_SECONDS),
)
exec(
"execute.StepExecutor.CODEX_TIMEOUT_SECONDS = 14400; execute.main()",
{"execute": execute},
)
assert main_calls == [14400]
-142
View File
@@ -1,142 +0,0 @@
import io
import json
from scripts.hooks import pre_tool_use, stop_validation
def edit_payload(root, path, *, session_id="session-1", turn_id="turn-1"):
return {
"session_id": session_id,
"turn_id": turn_id,
"cwd": str(root),
"tool_name": "apply_patch",
"tool_input": {
"patch": (
"*** Begin Patch\n"
f"*** Update File: {path}\n"
"@@\n"
"-old\n"
"+new\n"
"*** End Patch\n"
)
},
}
def stop_payload(root, *, session_id="session-1", turn_id="turn-1"):
return {
"session_id": session_id,
"turn_id": turn_id,
"cwd": str(root),
"hook_event_name": "Stop",
"stop_hook_active": False,
}
def run_pre_tool_use(monkeypatch, payload):
monkeypatch.setattr(
pre_tool_use.sys,
"stdin",
io.StringIO(json.dumps(payload)),
)
assert pre_tool_use.main() == 0
def run_stop(monkeypatch, payload, root, validation_calls):
monkeypatch.setattr(
stop_validation.sys,
"stdin",
io.StringIO(json.dumps(payload)),
)
monkeypatch.setattr(
stop_validation,
"_project_root",
lambda *args, **kwargs: root,
)
def capture_validation(*args, **kwargs):
validation_calls.append(root)
return True, ""
monkeypatch.setattr(stop_validation, "run_validations", capture_validation)
assert stop_validation.main() == 0
def test_stop_skips_validation_after_markdown_only_edit(tmp_path, monkeypatch):
run_pre_tool_use(monkeypatch, edit_payload(tmp_path, "docs/guide.md"))
validation_calls = []
run_stop(monkeypatch, stop_payload(tmp_path), tmp_path, validation_calls)
assert validation_calls == []
def test_stop_validates_after_non_document_edit(tmp_path, monkeypatch):
run_pre_tool_use(monkeypatch, edit_payload(tmp_path, "scripts/tool.py"))
validation_calls = []
run_stop(monkeypatch, stop_payload(tmp_path), tmp_path, validation_calls)
assert validation_calls == [tmp_path]
def test_stop_validates_after_mixed_document_and_code_edits(tmp_path, monkeypatch):
run_pre_tool_use(monkeypatch, edit_payload(tmp_path, "scripts/tool.py"))
run_pre_tool_use(monkeypatch, edit_payload(tmp_path, "docs/guide.md"))
validation_calls = []
run_stop(monkeypatch, stop_payload(tmp_path), tmp_path, validation_calls)
assert validation_calls == [tmp_path]
def test_stop_validates_when_turn_scope_is_missing(tmp_path, monkeypatch):
validation_calls = []
run_stop(monkeypatch, stop_payload(tmp_path), tmp_path, validation_calls)
assert validation_calls == [tmp_path]
def test_document_scope_is_isolated_by_turn(tmp_path, monkeypatch):
run_pre_tool_use(
monkeypatch,
edit_payload(tmp_path, "docs/guide.md", turn_id="document-turn"),
)
validation_calls = []
run_stop(
monkeypatch,
stop_payload(tmp_path, turn_id="different-turn"),
tmp_path,
validation_calls,
)
assert validation_calls == [tmp_path]
def test_stop_validates_when_turn_scope_is_corrupt(tmp_path, monkeypatch):
run_pre_tool_use(monkeypatch, edit_payload(tmp_path, "docs/guide.md"))
state_files = list((tmp_path / ".harness" / "hook-state").glob("*.json"))
assert len(state_files) == 1
state_files[0].write_text("not json", encoding="utf-8")
validation_calls = []
run_stop(monkeypatch, stop_payload(tmp_path), tmp_path, validation_calls)
assert validation_calls == [tmp_path]
def test_later_document_edit_does_not_repair_malformed_scope(
tmp_path, monkeypatch
):
payload = edit_payload(tmp_path, "docs/guide.md")
run_pre_tool_use(monkeypatch, payload)
state_files = list((tmp_path / ".harness" / "hook-state").glob("*.json"))
assert len(state_files) == 1
state_files[0].write_text('{"version": 1}', encoding="utf-8")
run_pre_tool_use(monkeypatch, payload)
validation_calls = []
run_stop(monkeypatch, stop_payload(tmp_path), tmp_path, validation_calls)
assert validation_calls == [tmp_path]
-268
View File
@@ -1,268 +0,0 @@
"""Exercise process boundaries and preserved raw evidence, not solver numerics."""
import importlib.util
import json
from pathlib import Path
import subprocess
import sys
import pytest
ROOT = Path(__file__).resolve().parents[1]
SPEC = importlib.util.spec_from_file_location("measure", ROOT / "scripts/performance/measure.py")
M = importlib.util.module_from_spec(SPEC)
SPEC.loader.exec_module(M)
@pytest.fixture
def source(tmp_path):
path = tmp_path / "model.inp"
path.write_text("from pathlib import Path\nimport time\ntime.sleep(0.08)\n"
"Path(__file__).with_name('model_results.h5').write_bytes(b'fresh')\n")
return path
def test_real_spawn_isolated_bytes_and_pending_comparison(tmp_path, source, monkeypatch):
monkeypatch.setattr(M, "validate_result", lambda *a: {"frames": 101})
before = source.read_bytes()
rows = [M.run_once(Path(sys.executable), source, "fixture", tmp_path / str(i), i)
for i in range(2)]
assert source.read_bytes() == before
for i, row in enumerate(rows):
assert row["native_exit"] == 0 and row["execution_state"] == "completed"
assert row["elapsed_seconds"] >= 0.08
assert row["comparison_status"] == "pending"
assert row["measurement_index"] == i
native = json.loads(Path(row["native_record"]).read_text())
assert len(native["command"]) == 2
assert native["original_sha256"] == native["scratch_sha256"]
assert native["scratch_unchanged"] and native["source_unchanged"]
assert rows[0]["scratch_input"] != rows[1]["scratch_input"]
def test_timer_excludes_preparation_and_validation(tmp_path, source, monkeypatch):
events = []
real_copy, real_spawn = M.shutil.copyfile, M.subprocess.Popen
monkeypatch.setattr(M.shutil, "copyfile", lambda *a: (events.append("copy"), real_copy(*a))[1])
monkeypatch.setattr(M.time, "perf_counter", lambda: (events.append("clock"), float(len(events)))[1])
monkeypatch.setattr(M.subprocess, "Popen", lambda *a, **kw: (events.append("spawn"), real_spawn(*a, **kw))[1])
monkeypatch.setattr(M, "validate_result", lambda *a: (events.append("validate"), {})[1])
M.run_once(Path(sys.executable), source, "fixture", tmp_path / "run", 0)
assert events == ["copy", "clock", "spawn", "clock", "validate"]
def test_nonzero_with_output_is_not_success(tmp_path, source, monkeypatch):
source.write_text(source.read_text() + "raise SystemExit(7)\n")
monkeypatch.setattr(M, "validate_result", lambda *a: pytest.fail("invalid native exit validated"))
row = M.run_once(Path(sys.executable), source, "fixture", tmp_path / "run", 0)
assert row["native_exit"] == 7 and not row["execution_valid"]
assert row["result_sha256"] is not None
def test_existing_directory_and_old_output_rejected(tmp_path, source):
run = tmp_path / "run"
run.mkdir()
old = run / "model_results.h5"
old.write_bytes(b"old")
with pytest.raises(FileExistsError):
M.run_once(Path(sys.executable), source, "fixture", run, 0)
assert old.read_bytes() == b"old"
def test_launch_failure_never_invents_result_hash(tmp_path, source):
row = M.run_once(tmp_path / "absent.exe", source, "fixture", tmp_path / "run", 0)
assert row["execution_state"] == "launch-failed" and row["native_exit"] is None
assert row["result_sha256"] is None and not row["execution_valid"]
def test_input_mutation_invalidates_result(tmp_path, source, monkeypatch):
source.write_text(source.read_text() + "Path(__file__).write_text('changed')\n")
monkeypatch.setattr(M, "validate_result", lambda *a: {})
row = M.run_once(Path(sys.executable), source, "fixture", tmp_path / "run", 0)
assert not row["execution_valid"]
assert not json.loads(Path(row["native_record"]).read_text())["scratch_unchanged"]
def test_interrupted_run_and_prior_case_are_preserved(tmp_path, source, monkeypatch):
real = M.subprocess.Popen
calls = 0
class Interrupted:
def __init__(self, *a, **kw):
self.proc = real(*a, **kw)
def wait(self, timeout=None):
raise KeyboardInterrupt
def kill(self):
self.proc.kill()
def communicate(self):
return self.proc.communicate()
@property
def returncode(self):
return self.proc.returncode
def spawn(*a, **kw):
nonlocal calls
calls += 1
return real(*a, **kw) if calls == 1 else Interrupted(*a, **kw)
monkeypatch.setattr(M.subprocess, "Popen", spawn)
monkeypatch.setattr(M, "validate_result", lambda *a: {})
monkeypatch.setattr(M, "collect_environment", lambda *a: {"fixed": True})
out = tmp_path / "artifacts"
assert M.measure(Path(sys.executable), out, cases=[("first", source, 1), ("second", source, 1)]) == 1
report = json.loads((out / "measurements.json").read_text())
assert len(report["cases"]) == 2
assert report["cases"][0]["runs"][0]["execution_state"] == "completed"
assert report["cases"][1]["runs"][0]["execution_state"] == "interrupted"
assert not report["cases"][1]["all_measured_runs_within_target"]
def test_timeout_preserves_actual_exit(tmp_path, source, monkeypatch):
monkeypatch.setattr(M, "validate_result", lambda *a: pytest.fail("timed out"))
row = M.run_once(Path(sys.executable), source, "fixture", tmp_path / "run", 0, timeout=0.001)
assert row["execution_state"] == "timed-out"
assert type(row["native_exit"]) is int and not row["execution_valid"]
def test_native_zero_missing_or_invalid_result_fails(tmp_path, source, monkeypatch):
source.write_text("pass\n")
row = M.run_once(Path(sys.executable), source, "fixture", tmp_path / "missing", 0)
assert row["native_exit"] == 0 and not row["execution_valid"]
assert row["result_sha256"] is None
source.write_text("from pathlib import Path\nPath(__file__).with_name('model_results.h5').write_bytes(b'bad schema')\n")
def reject(*a):
raise ValueError("invalid inventory")
monkeypatch.setattr(M, "validate_result", reject)
row = M.run_once(Path(sys.executable), source, "fixture", tmp_path / "invalid", 0)
assert row["native_exit"] == 0 and not row["execution_valid"]
assert row["error"] == "invalid inventory"
def test_slow_runs_are_retained_and_exit_zero_is_not_time_pass(tmp_path, source, monkeypatch):
monkeypatch.setattr(M, "validate_result", lambda *a: {})
monkeypatch.setattr(M, "collect_environment", lambda *a: {"fixed": True})
out = tmp_path / "artifacts"
assert M.measure(Path(sys.executable), out, cases=[("fixture", source, 0.001)]) == 0
report = json.loads((out / "measurements.json").read_text())
case = report["cases"][0]
assert report["execution_complete"] and len(case["runs"]) == 1
assert not case["all_measured_runs_within_target"]
assert case["runs"][0]["kind"] == "measurement"
assert case["runs"][0]["measurement_index"] == 1
assert all(r["comparison_status"] == "pending" for r in case["runs"])
def test_changed_binary_or_source_blocks_group(tmp_path, source, monkeypatch):
monkeypatch.setattr(M, "validate_result", lambda *a: {})
captures = iter([{"sha": "before"}, {"sha": "before"}, {"sha": "after"}])
monkeypatch.setattr(M, "collect_environment", lambda *a: next(captures))
out = tmp_path / "artifacts"
assert M.measure(Path(sys.executable), out, cases=[("fixture", source, 1)]) == 1
row = json.loads((out / "measurements.json").read_text())["cases"][0]["runs"][0]
assert row["native_exit"] == 0 and not row["execution_valid"]
def test_cli_bad_arguments_are_exit_two(tmp_path):
proc = subprocess.run([sys.executable, str(ROOT / "scripts/performance/measure.py"),
"--exe", str(tmp_path / "missing.exe"), "--output-dir", str(tmp_path)],
capture_output=True)
assert proc.returncode == 2
def test_comparison_link_preserves_raw_time_and_rejects_wrong_result(tmp_path, source, monkeypatch):
monkeypatch.setattr(M, "validate_result", lambda *a: {})
monkeypatch.setattr(M, "collect_environment", lambda *a: {"fixed": True})
out = tmp_path / "artifacts"
assert M.measure(Path(sys.executable), out, cases=[("fixture", source, 1)]) == 0
before = json.loads((out / "measurements.json").read_text())
row = before["cases"][0]["runs"][0]
comparison_path = Path(row["native_record"]).parent / "comparison.json"
comparison = {"case": "fixture", "identities": {
"input": {"sha256": row["scratch_sha256"]}, "results": {"sha256": row["result_sha256"]}},
"precheck": {"passed": True}, "blocking_passed": True, "verdict": "passed"}
M.publish(comparison_path, comparison)
assert M.link_comparisons(out) == 0
after = json.loads((out / "measurements.json").read_text())
assert after["cases"][0]["runs"][0]["comparison_status"] == "passed"
assert [r["elapsed_seconds"] for r in after["cases"][0]["runs"]] == [
r["elapsed_seconds"] for r in before["cases"][0]["runs"]]
comparison["identities"]["results"]["sha256"] = "other run"
M.publish(comparison_path, comparison)
with pytest.raises(ValueError):
M.link_comparisons(out)
assert json.loads((out / "measurements.json").read_text()) == after
def test_result_validation_interrupt_preserves_completed_native_and_prior_case(tmp_path, source, monkeypatch):
calls = 0
def validate(*a):
nonlocal calls
calls += 1
if calls == 2:
raise KeyboardInterrupt
return {}
monkeypatch.setattr(M, "validate_result", validate)
monkeypatch.setattr(M, "collect_environment", lambda *a: {"fixed": True})
out = tmp_path / "artifacts"
try:
code = M.measure(Path(sys.executable), out, cases=[("first", source, 1), ("second", source, 1)])
except KeyboardInterrupt:
pytest.fail("Result validation interrupt escaped without a completed native row")
assert code == 1
report = json.loads((out / "measurements.json").read_text())
assert len(report["cases"]) == 2 and not report["execution_complete"]
assert report["cases"][0]["runs"][0]["execution_valid"]
row = report["cases"][1]["runs"][0]
native = json.loads(Path(row["native_record"]).read_text())
assert row["native_exit"] == native["native_exit"] == 0
assert row["execution_state"] == native["execution_state"] == "completed"
assert row["elapsed_seconds"] == native["elapsed_seconds"]
assert row["result_sha256"] == native["result_sha256"] is not None
assert row["result_validation_state"] == "interrupted" and not row["execution_valid"]
assert row["comparison_status"] == "pending"
@pytest.mark.parametrize("exception,state", [(KeyboardInterrupt, "interrupted"), (OSError, "failed")])
def test_post_native_environment_failure_preserves_current_and_prior_case(tmp_path, source, monkeypatch, exception, state):
monkeypatch.setattr(M, "validate_result", lambda *a: {})
out = tmp_path / "artifacts"
def collect(*a):
if (out / "second/run-1/native.json").is_file():
raise exception("post-native environment unavailable")
return {"fixed": True}
monkeypatch.setattr(M, "collect_environment", collect)
try:
code = M.measure(Path(sys.executable), out, cases=[("first", source, 1), ("second", source, 1)])
except (KeyboardInterrupt, OSError):
pytest.fail("Post-native environment failure escaped without preserving the current row")
assert code == 1
report = json.loads((out / "measurements.json").read_text())
assert report["cases"][0]["runs"][0]["execution_valid"]
row = report["cases"][1]["runs"][0]
native = json.loads(Path(row["native_record"]).read_text())
assert row["native_exit"] == native["native_exit"] == 0
assert row["execution_state"] == native["execution_state"] == "completed"
assert row["elapsed_seconds"] == native["elapsed_seconds"]
assert row["result_sha256"] == native["result_sha256"] is not None
assert row["environment_validation_state"] == state
assert not row["execution_valid"] and row["error"]
assert not row["within_target"] and not report["execution_complete"]
assert json.loads((out / "second/run-1/run.json").read_text()) == row
@pytest.mark.parametrize("case,value,code", [
("hexa_nonlinearstatic", float("nan"), "nonfinite-result-value"),
("hexa_nonlinearstatic", float("inf"), "nonfinite-result-value"),
("hexa_dynnonlinear", 1e-12, "nonzero-initial-result-state"),
])
def test_hex8_precheck_rejects_invalid_physical_values(tmp_path, case, value, code):
import h5py
helper_spec = importlib.util.spec_from_file_location("hex8_fixture", ROOT / "tests/test_compare_hex8.py")
helper = importlib.util.module_from_spec(helper_spec)
helper_spec.loader.exec_module(helper)
scratch = tmp_path / f"{case}.inp"
M.shutil.copyfile(ROOT / "reference/hexa" / scratch.name, scratch)
result = tmp_path / f"{case}_results.h5"
model = helper.COMPARE.input_inventory(scratch, case)
helper.make_result(result, model)
with h5py.File(result, "r+") as file:
file["steps/Step-1/frames/0/nodal/displacement"][0, 0] = value
with pytest.raises(Exception) as error:
M.validate_result(case, scratch, result)
assert error.value.code == code