""" Unit tests for the ExceptionAgent read-only context gatherer and its stall counter, plus the tz helpers. Focus is the *degradation* behaviour: a missing pool, a failing query, or a Redis hiccup must never raise — the gatherer returns whatever facts it could collect so the decision still runs (skewing safe). No real Postgres/Redis/NATS is touched: the agent is built with __new__ and handed in-memory fakes. Run: python -m unittest discover -s tests """ import unittest from datetime import datetime, timedelta, timezone from agents.exception_agent import ExceptionAgent, _parse_ts, _utcnow # ── Fakes ────────────────────────────────────────────────────────────────── class FakePGConn: def __init__(self, row=None, raise_err=False): self._row = row self._raise = raise_err async def fetchrow(self, query, *args): if self._raise: raise RuntimeError("pg down") return self._row class _AcquireCtx: def __init__(self, conn): self._conn = conn async def __aenter__(self): return self._conn async def __aexit__(self, *exc): return False class FakePGPool: def __init__(self, row=None, raise_err=False): self._conn = FakePGConn(row, raise_err) def acquire(self): return _AcquireCtx(self._conn) class FakeRedis: """decode_responses=True style — returns str values.""" def __init__(self, movement=None, counter=None, raise_on=()): self._movement = movement or {} self._counter = counter # str or None self._raise_on = set(raise_on) self.incr_calls = [] self.expire_calls = [] async def hgetall(self, key): if "hgetall" in self._raise_on: raise RuntimeError("redis down") return dict(self._movement) async def get(self, key): if "get" in self._raise_on: raise RuntimeError("redis down") return self._counter async def incr(self, key): if "incr" in self._raise_on: raise RuntimeError("redis down") self.incr_calls.append(key) self._counter = str(int(self._counter or 0) + 1) return int(self._counter) async def expire(self, key, ttl): self.expire_calls.append((key, ttl)) return True def make_agent(pg, redis): agent = ExceptionAgent.__new__(ExceptionAgent) # skip heavy __init__ / message-bus registration agent._pg = pg agent._redis = redis return agent # ── Tz helpers ─────────────────────────────────────────────────────────────── class TestTimeHelpers(unittest.TestCase): def test_utcnow_is_aware(self): self.assertIsNotNone(_utcnow().tzinfo) def test_parse_ts_none_and_blank(self): self.assertIsNone(_parse_ts(None)) self.assertIsNone(_parse_ts("")) def test_parse_ts_invalid(self): self.assertIsNone(_parse_ts("not-a-timestamp")) def test_parse_ts_naive_assumed_utc(self): dt = _parse_ts("2024-01-01T00:00:00") self.assertIsNotNone(dt) self.assertEqual(dt.utcoffset(), timedelta(0)) def test_parse_ts_aware_preserved(self): dt = _parse_ts("2024-01-01T00:00:00+00:00") self.assertEqual(dt.utcoffset(), timedelta(0)) # ── Gatherer ─────────────────────────────────────────────────────────────── class TestGatherStallFacts(unittest.IsolatedAsyncioTestCase): async def test_full_context(self): now = _utcnow() pg = FakePGPool(row={"status": "Miler_Assigned", "createdat": now - timedelta(minutes=30)}) redis = FakeRedis( movement={ "updated_at": (now - timedelta(minutes=5)).isoformat(), "position_unchanged_since": (now - timedelta(minutes=15)).isoformat(), }, counter="3", ) facts = await make_agent(pg, redis)._gather_stall_facts("m1", "b1") self.assertEqual(facts["booking_status"], "Miler_Assigned") self.assertAlmostEqual(facts["minutes_since_booking_created"], 30.0, delta=0.2) self.assertAlmostEqual(facts["last_gps_ping_minutes_ago"], 5.0, delta=0.2) self.assertAlmostEqual(facts["position_unchanged_minutes"], 15.0, delta=0.2) self.assertEqual(facts["stalls_today"], 3) async def test_naive_createdat_treated_as_utc(self): naive_created = _utcnow().replace(tzinfo=None) - timedelta(minutes=20) # naive UTC, as a `timestamp` column returns pg = FakePGPool(row={"status": "Pickup_Scheduled", "createdat": naive_created}) facts = await make_agent(pg, FakeRedis())._gather_stall_facts("m1", "b1") self.assertAlmostEqual(facts["minutes_since_booking_created"], 20.0, delta=0.2) # no tz error async def test_no_pg_pool(self): redis = FakeRedis(movement={"updated_at": _utcnow().isoformat()}, counter="1") facts = await make_agent(None, redis)._gather_stall_facts("m1", "b1") self.assertNotIn("booking_status", facts) # no booking facts self.assertIn("last_gps_ping_minutes_ago", facts) # redis facts still gathered self.assertEqual(facts["stalls_today"], 1) async def test_pg_error_is_swallowed(self): pg = FakePGPool(raise_err=True) redis = FakeRedis(counter="2") facts = await make_agent(pg, redis)._gather_stall_facts("m1", "b1") # must not raise self.assertNotIn("booking_status", facts) self.assertEqual(facts["stalls_today"], 2) async def test_redis_hgetall_error_is_swallowed(self): pg = FakePGPool(row={"status": "Miler_Assigned", "createdat": _utcnow()}) redis = FakeRedis(raise_on={"hgetall"}, counter="1") facts = await make_agent(pg, redis)._gather_stall_facts("m1", "b1") # must not raise self.assertEqual(facts["booking_status"], "Miler_Assigned") # pg facts still gathered self.assertNotIn("position_unchanged_minutes", facts) # movement facts skipped self.assertEqual(facts["stalls_today"], 1) # counter still read async def test_counter_read_error_is_swallowed(self): redis = FakeRedis(movement={"updated_at": _utcnow().isoformat()}, raise_on={"get"}) facts = await make_agent(None, redis)._gather_stall_facts("m1", "b1") # must not raise self.assertIn("last_gps_ping_minutes_ago", facts) self.assertNotIn("stalls_today", facts) async def test_empty_everywhere(self): pg = FakePGPool(row=None) facts = await make_agent(pg, FakeRedis())._gather_stall_facts("m1", "b1") self.assertEqual(facts, {}) # nothing available → empty, no crash # ── Stall counter ──────────────────────────────────────────────────────────── class TestStallCounter(unittest.IsolatedAsyncioTestCase): async def test_counter_key_is_daily(self): agent = make_agent(None, FakeRedis()) key = agent._stall_counter_key("m42") self.assertTrue(key.startswith("miler:m42:stalls:")) self.assertEqual(key.split(":")[-1], _utcnow().strftime("%Y-%m-%d")) async def test_incr_sets_ttl(self): redis = FakeRedis(counter="0") agent = make_agent(None, redis) await agent._incr_stall_counter("m1") self.assertEqual(len(redis.incr_calls), 1) self.assertEqual(len(redis.expire_calls), 1) self.assertEqual(redis.expire_calls[0][1], 172800) # 48h TTL self.assertEqual(redis._counter, "1") async def test_incr_error_is_swallowed(self): redis = FakeRedis(raise_on={"incr"}) agent = make_agent(None, redis) await agent._incr_stall_counter("m1") # must not raise self.assertEqual(redis.incr_calls, []) if __name__ == "__main__": unittest.main()