feat: add P2 harness machine contracts

This commit is contained in:
2026-08-11 05:24:44 +02:00
parent 783cc3bb34
commit b235132d89
6 changed files with 463 additions and 199 deletions
@@ -1,3 +1,4 @@
# ruff: noqa: DTZ001
from datetime import datetime
import yaml
@@ -221,3 +222,62 @@ def test_suggest_fks_write_merges_and_is_idempotent(tmp_path):
assert "nessuna FK da suggerire" in res2.output
ann2 = Annotations.from_yaml(ann_path)
assert len(ann2.tables["fact_ablazione"].foreign_keys) == 2
def test_suggest_fks_json_is_single_deterministic_document_without_writing(tmp_path):
import hashlib
import json
cfg = _write_workspace(tmp_path)
staged = tmp_path / "staged"
staged.mkdir()
(staged / "z.sql").write_text(
"SELECT * FROM fact_ablazione f JOIN dim_patient p ON f.cod_paz = p.cod_paz"
)
(staged / "a.sql").write_text(
"SELECT * FROM fact_ablazione f JOIN dim_time p ON f.data_time_key = p.day_key"
)
response = CliRunner().invoke(
app, ["schema", "suggest-fks", "--json", "-c", str(cfg), "--from-sql", str(staged)]
)
assert response.exit_code == 0, response.output
assert response.stdout.count("\n") == 1
payload = json.loads(response.stdout)
assert payload["status"] == "succeeded"
assert payload["code"] == "ok"
assert payload["candidate_count"] == 2
assert payload["candidate_digest"].startswith("sha256:")
assert payload["orphan_count"] == 0
# The digest is over the canonical candidate export and --json never writes annotations.
expected = json.dumps(payload["candidates"], ensure_ascii=False, sort_keys=True, separators=(",", ":"))
assert payload["candidate_digest"] == "sha256:" + hashlib.sha256(expected.encode()).hexdigest()
assert not (tmp_path / "artifacts" / "mschema" / "annotations.yaml").exists()
def test_suggest_fks_json_failure_is_safe_and_single_document(tmp_path):
import json
cfg = _write_workspace(tmp_path)
(tmp_path / "artifacts" / "mschema" / "physical.yaml").unlink()
response = CliRunner().invoke(app, ["schema", "suggest-fks", "--json", "-c", str(cfg)])
assert response.exit_code != 0
assert response.stdout.count("\n") == 1
payload = json.loads(response.stdout)
assert payload == {"status": "failed", "code": "physical_schema_missing"}
assert str(tmp_path) not in response.stdout
def test_schema_check_json_reports_orphan_count_without_prose(tmp_path):
import json
cfg = _write_workspace(tmp_path)
Annotations(tables={"gone": TableAnnotation(description="x")}).to_yaml(
tmp_path / "artifacts" / "mschema" / "annotations.yaml"
)
response = CliRunner().invoke(app, ["schema", "check", "--json", "-c", str(cfg)])
assert response.exit_code == 3
assert response.stdout.count("\n") == 1
assert json.loads(response.stdout) == {
"status": "failed", "code": "annotation_orphans", "orphan_count": 1,
"orphans": ["gone"],
}