DataOps-platform/tests/test_datasource_pool_diagnostics.py

from flask import Flask

from app.core.data_source.models import PoolStatus


UID = "01900000-0000-7000-8000-000000000010"


def healthy_status():
    return PoolStatus(
        data_source_uid=UID,
        credential_version=2,
        config_fingerprint="must-not-be-exposed",
        pool_state="healthy",
        leases=0,
        draining=False,
        created_at=1,
        last_used_at=2,
        pool_size=2,
        checked_out=0,
        checked_in=2,
        overflow=0,
    )


class FakeDiagnosticService:
    def pool_status(self, uid):
        assert uid == UID
        return healthy_status()

    def pool_statuses(self):
        return [healthy_status()]

    def invalidate_pool(self, uid, *, reason, actor_uid):
        assert uid == UID
        assert reason == "admin_reset"
        del actor_uid
        return {"uid": uid, "invalidated": True}


def make_client(monkeypatch):
    from app.api.data_source import routes

    monkeypatch.setattr(
        routes,
        "get_data_source_service",
        FakeDiagnosticService,
    )
    app = Flask(__name__)
    app.register_blueprint(routes.bp, url_prefix="/api/datasource")
    return app.test_client()


def test_admin_pool_status_is_safe(monkeypatch):
    client = make_client(monkeypatch)

    response = client.get(f"/api/datasource/{UID}/pool")
    body = response.get_json()["data"]
    text = response.get_data(as_text=True).lower()

    assert body["data_source_uid"] == UID
    assert body["pool_state"] == "healthy"
    assert "config_fingerprint" not in body
    assert "host" not in body
    assert "username" not in body
    assert "password" not in text


def test_pool_invalidation_accepts_only_fixed_reason(monkeypatch):
    client = make_client(monkeypatch)

    accepted = client.post(
        f"/api/datasource/{UID}/pool/invalidate",
        json={"reason": "admin_reset"},
    )
    rejected = client.post(
        f"/api/datasource/{UID}/pool/invalidate",
        json={"reason": "arbitrary text"},
    )

    assert accepted.status_code == 200
    assert rejected.status_code == 400


def test_platform_health_contains_only_pool_aggregates(monkeypatch):
    from app.core.data_source import runtime
    from app.core.system.health import data_source_pool_health

    class FakeManager:
        def snapshot(self):
            return [healthy_status()]

    monkeypatch.setattr(
        runtime,
        "peek_data_source_manager",
        lambda: FakeManager(),
    )

    assert data_source_pool_health() == {
        "active_pool_count": 1,
        "degraded_pool_count": 0,
        "open_circuit_count": 0,
        "checked_out_total": 0,
        "pool_timeout_total": 0,
    }


def test_platform_health_does_not_create_runtime_when_unused(monkeypatch):
    from app.core.data_source import runtime
    from app.core.system.health import data_source_pool_health

    monkeypatch.setattr(
        runtime,
        "peek_data_source_manager",
        lambda: None,
    )

    assert data_source_pool_health()["active_pool_count"] == 0


def test_pool_routes_are_admin_only_in_permission_matrix():
    from app.core.system.permissions import (
        DATASOURCE_POOL_MANAGE,
        permission_for_request,
    )

    assert permission_for_request(
        "/api/datasource/pools",
        "GET",
    ) == (DATASOURCE_POOL_MANAGE,)
    assert permission_for_request(
        f"/api/datasource/{UID}/pool/invalidate",
        "POST",
    ) == (DATASOURCE_POOL_MANAGE,)