feat add connections management and health summary improvements

backend/src/api/routes/__tests__/test_connections_routes.py

@@ -0,0 +1,72 @@
+# [DEF:backend.src.api.routes.__tests__.test_connections_routes:Module]
+# @TIER: STANDARD
+# @PURPOSE: Verifies connection routes bootstrap their table before CRUD access.
+# @LAYER: API
+# @RELATION: VERIFIES -> backend.src.api.routes.connections
+
+import os
+import sys
+import asyncio
+from pathlib import Path
+
+import pytest
+from sqlalchemy import create_engine, inspect
+from sqlalchemy.orm import sessionmaker
+from sqlalchemy.pool import StaticPool
+
+# Force SQLite in-memory for database module imports.
+os.environ["DATABASE_URL"] = "sqlite:///:memory:"
+os.environ["TASKS_DATABASE_URL"] = "sqlite:///:memory:"
+os.environ["AUTH_DATABASE_URL"] = "sqlite:///:memory:"
+os.environ["ENVIRONMENT"] = "testing"
+
+backend_dir = str(Path(__file__).parent.parent.parent.parent.resolve())
+if backend_dir not in sys.path:
+    sys.path.insert(0, backend_dir)
+
+
+@pytest.fixture
+def db_session():
+    engine = create_engine(
+        "sqlite:///:memory:",
+        connect_args={"check_same_thread": False},
+        poolclass=StaticPool,
+    )
+    session = sessionmaker(bind=engine)()
+    try:
+        yield session
+    finally:
+        session.close()
+
+
+def test_list_connections_bootstraps_missing_table(db_session):
+    from src.api.routes.connections import list_connections
+
+    result = asyncio.run(list_connections(db=db_session))
+
+    inspector = inspect(db_session.get_bind())
+    assert result == []
+    assert "connection_configs" in inspector.get_table_names()
+
+
+def test_create_connection_bootstraps_missing_table(db_session):
+    from src.api.routes.connections import ConnectionCreate, create_connection
+
+    payload = ConnectionCreate(
+        name="Analytics Warehouse",
+        type="postgres",
+        host="warehouse.internal",
+        port=5432,
+        database="analytics",
+        username="reporter",
+        password="secret",
+    )
+
+    created = asyncio.run(create_connection(connection=payload, db=db_session))
+
+    inspector = inspect(db_session.get_bind())
+    assert created.name == "Analytics Warehouse"
+    assert created.host == "warehouse.internal"
+    assert "connection_configs" in inspector.get_table_names()
+
+# [/DEF:backend.src.api.routes.__tests__.test_connections_routes:Module]