NexaPG/backend/tests/e2e/test_api_smoke.py

import asyncio
import os
from datetime import datetime, timedelta, timezone
from uuid import uuid4

from fastapi.testclient import TestClient

from app.core.db import SessionLocal
from app.main import app
from app.models.models import Metric


def _admin_credentials() -> tuple[str, str]:
    return (
        os.getenv("INIT_ADMIN_EMAIL", "admin@example.com"),
        os.getenv("INIT_ADMIN_PASSWORD", "ChangeMe123!"),
    )


def _auth_headers(access_token: str) -> dict[str, str]:
    return {"Authorization": f"Bearer {access_token}"}


async def _insert_metric(target_id: int, metric_name: str, value: float) -> None:
    async with SessionLocal() as db:
        db.add(
            Metric(
                target_id=target_id,
                ts=datetime.now(timezone.utc),
                metric_name=metric_name,
                value=value,
                labels={},
            )
        )
        await db.commit()


def test_core_api_smoke_suite() -> None:
    admin_email, admin_password = _admin_credentials()
    unique = uuid4().hex[:8]
    target_name = f"smoke-target-{unique}"
    user_email = f"smoke-user-{unique}@example.com"

    with TestClient(app) as client:
        # Auth: login
        login_res = client.post(
            "/api/v1/auth/login",
            json={"email": admin_email, "password": admin_password},
        )
        assert login_res.status_code == 200, login_res.text
        tokens = login_res.json()
        assert tokens.get("access_token")
        assert tokens.get("refresh_token")
        headers = _auth_headers(tokens["access_token"])

        # Auth: me
        me_res = client.get("/api/v1/me", headers=headers)
        assert me_res.status_code == 200, me_res.text
        assert me_res.json()["email"] == admin_email

        # Targets: create
        create_target_res = client.post(
            "/api/v1/targets",
            headers=headers,
            json={
                "name": target_name,
                "host": "127.0.0.1",
                "port": 5432,
                "dbname": "postgres",
                "username": "postgres",
                "password": "postgres",
                "sslmode": "disable",
                "use_pg_stat_statements": False,
                "owner_user_ids": [],
                "tags": {"suite": "e2e-smoke"},
            },
        )
        assert create_target_res.status_code == 201, create_target_res.text
        target = create_target_res.json()
        target_id = target["id"]

        # Targets: list/get/update
        list_targets_res = client.get("/api/v1/targets", headers=headers)
        assert list_targets_res.status_code == 200, list_targets_res.text
        assert any(item["id"] == target_id for item in list_targets_res.json())

        get_target_res = client.get(f"/api/v1/targets/{target_id}", headers=headers)
        assert get_target_res.status_code == 200, get_target_res.text

        update_target_res = client.put(
            f"/api/v1/targets/{target_id}",
            headers=headers,
            json={"name": f"{target_name}-updated"},
        )
        assert update_target_res.status_code == 200, update_target_res.text
        assert update_target_res.json()["name"].endswith("-updated")

        # Metrics access
        asyncio.run(_insert_metric(target_id, "connections_total", 7.0))
        now = datetime.now(timezone.utc)
        from_ts = (now - timedelta(minutes=5)).isoformat()
        to_ts = (now + timedelta(minutes=5)).isoformat()
        metrics_res = client.get(
            f"/api/v1/targets/{target_id}/metrics",
            headers=headers,
            params={"metric": "connections_total", "from": from_ts, "to": to_ts},
        )
        assert metrics_res.status_code == 200, metrics_res.text
        assert isinstance(metrics_res.json(), list)
        assert len(metrics_res.json()) >= 1

        # Alerts status
        alerts_status_res = client.get("/api/v1/alerts/status", headers=headers)
        assert alerts_status_res.status_code == 200, alerts_status_res.text
        payload = alerts_status_res.json()
        assert "warnings" in payload
        assert "alerts" in payload

        # Admin users: list/create/update/delete
        users_res = client.get("/api/v1/admin/users", headers=headers)
        assert users_res.status_code == 200, users_res.text
        assert isinstance(users_res.json(), list)

        create_user_res = client.post(
            "/api/v1/admin/users",
            headers=headers,
            json={
                "email": user_email,
                "first_name": "Smoke",
                "last_name": "User",
                "password": "SmokePass123!",
                "role": "viewer",
            },
        )
        assert create_user_res.status_code == 201, create_user_res.text
        created_user_id = create_user_res.json()["id"]

        update_user_res = client.put(
            f"/api/v1/admin/users/{created_user_id}",
            headers=headers,
            json={"role": "operator", "first_name": "SmokeUpdated"},
        )
        assert update_user_res.status_code == 200, update_user_res.text
        assert update_user_res.json()["role"] == "operator"

        delete_user_res = client.delete(f"/api/v1/admin/users/{created_user_id}", headers=headers)
        assert delete_user_res.status_code == 200, delete_user_res.text
        assert delete_user_res.json().get("status") == "deleted"

        # Cleanup target
        delete_target_res = client.delete(f"/api/v1/targets/{target_id}", headers=headers)
        assert delete_target_res.status_code == 200, delete_target_res.text
        assert delete_target_res.json().get("status") == "deleted"