#!/usr/bin/env python3 """ Tests for Story 6.01: SwarmResult — Data Model + Conflict Detector AIVA RLM Nexus PRD v2 — Track A, Module 6 Black box tests (BB1-BB4): verify public API behaviour from the outside. White box tests (WB1-WB4): verify internal invariants and structural properties. Integration test (IT1): import in a fresh process produces no errors. All tests use pure in-memory objects — zero real I/O. """ import dataclasses import inspect import sys from datetime import datetime, timezone import pytest sys.path.insert(0, "/mnt/e/genesis-system") from core.merge.swarm_result import SwarmConflictDetector, SwarmResult # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _ts() -> datetime: """Return a fixed UTC timestamp for test stability.""" return datetime(2026, 2, 25, 10, 0, 0, tzinfo=timezone.utc) def _make_result( session_id: str = "sess-001", worker_name: str = "worker-A", output: dict | None = None, confidence: float = 0.9, ) -> SwarmResult: """Return a minimal valid SwarmResult with optional overrides.""" return SwarmResult( session_id=session_id, worker_name=worker_name, output=output if output is not None else {}, completed_at=_ts(), confidence=confidence, ) # --------------------------------------------------------------------------- # BB1: Two results with same key and different non-None value → detect() True # --------------------------------------------------------------------------- class TestBB1_ConflictDetected: """BB1: detect() returns True when same key has different values.""" def test_same_key_different_value_returns_true(self): r_a = _make_result(worker_name="A", output={"status": "confirmed"}) r_b = _make_result(worker_name="B", output={"status": "pending"}) detector = SwarmConflictDetector() assert detector.detect([r_a, r_b]) is True def test_same_key_different_numeric_value_returns_true(self): r_a = _make_result(worker_name="A", output={"score": 1}) r_b = _make_result(worker_name="B", output={"score": 2}) detector = SwarmConflictDetector() assert detector.detect([r_a, r_b]) is True def test_multiple_conflicting_keys_still_true(self): r_a = _make_result(worker_name="A", output={"x": "foo", "y": "bar"}) r_b = _make_result(worker_name="B", output={"x": "baz", "y": "qux"}) detector = SwarmConflictDetector() assert detector.detect([r_a, r_b]) is True # --------------------------------------------------------------------------- # BB2: Two results with different keys → detect() False (complementary) # --------------------------------------------------------------------------- class TestBB2_ComplementaryOutputs: """BB2: detect() returns False when results use entirely different keys.""" def test_different_keys_returns_false(self): r_a = _make_result(worker_name="A", output={"name": "George"}) r_b = _make_result(worker_name="B", output={"location": "Cairns"}) detector = SwarmConflictDetector() assert detector.detect([r_a, r_b]) is False def test_empty_outputs_returns_false(self): r_a = _make_result(worker_name="A", output={}) r_b = _make_result(worker_name="B", output={}) detector = SwarmConflictDetector() assert detector.detect([r_a, r_b]) is False def test_one_output_empty_one_populated_returns_false(self): r_a = _make_result(worker_name="A", output={}) r_b = _make_result(worker_name="B", output={"key": "value"}) detector = SwarmConflictDetector() assert detector.detect([r_a, r_b]) is False # --------------------------------------------------------------------------- # BB3: get_conflicts() returns correct (a, b, key) tuple # --------------------------------------------------------------------------- class TestBB3_GetConflicts: """BB3: get_conflicts() returns correct (result_a, result_b, key) tuples.""" def test_returns_correct_tuple_structure(self): r_a = _make_result(worker_name="A", output={"status": "confirmed"}) r_b = _make_result(worker_name="B", output={"status": "pending"}) detector = SwarmConflictDetector() conflicts = detector.get_conflicts([r_a, r_b]) assert len(conflicts) == 1 conflict = conflicts[0] assert len(conflict) == 3 result_x, result_y, key = conflict assert key == "status" assert {result_x.worker_name, result_y.worker_name} == {"A", "B"} def test_empty_list_when_no_conflicts(self): r_a = _make_result(worker_name="A", output={"name": "George"}) r_b = _make_result(worker_name="B", output={"location": "Cairns"}) detector = SwarmConflictDetector() assert detector.get_conflicts([r_a, r_b]) == [] def test_two_conflicting_keys_produce_two_tuples(self): r_a = _make_result(worker_name="A", output={"x": "foo", "y": "bar"}) r_b = _make_result(worker_name="B", output={"x": "baz", "y": "qux"}) detector = SwarmConflictDetector() conflicts = detector.get_conflicts([r_a, r_b]) # One tuple per conflicting key assert len(conflicts) == 2 keys = {c[2] for c in conflicts} assert keys == {"x", "y"} # --------------------------------------------------------------------------- # BB4: 3+ results with overlapping conflicts → all pairs found # --------------------------------------------------------------------------- class TestBB4_MultipleResultConflicts: """BB4: get_conflicts() finds all conflicting pairs when 3+ results given.""" def test_three_results_two_conflicting_pairs(self): # r_a vs r_b conflict on "status" # r_a vs r_c conflict on "status" # r_b vs r_c conflict on "status" (all three have different status) r_a = _make_result(worker_name="A", output={"status": "confirmed"}) r_b = _make_result(worker_name="B", output={"status": "pending"}) r_c = _make_result(worker_name="C", output={"status": "cancelled"}) detector = SwarmConflictDetector() conflicts = detector.get_conflicts([r_a, r_b, r_c]) # 3 pairs: (A,B), (A,C), (B,C) — each conflicting on "status" assert len(conflicts) == 3 def test_results_from_different_sessions_not_compared(self): # Workers in session-001 conflict; worker in session-002 should not # be compared against session-001 workers at all. r_a = _make_result(session_id="sess-001", worker_name="A", output={"v": 1}) r_b = _make_result(session_id="sess-001", worker_name="B", output={"v": 2}) r_c = _make_result(session_id="sess-002", worker_name="C", output={"v": 99}) detector = SwarmConflictDetector() conflicts = detector.get_conflicts([r_a, r_b, r_c]) # Only A vs B conflict; C is in a different session assert len(conflicts) == 1 _, _, key = conflicts[0] assert key == "v" # --------------------------------------------------------------------------- # WB1: SwarmResult dataclass has __eq__ via @dataclass # --------------------------------------------------------------------------- class TestWB1_DataclassEquality: """WB1: @dataclass generates __eq__ so identical instances compare equal.""" def test_equal_instances(self): ts = _ts() a = SwarmResult( session_id="s1", worker_name="w1", output={"k": "v"}, completed_at=ts, confidence=0.8, ) b = SwarmResult( session_id="s1", worker_name="w1", output={"k": "v"}, completed_at=ts, confidence=0.8, ) assert a == b def test_different_confidence_not_equal(self): ts = _ts() a = SwarmResult("s", "w", {}, ts, 0.5) b = SwarmResult("s", "w", {}, ts, 0.9) assert a != b def test_is_dataclass(self): assert dataclasses.is_dataclass(SwarmResult) def test_has_five_fields(self): fields = {f.name for f in dataclasses.fields(SwarmResult)} expected = {"session_id", "worker_name", "output", "completed_at", "confidence"} assert fields == expected # --------------------------------------------------------------------------- # WB2: SwarmConflictDetector works with empty list — returns False # --------------------------------------------------------------------------- class TestWB2_EmptyList: """WB2: detect() and get_conflicts() handle empty input gracefully.""" def test_detect_empty_list_returns_false(self): assert SwarmConflictDetector().detect([]) is False def test_get_conflicts_empty_list_returns_empty(self): assert SwarmConflictDetector().get_conflicts([]) == [] def test_detect_single_result_returns_false(self): r = _make_result(output={"key": "value"}) assert SwarmConflictDetector().detect([r]) is False def test_get_conflicts_single_result_returns_empty(self): r = _make_result(output={"key": "value"}) assert SwarmConflictDetector().get_conflicts([r]) == [] # --------------------------------------------------------------------------- # WB3: Conflict check uses key equality, not pointer equality # --------------------------------------------------------------------------- class TestWB3_KeyEqualityNotPointer: """WB3: Two strings with the same content are treated as the same key.""" def test_equal_string_values_are_not_a_conflict(self): # Both workers produce the same value — not a conflict. r_a = _make_result(worker_name="A", output={"status": "confirmed"}) r_b = _make_result(worker_name="B", output={"status": "confirmed"}) detector = SwarmConflictDetector() assert detector.detect([r_a, r_b]) is False assert detector.get_conflicts([r_a, r_b]) == [] def test_structurally_equal_dicts_not_a_conflict(self): # Identical nested dicts → no conflict. r_a = _make_result(worker_name="A", output={"data": {"x": 1}}) r_b = _make_result(worker_name="B", output={"data": {"x": 1}}) assert SwarmConflictDetector().detect([r_a, r_b]) is False # --------------------------------------------------------------------------- # WB4: None values are NOT treated as conflicts # --------------------------------------------------------------------------- class TestWB4_NoneValuesNotConflicts: """WB4: None on either side of a key comparison is skipped (complementary).""" def test_none_vs_value_not_a_conflict(self): r_a = _make_result(worker_name="A", output={"status": None}) r_b = _make_result(worker_name="B", output={"status": "confirmed"}) assert SwarmConflictDetector().detect([r_a, r_b]) is False def test_none_vs_none_not_a_conflict(self): r_a = _make_result(worker_name="A", output={"status": None}) r_b = _make_result(worker_name="B", output={"status": None}) assert SwarmConflictDetector().detect([r_a, r_b]) is False def test_value_vs_none_not_a_conflict(self): r_a = _make_result(worker_name="A", output={"status": "pending"}) r_b = _make_result(worker_name="B", output={"status": None}) assert SwarmConflictDetector().detect([r_a, r_b]) is False def test_none_key_plus_real_conflict_on_other_key(self): # "status" has None on one side (not a conflict), but "score" conflicts. r_a = _make_result(worker_name="A", output={"status": None, "score": 1}) r_b = _make_result(worker_name="B", output={"status": "ok", "score": 2}) assert SwarmConflictDetector().detect([r_a, r_b]) is True conflicts = SwarmConflictDetector().get_conflicts([r_a, r_b]) assert len(conflicts) == 1 assert conflicts[0][2] == "score" # --------------------------------------------------------------------------- # IT1: Import in a fresh process → no errors # --------------------------------------------------------------------------- class TestIT1_CleanImport: """IT1: The module can be imported cleanly without any side-effects.""" def test_swarm_result_importable(self): from core.merge.swarm_result import SwarmResult as SR assert SR is SwarmResult def test_swarm_conflict_detector_importable(self): from core.merge.swarm_result import SwarmConflictDetector as SCD assert SCD is SwarmConflictDetector def test_no_sqlite3_in_module(self): import core.merge.swarm_result as module source = inspect.getsource(module) assert "import sqlite3" not in source, ( "sqlite3 is FORBIDDEN in swarm_result.py (Rule 7)" ) def test_class_names_not_conflicting_with_track_b(self): """SwarmConflictDetector must NOT shadow ConflictDetector from Track B.""" from core.merge.conflict_detector import ConflictDetector from core.merge.swarm_result import SwarmConflictDetector assert SwarmConflictDetector is not ConflictDetector assert SwarmConflictDetector.__name__ == "SwarmConflictDetector" assert ConflictDetector.__name__ == "ConflictDetector" # --------------------------------------------------------------------------- # Run summary # --------------------------------------------------------------------------- if __name__ == "__main__": result = pytest.main([__file__, "-v", "--tb=short"]) sys.exit(result)