"""Unit tests for src/my_deepagent/binding.py.""" from __future__ import annotations import fcntl import json import re from pathlib import Path import pytest from my_deepagent.binding import ( BackendAvailability, Binding, BindingOverride, PersonaConsentStore, bind_personas, filter_consented_personas, is_persona_eligible_for_role, ) from my_deepagent.enums import Backend, Capability from my_deepagent.errors import MyDeepAgentError from my_deepagent.persona import Persona, load_personas_from_dir from my_deepagent.workflow import WorkflowTemplate, load_workflows_from_dir # --------------------------------------------------------------------------- # PersonaConsentStore file-lock (fcntl.flock) verification # --------------------------------------------------------------------------- def test_consent_store_set_acquires_exclusive_lock( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: """set() must take an exclusive flock and release it.""" ops: list[int] = [] orig_flock = fcntl.flock def spy(fd: int, op: int) -> None: ops.append(op) orig_flock(fd, op) monkeypatch.setattr(fcntl, "flock", spy) store = PersonaConsentStore(tmp_path / "consents.json") store.set("hash_abc", "approve") assert fcntl.LOCK_EX in ops assert fcntl.LOCK_UN in ops def test_consent_store_revoke_acquires_exclusive_lock( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: ops: list[int] = [] orig_flock = fcntl.flock def spy(fd: int, op: int) -> None: ops.append(op) orig_flock(fd, op) monkeypatch.setattr(fcntl, "flock", spy) store = PersonaConsentStore(tmp_path / "consents.json") store.set("h", "approve") ops.clear() store.revoke("h") assert fcntl.LOCK_EX in ops assert fcntl.LOCK_UN in ops def test_consent_store_get_acquires_shared_lock( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: """get() takes a shared lock (LOCK_SH) so multiple readers don't serialise.""" ops: list[int] = [] orig_flock = fcntl.flock def spy(fd: int, op: int) -> None: ops.append(op) orig_flock(fd, op) monkeypatch.setattr(fcntl, "flock", spy) store = PersonaConsentStore(tmp_path / "consents.json") store.set("h", "approve") ops.clear() _ = store.get("h") assert fcntl.LOCK_SH in ops assert fcntl.LOCK_UN in ops def test_consent_store_lock_file_created(tmp_path: Path) -> None: """A .lock sidecar file is created next to the consent store on first write.""" path = tmp_path / "consents.json" store = PersonaConsentStore(path) store.set("h", "approve") assert (tmp_path / "consents.json.lock").is_file() # --------------------------------------------------------------------------- # Fixtures / helpers # --------------------------------------------------------------------------- PERSONAS_DIR = Path(__file__).parent.parent.parent / "docs" / "schemas" / "personas" WORKFLOWS_DIR = Path(__file__).parent.parent.parent / "docs" / "schemas" / "workflows" def _minimal_persona(**overrides: object) -> Persona: base: dict[str, object] = { "name": "test-persona", "version": 1, "backend": "openrouter", "model": "openrouter:anthropic/claude-sonnet-4-6", "provider_origin": "US/Anthropic", "capabilities": ["spec_write", "phase_planning"], "max_risk_level": "low", "system_prompt": "You are a test persona for unit tests.", } base.update(overrides) return Persona.model_validate(base) def _all_available() -> BackendAvailability: return BackendAvailability(available_backends=frozenset(Backend)) def _none_available() -> BackendAvailability: return BackendAvailability(available_backends=frozenset()) @pytest.fixture() def consent_store(tmp_path: Path) -> PersonaConsentStore: return PersonaConsentStore(tmp_path / "consents.json") @pytest.fixture() def seed_personas() -> list[Persona]: return load_personas_from_dir(PERSONAS_DIR) @pytest.fixture() def spec_and_review() -> WorkflowTemplate: workflows = load_workflows_from_dir(WORKFLOWS_DIR) return next(w for w in workflows if w.name == "spec-and-review") # --------------------------------------------------------------------------- # is_persona_eligible_for_role # --------------------------------------------------------------------------- def test_eligible_all_ok(spec_and_review: WorkflowTemplate) -> None: spec_writer_role = next(r for r in spec_and_review.roles if r.id == "spec_writer") p = _minimal_persona(capabilities=["spec_write", "phase_planning"], max_risk_level="low") ok, reason = is_persona_eligible_for_role(p, spec_writer_role, spec_and_review) assert ok is True assert reason is None def test_eligible_missing_capability(spec_and_review: WorkflowTemplate) -> None: spec_writer_role = next(r for r in spec_and_review.roles if r.id == "spec_writer") # only spec_write, missing phase_planning p = _minimal_persona(capabilities=["spec_write"], max_risk_level="low") ok, reason = is_persona_eligible_for_role(p, spec_writer_role, spec_and_review) assert ok is False assert reason is not None assert "phase_planning" in reason def test_eligible_allowed_roles_mismatch(spec_and_review: WorkflowTemplate) -> None: spec_writer_role = next(r for r in spec_and_review.roles if r.id == "spec_writer") p = _minimal_persona( capabilities=["spec_write", "phase_planning"], max_risk_level="low", allowed_roles=["reviewer"], # does not include spec_writer ) ok, reason = is_persona_eligible_for_role(p, spec_writer_role, spec_and_review) assert ok is False assert reason is not None assert "allowed_roles" in reason def test_eligible_allowed_roles_matches(spec_and_review: WorkflowTemplate) -> None: spec_writer_role = next(r for r in spec_and_review.roles if r.id == "spec_writer") p = _minimal_persona( capabilities=["spec_write", "phase_planning"], max_risk_level="low", allowed_roles=["spec_writer"], ) ok, reason = is_persona_eligible_for_role(p, spec_writer_role, spec_and_review) assert ok is True assert reason is None def test_eligible_risk_too_high(spec_and_review: WorkflowTemplate) -> None: """bug-fix workflow has a 'medium' risk phase; a low-only persona is ineligible for it.""" bug_fix = load_workflows_from_dir(WORKFLOWS_DIR) bug_fix_wf = next(w for w in bug_fix if w.name == "bug-fix-with-reproduction") fixer_role = next(r for r in bug_fix_wf.roles if r.id == "fixer") # fixer role has a 'medium' risk phase p = _minimal_persona( capabilities=["code_edit", "test_first_development"], max_risk_level="low", # too low for medium phase ) ok, reason = is_persona_eligible_for_role(p, fixer_role, bug_fix_wf) assert ok is False assert reason is not None assert "medium" in reason def test_eligible_risk_exact_match(spec_and_review: WorkflowTemplate) -> None: spec_writer_role = next(r for r in spec_and_review.roles if r.id == "spec_writer") p = _minimal_persona(capabilities=["spec_write", "phase_planning"], max_risk_level="low") ok, _ = is_persona_eligible_for_role(p, spec_writer_role, spec_and_review) assert ok is True # --------------------------------------------------------------------------- # bind_personas: end-to-end with seed data # --------------------------------------------------------------------------- def test_bind_personas_spec_and_review_success( seed_personas: list[Persona], spec_and_review: WorkflowTemplate, consent_store: PersonaConsentStore, ) -> None: bindings = bind_personas(spec_and_review, seed_personas, _all_available(), consent_store) assert set(bindings.keys()) == {"spec_writer", "reviewer", "verifier"} for role_id, binding in bindings.items(): assert isinstance(binding, Binding) assert binding.role_id == role_id assert re.fullmatch(r"[0-9a-f]{64}", binding.binding_hash) def test_bind_personas_binding_hash_deterministic( seed_personas: list[Persona], spec_and_review: WorkflowTemplate, consent_store: PersonaConsentStore, ) -> None: b1 = bind_personas(spec_and_review, seed_personas, _all_available(), consent_store) b2 = bind_personas(spec_and_review, seed_personas, _all_available(), consent_store) for role_id in b1: assert b1[role_id].binding_hash == b2[role_id].binding_hash def test_bind_personas_spec_writer_is_spec_writer( seed_personas: list[Persona], spec_and_review: WorkflowTemplate, consent_store: PersonaConsentStore, ) -> None: bindings = bind_personas(spec_and_review, seed_personas, _all_available(), consent_store) spec_persona = bindings["spec_writer"].persona assert Capability.SPEC_WRITE in spec_persona.capabilities assert Capability.PHASE_PLANNING in spec_persona.capabilities # --------------------------------------------------------------------------- # bind_personas: override # --------------------------------------------------------------------------- def test_bind_personas_override_picks_pinned( seed_personas: list[Persona], spec_and_review: WorkflowTemplate, consent_store: PersonaConsentStore, ) -> None: override = BindingOverride.parse({"spec_writer": "openrouter-claude-spec-writer@1"}) bindings = bind_personas( spec_and_review, seed_personas, _all_available(), consent_store, override ) assert bindings["spec_writer"].persona.name == "openrouter-claude-spec-writer" def test_bind_personas_override_invalid_persona_raises( seed_personas: list[Persona], spec_and_review: WorkflowTemplate, consent_store: PersonaConsentStore, ) -> None: override = BindingOverride.parse({"spec_writer": "nonexistent-persona@1"}) with pytest.raises(MyDeepAgentError) as exc_info: bind_personas(spec_and_review, seed_personas, _all_available(), consent_store, override) assert exc_info.value.code == "no_eligible_persona" # --------------------------------------------------------------------------- # bind_personas: backend unavailable # --------------------------------------------------------------------------- def test_bind_personas_backend_unavailable_raises( seed_personas: list[Persona], spec_and_review: WorkflowTemplate, consent_store: PersonaConsentStore, ) -> None: with pytest.raises(MyDeepAgentError) as exc_info: bind_personas(spec_and_review, seed_personas, _none_available(), consent_store) assert exc_info.value.code == "backend_unavailable" # --------------------------------------------------------------------------- # bind_personas: model_unavailable for openrouter with empty model # --------------------------------------------------------------------------- def test_bind_personas_model_unavailable_raises( spec_and_review: WorkflowTemplate, consent_store: PersonaConsentStore, ) -> None: """Verify FAKE backend binds successfully (positive path for non-openrouter backends). We cannot construct an openrouter persona with empty model via model_validate because the validator rejects it. Instead verify the happy path: FAKE backend + non-empty model should bind without errors when the FAKE backend is available. """ from my_deepagent.workflow import WorkflowPhase, WorkflowRole role = WorkflowRole.model_validate( { "id": "spec_writer", "required_capabilities": ["spec_write", "phase_planning"], "preferred_backends": ["fake"], } ) phase = WorkflowPhase.model_validate( { "key": "spec", "title": "Write spec", "risk": "low", "role": "spec_writer", "instructions": "Write the specification document.", } ) tmpl = WorkflowTemplate.model_validate( { "name": "fake-wf", "version": 1, "roles": [role.model_dump()], "phases": [phase.model_dump()], } ) fake_persona = _minimal_persona( backend="fake", model="fake-model", capabilities=["spec_write", "phase_planning"], ) fake_avail = BackendAvailability(available_backends=frozenset({Backend.FAKE})) # Should succeed with FAKE backend + non-empty model bindings = bind_personas(tmpl, [fake_persona], fake_avail, consent_store) assert "spec_writer" in bindings # --------------------------------------------------------------------------- # bind_personas: no eligible persona # --------------------------------------------------------------------------- def test_bind_personas_no_eligible_raises( spec_and_review: WorkflowTemplate, consent_store: PersonaConsentStore, ) -> None: # Provide a persona with wrong capabilities bad_persona = _minimal_persona(capabilities=["backtest_run"]) with pytest.raises(MyDeepAgentError) as exc_info: bind_personas(spec_and_review, [bad_persona], _all_available(), consent_store) assert exc_info.value.code == "no_eligible_persona" # --------------------------------------------------------------------------- # PersonaConsentStore: get / set / revoke # --------------------------------------------------------------------------- def test_consent_store_get_none_when_absent(consent_store: PersonaConsentStore) -> None: assert consent_store.get("abc123") is None def test_consent_store_set_and_get(consent_store: PersonaConsentStore) -> None: consent_store.set("abc123", "approve") assert consent_store.get("abc123") == "approve" def test_consent_store_block(consent_store: PersonaConsentStore) -> None: consent_store.set("abc123", "block") assert consent_store.get("abc123") == "block" def test_consent_store_once(consent_store: PersonaConsentStore) -> None: consent_store.set("abc123", "once") assert consent_store.get("abc123") == "once" def test_consent_store_revoke(consent_store: PersonaConsentStore) -> None: consent_store.set("abc123", "approve") consent_store.revoke("abc123") assert consent_store.get("abc123") is None def test_consent_store_revoke_absent_is_noop(consent_store: PersonaConsentStore) -> None: consent_store.revoke("not_present") # must not raise def test_consent_store_overwrite(consent_store: PersonaConsentStore) -> None: consent_store.set("abc123", "approve") consent_store.set("abc123", "block") assert consent_store.get("abc123") == "block" def test_consent_store_unknown_decision_returns_none( consent_store: PersonaConsentStore, tmp_path: Path, ) -> None: """Corrupt decision value (not approve/block/once) returns None, not raise.""" path = tmp_path / "consents.json" path.write_text( json.dumps({"abc123": {"decision": "foobar", "decided_at": "2026-01-01T00:00:00+00:00"}}), encoding="utf-8", ) store = PersonaConsentStore(path) assert store.get("abc123") is None def test_consent_store_corrupted_json_raises_fatal(tmp_path: Path) -> None: path = tmp_path / "consents.json" path.write_text("{invalid json", encoding="utf-8") store = PersonaConsentStore(path) with pytest.raises(MyDeepAgentError) as exc_info: store.get("abc123") assert exc_info.value.code == "internal_state_corruption" def test_consent_store_atomic_write(consent_store: PersonaConsentStore) -> None: """The .tmp file must not remain after a successful write.""" consent_store.set("abc", "approve") tmp_file = consent_store._path.with_suffix(".json.tmp") assert not tmp_file.exists(), ".tmp leftover after successful write" def test_consent_store_json_format(consent_store: PersonaConsentStore) -> None: """Stored JSON must be valid and contain decision + decided_at.""" consent_store.set("myhash", "once") raw = consent_store._path.read_text(encoding="utf-8") data = json.loads(raw) assert "myhash" in data assert data["myhash"]["decision"] == "once" assert "decided_at" in data["myhash"] # --------------------------------------------------------------------------- # filter_consented_personas # --------------------------------------------------------------------------- def test_filter_removes_blocked(consent_store: PersonaConsentStore) -> None: p1 = _minimal_persona(name="p1") p2 = _minimal_persona(name="p2") consent_store.set(p2.compute_hash(), "block") result = filter_consented_personas([p1, p2], consent_store) assert len(result) == 1 assert result[0].name == "p1" def test_filter_keeps_approved(consent_store: PersonaConsentStore) -> None: p = _minimal_persona() consent_store.set(p.compute_hash(), "approve") result = filter_consented_personas([p], consent_store) assert len(result) == 1 def test_filter_keeps_once(consent_store: PersonaConsentStore) -> None: p = _minimal_persona() consent_store.set(p.compute_hash(), "once") result = filter_consented_personas([p], consent_store) assert len(result) == 1 def test_filter_keeps_none_decision(consent_store: PersonaConsentStore) -> None: """Persona with no stored decision passes through.""" p = _minimal_persona() result = filter_consented_personas([p], consent_store) assert len(result) == 1 def test_filter_empty_list(consent_store: PersonaConsentStore) -> None: result = filter_consented_personas([], consent_store) assert result == [] # --------------------------------------------------------------------------- # bind_personas: consent-blocked persona detection # --------------------------------------------------------------------------- def test_bind_personas_all_eligible_blocked_raises( seed_personas: list[Persona], spec_and_review: WorkflowTemplate, consent_store: PersonaConsentStore, ) -> None: # Block all spec_writer-eligible personas for p in seed_personas: if Capability.SPEC_WRITE in p.capabilities and Capability.PHASE_PLANNING in p.capabilities: consent_store.set(p.compute_hash(), "block") with pytest.raises(MyDeepAgentError) as exc_info: bind_personas(spec_and_review, seed_personas, _all_available(), consent_store) assert exc_info.value.code in ("persona_blocked_by_user", "no_eligible_persona") def test_bind_personas_override_blocked_raises( seed_personas: list[Persona], spec_and_review: WorkflowTemplate, consent_store: PersonaConsentStore, ) -> None: spec_writer = next(p for p in seed_personas if p.name == "openrouter-claude-spec-writer") consent_store.set(spec_writer.compute_hash(), "block") override = BindingOverride.parse({"spec_writer": "openrouter-claude-spec-writer@1"}) with pytest.raises(MyDeepAgentError) as exc_info: bind_personas(spec_and_review, seed_personas, _all_available(), consent_store, override) assert exc_info.value.code == "persona_blocked_by_user" # --------------------------------------------------------------------------- # _auto_select: preferred_backends order # --------------------------------------------------------------------------- def test_auto_select_prefers_preferred_backend(spec_and_review: WorkflowTemplate) -> None: """Persona with preferred backend wins over non-preferred even if alphabetically later.""" from my_deepagent.binding import _auto_select spec_writer_role = next(r for r in spec_and_review.roles if r.id == "spec_writer") # preferred_backends = ["openrouter"] p_openrouter = _minimal_persona( name="z-openrouter-persona", backend="openrouter", capabilities=["spec_write", "phase_planning"], ) p_fake = _minimal_persona( name="a-fake-persona", backend="fake", capabilities=["spec_write", "phase_planning"], ) chosen = _auto_select([p_openrouter, p_fake], spec_writer_role) assert chosen.name == "z-openrouter-persona" def test_auto_select_higher_version_wins(spec_and_review: WorkflowTemplate) -> None: from my_deepagent.binding import _auto_select spec_writer_role = next(r for r in spec_and_review.roles if r.id == "spec_writer") p_v1 = _minimal_persona(version=1, capabilities=["spec_write", "phase_planning"]) p_v2 = _minimal_persona(version=2, capabilities=["spec_write", "phase_planning"]) chosen = _auto_select([p_v1, p_v2], spec_writer_role) assert chosen.version == 2 def test_auto_select_name_asc_tiebreak(spec_and_review: WorkflowTemplate) -> None: from my_deepagent.binding import _auto_select spec_writer_role = next(r for r in spec_and_review.roles if r.id == "spec_writer") caps = ["spec_write", "phase_planning"] p_b = _minimal_persona(name="b-persona", version=1, capabilities=caps) p_a = _minimal_persona(name="a-persona", version=1, capabilities=caps) chosen = _auto_select([p_b, p_a], spec_writer_role) assert chosen.name == "a-persona" # --------------------------------------------------------------------------- # Step 2 patch: FAKE backend recovery hint # --------------------------------------------------------------------------- def test_backend_recovery_hint_fake() -> None: """FAKE backend recovery hint must mention 'fake' and 'tests only'.""" from my_deepagent.binding import _backend_recovery_hint hint = _backend_recovery_hint(Backend.FAKE) assert "fake" in hint.lower() assert "tests only" in hint.lower() or "test harness" in hint.lower() # --------------------------------------------------------------------------- # Step 2 patch: override with non-integer version raises with diagnostic # --------------------------------------------------------------------------- def test_bind_personas_override_non_integer_version_raises( seed_personas: list[Persona], spec_and_review: WorkflowTemplate, consent_store: PersonaConsentStore, ) -> None: """An override spec with a non-integer version must raise with clear diagnostic.""" override = BindingOverride(persona_pinned={"spec_writer": "openrouter-claude-spec-writer@abc"}) with pytest.raises(MyDeepAgentError) as exc_info: bind_personas(spec_and_review, seed_personas, _all_available(), consent_store, override) assert exc_info.value.code == "no_eligible_persona" assert "non-integer version" in str(exc_info.value) # --------------------------------------------------------------------------- # Step 2 patch: override with ineligible persona surfaces reason # --------------------------------------------------------------------------- def test_bind_personas_override_ineligible_persona_surfaces_reason( seed_personas: list[Persona], spec_and_review: WorkflowTemplate, consent_store: PersonaConsentStore, ) -> None: """Override that names an ineligible persona must surface the ineligibility reason.""" # 'spec_writer' role needs spec_write + phase_planning. # Find a persona in seed that does NOT have those caps so we can force it. ineligible = next( p for p in seed_personas if "spec_write" not in [c.value for c in p.capabilities] ) override = BindingOverride( persona_pinned={"spec_writer": f"{ineligible.name}@{ineligible.version}"} ) with pytest.raises(MyDeepAgentError) as exc_info: bind_personas(spec_and_review, seed_personas, _all_available(), consent_store, override) assert exc_info.value.code == "no_eligible_persona" err_str = str(exc_info.value) # The error message must say the persona is ineligible with a reason. assert "ineligible" in err_str or "missing" in err_str # --------------------------------------------------------------------------- # Step 2 patch: PersonaConsentStore atomic write calls os.fsync # --------------------------------------------------------------------------- def test_consent_store_write_calls_fsync(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None: """PersonaConsentStore.set() must call os.fsync() for atomic durability.""" import os called: list[int] = [] orig_fsync = os.fsync def spy(fd: int) -> None: called.append(fd) orig_fsync(fd) monkeypatch.setattr(os, "fsync", spy) store = PersonaConsentStore(tmp_path / "consents.json") store.set("hash_abc", "approve") assert len(called) >= 1, "os.fsync must be called at least once during write"