#!/usr/bin/env python3 """ Tests for Story 4.02: BinduHydrator — Redis Scatter Task (L1 Fetch) AIVA RLM Nexus PRD v2 — Track A Black box tests (BB1-BB3): public API behaviour — what comes out. White box tests (WB1-WB3): internal structure — how it works inside. ALL tests use mocks — NO real Redis connection. """ import asyncio import json import sys from unittest.mock import AsyncMock, MagicMock, call, patch import pytest sys.path.insert(0, "/mnt/e/genesis-system") from core.hydrators.bindu_hydrator import BinduHydrator # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _make_redis_mock( aiva_state_value=None, directives_value=None, side_effect=None, ) -> AsyncMock: """ Build an AsyncMock redis client. aiva_state_value: raw bytes/str returned by redis.get("aiva:state:*"), or None directives_value: raw bytes/str returned by redis.get("kinan:directives:active"), or None side_effect: if set, redis.get raises this exception """ mock = AsyncMock() mock.setex = AsyncMock(return_value=True) if side_effect is not None: mock.get = AsyncMock(side_effect=side_effect) else: # asyncio.gather calls get() twice; return different values per call mock.get = AsyncMock(side_effect=[aiva_state_value, directives_value]) return mock def run(coro): """Run coroutine synchronously (Python 3.10-safe).""" return asyncio.get_event_loop().run_until_complete(coro) # --------------------------------------------------------------------------- # BB1: Both keys exist → returns both values correctly parsed # --------------------------------------------------------------------------- class TestBB1_BothKeysExist: """BB1: When both Redis keys exist, returns correctly parsed dict + list.""" def test_aiva_state_returned_as_dict(self): state = {"mood": "focused", "active_call": "c123"} redis = _make_redis_mock( aiva_state_value=json.dumps(state), directives_value=json.dumps([]), ) hydrator = BinduHydrator(redis_client=redis) result = run(hydrator._scatter_redis_tasks("s1")) assert result["aiva_state"] == state def test_kinan_directives_returned_as_list(self): directives = ["respond_in_english", "prioritise_bookings"] redis = _make_redis_mock( aiva_state_value=json.dumps({}), directives_value=json.dumps(directives), ) hydrator = BinduHydrator(redis_client=redis) result = run(hydrator._scatter_redis_tasks("s1")) assert result["kinan_directives"] == directives def test_both_keys_parsed_correctly_together(self): state = {"status": "ready"} directives = ["be_brief", "log_all"] redis = _make_redis_mock( aiva_state_value=json.dumps(state), directives_value=json.dumps(directives), ) hydrator = BinduHydrator(redis_client=redis) result = run(hydrator._scatter_redis_tasks("session-abc")) assert result["aiva_state"] == state assert result["kinan_directives"] == directives def test_result_has_exactly_two_keys(self): redis = _make_redis_mock( aiva_state_value=json.dumps({"x": 1}), directives_value=json.dumps(["d1"]), ) hydrator = BinduHydrator(redis_client=redis) result = run(hydrator._scatter_redis_tasks("s1")) assert set(result.keys()) == {"aiva_state", "kinan_directives"} # --------------------------------------------------------------------------- # BB2: aiva:state missing → aiva_state defaults to {}, directives still parsed # --------------------------------------------------------------------------- class TestBB2_AivaStateMissing: """BB2: When aiva:state key is None (missing), aiva_state defaults to {}.""" def test_aiva_state_missing_returns_empty_dict(self): directives = ["directive_one"] redis = _make_redis_mock( aiva_state_value=None, # key not found in Redis directives_value=json.dumps(directives), ) hydrator = BinduHydrator(redis_client=redis) result = run(hydrator._scatter_redis_tasks("s2")) assert result["aiva_state"] == {} def test_directives_still_parsed_when_aiva_state_missing(self): directives = ["active_directive"] redis = _make_redis_mock( aiva_state_value=None, directives_value=json.dumps(directives), ) hydrator = BinduHydrator(redis_client=redis) result = run(hydrator._scatter_redis_tasks("s2")) assert result["kinan_directives"] == directives # --------------------------------------------------------------------------- # BB3: Both keys missing → {"aiva_state": {}, "kinan_directives": []} # --------------------------------------------------------------------------- class TestBB3_BothKeysMissing: """BB3: Both Redis keys absent → full safe defaults returned.""" def test_both_missing_returns_empty_defaults(self): redis = _make_redis_mock( aiva_state_value=None, directives_value=None, ) hydrator = BinduHydrator(redis_client=redis) result = run(hydrator._scatter_redis_tasks("s3")) assert result == {"aiva_state": {}, "kinan_directives": []} def test_both_missing_no_exception_raised(self): redis = _make_redis_mock( aiva_state_value=None, directives_value=None, ) hydrator = BinduHydrator(redis_client=redis) # Must not raise any exception result = run(hydrator._scatter_redis_tasks("s3")) assert isinstance(result, dict) # --------------------------------------------------------------------------- # WB1: asyncio.gather called with 2 coroutines (parallel, not sequential) # --------------------------------------------------------------------------- class TestWB1_AsyncioGatherUsed: """WB1: Verify asyncio.gather is used so both reads fire in parallel.""" def test_gather_called_once(self): redis = _make_redis_mock( aiva_state_value=json.dumps({}), directives_value=json.dumps([]), ) hydrator = BinduHydrator(redis_client=redis) with patch("core.hydrators.bindu_hydrator.asyncio.gather", wraps=asyncio.gather) as mock_gather: run(hydrator._scatter_redis_tasks("s4")) assert mock_gather.call_count == 1, ( f"asyncio.gather should be called exactly once, got {mock_gather.call_count}" ) def test_redis_get_called_twice(self): """Two redis.get() calls — one per key.""" redis = _make_redis_mock( aiva_state_value=json.dumps({"a": 1}), directives_value=json.dumps(["d"]), ) hydrator = BinduHydrator(redis_client=redis) run(hydrator._scatter_redis_tasks("s4")) assert redis.get.call_count == 2, ( f"redis.get should be called exactly twice, got {redis.get.call_count}" ) def test_correct_keys_passed_to_redis_get(self): """Verify the exact Redis key strings used.""" redis = _make_redis_mock( aiva_state_value=json.dumps({}), directives_value=json.dumps([]), ) hydrator = BinduHydrator(redis_client=redis) run(hydrator._scatter_redis_tasks("my-session-99")) call_args = [c.args[0] for c in redis.get.call_args_list] assert "aiva:state:my-session-99" in call_args, ( f"Expected 'aiva:state:my-session-99' in get calls: {call_args}" ) assert "kinan:directives:active" in call_args, ( f"Expected 'kinan:directives:active' in get calls: {call_args}" ) # --------------------------------------------------------------------------- # WB2: JSON parse error on aiva_state → defaults to {} (no crash) # --------------------------------------------------------------------------- class TestWB2_JsonParseError: """WB2: Malformed JSON in Redis values must not crash the hydration pipeline.""" def test_aiva_state_bad_json_defaults_to_empty_dict(self): redis = _make_redis_mock( aiva_state_value="not-valid-json{{{", directives_value=json.dumps(["ok_directive"]), ) hydrator = BinduHydrator(redis_client=redis) result = run(hydrator._scatter_redis_tasks("s5")) assert result["aiva_state"] == {} def test_directives_bad_json_defaults_to_empty_list(self): redis = _make_redis_mock( aiva_state_value=json.dumps({"fine": True}), directives_value="[[[bad json", ) hydrator = BinduHydrator(redis_client=redis) result = run(hydrator._scatter_redis_tasks("s5")) assert result["kinan_directives"] == [] def test_both_bad_json_returns_safe_defaults(self): redis = _make_redis_mock( aiva_state_value="bad{{{", directives_value="[bad", ) hydrator = BinduHydrator(redis_client=redis) result = run(hydrator._scatter_redis_tasks("s5")) assert result == {"aiva_state": {}, "kinan_directives": []} def test_aiva_state_wrong_type_json_defaults_to_dict(self): """If aiva_state parses to a list instead of dict → default to {}.""" redis = _make_redis_mock( aiva_state_value=json.dumps([1, 2, 3]), # valid JSON but wrong type directives_value=json.dumps([]), ) hydrator = BinduHydrator(redis_client=redis) result = run(hydrator._scatter_redis_tasks("s5")) assert result["aiva_state"] == {} def test_directives_wrong_type_json_defaults_to_list(self): """If directives parses to a dict instead of list → default to [].""" redis = _make_redis_mock( aiva_state_value=json.dumps({}), directives_value=json.dumps({"bad": "type"}), # valid JSON but wrong type ) hydrator = BinduHydrator(redis_client=redis) result = run(hydrator._scatter_redis_tasks("s5")) assert result["kinan_directives"] == [] # --------------------------------------------------------------------------- # WB3: Redis connection error → returns defaults (non-fatal) # --------------------------------------------------------------------------- class TestWB3_RedisConnectionError: """WB3: Any Redis exception must be caught; safe defaults returned.""" def test_connection_error_returns_safe_defaults(self): redis = _make_redis_mock(side_effect=ConnectionError("Redis down")) hydrator = BinduHydrator(redis_client=redis) result = run(hydrator._scatter_redis_tasks("s6")) assert result == {"aiva_state": {}, "kinan_directives": []} def test_timeout_error_returns_safe_defaults(self): redis = _make_redis_mock(side_effect=TimeoutError("Redis timed out")) hydrator = BinduHydrator(redis_client=redis) result = run(hydrator._scatter_redis_tasks("s6")) assert result == {"aiva_state": {}, "kinan_directives": []} def test_os_error_returns_safe_defaults(self): redis = _make_redis_mock(side_effect=OSError("Network unreachable")) hydrator = BinduHydrator(redis_client=redis) result = run(hydrator._scatter_redis_tasks("s6")) assert result == {"aiva_state": {}, "kinan_directives": []} def test_redis_error_does_not_raise(self): """Confirm no exception propagates out of _scatter_redis_tasks.""" redis = _make_redis_mock(side_effect=RuntimeError("Unexpected Redis failure")) hydrator = BinduHydrator(redis_client=redis) try: result = run(hydrator._scatter_redis_tasks("s6")) except Exception as exc: pytest.fail(f"_scatter_redis_tasks raised unexpectedly: {exc}") assert isinstance(result, dict) # --------------------------------------------------------------------------- # Run summary # --------------------------------------------------------------------------- if __name__ == "__main__": result = pytest.main([__file__, "-v", "--tb=short"]) sys.exit(result)