atlas-iac/testing/tests/test_local_planning_policy.py

131 lines
5.5 KiB
Python

"""Keep analysis-only source and generated text out of every export surface."""
import copy
import json
import pytest
from scripts.ops.test_planning import policy, planning
from scripts.ops.test_planning.contracts import DIMENSIONS
@pytest.fixture
def source():
return {"campaign": "FA01", "campaign_family": "SYNTHETIC", "suite": "S1", "case_id": "SYN-01",
"description": "ANALYSIS_ONLY description", "operating_condition": "ANALYSIS_ONLY condition",
"preconditions": "ANALYSIS_ONLY setup", "success_criteria": "ANALYSIS_ONLY assertion",
"case_type": "nominal", "verification_method": "test", "swci": "SYN-COMPONENT",
"witness": "WITNESS_ONLY", "atp": "REPORTING_ONLY", "qtp": "REPORTING_ONLY",
"fqt": "REPORTING_ONLY", "etr_suite": "REPORTING_ONLY", "source_row": 42,
"raw": {"private_column": "RAW_ONLY"}, "unmapped": "UNMAPPED_ONLY"}
@pytest.mark.parametrize("operation", ["profile", "group"])
def test_inference_is_minimized_and_never_includes_raw_or_unmapped_fields(source, operation):
original = copy.deepcopy(source)
encoded = json.dumps(policy.analysis_view(source, operation))
assert "ANALYSIS_ONLY" in encoded
for marker in ("WITNESS_ONLY", "REPORTING_ONLY", "RAW_ONLY", "UNMAPPED_ONLY", "source_row"):
assert marker not in encoded
assert source == original
def test_export_whitelist_and_source_row_switch_are_independent(source):
preview = policy.export_view(source, enabled_fields=set(source))
encoded = json.dumps(preview)
for marker in ("ANALYSIS_ONLY", "RAW_ONLY", "UNMAPPED_ONLY", "source_row"):
assert marker not in encoded
assert "WITNESS_ONLY" in encoded and "REPORTING_ONLY" in encoded
assert policy.export_view(source, export_source_rows=True)["source_row"] == 42
def test_generated_text_cannot_enter_titles_descriptions_objectives_or_attachments(source):
family = {"member_case_ids": [source["case_id"]], "family_name": "DERIVED_RESTRICTED",
"implementation_task_title": "DERIVED_RESTRICTED", "why_together": "DERIVED_RESTRICTED",
"objectives": [{"case_id": source["case_id"], "objective": "DERIVED_RESTRICTED"}],
"attachments": ["DERIVED_RESTRICTED"]}
result = {"campaign_id": "FA01", "suite_id": "S1", "families": [family]}
previews = policy.safe_family_previews(result, [source])
encoded = json.dumps(previews)
for marker in ("ANALYSIS_ONLY", "DERIVED_RESTRICTED", "RAW_ONLY", "UNMAPPED_ONLY"):
assert marker not in encoded
assert previews[0]["title"] == "Implementation family 001"
assert previews[0]["publishing_enabled"] is False
assert previews[0]["attachments"] == []
family["member_case_ids"].append("INVENTED")
with pytest.raises(ValueError):
policy.safe_family_previews(result, [source])
def test_exact_fa01_selection_cannot_match_substrings(source):
records = [source, {**source, "campaign": "FA010"}, {**source, "campaign": "prefix FA01"}]
assert policy.select_fa01(records) == [source]
with pytest.raises(ValueError):
policy.select_fa01(records[1:])
def test_nested_values_cannot_smuggle_raw_content_through_an_allowed_field(source):
source["description"] = {"raw": "unexpected source dictionary"}
with pytest.raises(ValueError):
policy.analysis_view(source, "profile")
class StopBeforeNetwork(Exception):
pass
class CaptureClient:
def __init__(self):
self.envelopes = []
def run(self, envelope, validator):
self.envelopes.append(envelope)
raise StopBeforeNetwork
def planner():
instance = planning.Planner.__new__(planning.Planner)
instance.client = CaptureClient()
return instance
def profile(source):
return {"campaign_id": source["campaign"], "suite_id": source["suite"], "case_id": source["case_id"],
"facts": {name: None for name in DIMENSIONS}, "inferred_suggestions": []}
def test_profile_envelope_tracks_identity_without_sending_unneeded_fields(source):
instance = planner()
with pytest.raises(StopBeforeNetwork):
instance.profile_case(source)
envelope = instance.client.envelopes[0]
assert envelope["case_ids"] == [source["case_id"]]
assert "WITNESS_ONLY" not in envelope["request"]["prompt"]
assert "RAW_ONLY" not in envelope["request"]["prompt"]
with pytest.raises(ValueError):
instance.profile_case({**source, "campaign": "FA02"})
assert len(instance.client.envelopes) == 1
def test_suite_request_uses_all_and_only_owned_cases(source):
second = {**source, "case_id": "SYN-02"}
foreign = {**source, "campaign": "FA02", "description": "FOREIGN_CAMPAIGN"}
other_suite = {**source, "suite": "S2", "description": "FOREIGN_SUITE"}
instance = planner()
with pytest.raises(StopBeforeNetwork):
instance.group_suite([source, second, foreign, other_suite], "S1", [profile(source), profile(second)])
envelope = instance.client.envelopes[0]
assert envelope["case_ids"] == ["SYN-01", "SYN-02"]
assert "FOREIGN_" not in envelope["request"]["prompt"]
with pytest.raises(ValueError):
instance.group_suite([source, second], "S1", [profile(source)])
assert len(instance.client.envelopes) == 1
def test_oversized_suite_stops_before_network_instead_of_truncating(source):
instance = planner()
source["description"] = "x" * 70000
with pytest.raises(ValueError, match="never truncate"):
instance.group_suite([source], "S1", [profile(source)])
assert not instance.client.envelopes