"""repair_composition: normalize and safely repair an AI-written `compose` entry, keep the original, report every change, and NEVER guess anything security- or data-relevant.""" from __future__ import annotations import copy import random import pytest from src.domain.feature_packs.manifest_repair import repair_composition, to_slug from src.domain.feature_packs.requirements_spec import CompositionRequirement def good(**overrides): entry = { "name": "contact_submission", "actors": ["anonymous", "admin"], "requirements": [ {"action": "create", "resource": "contact_submission", "actor": "anonymous"}, {"action": "read", "resource": "contact_submission", "actor": "admin"}, ], "fields": [ {"key": "email", "label": "Email", "type": "text", "required": True}, {"key": "message", "label": "Message", "type": "text"}, ], "criticality": "required", } entry.update(overrides) return entry def reasons(result): return {(r["path"], r["reason"]) for r in result.repairs} # ---- the shape an agent actually writes --------------------------------------------- def test_an_already_canonical_entry_needs_no_repair(): result = repair_composition(good()) assert result.status == "valid" and result.repairs == [] and result.errors == [] assert isinstance(result.requirement, CompositionRequirement) def test_a_cloud_shaped_entry_is_repaired_into_a_valid_requirement(): """Fields copied from Cloud's own entity shape (`name` next to `key`, no `label`), a display-cased entity name, capitalized actions and no actors/criticality: the everyday near-miss that used to block a run outright.""" entry = { "entity_key": "Contact Submission", "description": "Public contact form", "requirements": [ {"action": "Create", "resource": "Contact Submission", "actor": "Anonymous"}, {"action": "READ", "resource": "Contact Submission", "actor": " admin "}, ], "fields": [ { "key": "name", "name": "Name", "type": "string", "required": True, "placeholder": "Your name", }, {"key": "email", "name": "Email", "type": "email", "required": True}, {"key": "message", "label": "Message", "name": "Message", "type": "text"}, ], } result = repair_composition(entry) assert result.status == "repaired", result.errors req = result.requirement assert req.name == "contact_submission" assert req.actors == ["anonymous", "admin"] assert req.criticality == "required" assert [(r.action, r.resource, r.actor) for r in req.requirements] == [ ("create", "contact_submission", "anonymous"), ("read", "contact_submission", "admin"), ] assert [(f.key, f.label, f.type) for f in req.fields] == [ ("name", "Name", "text"), ("email", "Email", "text"), ("message", "Message", "text"), ] assert result.renames # the frontend has to be told the entity name changed # ---- each repair, individually --------------------------------------------------------- @pytest.mark.parametrize( "mutate,expected", [ (lambda e: e.update(name="Contact Submission"), ("name", "normalized_to_slug")), (lambda e: e.update(name="contactSubmission"), ("name", "normalized_to_slug")), (lambda e: e.update(name="contact-submission"), ("name", "normalized_to_slug")), ( lambda e: e["fields"][0].update(key="Email Address"), ("fields.0.key", "normalized_to_slug"), ), ( lambda e: e["requirements"][0].update(resource="Contact Submission"), ("requirements.0.resource", "normalized_to_slug"), ), ( lambda e: e["requirements"][0].update(action="Create"), ("requirements.0.action", "normalized_case"), ), ( lambda e: e["requirements"][0].update(actor="Anonymous"), ("requirements.0.actor", "normalized_case"), ), (lambda e: e.update(criticality="Required"), ("criticality", "normalized_case")), (lambda e: e.pop("criticality"), ("criticality", "default_made_explicit")), (lambda e: e["fields"][0].update(type="string"), ("fields.0.type", "type_synonym")), ( lambda e: e["fields"][0].update(placeholder="x"), ("fields.0.placeholder", "cosmetic_key_removed"), ), (lambda e: e.update(description="A form"), ("description", "cosmetic_key_removed")), ( lambda e: e.update(entity="contact_submission") or e.pop("name"), ("name", "alias_to_name"), ), ( lambda e: e["fields"][0].update(name="Email"), ("fields.0.name", "duplicate_display_name_removed"), ), ], ) def test_each_safe_repair_is_applied_and_recorded(mutate, expected): entry = good() mutate(entry) result = repair_composition(entry) assert result.status == "repaired", result.errors assert expected in reasons(result) def test_a_field_name_alias_becomes_the_label_when_there_is_no_label(): entry = good() entry["fields"][0] = {"key": "email", "name": "Email", "type": "text"} result = repair_composition(entry) assert result.requirement.fields[0].label == "Email" assert ("fields.0.label", "alias_to_label") in reasons(result) def test_actors_are_inferred_or_completed_from_what_the_requirements_already_name(): inferred = repair_composition(good(actors=[])) assert inferred.requirement.actors == ["anonymous", "admin"] assert ("actors", "actors_inferred_from_requirements") in reasons(inferred) completed = repair_composition(good(actors=["anonymous"])) assert completed.requirement.actors == ["anonymous", "admin"] assert ("actors", "actor_added_from_requirements") in reasons(completed) # ---- what is NEVER repaired ---------------------------------------------------------------- def test_an_unknown_key_is_reported_not_silently_dropped(): """`public: true` could be an access intent; dropping it would change the meaning.""" entry = good() entry["fields"][0]["public"] = True result = repair_composition(entry) assert result.status == "invalid" assert {"path": "fields.0.public", "code": "unknown_key"}.items() <= result.errors[0].items() assert result.repaired["fields"][0]["public"] is True # still there @pytest.mark.parametrize( "mutate,code_path", [ (lambda e: e["fields"][0].update(type="geo"), "fields.0.type"), (lambda e: e["fields"][0].update(type="datetime"), "fields.0.type"), (lambda e: e["fields"][0].pop("key"), "fields.0.key"), (lambda e: e["requirements"][0].update(action="write"), "requirements.0.action"), (lambda e: e["requirements"][0].update(actor="guest"), "requirements.0.actor"), (lambda e: e["requirements"][0].update(actor="user"), "requirements.0.actor"), (lambda e: e.update(name="9lives"), "name"), (lambda e: e.update(name="x" * 60), "name"), (lambda e: e.update(requirements=[]), "requirements"), (lambda e: e.update(fields=[]), "fields"), (lambda e: e.update(criticality="maybe"), "criticality"), ( lambda e: e["fields"].append({"key": "pick", "label": "Pick", "type": "select"}), "fields.2", ), ], ) def test_ambiguous_or_missing_meaning_is_reported_never_invented(mutate, code_path): entry = good() mutate(entry) result = repair_composition(entry) assert result.status == "invalid" assert any(e["path"].startswith(code_path) for e in result.errors), result.errors def test_conflicting_name_aliases_are_a_conflict_not_a_guess(): result = repair_composition(good(entity_key="something_else")) assert result.status == "invalid" assert any(e["code"] == "unknown_key" and e["path"] == "entity_key" for e in result.errors) def test_workflow_transitions_functions_and_requires_are_left_exactly_as_written(): entry = good( fields=[ {"key": "status", "label": "Status", "type": "select", "options": ["New", "Done"]}, ], workflow={"field": "Status", "transitions": {"New": ["Done"], "Done": []}}, function_slugs=["Send-Email"], requires=["email.send"], ) result = repair_composition(entry) assert result.requirement.workflow.field == "status" # identifier normalized assert result.repaired["workflow"]["transitions"] == {"New": ["Done"], "Done": []} assert result.repaired["function_slugs"] == ["Send-Email"] # not "fixed" to a slug assert result.repaired["requires"] == ["email.send"] assert result.repaired["fields"][0]["options"] == ["New", "Done"] def test_a_non_object_entry_is_reported(): for bad in ("contact form", 7, None, ["a"]): result = repair_composition(bad) assert result.status == "invalid" assert result.errors[0]["code"] == "not_an_object" assert result.original == bad # ---- invariants ---------------------------------------------------------------------------- def test_the_original_entry_is_preserved_and_the_input_is_never_mutated(): entry = { "name": "Contact Submission", "requirements": [ {"action": "Create", "resource": "Contact Submission", "actor": "Anonymous"} ], "fields": [{"key": "Email", "name": "Email", "type": "string"}], } snapshot = copy.deepcopy(entry) result = repair_composition(entry) assert entry == snapshot assert result.original == snapshot assert result.original is not entry assert result.repaired["name"] == "contact_submission" def test_repairing_twice_changes_nothing_more(): entry = { "entity": "Job Application", "requirements": [ {"action": "Create", "resource": "Job Application", "actor": "Anonymous"}, {"action": "Read", "resource": "Job Application", "actor": "Admin"}, ], "fields": [{"key": "Full Name", "name": "Full name", "type": "string", "required": True}], "notes": "x", } first = repair_composition(entry) assert first.status == "repaired", first.errors second = repair_composition(first.repaired) assert second.status == "valid" and second.repairs == [] assert second.repaired == first.repaired def test_repair_never_changes_what_the_entry_says_over_many_random_spellings(): rng = random.Random(1234) def respell(text: str) -> str: words = text.split("_") style = rng.choice(["Title Case", "camelCase", "kebab-case", "UPPER", "spaced out"]) if style == "Title Case": return " ".join(w.capitalize() for w in words) if style == "camelCase": return words[0] + "".join(w.capitalize() for w in words[1:]) if style == "kebab-case": return "-".join(words) if style == "UPPER": return "_".join(w.upper() for w in words) return " ".join(words) for _ in range(200): base = good(name="contact_submission") entry = copy.deepcopy(base) entry["name"] = respell("contact_submission") for requirement in entry["requirements"]: requirement["resource"] = respell("contact_submission") requirement["action"] = rng.choice( [requirement["action"], requirement["action"].upper()] ) requirement["actor"] = rng.choice([requirement["actor"], requirement["actor"].title()]) for item in entry["fields"]: item["key"] = respell(item["key"]) result = repair_composition(entry) assert result.status in {"valid", "repaired"}, (entry, result.errors) req = result.requirement assert req.name == "contact_submission" assert sorted((r.action, r.resource, r.actor) for r in req.requirements) == [ ("create", "contact_submission", "anonymous"), ("read", "contact_submission", "admin"), ] assert sorted(f.key for f in req.fields) == ["email", "message"] def test_a_repaired_entry_can_never_grant_an_actor_more_than_the_requirements_asked(): """Adding declared actors is safe only because access comes from the requirements.""" result = repair_composition(good(actors=["anonymous"])) assert {(r.actor, r.action) for r in result.requirement.requirements} == { ("anonymous", "create"), ("admin", "read"), } def test_the_record_keeps_original_repaired_repairs_and_the_verdict(): result = repair_composition({**good(), "name": "Contact Submission"}) record = result.record() assert record["original_manifest"]["name"] == "Contact Submission" assert record["repaired_manifest"]["name"] == "contact_submission" assert { "path": "name", "from": "Contact Submission", "to": "contact_submission", "reason": "normalized_to_slug", } in record["repairs"] assert record["validation"] == "passed" and record["errors"] == [] failed = repair_composition({**good(), "fields": []}).record() assert failed["validation"] == "failed" and failed["errors"] def test_slug_helper(): assert to_slug("Contact Submission") == "contact_submission" assert to_slug("contactSubmission") == "contact_submission" assert to_slug("Café Menü") == "cafe_menu" assert to_slug("!!!") is None