#!/usr/bin/env python3 """ Tests for Story 2.01: QueenRegistry — Identity Read AIVA RLM Nexus — Track A Module: core/registry/queen_registry.py Black Box + White Box tests — all using mocks (no live infra required). Run with: python -m pytest tests/track_a/test_story_2_01.py -v """ import sys import json from datetime import datetime, timezone from unittest.mock import MagicMock, patch, call sys.path.insert(0, '/mnt/e/genesis-system') import pytest from core.registry.queen_registry import ( QueenRegistry, RegistryError, CACHE_KEY, CACHE_TTL, REQUIRED_FIELDS, ) # --------------------------------------------------------------------------- # Helpers / fixtures # --------------------------------------------------------------------------- def _make_mock_cf( redis_get_return=None, redis_raises=None, pg_count=42, pg_rows=None, pg_raises=None, ): """ Build a mock ConnectionFactory. Args: redis_get_return: value returned by redis.get(CACHE_KEY) redis_raises: if set, redis.get() raises this exception pg_count: COUNT(*) returned from royal_conversations pg_rows: rows returned from aiva_capability_log (list of (desc, ts)) pg_raises: if set, conn.cursor() raises this exception """ if pg_rows is None: ts = datetime(2026, 2, 20, 12, 0, 0, tzinfo=timezone.utc) pg_rows = [ ("Learned voice routing", ts), ("Improved memory recall", ts), ] # -- Redis mock -- redis_mock = MagicMock() if redis_raises: redis_mock.get.side_effect = redis_raises redis_mock.ping.side_effect = redis_raises else: redis_mock.get.return_value = redis_get_return redis_mock.ping.return_value = True # -- Postgres mock -- cur_mock = MagicMock() if pg_raises: cur_mock.execute.side_effect = pg_raises else: # fetchone() called for COUNT, fetchall() for capability log cur_mock.fetchone.return_value = (pg_count,) cur_mock.fetchall.return_value = pg_rows conn_mock = MagicMock() if pg_raises: conn_mock.cursor.side_effect = pg_raises else: conn_mock.cursor.return_value = cur_mock cf = MagicMock() cf.get_redis.return_value = redis_mock cf.get_postgres.return_value = conn_mock return cf, redis_mock, conn_mock, cur_mock # --------------------------------------------------------------------------- # BB1: get_identity() returns dict with all 7 required fields + correct types # --------------------------------------------------------------------------- def test_bb1_get_identity_all_fields_present(): """BB1: get_identity() returns a dict containing all 7 required keys with correct types.""" cf, redis_mock, conn_mock, cur_mock = _make_mock_cf(redis_get_return=None) registry = QueenRegistry(connection_factory=cf) identity = registry.get_identity() # All 7 fields present assert REQUIRED_FIELDS.issubset(identity.keys()), ( f"Missing fields: {REQUIRED_FIELDS - identity.keys()}" ) # Type checks assert isinstance(identity["name"], str), "name must be str" assert isinstance(identity["role"], str), "role must be str" assert isinstance(identity["voice_model"], str), "voice_model must be str" assert isinstance(identity["total_conversations"], int), "total_conversations must be int" assert isinstance(identity["last_evolved"], str), "last_evolved must be str" assert isinstance(identity["active_capabilities"], list), "active_capabilities must be list" assert isinstance(identity["recent_improvements"], list), "recent_improvements must be list" # Sentinel values assert identity["name"] == "AIVA" assert "Queen" in identity["role"] assert identity["total_conversations"] == 42 print("BB1 PASS: all 7 fields present with correct types") # --------------------------------------------------------------------------- # BB2: Second call uses Redis cache; Postgres is NOT queried twice # --------------------------------------------------------------------------- def test_bb2_second_call_uses_redis_cache(): """BB2: After the first call populates Redis, the second call reads from cache (no Postgres query).""" cf, redis_mock, conn_mock, cur_mock = _make_mock_cf(redis_get_return=None) registry = QueenRegistry(connection_factory=cf) # First call — Redis miss, hits Postgres, writes cache first = registry.get_identity() # Simulate Redis now returning the cached value cached_json = json.dumps(first, default=str) redis_mock.get.return_value = cached_json # Second call — should hit Redis, NOT Postgres second = registry.get_identity() # Postgres cursor should only have been called ONCE (from first call) assert conn_mock.cursor.call_count == 1, ( f"Expected Postgres cursor called 1 time, got {conn_mock.cursor.call_count}" ) # Both calls return the same data assert first == second, "First and second call should return identical identity dicts" print("BB2 PASS: second call uses Redis cache; Postgres queried only once") # --------------------------------------------------------------------------- # BB3: total_conversations matches the COUNT returned from Postgres # --------------------------------------------------------------------------- def test_bb3_total_conversations_matches_count(): """BB3: total_conversations equals the COUNT(*) returned by Postgres.""" for expected_count in [0, 1, 999, 10_000]: cf, _, _, _ = _make_mock_cf(redis_get_return=None, pg_count=expected_count) registry = QueenRegistry(connection_factory=cf) identity = registry.get_identity() assert identity["total_conversations"] == expected_count, ( f"Expected {expected_count}, got {identity['total_conversations']}" ) print("BB3 PASS: total_conversations matches Postgres COUNT for values [0, 1, 999, 10000]") # --------------------------------------------------------------------------- # WB1: Redis unavailable → falls back to Postgres, no crash # --------------------------------------------------------------------------- def test_wb1_redis_unavailable_falls_back_to_postgres(): """WB1: When Redis raises an exception, get_identity() silently falls back to Postgres.""" import redis as redis_lib cf, redis_mock, conn_mock, cur_mock = _make_mock_cf( redis_raises=redis_lib.ConnectionError("Redis is down"), ) registry = QueenRegistry(connection_factory=cf) # Must NOT raise — Redis failure is non-fatal identity = registry.get_identity() # Still returns a valid dict assert REQUIRED_FIELDS.issubset(identity.keys()), "Identity dict incomplete after Redis failure" assert identity["total_conversations"] == 42 # from pg_count default # Postgres WAS queried assert conn_mock.cursor.call_count >= 1, "Postgres should have been queried as fallback" print("WB1 PASS: Redis unavailable → fell back to Postgres without crash") # --------------------------------------------------------------------------- # WB2: Postgres unavailable → raises RegistryError # --------------------------------------------------------------------------- def test_wb2_postgres_unavailable_raises_registry_error(): """WB2: When Postgres raises an exception, get_identity() raises RegistryError.""" import redis as redis_lib # Redis miss AND Postgres down cf, redis_mock, conn_mock, cur_mock = _make_mock_cf( redis_get_return=None, pg_raises=Exception("PG connection refused"), ) registry = QueenRegistry(connection_factory=cf) with pytest.raises(RegistryError) as exc_info: registry.get_identity() assert "Failed to read Queen identity" in str(exc_info.value), ( f"RegistryError message unexpected: {exc_info.value}" ) print("WB2 PASS: Postgres unavailable → RegistryError raised with correct message") # --------------------------------------------------------------------------- # WB3: invalidate_cache() deletes the Redis key # --------------------------------------------------------------------------- def test_wb3_invalidate_cache_deletes_redis_key(): """WB3: invalidate_cache() calls redis.delete(CACHE_KEY).""" cf, redis_mock, _, _ = _make_mock_cf() registry = QueenRegistry(connection_factory=cf) registry.invalidate_cache() redis_mock.delete.assert_called_once_with(CACHE_KEY) print(f"WB3 PASS: invalidate_cache() called redis.delete('{CACHE_KEY}')") # --------------------------------------------------------------------------- # BONUS WB4: Cache write uses correct TTL # --------------------------------------------------------------------------- def test_wb4_cache_write_uses_correct_ttl(): """WB4: After a Postgres read, Redis.setex is called with CACHE_TTL.""" cf, redis_mock, _, _ = _make_mock_cf(redis_get_return=None) registry = QueenRegistry(connection_factory=cf) registry.get_identity() # setex should have been called once with (CACHE_KEY, CACHE_TTL, ) assert redis_mock.setex.call_count == 1, ( f"Expected setex called once, got {redis_mock.setex.call_count}" ) args = redis_mock.setex.call_args[0] assert args[0] == CACHE_KEY, f"setex key: expected '{CACHE_KEY}', got '{args[0]}'" assert args[1] == CACHE_TTL, f"setex TTL: expected {CACHE_TTL}, got {args[1]}" # Third arg is the JSON string — verify it round-trips decoded = json.loads(args[2]) assert "name" in decoded, "Cached JSON missing 'name' field" print(f"WB4 PASS: Redis.setex called with key='{CACHE_KEY}' TTL={CACHE_TTL}") # --------------------------------------------------------------------------- # BONUS WB5: invalidate_cache() is non-fatal when Redis is down # --------------------------------------------------------------------------- def test_wb5_invalidate_cache_non_fatal_when_redis_down(): """WB5: invalidate_cache() swallows Redis exceptions silently.""" import redis as redis_lib cf, redis_mock, _, _ = _make_mock_cf( redis_raises=redis_lib.ConnectionError("Redis is down"), ) registry = QueenRegistry(connection_factory=cf) # Must not raise registry.invalidate_cache() print("WB5 PASS: invalidate_cache() is non-fatal when Redis is down") # --------------------------------------------------------------------------- # BONUS WB6: Package import — core.registry exports QueenRegistry + RegistryError # --------------------------------------------------------------------------- def test_wb6_package_import(): """WB6: core.registry package exports QueenRegistry and RegistryError correctly.""" from core.registry import QueenRegistry as QR, RegistryError as RE assert QR is QueenRegistry, "core.registry.QueenRegistry identity mismatch" assert RE is RegistryError, "core.registry.RegistryError identity mismatch" print("WB6 PASS: core.registry package exports are correct") # --------------------------------------------------------------------------- # Standalone runner # --------------------------------------------------------------------------- if __name__ == "__main__": test_bb1_get_identity_all_fields_present() test_bb2_second_call_uses_redis_cache() test_bb3_total_conversations_matches_count() test_wb1_redis_unavailable_falls_back_to_postgres() test_wb2_postgres_unavailable_raises_registry_error() test_wb3_invalidate_cache_deletes_redis_key() test_wb4_cache_write_uses_correct_ttl() test_wb5_invalidate_cache_non_fatal_when_redis_down() test_wb6_package_import() print("\nALL TESTS PASSED — Story 2.01") print("Tests: 9/9 PASS (BB1, BB2, BB3, WB1, WB2, WB3, WB4, WB5, WB6)")