""" tests/track_b/test_story_8_06.py Story 8.06: CodeProposer — Structural Fix Generator Black Box Tests (BB1–BB4): BB1 Generated interceptor code contains "BaseInterceptor" (string inspection) BB2 Proposal fails axiomatic test → validate_proposal returns False BB3 Proposal written to data/evolution/proposals/ directory (tmp_path) BB4 Valid opus response → CodeProposal with all 5 fields populated White Box Tests (WB1–WB4): WB1 Opus prompt contains existing interceptor code as reference WB2 test_file_path follows naming convention "tests/..." WB3 Invalid JSON from opus → ValueError raised WB4 validate_proposal calls axiomatic_tests.run_all (mock verified) ALL tests use mocks for opus_client and axiomatic_tests. ALL file I/O uses tmp_path. """ from __future__ import annotations import dataclasses import json import sys from pathlib import Path from unittest.mock import MagicMock, call, patch import pytest # --------------------------------------------------------------------------- # Path setup # --------------------------------------------------------------------------- GENESIS_ROOT = "/mnt/e/genesis-system" if GENESIS_ROOT not in sys.path: sys.path.insert(0, GENESIS_ROOT) # --------------------------------------------------------------------------- # Imports under test # --------------------------------------------------------------------------- from core.evolution.code_proposer import ( # noqa: E402 CodeProposer, CodeProposal, PROPOSE_PROMPT, _DEFAULT_PROPOSALS_DIR, ) from core.evolution.meta_architect import Bottleneck # noqa: E402 from core.evolution.axiomatic_tests import AxiomaticTests, AxiomResult # noqa: E402 # --------------------------------------------------------------------------- # Helpers — shared fixtures and factory functions # --------------------------------------------------------------------------- def _make_bottleneck( description: str = "DB connection pooling code module fails under load", frequency: int = 5, saga_ids: list[str] | None = None, scar_ids: list[str] | None = None, ) -> Bottleneck: """Return a populated Bottleneck for use in tests.""" return Bottleneck( description=description, frequency=frequency, affected_saga_ids=saga_ids or ["saga-001", "saga-002"], scar_ids=scar_ids or ["scar-a1", "scar-a2"], ) def _make_valid_opus_response( file_path: str = "core/interceptors/pool_guard_interceptor.py", base_interceptor: bool = True, test_file_path: str = "tests/interceptors/test_pool_guard_interceptor.py", ) -> str: """Return a valid JSON string that CodeProposer._parse_response() accepts.""" code_content = ( "from core.interceptors.base_interceptor import BaseInterceptor\n\n" "class PoolGuardInterceptor(BaseInterceptor):\n" " async def pre_execute(self, task_payload):\n" " return task_payload\n" " async def post_execute(self, result, task_payload):\n" " pass\n" " async def on_error(self, error, task_payload):\n" " return {}\n" " async def on_correction(self, correction_payload):\n" " return correction_payload\n" ) if base_interceptor else ( "class BadInterceptor:\n" " pass\n" ) data = { "file_path": file_path, "code_content": code_content, "test_file_path": test_file_path, "test_content": "def test_placeholder(): pass\n", "config_changes": {}, } return json.dumps(data) def _make_proposer(tmp_path, opus_response: str | None = None, axiomatic_tests=None) -> CodeProposer: """Return a CodeProposer with mocked opus_client and tmp proposals_dir.""" if opus_response is not None: opus_client = MagicMock(return_value=opus_response) else: opus_client = MagicMock(return_value=_make_valid_opus_response()) return CodeProposer( opus_client=opus_client, axiomatic_tests=axiomatic_tests, proposals_dir=str(tmp_path / "proposals"), ) # --------------------------------------------------------------------------- # BB Tests — Black Box # --------------------------------------------------------------------------- def test_bb1_generated_code_contains_base_interceptor(tmp_path): """BB1: Generated interceptor code contains "BaseInterceptor" — verified by string inspection.""" opus_response = _make_valid_opus_response(base_interceptor=True) proposer = _make_proposer(tmp_path, opus_response=opus_response) bottleneck = _make_bottleneck() proposal = proposer.propose(bottleneck, existing_code_context="# existing interceptor code") assert isinstance(proposal, CodeProposal), "propose() must return a CodeProposal" assert "BaseInterceptor" in proposal.code_content, ( f"Expected 'BaseInterceptor' in code_content, got:\n{proposal.code_content}" ) def test_bb2_proposal_fails_axiomatic_test_returns_false(tmp_path): """BB2: Proposal fails axiomatic test → validate_proposal returns False.""" # Build a proposal whose code_content contains SQLite (axiom violation) bad_code = ( "import sqlite3\n" "from core.interceptors.base_interceptor import BaseInterceptor\n\n" "class SqliteInterceptor(BaseInterceptor):\n" " pass\n" ) bad_response = json.dumps({ "file_path": "core/interceptors/bad.py", "code_content": bad_code, "test_file_path": "tests/interceptors/test_bad.py", "test_content": "def test_nothing(): pass\n", "config_changes": {}, }) # Use real AxiomaticTests — it will catch the sqlite3 import proposer = CodeProposer( opus_client=MagicMock(return_value=bad_response), axiomatic_tests=AxiomaticTests(), proposals_dir=str(tmp_path / "proposals"), ) bottleneck = _make_bottleneck() proposal = proposer.propose(bottleneck, existing_code_context="") result = proposer.validate_proposal(proposal) assert result is False, ( "validate_proposal should return False when code contains SQLite (AXIOM_NO_SQLITE)" ) def test_bb3_proposal_written_to_proposals_directory(tmp_path): """BB3: Proposal written to proposals_dir/{epoch_id}/ directory using tmp_path.""" opus_response = _make_valid_opus_response() proposer = _make_proposer(tmp_path, opus_response=opus_response) bottleneck = _make_bottleneck() proposal = proposer.propose(bottleneck, existing_code_context="# ref code") manifest_path = proposer.write_proposal(proposal, epoch_id="epoch_test_001") # Manifest file must exist manifest = Path(manifest_path) assert manifest.exists(), f"proposal.json not found at {manifest_path}" assert manifest.name == "proposal.json" # Interceptor source must be written epoch_dir = manifest.parent interceptor_file = epoch_dir / proposal.file_path assert interceptor_file.exists(), f"Interceptor file not found: {interceptor_file}" assert interceptor_file.read_text(encoding="utf-8") == proposal.code_content # Test file must be written test_file = epoch_dir / proposal.test_file_path assert test_file.exists(), f"Test file not found: {test_file}" assert test_file.read_text(encoding="utf-8") == proposal.test_content # Manifest content must be valid JSON with required keys meta = json.loads(manifest.read_text(encoding="utf-8")) assert meta["epoch_id"] == "epoch_test_001" assert meta["file_path"] == proposal.file_path assert meta["test_file_path"] == proposal.test_file_path assert "config_changes" in meta def test_bb4_valid_opus_response_populates_all_five_fields(tmp_path): """BB4: Valid opus response → CodeProposal with all 5 fields populated.""" opus_response = _make_valid_opus_response( file_path="core/interceptors/retry_interceptor.py", test_file_path="tests/interceptors/test_retry_interceptor.py", ) proposer = _make_proposer(tmp_path, opus_response=opus_response) bottleneck = _make_bottleneck() proposal = proposer.propose(bottleneck, existing_code_context="") # All 5 fields must be populated (non-empty strings / non-None dicts) assert dataclasses.is_dataclass(proposal), "CodeProposal must be a dataclass" assert isinstance(proposal.file_path, str) and proposal.file_path, "file_path must be non-empty" assert isinstance(proposal.code_content, str) and proposal.code_content, "code_content must be non-empty" assert isinstance(proposal.test_file_path, str) and proposal.test_file_path, "test_file_path must be non-empty" assert isinstance(proposal.test_content, str) and proposal.test_content, "test_content must be non-empty" assert isinstance(proposal.config_changes, dict), "config_changes must be a dict" # Field values match the mock response assert proposal.file_path == "core/interceptors/retry_interceptor.py" assert proposal.test_file_path == "tests/interceptors/test_retry_interceptor.py" # --------------------------------------------------------------------------- # WB Tests — White Box # --------------------------------------------------------------------------- def test_wb1_opus_prompt_contains_existing_code_reference(tmp_path): """WB1: Opus prompt contains existing interceptor code as reference.""" opus_response = _make_valid_opus_response() opus_client = MagicMock(return_value=opus_response) proposer = CodeProposer( opus_client=opus_client, proposals_dir=str(tmp_path / "proposals"), ) bottleneck = _make_bottleneck() existing_code = "class ExistingInterceptor(BaseInterceptor):\n pass\n" proposer.propose(bottleneck, existing_code_context=existing_code) # Verify opus_client was called once assert opus_client.call_count == 1, "opus_client must be called exactly once" # Extract the prompt that was passed prompt_used: str = opus_client.call_args[0][0] # The existing code must appear verbatim in the prompt assert existing_code in prompt_used, ( f"Existing code not found in prompt. Prompt (first 500 chars):\n{prompt_used[:500]}" ) # Bottleneck fields must also be in the prompt assert bottleneck.description in prompt_used, "Bottleneck description missing from prompt" assert str(bottleneck.frequency) in prompt_used, "Bottleneck frequency missing from prompt" def test_wb2_test_file_path_starts_with_tests_prefix(tmp_path): """WB2: test_file_path follows naming convention 'tests/...'.""" # Craft a response where test_file_path starts with "tests/" opus_response = _make_valid_opus_response( test_file_path="tests/interceptors/test_my_interceptor.py" ) proposer = _make_proposer(tmp_path, opus_response=opus_response) bottleneck = _make_bottleneck() proposal = proposer.propose(bottleneck, existing_code_context="") assert proposal.test_file_path.startswith("tests/"), ( f"test_file_path must start with 'tests/', got: {proposal.test_file_path!r}" ) def test_wb3_invalid_json_from_opus_raises_value_error(tmp_path): """WB3: Invalid JSON from opus → ValueError raised.""" # Various invalid JSON strings invalid_responses = [ "This is not JSON at all", "{'single_quotes': 'are not JSON'}", '{"incomplete": ', "", "null", # valid JSON but not a dict with required keys ] for bad_response in invalid_responses: opus_client = MagicMock(return_value=bad_response) proposer = CodeProposer( opus_client=opus_client, proposals_dir=str(tmp_path / "proposals"), ) bottleneck = _make_bottleneck() with pytest.raises((ValueError, json.JSONDecodeError)): proposer.propose(bottleneck, existing_code_context="") def test_wb3_missing_required_fields_raises_value_error(tmp_path): """WB3 (extended): JSON missing required fields → ValueError raised.""" # JSON is parseable but missing 'code_content' incomplete_response = json.dumps({ "file_path": "core/interceptors/x.py", # "code_content" is intentionally absent "test_file_path": "tests/interceptors/test_x.py", "test_content": "pass", }) opus_client = MagicMock(return_value=incomplete_response) proposer = CodeProposer( opus_client=opus_client, proposals_dir=str(tmp_path / "proposals"), ) bottleneck = _make_bottleneck() with pytest.raises(ValueError, match="missing required fields"): proposer.propose(bottleneck, existing_code_context="") def test_wb4_validate_proposal_calls_axiomatic_tests_run_all(tmp_path): """WB4: validate_proposal calls axiomatic_tests.run_all (mock verified).""" opus_response = _make_valid_opus_response() proposer = _make_proposer(tmp_path, opus_response=opus_response) bottleneck = _make_bottleneck() # Build a proposal manually (bypass propose() to isolate validate_proposal) proposal = CodeProposal( file_path="core/interceptors/guard.py", code_content=( "from core.interceptors.base_interceptor import BaseInterceptor\n\n" "class GuardInterceptor(BaseInterceptor):\n" " async def pre_execute(self, task_payload): return task_payload\n" " async def post_execute(self, result, task_payload): pass\n" " async def on_error(self, error, task_payload): return {}\n" " async def on_correction(self, correction_payload): return correction_payload\n" ), test_file_path="tests/interceptors/test_guard.py", test_content="def test_placeholder(): pass\n", config_changes={}, ) # Mock the AxiomaticTests instance injected into proposer mock_axiomatic = MagicMock(spec=AxiomaticTests) mock_result = AxiomResult(passed=True, violations=[]) mock_axiomatic.run_all.return_value = mock_result proposer.axiomatic_tests = mock_axiomatic result = proposer.validate_proposal(proposal) # Confirm run_all was called with code_content and an empty state dict assert mock_axiomatic.run_all.call_count == 1, ( "axiomatic_tests.run_all must be called exactly once" ) call_kwargs = mock_axiomatic.run_all.call_args # Accept both positional and keyword call signatures if call_kwargs[0]: # positional args code_arg = call_kwargs[0][0] else: code_arg = call_kwargs[1].get("code_content", "") assert code_arg == proposal.code_content, ( "run_all must receive the proposal code_content" ) assert result is True # --------------------------------------------------------------------------- # Additional edge-case and integration tests # --------------------------------------------------------------------------- def test_code_proposal_is_dataclass(): """CodeProposal is a proper dataclass with all 5 required fields.""" assert dataclasses.is_dataclass(CodeProposal) field_names = {f.name for f in dataclasses.fields(CodeProposal)} assert "file_path" in field_names assert "code_content" in field_names assert "test_file_path" in field_names assert "test_content" in field_names assert "config_changes" in field_names def test_code_proposal_config_changes_defaults_to_empty_dict(): """config_changes defaults to an empty dict (not None, not shared state).""" p1 = CodeProposal( file_path="a.py", code_content="pass", test_file_path="tests/test_a.py", test_content="pass", ) p2 = CodeProposal( file_path="b.py", code_content="pass", test_file_path="tests/test_b.py", test_content="pass", ) assert p1.config_changes == {} assert p2.config_changes == {} # Must not share the same dict instance (dataclass field default_factory) p1.config_changes["key"] = "value" assert p2.config_changes == {}, "config_changes must not be a shared mutable default" def test_propose_without_opus_client_raises_runtime_error(tmp_path): """propose() without an opus_client raises RuntimeError.""" proposer = CodeProposer( opus_client=None, proposals_dir=str(tmp_path / "proposals"), ) bottleneck = _make_bottleneck() with pytest.raises(RuntimeError, match="opus_client"): proposer.propose(bottleneck, existing_code_context="") def test_validate_proposal_fails_when_base_interceptor_absent(tmp_path): """validate_proposal returns False when code_content lacks 'BaseInterceptor'.""" # Use a passing AxiomResult — so the only failure is the missing BaseInterceptor mock_axiomatic = MagicMock(spec=AxiomaticTests) mock_axiomatic.run_all.return_value = AxiomResult(passed=True, violations=[]) proposer = CodeProposer( opus_client=MagicMock(), axiomatic_tests=mock_axiomatic, proposals_dir=str(tmp_path / "proposals"), ) proposal = CodeProposal( file_path="core/interceptors/bad.py", code_content="class SomeClass:\n pass\n", # no BaseInterceptor test_file_path="tests/interceptors/test_bad.py", test_content="def test_nothing(): pass\n", ) result = proposer.validate_proposal(proposal) assert result is False, "Should fail when BaseInterceptor not in code_content" def test_validate_proposal_fails_when_test_path_missing_prefix(tmp_path): """validate_proposal returns False when test_file_path doesn't start with 'tests/'.""" mock_axiomatic = MagicMock(spec=AxiomaticTests) mock_axiomatic.run_all.return_value = AxiomResult(passed=True, violations=[]) proposer = CodeProposer( opus_client=MagicMock(), axiomatic_tests=mock_axiomatic, proposals_dir=str(tmp_path / "proposals"), ) proposal = CodeProposal( file_path="core/interceptors/x.py", code_content=( "from core.interceptors.base_interceptor import BaseInterceptor\n\n" "class X(BaseInterceptor): pass\n" ), test_file_path="src/test_x.py", # missing "tests/" prefix test_content="def test_nothing(): pass\n", ) result = proposer.validate_proposal(proposal) assert result is False, "Should fail when test_file_path doesn't start with 'tests/'" def test_write_proposal_creates_epoch_subdirectory(tmp_path): """write_proposal creates the epoch directory structure automatically.""" opus_response = _make_valid_opus_response() proposer = _make_proposer(tmp_path, opus_response=opus_response) bottleneck = _make_bottleneck() proposal = proposer.propose(bottleneck, existing_code_context="") manifest_path = proposer.write_proposal(proposal, epoch_id="epoch_xyz") # The epoch directory must be nested under proposals_dir manifest = Path(manifest_path) epoch_dir = manifest.parent assert epoch_dir.name == "epoch_xyz" assert epoch_dir.parent.name == "proposals" def test_write_proposal_multiple_epochs_independent(tmp_path): """Multiple write_proposal calls with different epoch_ids produce independent dirs.""" opus_response = _make_valid_opus_response() proposer = _make_proposer(tmp_path, opus_response=opus_response) bottleneck = _make_bottleneck() proposal = proposer.propose(bottleneck, existing_code_context="") path1 = proposer.write_proposal(proposal, epoch_id="epoch_001") path2 = proposer.write_proposal(proposal, epoch_id="epoch_002") assert path1 != path2 assert Path(path1).exists() assert Path(path2).exists() assert Path(path1).parent.name == "epoch_001" assert Path(path2).parent.name == "epoch_002" def test_parse_response_strips_markdown_fences(tmp_path): """_parse_response correctly handles markdown-fenced JSON from the LLM.""" data = { "file_path": "core/interceptors/fenced.py", "code_content": "from core.interceptors.base_interceptor import BaseInterceptor\nclass X(BaseInterceptor): pass\n", "test_file_path": "tests/interceptors/test_fenced.py", "test_content": "def test_fenced(): pass\n", "config_changes": {}, } raw_with_fences = f"```json\n{json.dumps(data)}\n```" proposer = CodeProposer(proposals_dir=str(tmp_path / "proposals")) proposal = proposer._parse_response(raw_with_fences) assert proposal.file_path == "core/interceptors/fenced.py" assert "BaseInterceptor" in proposal.code_content def test_package_init_exports(tmp_path): """core.evolution.__init__.py exports CodeProposer and CodeProposal.""" from core.evolution import ( # noqa: F401 CodeProposer as CP, CodeProposal as CProposal, ) assert CP is CodeProposer assert CProposal is CodeProposal def test_propose_prompt_template_contains_required_placeholders(): """PROPOSE_PROMPT template has all required format keys.""" required_keys = {"description", "frequency", "saga_ids", "scar_ids", "existing_code"} for key in required_keys: assert f"{{{key}}}" in PROPOSE_PROMPT, ( f"PROPOSE_PROMPT is missing placeholder {{{key}}}" )