""" Tests for PostgresIncidentStore, AutoIncidentStore, and INCIDENT_BACKEND=auto logic. """ import os import sys import json import time import threading import unittest from pathlib import Path from unittest.mock import MagicMock, patch, PropertyMock import tempfile ROOT = Path(__file__).resolve().parent.parent ROUTER = ROOT / "services" / "router" if str(ROUTER) not in sys.path: sys.path.insert(0, str(ROUTER)) class TestPostgresIncidentStore: """Unit tests for PostgresIncidentStore using mocked psycopg2.""" def _make_store(self): """Create a PostgresIncidentStore with a mocked DB connection.""" mock_psycopg2 = MagicMock() mock_conn = MagicMock() mock_conn.closed = False mock_psycopg2.connect.return_value = mock_conn mock_cursor = MagicMock() mock_conn.cursor.return_value = mock_cursor with patch.dict("sys.modules", {"psycopg2": mock_psycopg2, "psycopg2.extras": MagicMock()}): from importlib import reload import incident_store reload(incident_store) store = incident_store.PostgresIncidentStore("postgresql://test:test@localhost/test") store._local = threading.local() store._local.conn = mock_conn return store, mock_cursor, mock_conn def test_create_incident(self): store, mock_cursor, _ = self._make_store() result = store.create_incident({ "service": "gateway", "severity": "P1", "title": "Test incident", "started_at": "2025-01-01T00:00:00Z", }) assert result["status"] == "open" assert result["id"].startswith("inc_") assert mock_cursor.execute.called def test_get_incident_not_found(self): store, mock_cursor, _ = self._make_store() mock_cursor.fetchone.return_value = None result = store.get_incident("nonexistent") assert result is None def test_list_incidents_with_filters(self): store, mock_cursor, _ = self._make_store() mock_cursor.description = [("id",), ("workspace_id",), ("service",), ("env",), ("severity",), ("status",), ("title",), ("summary",), ("started_at",), ("ended_at",), ("created_by",), ("created_at",), ("updated_at",)] mock_cursor.fetchall.return_value = [] result = store.list_incidents({"service": "gateway", "status": "open"}, limit=10) assert isinstance(result, list) sql_call = mock_cursor.execute.call_args[0][0] assert "service=%s" in sql_call assert "status=%s" in sql_call def test_close_incident(self): store, mock_cursor, _ = self._make_store() mock_cursor.fetchone.return_value = ("inc_test",) result = store.close_incident("inc_test", "2025-01-02T00:00:00Z", "Fixed") assert result["status"] == "closed" def test_append_event(self): store, mock_cursor, _ = self._make_store() result = store.append_event("inc_test", "note", "test message", {"key": "val"}) assert result["type"] == "note" assert mock_cursor.execute.called class TestAutoIncidentStore: """Tests for AutoIncidentStore with Postgres → JSONL fallback.""" def test_fallback_on_pg_failure(self): from incident_store import AutoIncidentStore with tempfile.TemporaryDirectory() as tmpdir: store = AutoIncidentStore( pg_dsn="postgresql://invalid:invalid@localhost:1/none", jsonl_dir=tmpdir, ) result = store.create_incident({ "service": "test-svc", "title": "Test fallback", "severity": "P2", "started_at": "2025-01-01T00:00:00Z", }) assert result["id"].startswith("inc_") assert store._using_fallback is True assert store.active_backend() == "jsonl_fallback" def test_recovery_resets_after_interval(self): from incident_store import AutoIncidentStore with tempfile.TemporaryDirectory() as tmpdir: store = AutoIncidentStore( pg_dsn="postgresql://invalid:invalid@localhost:1/none", jsonl_dir=tmpdir, ) store.create_incident({ "service": "test", "title": "Initial fail", "severity": "P2", }) assert store._using_fallback is True store._fallback_since = time.monotonic() - 400 store._maybe_recover() assert store._using_fallback is False def test_active_backend_reflects_state(self): from incident_store import AutoIncidentStore with tempfile.TemporaryDirectory() as tmpdir: store = AutoIncidentStore( pg_dsn="postgresql://invalid:invalid@localhost:1/none", jsonl_dir=tmpdir, ) assert store.active_backend() == "postgres" store._using_fallback = True assert store.active_backend() == "jsonl_fallback" def test_list_and_get_after_fallback(self): from incident_store import AutoIncidentStore with tempfile.TemporaryDirectory() as tmpdir: store = AutoIncidentStore( pg_dsn="postgresql://invalid:invalid@localhost:1/none", jsonl_dir=tmpdir, ) inc = store.create_incident({ "service": "api", "title": "Test list", "severity": "P1", }) inc_id = inc["id"] store.append_event(inc_id, "note", "some event") fetched = store.get_incident(inc_id) assert fetched is not None assert fetched["service"] == "api" listed = store.list_incidents() assert any(i["id"] == inc_id for i in listed) class TestCreateStoreFactory: """Tests for _create_store() with INCIDENT_BACKEND env var.""" def test_backend_memory(self): from incident_store import _create_store, MemoryIncidentStore with patch.dict(os.environ, {"INCIDENT_BACKEND": "memory"}): store = _create_store() assert isinstance(store, MemoryIncidentStore) def test_backend_jsonl_default(self): from incident_store import _create_store, JsonlIncidentStore env = {"INCIDENT_BACKEND": "jsonl", "INCIDENT_JSONL_DIR": "/tmp/test_inc"} with patch.dict(os.environ, env, clear=False): store = _create_store() assert isinstance(store, JsonlIncidentStore) def test_backend_auto_with_dsn(self): from incident_store import _create_store, AutoIncidentStore env = {"INCIDENT_BACKEND": "auto", "DATABASE_URL": "postgresql://x:x@localhost/test"} with patch.dict(os.environ, env, clear=False): store = _create_store() assert isinstance(store, AutoIncidentStore) def test_backend_auto_without_dsn(self): from incident_store import _create_store, JsonlIncidentStore env = {"INCIDENT_BACKEND": "auto"} env_clear = {k: v for k, v in os.environ.items() if k not in ("DATABASE_URL", "INCIDENT_DATABASE_URL")} env_clear["INCIDENT_BACKEND"] = "auto" with patch.dict(os.environ, env_clear, clear=True): store = _create_store() assert isinstance(store, JsonlIncidentStore) def test_backend_postgres_without_dsn_falls_back(self): from incident_store import _create_store, JsonlIncidentStore env = {"INCIDENT_BACKEND": "postgres", "INCIDENT_JSONL_DIR": "/tmp/test_inc_pg"} env_clear = {k: v for k, v in os.environ.items() if k not in ("DATABASE_URL", "INCIDENT_DATABASE_URL")} env_clear.update(env) with patch.dict(os.environ, env_clear, clear=True): store = _create_store() assert isinstance(store, JsonlIncidentStore)