"""U-015 tests: c_bug_record_create + the human_reported claim/complete mechanics. All real (no mocking needed -- sqlite is fast/free/local). Run: python3 -m unittest src.test_bug_record -v """ import json import unittest from src import bug_record, schema class BugRecordTestBase(unittest.TestCase): def setUp(self): self.conn = schema.connect(":memory:") schema.create_schema(self.conn) self.conn.execute( "INSERT INTO project (project_id, name, target_folder_path, has_git_repo, status) " "VALUES ('proj-1', 'P', '/tmp/proj-1', 1, 'active')" ) self.conn.execute( "INSERT INTO task (task_id, project_id, goal_text, task_type, status, assigned_org_node) " "VALUES ('task-1', 'proj-1', 'goal', 'goal', 'validating', 'validator')" ) self.conn.commit() def tearDown(self): self.conn.close() class CreateBugRecordTest(BugRecordTestBase): def test_happy_path_creates_routed_validator_record(self): result = bug_record.create_bug_record( self.conn, task_id="task-1", description="wall collision misjudged", root_cause_category="coder", root_cause_detail="SDD §3.2 collision check off-by-one", suggested_route="fast_lane", ) row = self.conn.execute( "SELECT source, status, root_cause_category, root_cause_detail, suggested_route " "FROM bug_record WHERE bug_id = ?", (result["bug_id"],), ).fetchone() self.assertEqual(row, ("validator", "routed", "coder", "SDD §3.2 collision check off-by-one", "fast_lane")) def test_related_artifact_ids_stored_as_json(self): result = bug_record.create_bug_record( self.conn, task_id="task-1", description="x", root_cause_category="planning", root_cause_detail="d", suggested_route="full_replan", related_artifact_ids=["art-1", "art-2"], ) row = self.conn.execute( "SELECT related_artifact_ids FROM bug_record WHERE bug_id = ?", (result["bug_id"],) ).fetchone() self.assertEqual(json.loads(row[0]), ["art-1", "art-2"]) def test_invalid_root_cause_category_rejected(self): with self.assertRaises(bug_record.InvalidRootCauseCategoryError): bug_record.create_bug_record( self.conn, task_id="task-1", description="x", root_cause_category="not_a_real_category", root_cause_detail="d", suggested_route="fast_lane", ) def test_core_engine_rejected_pre_ph5(self): # core_engine is PH5-scoped (U-034) -- not accepted yet, matching the # schema CHECK constraint's own exclusion (see src/schema.py) with self.assertRaises(bug_record.InvalidRootCauseCategoryError): bug_record.create_bug_record( self.conn, task_id="task-1", description="x", root_cause_category="core_engine", root_cause_detail="d", suggested_route="fast_lane", ) class CreateBugReportTest(BugRecordTestBase): """U-023: c_bug_report_create (03-C §3.1.1) -- console-submitted human bug reports, distinct from create_bug_record's Validator-judged path.""" def test_happy_path_creates_open_human_record(self): result = bug_record.create_bug_report( self.conn, project_id="proj-1", description="snake doesn't wrap around edges", ) row = self.conn.execute( "SELECT source, status, task_id, description FROM bug_record WHERE bug_id = ?", (result["bug_id"],), ).fetchone() self.assertEqual(row, ("human", "open", None, "snake doesn't wrap around edges")) self.assertEqual(result["status"], "open") def test_related_task_id_is_optional_and_stored_when_given(self): result = bug_record.create_bug_report( self.conn, project_id="proj-1", description="x", related_task_id="task-1", ) row = self.conn.execute( "SELECT task_id FROM bug_record WHERE bug_id = ?", (result["bug_id"],) ).fetchone() self.assertEqual(row[0], "task-1") def test_repro_steps_optional_and_stored_when_given(self): result = bug_record.create_bug_report( self.conn, project_id="proj-1", description="x", repro_steps="1. start game 2. hit wall", ) row = self.conn.execute( "SELECT repro_steps FROM bug_record WHERE bug_id = ?", (result["bug_id"],) ).fetchone() self.assertEqual(row[0], "1. start game 2. hit wall") def test_empty_description_rejected(self): with self.assertRaises(bug_record.InvalidParamError): bug_record.create_bug_report(self.conn, project_id="proj-1", description="") def test_nonexistent_project_id_rejected(self): # a raw FK violation surfacing as an uncaught IntegrityError instead # of this is exactly what a real backend_server.py smoke test caught # (a bad/stale project_id would otherwise 500 the request) with self.assertRaises(bug_record.InvalidParamError): bug_record.create_bug_report(self.conn, project_id="no-such-project", description="x") def test_nonexistent_related_task_id_rejected(self): with self.assertRaises(bug_record.InvalidParamError): bug_record.create_bug_report( self.conn, project_id="proj-1", description="x", related_task_id="no-such-task" ) def test_writes_bug_recorded_execution_event(self): result = bug_record.create_bug_report(self.conn, project_id="proj-1", description="x") row = self.conn.execute( "SELECT project_id, event_type FROM execution_event WHERE payload LIKE ?", ('%' + result["bug_id"] + '%',), ).fetchone() self.assertEqual(row, ("proj-1", "bug_recorded")) def test_no_related_task_still_writes_event_with_null_task_id(self): # this is exactly the case 03-C's own input_contract can't satisfy # without project_id -- see create_bug_report's docstring result = bug_record.create_bug_report(self.conn, project_id="proj-1", description="x") row = self.conn.execute( "SELECT task_id FROM execution_event WHERE payload LIKE ?", ('%' + result["bug_id"] + '%',), ).fetchone() self.assertIsNone(row[0]) class HumanReportedFlowTest(BugRecordTestBase): def _insert_human_bug(self, bug_id="bug-1", status="open"): self.conn.execute( "INSERT INTO bug_record (bug_id, task_id, source, description, status) " "VALUES (?, 'task-1', 'human', 'something looks broken', ?)", (bug_id, status), ) self.conn.commit() def test_claim_transitions_open_to_triaging(self): self._insert_human_bug() bug_record.claim_human_reported_bug(self.conn, "bug-1") status = self.conn.execute("SELECT status FROM bug_record WHERE bug_id = 'bug-1'").fetchone()[0] self.assertEqual(status, "triaging") def test_claim_rejects_already_claimed_bug(self): self._insert_human_bug(status="triaging") with self.assertRaises(ValueError): bug_record.claim_human_reported_bug(self.conn, "bug-1") def test_claim_rejects_unknown_bug_id(self): with self.assertRaises(ValueError): bug_record.claim_human_reported_bug(self.conn, "no-such-bug") def test_complete_triage_updates_in_place_not_new_record(self): self._insert_human_bug() bug_record.claim_human_reported_bug(self.conn, "bug-1") bug_record.complete_human_reported_triage( self.conn, "bug-1", root_cause_category="coder", root_cause_detail="found it", suggested_route="fast_lane", ) count = self.conn.execute("SELECT COUNT(*) FROM bug_record").fetchone()[0] self.assertEqual(count, 1) row = self.conn.execute( "SELECT status, root_cause_category, root_cause_detail, suggested_route " "FROM bug_record WHERE bug_id = 'bug-1'" ).fetchone() self.assertEqual(row, ("routed", "coder", "found it", "fast_lane")) def test_complete_triage_rejects_invalid_category(self): self._insert_human_bug() bug_record.claim_human_reported_bug(self.conn, "bug-1") with self.assertRaises(bug_record.InvalidRootCauseCategoryError): bug_record.complete_human_reported_triage( self.conn, "bug-1", root_cause_category="nonsense", root_cause_detail="x", suggested_route="fast_lane", ) if __name__ == "__main__": unittest.main()