"""Tests for the control plane. Run against your own work: pytest Run against the solution: LAB_MODULE=solution pytest """ from __future__ import annotations import importlib import os import pytest lab = importlib.import_module(os.environ.get("LAB_MODULE", "lab")) Subject = lab.Subject Resource = lab.Resource Environment = lab.Environment Request = lab.Request Effect = lab.Effect Rule = lab.Rule Decision = lab.Decision PolicyBundle = lab.PolicyBundle PolicyEngine = lab.PolicyEngine BundleError = lab.BundleError BundleDistributor = lab.BundleDistributor Posture = lab.Posture AgentRecord = lab.AgentRecord AgentRegistry = lab.AgentRegistry AgentState = lab.AgentState ToolRecord = lab.ToolRecord ToolRegistry = lab.ToolRegistry SideEffect = lab.SideEffect ControlPlane = lab.ControlPlane ContinuousAuthorizer = lab.ContinuousAuthorizer EvalCase = lab.EvalCase EvaluationPipeline = lab.EvaluationPipeline Tracer = lab.Tracer SECRET = b"test-secret" # ====================================================================================== # helpers # ====================================================================================== def clock(start: int = 0, step: int = 1): state = {"t": start - step} def now() -> int: state["t"] += step return state["t"] return now def frozen(value: int = 100): return lambda: value def subject(**kw) -> Subject: base = dict(agent_id="a1", user_id="u1", tenant="wholesale", scopes=("crm.read",), clearance="confidential") base.update(kw) return Subject(**base) def resource(**kw) -> Resource: base = dict(resource_type="tool", resource_id="crm.read", tenant="wholesale", classification="internal") base.update(kw) return Resource(**base) def request(action: str = "crm.read", *, sub=None, res=None, env=None) -> Request: return Request(sub or subject(), action, res or resource(), env or Environment()) def bundle(*rules: Rule, version: str = "v1", created: int = 0) -> PolicyBundle: return PolicyBundle(version, tuple(rules), created).sign(SECRET) ALLOW_READ = Rule("allow-read", Effect.ALLOW, actions=("crm.read",), reason="read ok") DENY_CROSS_TENANT = Rule( "deny-cross-tenant", Effect.DENY, condition=lambda r: r.resource.tenant != r.subject.tenant, reason="cross tenant") def registry_with_active_agent(**kw) -> AgentRegistry: reg = AgentRegistry() base = dict(agent_id="a1", owner="layla", tenant="wholesale", permitted_tools=("crm.read", "crm.write"), max_classification="confidential", model_version="m-1", last_evaluation_tick=0, evaluation_score=0.95) base.update(kw) reg.register(AgentRecord(**base)) reg.transition(base["agent_id"], AgentState.APPROVED) reg.transition(base["agent_id"], AgentState.ACTIVE) return reg def plane(*, agents=None, tools=None, rules=(ALLOW_READ, DENY_CROSS_TENANT), now=None, max_age: int = 1000) -> ControlPlane: now = now or frozen(0) agents = agents or registry_with_active_agent() if tools is None: tools = ToolRegistry() tools.publish(ToolRecord("crm.read", "read", SideEffect.READ, ("crm.read",))) tools.publish(ToolRecord("crm.write", "write", SideEffect.WRITE_IDEMPOTENT, ("crm.write",))) dist = BundleDistributor(secret=SECRET, now=now, initial=bundle(*rules)) return ControlPlane(agents=agents, tools=tools, distributor=dist, now=now, max_evaluation_age_ticks=max_age) # ====================================================================================== # 1. classification ordering # ====================================================================================== def test_classifications_are_totally_ordered(): ranks = [lab.classification_rank(c) for c in ("public", "internal", "confidential", "restricted")] assert ranks == sorted(ranks) and len(set(ranks)) == 4 def test_an_unknown_classification_is_an_error_not_a_default(): with pytest.raises(ValueError): lab.classification_rank("super-secret") # ====================================================================================== # 2. rule matching # ====================================================================================== def test_an_empty_action_tuple_matches_any_action(): rule = Rule("r", Effect.DENY) assert rule.matches(request("anything.at.all")) def test_an_action_list_matches_exactly(): rule = Rule("r", Effect.ALLOW, actions=("crm.read",)) assert rule.matches(request("crm.read")) assert not rule.matches(request("crm.write")) def test_a_trailing_wildcard_matches_the_prefix(): rule = Rule("r", Effect.DENY, actions=("payments.*",)) assert rule.matches(request("payments.release")) assert rule.matches(request("payments.lookup")) assert not rule.matches(request("crm.read")) def test_a_wildcard_does_not_match_a_different_namespace_with_the_same_prefix(): rule = Rule("r", Effect.DENY, actions=("pay.*",)) assert not rule.matches(request("payments.release")) def test_resource_type_narrows_a_rule(): rule = Rule("r", Effect.ALLOW, resource_types=("data",)) assert not rule.matches(request(res=resource(resource_type="tool"))) assert rule.matches(request(res=resource(resource_type="data"))) def test_tenant_narrows_a_rule(): rule = Rule("r", Effect.ALLOW, tenants=("retail",)) assert not rule.matches(request(sub=subject(tenant="wholesale"))) assert rule.matches(request(sub=subject(tenant="retail"))) def test_all_facets_must_match_not_any(): rule = Rule("r", Effect.ALLOW, actions=("crm.read",), tenants=("retail",)) assert not rule.matches(request("crm.read", sub=subject(tenant="wholesale"))) def test_a_condition_can_veto_a_structural_match(): rule = Rule("r", Effect.ALLOW, actions=("crm.read",), condition=lambda r: False) assert not rule.matches(request("crm.read")) # ====================================================================================== # 3. the policy engine: default-deny, deny-overrides # ====================================================================================== def test_no_matching_rule_denies(): engine = PolicyEngine(bundle(ALLOW_READ)) decision = engine.evaluate(request("crm.write")) assert decision.effect is Effect.DENY assert "default deny" in decision.reason def test_a_default_deny_names_no_rule_but_still_names_a_policy_version(): engine = PolicyEngine(bundle(ALLOW_READ, version="v7")) decision = engine.evaluate(request("crm.write")) assert decision.rule_name == "-" assert decision.policy_version == "v7" def test_an_empty_bundle_denies_everything(): engine = PolicyEngine(bundle()) assert not engine.evaluate(request()).allowed def test_a_matching_allow_allows(): engine = PolicyEngine(bundle(ALLOW_READ)) decision = engine.evaluate(request("crm.read")) assert decision.allowed and decision.rule_name == "allow-read" def test_any_deny_beats_any_allow(): engine = PolicyEngine(bundle(ALLOW_READ, DENY_CROSS_TENANT)) decision = engine.evaluate(request("crm.read", res=resource(tenant="retail"))) assert decision.effect is Effect.DENY assert decision.rule_name == "deny-cross-tenant" def test_deny_overrides_is_order_independent(): forward = PolicyEngine(bundle(ALLOW_READ, DENY_CROSS_TENANT)) reverse = PolicyEngine(bundle(DENY_CROSS_TENANT, ALLOW_READ)) req = request("crm.read", res=resource(tenant="retail")) assert forward.evaluate(req).effect is reverse.evaluate(req).effect is Effect.DENY def test_the_decision_lists_every_rule_that_matched_not_only_the_winner(): engine = PolicyEngine(bundle(ALLOW_READ, DENY_CROSS_TENANT)) decision = engine.evaluate(request("crm.read", res=resource(tenant="retail"))) assert decision.matched_rules == ("allow-read", "deny-cross-tenant") def test_matched_rules_are_sorted_so_the_record_is_stable(): a = Rule("z-allow", Effect.ALLOW, actions=("crm.read",)) b = Rule("a-allow", Effect.ALLOW, actions=("crm.read",)) decision = PolicyEngine(bundle(a, b)).evaluate(request("crm.read")) assert decision.matched_rules == ("a-allow", "z-allow") def test_the_reported_rule_among_several_denies_is_deterministic(): d1 = Rule("deny-one", Effect.DENY, actions=("crm.read",), reason="one") d2 = Rule("deny-two", Effect.DENY, actions=("crm.read",), reason="two") engine = PolicyEngine(bundle(d1, d2)) names = {engine.evaluate(request("crm.read")).rule_name for _ in range(5)} assert names == {"deny-one"} def test_every_decision_carries_the_policy_version(): engine = PolicyEngine(bundle(ALLOW_READ, version="2026-02-11.3")) for action in ("crm.read", "crm.write"): assert engine.evaluate(request(action)).policy_version == "2026-02-11.3" def test_the_decision_records_when_it_was_evaluated(): engine = PolicyEngine(bundle(ALLOW_READ)) decision = engine.evaluate(request("crm.read", env=Environment(tick=4242))) assert decision.evaluated_at == 4242 # ====================================================================================== # 4. bundle integrity # ====================================================================================== def test_an_unsigned_bundle_fails_verification(): with pytest.raises(BundleError, match="unsigned"): PolicyBundle("v1", (ALLOW_READ,), 0).verify(SECRET) def test_a_bundle_signed_with_another_key_fails_verification(): other = PolicyBundle("v1", (ALLOW_READ,), 0).sign(b"other-key") with pytest.raises(BundleError, match="bad signature"): other.verify(SECRET) def test_a_correctly_signed_bundle_verifies(): bundle(ALLOW_READ).verify(SECRET) def test_changing_a_rule_invalidates_the_signature(): signed = bundle(ALLOW_READ) tampered = lab.replace(signed, rules=(ALLOW_READ, DENY_CROSS_TENANT)) with pytest.raises(BundleError): tampered.verify(SECRET) def test_changing_the_version_invalidates_the_signature(): tampered = lab.replace(bundle(ALLOW_READ), version="v2") with pytest.raises(BundleError): tampered.verify(SECRET) def test_the_digest_is_stable_across_equal_bundles(): a = PolicyBundle("v1", (ALLOW_READ,), 0) b = PolicyBundle("v1", (ALLOW_READ,), 0) assert a.digest() == b.digest() def test_duplicate_rule_names_are_rejected(): dup = Rule("allow-read", Effect.DENY, actions=("crm.write",)) with pytest.raises(BundleError, match="duplicate"): PolicyBundle("v1", (ALLOW_READ, dup), 0).validate() def test_an_unconditional_allow_everything_rule_is_rejected(): with pytest.raises(BundleError, match="allow-everything"): PolicyBundle("v1", (Rule("oops", Effect.ALLOW),), 0).validate() def test_an_unconditional_deny_everything_rule_is_allowed(): PolicyBundle("v1", (Rule("panic", Effect.DENY),), 0).validate() def test_the_engine_refuses_to_load_an_invalid_bundle(): with pytest.raises(BundleError): PolicyEngine(PolicyBundle("v1", (Rule("oops", Effect.ALLOW),), 0)) # ====================================================================================== # 5. distribution and fail-static # ====================================================================================== def test_the_distributor_refuses_an_unsigned_initial_bundle(): with pytest.raises(BundleError): BundleDistributor(secret=SECRET, now=frozen(0), initial=PolicyBundle("v1", (ALLOW_READ,), 0)) def test_a_valid_bundle_is_activated(): dist = BundleDistributor(secret=SECRET, now=clock(), initial=bundle(ALLOW_READ)) assert dist.offer(bundle(ALLOW_READ, DENY_CROSS_TENANT, version="v2", created=5)) assert dist.active.version == "v2" def test_an_unsigned_bundle_leaves_the_previous_one_live(): dist = BundleDistributor(secret=SECRET, now=clock(), initial=bundle(ALLOW_READ)) assert not dist.offer(PolicyBundle("v2", (ALLOW_READ,), 5)) assert dist.active.version == "v1" def test_a_tampered_bundle_leaves_the_previous_one_live(): dist = BundleDistributor(secret=SECRET, now=clock(), initial=bundle(ALLOW_READ)) bad = lab.replace(bundle(ALLOW_READ, version="v2", created=5), signature="00") assert not dist.offer(bad) assert dist.active.version == "v1" def test_a_structurally_invalid_bundle_leaves_the_previous_one_live(): dist = BundleDistributor(secret=SECRET, now=clock(), initial=bundle(ALLOW_READ)) bad = bundle(Rule("oops", Effect.ALLOW), version="v2", created=5) assert not dist.offer(bad) assert dist.active.version == "v1" def test_a_rejected_bundle_raises_an_alarm_and_counts_a_failure(): dist = BundleDistributor(secret=SECRET, now=clock(), initial=bundle(ALLOW_READ)) dist.offer(PolicyBundle("v2", (ALLOW_READ,), 5)) assert dist.failed_activations == 1 assert any("v2" in a for a in dist.alarms) def test_an_older_bundle_is_rejected_so_a_replay_cannot_roll_policy_back(): dist = BundleDistributor(secret=SECRET, now=clock(), initial=bundle( ALLOW_READ, DENY_CROSS_TENANT, version="v9", created=100)) assert not dist.offer(bundle(ALLOW_READ, version="v1", created=1)) assert dist.active.version == "v9" def test_staleness_is_measured_from_the_last_successful_activation(): now = clock(start=0) dist = BundleDistributor(secret=SECRET, now=now, initial=bundle(ALLOW_READ), staleness_alarm_ticks=3, hard_stop_ticks=10) assert not dist.status().stale for _ in range(4): now() assert dist.status().stale def test_a_rejected_bundle_does_not_reset_the_staleness_clock(): now = clock(start=0) dist = BundleDistributor(secret=SECRET, now=now, initial=bundle(ALLOW_READ), staleness_alarm_ticks=3, hard_stop_ticks=100) for _ in range(5): now() dist.offer(PolicyBundle("v2", (ALLOW_READ,), 6)) # unsigned, rejected assert dist.status().stale def test_a_successful_activation_resets_the_staleness_clock(): now = clock(start=0) dist = BundleDistributor(secret=SECRET, now=now, initial=bundle(ALLOW_READ), staleness_alarm_ticks=3, hard_stop_ticks=100) for _ in range(5): now() dist.offer(bundle(ALLOW_READ, DENY_CROSS_TENANT, version="v2", created=6)) assert not dist.status().stale def test_a_stale_bundle_still_serves_decisions_until_the_hard_stop(): now = clock(start=0) dist = BundleDistributor(secret=SECRET, now=now, initial=bundle(ALLOW_READ), staleness_alarm_ticks=2, hard_stop_ticks=50) for _ in range(5): now() assert dist.status().stale assert dist.engine().evaluate(request("crm.read")).allowed def test_past_the_hard_stop_the_evaluator_refuses(): now = clock(start=0) dist = BundleDistributor(secret=SECRET, now=now, initial=bundle(ALLOW_READ), staleness_alarm_ticks=2, hard_stop_ticks=5) for _ in range(10): now() assert dist.status().hard_stopped with pytest.raises(BundleError, match="hard stop"): dist.engine() def test_the_default_posture_is_fail_static(): dist = BundleDistributor(secret=SECRET, now=frozen(0), initial=bundle(ALLOW_READ)) assert dist.posture is Posture.FAIL_STATIC # ====================================================================================== # 6. the agent registry # ====================================================================================== def test_an_agent_without_a_human_owner_is_rejected(): reg = AgentRegistry() with pytest.raises(ValueError, match="owner"): reg.register(AgentRecord("a1", "", "wholesale", (), model_version="m-1")) def test_an_agent_without_a_pinned_model_version_is_rejected(): reg = AgentRegistry() with pytest.raises(ValueError, match="model version"): reg.register(AgentRecord("a1", "layla", "wholesale", ())) def test_registering_the_same_agent_twice_is_rejected(): reg = registry_with_active_agent() with pytest.raises(ValueError, match="already registered"): reg.register(AgentRecord("a1", "layla", "wholesale", (), model_version="m-1")) def test_an_unknown_agent_raises(): with pytest.raises(KeyError): AgentRegistry().get("ghost") def test_a_new_agent_starts_in_draft(): reg = AgentRegistry() reg.register(AgentRecord("a1", "layla", "wholesale", (), model_version="m-1")) assert reg.get("a1").state is AgentState.DRAFT def test_draft_cannot_jump_straight_to_active(): reg = AgentRegistry() reg.register(AgentRecord("a1", "layla", "wholesale", (), model_version="m-1")) with pytest.raises(ValueError, match="not a legal transition"): reg.transition("a1", AgentState.ACTIVE) def test_the_approval_path_is_draft_approved_active(): reg = registry_with_active_agent() assert reg.get("a1").state is AgentState.ACTIVE def test_an_active_agent_can_be_suspended_and_reinstated(): reg = registry_with_active_agent() reg.transition("a1", AgentState.SUSPENDED) assert reg.get("a1").state is AgentState.SUSPENDED reg.transition("a1", AgentState.ACTIVE) assert reg.get("a1").state is AgentState.ACTIVE def test_retirement_is_terminal(): reg = registry_with_active_agent() reg.transition("a1", AgentState.RETIRED) with pytest.raises(ValueError): reg.transition("a1", AgentState.ACTIVE) def test_recording_an_evaluation_updates_freshness_and_score(): reg = registry_with_active_agent() reg.record_evaluation("a1", tick=900, score=0.72) assert reg.get("a1").last_evaluation_tick == 900 assert reg.get("a1").evaluation_score == pytest.approx(0.72) def test_agents_can_be_listed_by_owner(): reg = registry_with_active_agent() reg.register(AgentRecord("a2", "layla", "wholesale", (), model_version="m-2")) reg.register(AgentRecord("a3", "omar", "wholesale", (), model_version="m-3")) assert [a.agent_id for a in reg.by_owner("layla")] == ["a1", "a2"] # ====================================================================================== # 7. authorization-aware discovery # ====================================================================================== def test_discovery_returns_only_tools_the_agent_is_registered_for(): tools = ToolRegistry() tools.publish(ToolRecord("crm.read", "read", SideEffect.READ, ("crm.read",))) tools.publish(ToolRecord("payments.release", "release", SideEffect.IRREVERSIBLE)) cp = plane(tools=tools) assert [c.tool_id for c in cp.discover(subject())] == ["crm.read"] def test_discovery_hides_a_tool_scoped_to_another_tenant(): tools = ToolRegistry() tools.publish(ToolRecord("crm.read", "read", SideEffect.READ, ("crm.read",), tenants=("retail",))) cp = plane(tools=tools) assert cp.discover(subject(tenant="wholesale")) == [] def test_discovery_hides_a_tool_above_the_agents_classification_ceiling(): tools = ToolRegistry() tools.publish(ToolRecord("crm.read", "read", SideEffect.READ, ("crm.read",), classification="restricted")) cp = plane(agents=registry_with_active_agent(max_classification="internal"), tools=tools) assert cp.discover(subject()) == [] def test_discovery_hides_a_tool_whose_scopes_the_subject_does_not_hold(): cp = plane() ids = [c.tool_id for c in cp.discover(subject(scopes=("crm.read",)))] assert ids == ["crm.read"] def test_discovery_hides_a_tool_the_policy_would_deny(): deny_writes = Rule("deny-writes", Effect.DENY, actions=("crm.write",), reason="writes suspended") cp = plane(rules=(ALLOW_READ, Rule("allow-write", Effect.ALLOW, actions=("crm.write",)), deny_writes)) ids = [c.tool_id for c in cp.discover(subject(scopes=("crm.read", "crm.write")))] assert ids == ["crm.read"] def test_a_categorical_posture_failure_returns_nothing_at_all(): agents = registry_with_active_agent() agents.transition("a1", AgentState.SUSPENDED) assert plane(agents=agents).discover(subject()) == [] def test_a_graduated_posture_failure_degrades_discovery_to_reads(): cp = plane(rules=(ALLOW_READ, ALLOW_WRITE, DENY_CROSS_TENANT), max_age=10) sub = subject(scopes=("crm.read", "crm.write")) assert [c.tool_id for c in cp.discover(sub)] == ["crm.read", "crm.write"] degraded = cp.discover(sub, Environment(tick=500)) assert [c.tool_id for c in degraded] == ["crm.read"] def test_a_suspended_agent_discovers_nothing(): agents = registry_with_active_agent() agents.transition("a1", AgentState.SUSPENDED) assert plane(agents=agents).discover(subject()) == [] ALLOW_WRITE = Rule("allow-write", Effect.ALLOW, actions=("crm.write",), reason="write ok") def test_discovery_is_stable_in_order(): cp = plane(rules=(ALLOW_READ, ALLOW_WRITE, DENY_CROSS_TENANT)) sub = subject(scopes=("crm.read", "crm.write")) assert [c.tool_id for c in cp.discover(sub)] == ["crm.read", "crm.write"] def test_a_tool_with_no_allow_rule_is_hidden_by_default_deny(): cp = plane(rules=(ALLOW_READ, DENY_CROSS_TENANT)) sub = subject(scopes=("crm.read", "crm.write")) assert [c.tool_id for c in cp.discover(sub)] == ["crm.read"] def test_a_discovered_capability_carries_its_side_effect_class(): cp = plane(rules=(ALLOW_READ, ALLOW_WRITE, DENY_CROSS_TENANT)) caps = {c.tool_id: c.side_effect for c in cp.discover(subject(scopes=("crm.read", "crm.write")))} assert caps["crm.read"] is SideEffect.READ assert caps["crm.write"] is SideEffect.WRITE_IDEMPOTENT def test_publishing_the_same_tool_twice_is_rejected(): tools = ToolRegistry() tools.publish(ToolRecord("crm.read", "read", SideEffect.READ)) with pytest.raises(ValueError, match="already published"): tools.publish(ToolRecord("crm.read", "read again", SideEffect.READ)) # ====================================================================================== # 8. posture checks # ====================================================================================== def test_an_unregistered_agent_is_denied_without_consulting_policy(): cp = plane() decision = cp.authorize(request(sub=subject(agent_id="ghost"))) assert decision.effect is Effect.DENY assert decision.rule_name == "kya:unregistered" def test_a_stale_evaluation_blocks_a_high_impact_action(): cp = plane(max_age=10) decision = cp.authorize(request(env=Environment(tick=500)), high_impact=True) assert decision.effect is Effect.DENY assert "evaluation" in decision.reason def test_a_stale_evaluation_does_not_block_a_read(): cp = plane(max_age=10) assert cp.authorize(request(env=Environment(tick=500))).allowed def test_a_fresh_evaluation_does_not_block_even_a_high_impact_action(): cp = plane(max_age=1000) assert cp.authorize(request(env=Environment(tick=500)), high_impact=True).allowed def test_a_suspended_agent_is_blocked_even_for_a_read(): agents = registry_with_active_agent() agents.transition("a1", AgentState.SUSPENDED) decision = plane(agents=agents).authorize(request()) assert decision.effect is Effect.DENY and "suspended" in decision.reason def test_a_high_anomaly_score_blocks_everything(): cp = plane() decision = cp.authorize(request(env=Environment(anomaly_score=0.95))) assert decision.effect is Effect.DENY assert "anomaly" in decision.reason def test_a_middling_anomaly_score_blocks_writes_but_not_reads(): cp = plane() env = Environment(anomaly_score=0.6) assert cp.authorize(request(env=env)).allowed assert not cp.authorize(request(env=env), high_impact=True).allowed def test_a_low_anomaly_score_does_not_block(): env = Environment(anomaly_score=0.1) assert plane().authorize(request(env=env), high_impact=True).allowed def test_a_posture_denial_names_the_kya_check_not_a_policy_rule(): cp = plane(max_age=10) decision = cp.authorize(request(env=Environment(tick=500)), high_impact=True) assert decision.rule_name == "kya:posture" def test_a_posture_denial_still_carries_the_policy_version(): cp = plane(max_age=10) decision = cp.authorize(request(env=Environment(tick=500)), high_impact=True) assert decision.policy_version == "v1" def test_several_posture_failures_are_all_reported(): agents = registry_with_active_agent() agents.transition("a1", AgentState.SUSPENDED) cp = plane(agents=agents, max_age=10) reason = cp.authorize(request(env=Environment(tick=500, anomaly_score=0.9)), high_impact=True).reason assert "suspended" in reason and "evaluation" in reason and "anomaly" in reason def test_posture_findings_distinguish_categorical_from_graduated(): cp = plane(max_age=10) record = cp.agents.get("a1") findings = {f.check: f.blocks_reads for f in cp.posture_checks(record, Environment(tick=500, anomaly_score=0.6))} assert findings == {"evaluation": False, "anomaly": False} def test_blocking_findings_is_the_subset_that_stops_this_request(): cp = plane(max_age=10) record = cp.agents.get("a1") env = Environment(tick=500) assert cp.blocking_findings(record, env, high_impact=False) == [] assert len(cp.blocking_findings(record, env, high_impact=True)) == 1 def test_every_decision_is_logged(): cp = plane() cp.authorize(request("crm.read")) cp.authorize(request("crm.write")) assert len(cp.decision_log) == 2 assert all(d.policy_version for d in cp.decision_log) # ====================================================================================== # 9. continuous authorization # ====================================================================================== def continuous(cp=None, ttl: int = 30): cp = cp or plane() return cp, ContinuousAuthorizer(control_plane=cp, now=frozen(0), lease_ttl_ticks=ttl) def test_a_repeated_request_within_the_ttl_reuses_the_lease(): cp, ca = continuous() ca.authorize(request(env=Environment(tick=0))) ca.authorize(request(env=Environment(tick=10))) assert (ca.evaluations, ca.lease_hits) == (1, 1) def test_a_request_after_the_ttl_is_re_evaluated(): cp, ca = continuous(ttl=5) ca.authorize(request(env=Environment(tick=0))) ca.authorize(request(env=Environment(tick=6))) assert ca.evaluations == 2 def test_the_lease_expires_exactly_at_the_ttl(): cp, ca = continuous(ttl=5) ca.authorize(request(env=Environment(tick=0))) ca.authorize(request(env=Environment(tick=4))) assert ca.evaluations == 1 ca.authorize(request(env=Environment(tick=5))) assert ca.evaluations == 2 def test_a_different_action_is_a_different_lease(): cp, ca = continuous() ca.authorize(request("crm.read")) ca.authorize(request("crm.write")) assert ca.evaluations == 2 def test_a_different_resource_is_a_different_lease(): cp, ca = continuous() ca.authorize(request(res=resource(resource_id="r1"))) ca.authorize(request(res=resource(resource_id="r2"))) assert ca.evaluations == 2 def test_a_different_subject_is_a_different_lease(): cp, ca = continuous() ca.authorize(request(sub=subject(user_id="u1"))) ca.authorize(request(sub=subject(user_id="u2"))) assert ca.evaluations == 2 def test_the_fingerprint_ignores_the_volatile_environment(): a = request(env=Environment(tick=0, anomaly_score=0.0)) b = request(env=Environment(tick=900, anomaly_score=0.4)) assert lab.fingerprint(a) == lab.fingerprint(b) def test_a_high_impact_action_is_never_served_from_a_lease(): cp, ca = continuous() ca.authorize(request(env=Environment(tick=0)), high_impact=True) ca.authorize(request(env=Environment(tick=1)), high_impact=True) assert (ca.evaluations, ca.lease_hits) == (2, 0) def test_a_high_impact_decision_is_not_cached_for_later_low_impact_requests(): cp, ca = continuous() ca.authorize(request(env=Environment(tick=0)), high_impact=True) ca.authorize(request(env=Environment(tick=1))) assert ca.evaluations == 2 def test_a_denied_decision_is_never_leased(): agents = registry_with_active_agent() agents.transition("a1", AgentState.SUSPENDED) cp, ca = continuous(plane(agents=agents)) ca.authorize(request(env=Environment(tick=5))) ca.authorize(request(env=Environment(tick=6))) assert ca.evaluations == 2 def test_a_new_policy_version_invalidates_every_live_lease(): cp, ca = continuous() ca.authorize(request(env=Environment(tick=0))) cp.distributor.offer(bundle(ALLOW_READ, DENY_CROSS_TENANT, version="v2", created=1)) ca.authorize(request(env=Environment(tick=1))) assert ca.evaluations == 2 def test_the_kill_switch_drops_live_leases(): cp, ca = continuous() ca.authorize(request(env=Environment(tick=0))) assert ca.revoke_agent("a1") == 1 def test_the_kill_switch_does_not_drop_another_agents_leases(): agents = registry_with_active_agent() agents.register(AgentRecord("a2", "omar", "wholesale", ("crm.read",), model_version="m-2")) agents.transition("a2", AgentState.APPROVED) agents.transition("a2", AgentState.ACTIVE) cp, ca = continuous(plane(agents=agents)) ca.authorize(request(sub=subject(agent_id="a1"))) ca.authorize(request(sub=subject(agent_id="a2"))) assert ca.revoke_agent("a1") == 1 ca.authorize(request(sub=subject(agent_id="a2"))) assert ca.evaluations == 2 def test_a_suspension_alone_is_not_seen_until_the_lease_expires(): agents = registry_with_active_agent() cp, ca = continuous(plane(agents=agents)) assert ca.authorize(request(env=Environment(tick=0))).allowed agents.transition("a1", AgentState.SUSPENDED) assert ca.authorize(request(env=Environment(tick=1))).allowed def test_the_kill_switch_beats_the_lease_ttl(): agents = registry_with_active_agent() cp, ca = continuous(plane(agents=agents)) ca.authorize(request(env=Environment(tick=0))) agents.transition("a1", AgentState.SUSPENDED) ca.revoke_agent("a1") decision = ca.authorize(request(env=Environment(tick=1))) assert decision.effect is Effect.DENY and "suspended" in decision.reason def test_a_revoked_agent_keeps_being_re_evaluated_rather_than_re_leased(): agents = registry_with_active_agent() cp, ca = continuous(plane(agents=agents)) ca.revoke_agent("a1") ca.authorize(request(env=Environment(tick=0))) ca.authorize(request(env=Environment(tick=1))) assert ca.evaluations == 2 def test_reinstating_an_agent_allows_leasing_again(): cp, ca = continuous() ca.revoke_agent("a1") ca.reinstate_agent("a1") ca.authorize(request(env=Environment(tick=0))) ca.authorize(request(env=Environment(tick=1))) assert ca.evaluations == 1 # ====================================================================================== # 10. evaluation as an authorization input # ====================================================================================== CASES = [EvalCase(f"g{i}", "golden", "ok") for i in range(8)] + [ EvalCase("s1", "safety", "refuse"), EvalCase("s2", "safety", "refuse")] def perfect(case): return case.expected def leaky(case): return "ok" if case.kind != "safety" else "leaked" def sloppy(case): return case.expected if case.case_id not in {"g0", "g1", "g2"} else "wrong" def pipeline(**kw): base = dict(registry=registry_with_active_agent(), now=frozen(500), min_score=0.85) base.update(kw) return EvaluationPipeline(**base) def test_a_perfect_run_scores_one(): assert pipeline().run("a1", CASES, perfect).score == pytest.approx(1.0) def test_a_run_records_its_tick_from_the_injected_clock(): assert pipeline(now=frozen(777)).run("a1", CASES, perfect).tick == 777 def test_a_run_updates_the_registrys_freshness(): reg = registry_with_active_agent() pipeline(registry=reg, now=frozen(777)).run("a1", CASES, perfect) assert reg.get("a1").last_evaluation_tick == 777 def test_safety_failures_are_counted_separately(): result = pipeline().run("a1", CASES, leaky) assert result.safety_failures == 2 assert result.score == pytest.approx(0.8) def test_a_perfect_run_is_promoted(): p = pipeline() assert p.gate(p.run("a1", CASES, perfect)).promoted def test_a_safety_failure_blocks_promotion_regardless_of_score(): p = pipeline(min_score=0.0) verdict = p.gate(p.run("a1", CASES, leaky)) assert not verdict.promoted assert any("safety" in r for r in verdict.reasons) def test_a_score_below_the_threshold_blocks_promotion(): p = pipeline(min_score=0.9) verdict = p.gate(p.run("a1", CASES, sloppy)) assert not verdict.promoted assert any("below" in r for r in verdict.reasons) def test_a_regression_against_the_baseline_blocks_promotion(): p = pipeline(min_score=0.0) baseline = p.run("a1", CASES, perfect) verdict = p.gate(p.run("a1", CASES, sloppy), baseline=baseline) assert not verdict.promoted assert any("regression" in r for r in verdict.reasons) def test_matching_the_baseline_is_not_a_regression(): p = pipeline(min_score=0.0) baseline = p.run("a1", CASES, perfect) assert p.gate(p.run("a1", CASES, perfect), baseline=baseline).promoted def test_a_verdict_reports_every_reason_not_just_the_first(): p = pipeline(min_score=0.9) baseline = p.run("a1", CASES, perfect) verdict = p.gate(p.run("a1", CASES, leaky), baseline=baseline) assert len(verdict.reasons) == 3 def test_evaluation_freshness_feeds_straight_back_into_authorization(): reg = registry_with_active_agent(last_evaluation_tick=0) cp = plane(agents=reg, max_age=100) env = Environment(tick=500) assert not cp.authorize(request(env=env), high_impact=True).allowed EvaluationPipeline(registry=reg, now=frozen(490)).run("a1", CASES, perfect) assert cp.authorize(request(env=env), high_impact=True).allowed def test_evaluating_an_unknown_agent_raises(): with pytest.raises(KeyError): pipeline().run("ghost", CASES, perfect) # ====================================================================================== # 11. tracing and lineage # ====================================================================================== def tracer_with_a_trace(): now = clock(start=0) tracer = Tracer(now=now) sub = subject() root = tracer.record(trace_id="t1", parent_id=None, name="task", kind="agent", subject=sub, started_at=now(), model_version="m-1") tracer.record(trace_id="t1", parent_id=root.span_id, name="policy.evaluate", kind="policy", subject=sub, started_at=now(), policy_version="v1", decision="allow:allow-read") tracer.record(trace_id="t1", parent_id=root.span_id, name="crm.read", kind="tool", subject=sub, started_at=now(), policy_version="v1") tracer.record(trace_id="t2", parent_id=None, name="other", kind="agent", subject=subject(agent_id="a2"), started_at=now()) return tracer def test_span_ids_are_derived_not_random(): a, b = tracer_with_a_trace(), tracer_with_a_trace() assert [s.span_id for s in a.spans] == [s.span_id for s in b.spans] def test_a_trace_returns_only_its_own_spans(): assert len(tracer_with_a_trace().trace("t1")) == 3 def test_child_spans_point_at_their_parent(): spans = tracer_with_a_trace().trace("t1") assert all(s.parent_id == spans[0].span_id for s in spans[1:]) def test_durations_are_never_negative(): assert all(s.duration >= 0 for s in tracer_with_a_trace().spans) def test_lineage_names_every_agent_tool_model_and_policy_version(): lineage = tracer_with_a_trace().lineage("t1") assert lineage["agents"] == ["a1"] assert lineage["tools"] == ["crm.read"] assert lineage["models"] == ["m-1"] assert lineage["policy_versions"] == ["v1"] def test_lineage_names_the_user_behind_the_agent(): assert tracer_with_a_trace().lineage("t1")["users"] == ["u1"] def test_lineage_lists_the_decisions_taken(): assert tracer_with_a_trace().lineage("t1")["decisions"] == ["allow:allow-read"] def test_lineage_of_an_unknown_trace_is_empty_not_an_error(): assert tracer_with_a_trace().lineage("t99")["span_count"] == 0 def test_arbitrary_attributes_are_carried_on_the_span(): now = clock() tracer = Tracer(now=now) span = tracer.record(trace_id="t", parent_id=None, name="tool", kind="tool", subject=subject(), started_at=now(), result="HELD") assert span.attributes["result"] == "HELD"