"""Tests for Issue #689: Deterministic MCP namespace attachment. Verifies cohort identity exposure, stale cohort refusal, parity matching, reconcile_id freshness, session context cleanup, and regression scenarios. """ from __future__ import annotations import os import unittest from unittest.mock import patch import master_parity_gate import mcp_namespace_health import post_restart_reconcile import session_context_binding as session_ctx class TestIssue689DeterministicCohortAttachment(unittest.TestCase): """Suite covering Issue #689 acceptance criteria.""" def setUp(self) -> None: session_ctx._reset_session_context_for_testing() def tearDown(self) -> None: session_ctx._reset_session_context_for_testing() def test_ac1_session_context_exposes_cohort_identity(self) -> None: """AC1: Session context exposes cohort ID, startup SHA, endpoint, and config fingerprint.""" ctx = session_ctx.bind_session_context( profile_name="prgs-author", remote="prgs", host="gitea.prgs.cc", identity="jcwalker3", repository="Gitea-Tools", org="Scaled-Tech-Consulting", role_kind="author", cohort_id="cohort-p1234-abc123456789", startup_sha="abc123456789def", endpoint="gitea.prgs.cc", config_fingerprint="fingerprint12345", ) self.assertEqual(ctx["cohort_id"], "cohort-p1234-abc123456789") self.assertEqual(ctx["startup_sha"], "abc123456789def") self.assertEqual(ctx["endpoint"], "gitea.prgs.cc") self.assertEqual(ctx["config_fingerprint"], "fingerprint12345") fetched = session_ctx.get_session_context() self.assertIsNotNone(fetched) self.assertEqual(fetched["cohort_id"], "cohort-p1234-abc123456789") self.assertEqual(fetched["startup_sha"], "abc123456789def") def test_ac2_stale_cohort_refused_by_probe_classifier(self) -> None: """AC2: Probe classifier refuses binding to a stale cohort as stale_cohort_refused.""" probe_res = { "success": True, "cohort": { "cohort_id": "cohort-obsolete-1", "startup_sha": "22698c1000000000000000000000000000000000", "endpoint": "gitea.prgs.cc", "config_fingerprint": "fp-old", }, } res = mcp_namespace_health.classify_namespace_probe( "gitea-author", probe_result=probe_res, probe_source="client_namespace", expected_parity_sha="a4c73766f4b0cc32f7c3808688eceeb6fee74335", ) self.assertFalse(res["healthy"]) self.assertFalse(res["success"]) self.assertEqual(res["error_type"], "stale_cohort_refused") self.assertIn("stale cohort refused", " ".join(res["reasons"])) self.assertEqual( res["diagnostics"]["startup_sha"], "22698c1000000000000000000000000000000000", ) def test_ac3_reconnection_parity_matching_and_fail_closed(self) -> None: """AC3: Parity gate fails closed when bound cohort startup SHA mismatches parity.""" startup = {"startup_head": "a4c73766f4b0cc32f7c3808688eceeb6fee74335"} current = "a4c73766f4b0cc32f7c3808688eceeb6fee74335" live_remote = "a4c73766f4b0cc32f7c3808688eceeb6fee74335" # Matching cohort matching_cohort = { "cohort_id": "cohort-fresh", "startup_sha": "a4c73766f4b0cc32f7c3808688eceeb6fee74335", } res_matching = master_parity_gate.assess_master_parity( startup, current, live_remote_head=live_remote, bound_cohort=matching_cohort ) self.assertTrue(res_matching["cohort_parity_match"]) self.assertFalse(res_matching["cohort_stale"]) self.assertTrue(res_matching["mutation_safe"]) # Mismatched obsolete cohort obsolete_cohort = { "cohort_id": "cohort-obsolete-22698c1", "startup_sha": "22698c1000000000000000000000000000000000", } res_stale = master_parity_gate.assess_master_parity( startup, current, live_remote_head=live_remote, bound_cohort=obsolete_cohort ) self.assertFalse(res_stale["cohort_parity_match"]) self.assertTrue(res_stale["cohort_stale"]) self.assertTrue(res_stale["restart_required"]) self.assertFalse(res_stale["mutation_safe"]) def test_ac4_reconcile_id_freshness(self) -> None: """AC4: Re-attachment distinguishes new reconcile_id from preserved binding.""" inventory = {"inventory_complete": True} # New attachment generates fresh reconcile_id proof1 = post_restart_reconcile.reconcile_after_restart(inventory) self.assertFalse(proof1.binding_unchanged) self.assertTrue(proof1.reconcile_id.startswith("reconcile-")) # Preserved binding reports binding_unchanged=True proof2 = post_restart_reconcile.reconcile_after_restart( inventory, reconcile_id=proof1.reconcile_id, prior_reconcile_id=proof1.reconcile_id, ) self.assertTrue(proof2.binding_unchanged) self.assertEqual(proof2.reconcile_id, proof1.reconcile_id) # Disconnected re-attachment gets new reconcile_id proof3 = post_restart_reconcile.reconcile_after_restart( inventory, prior_reconcile_id=proof1.reconcile_id, ) self.assertFalse(proof3.binding_unchanged) self.assertNotEqual(proof3.reconcile_id, proof1.reconcile_id) def test_ac5_session_disconnect_clears_bindings(self) -> None: """AC5: clear_session_context purges session context on disconnect.""" session_ctx.bind_session_context( profile_name="prgs-author", remote="prgs", host="gitea.prgs.cc", identity="jcwalker3", cohort_id="cohort-1", ) self.assertIsNotNone(session_ctx.get_session_context()) session_ctx.clear_session_context() self.assertIsNone(session_ctx.get_session_context()) def test_ac6_bound_cohort_in_diagnostics(self) -> None: """AC6: Bound cohort identity appears in audit diagnostics.""" session_ctx.bind_session_context( profile_name="prgs-author", remote="prgs", host="gitea.prgs.cc", identity="jcwalker3", cohort_id="cohort-test-99", startup_sha="sha99999", endpoint="gitea.prgs.cc", config_fingerprint="fp999", ) audit = session_ctx.mutation_context_audit_fields() self.assertTrue(audit["session_context_bound"]) self.assertEqual(audit["session_cohort_id"], "cohort-test-99") self.assertEqual(audit["session_startup_sha"], "sha99999") self.assertEqual(audit["session_endpoint"], "gitea.prgs.cc") self.assertEqual(audit["session_config_fingerprint"], "fp999") def test_ac7_regression_n_reconnects_never_bind_to_obsolete_daemon(self) -> None: """AC7: N reconnects against a daemon set containing obsolete daemons never bind obsolete ones.""" live_master = "master-head-latest-12345" daemons = [ {"id": "d1", "startup_sha": "obsolete-head-11111"}, {"id": "d2", "startup_sha": "obsolete-head-22698c1"}, {"id": "d3", "startup_sha": live_master}, {"id": "d4", "startup_sha": "obsolete-head-33333"}, ] for _ in range(5): for daemon in daemons: res = master_parity_gate.assess_master_parity( {"startup_head": live_master}, live_master, live_remote_head=live_master, bound_cohort=daemon, ) if daemon["startup_sha"] != live_master: self.assertFalse(res["mutation_safe"]) self.assertTrue(res["cohort_stale"]) else: self.assertTrue(res["mutation_safe"]) self.assertFalse(res["cohort_stale"]) def test_ac8_regression_incident_shape_reproduction(self) -> None: """AC8: Reproduce incident shape — obsolete cohort 22698c1 resident vs newer daemon.""" live_master = "a4c73766f4b0cc32f7c3808688eceeb6fee74335" obsolete_cohort = { "cohort_id": "cohort-resident-22698c1", "startup_sha": "22698c1000000000000000000000000000000000", } new_cohort = { "cohort_id": "cohort-spawned-new", "startup_sha": live_master, } # Obsolete cohort fails parity check obs_res = mcp_namespace_health.classify_namespace_probe( "gitea-author", probe_result={"success": True, "cohort": obsolete_cohort}, probe_source="client_namespace", expected_parity_sha=live_master, ) self.assertFalse(obs_res["healthy"]) self.assertEqual(obs_res["error_type"], "stale_cohort_refused") # Fresh cohort succeeds new_res = mcp_namespace_health.classify_namespace_probe( "gitea-author", probe_result={"success": True, "cohort": new_cohort}, probe_source="client_namespace", expected_parity_sha=live_master, ) self.assertTrue(new_res["healthy"]) def test_ac9_regression_bound_cohort_going_stale_detected(self) -> None: """AC9: A bound cohort that later goes stale is detected on next attachment check.""" initial_master = "sha-v1-initial" cohort = {"cohort_id": "c1", "startup_sha": initial_master} # Initial state: in parity res1 = master_parity_gate.assess_master_parity( {"startup_head": initial_master}, initial_master, live_remote_head=initial_master, bound_cohort=cohort, ) self.assertTrue(res1["mutation_safe"]) # Master advances to sha-v2-advanced while cohort remains at sha-v1-initial advanced_master = "sha-v2-advanced" res2 = master_parity_gate.assess_master_parity( {"startup_head": initial_master}, advanced_master, live_remote_head=advanced_master, bound_cohort=cohort, ) self.assertFalse(res2["mutation_safe"]) self.assertTrue(res2["restart_required"]) self.assertTrue(res2["cohort_stale"]) if __name__ == "__main__": unittest.main()