[NX-501 Issue] Add E2E API smoke test workflow and related test suite
Some checks failed
Container CVE Scan (development) / Scan backend/frontend images for CVEs (push) Successful in 2m45s
E2E API Smoke / Core API E2E Smoke (push) Failing after 24s
PostgreSQL Compatibility Matrix / PG14 smoke (push) Successful in 7s
PostgreSQL Compatibility Matrix / PG15 smoke (push) Successful in 8s
PostgreSQL Compatibility Matrix / PG16 smoke (push) Successful in 7s
PostgreSQL Compatibility Matrix / PG17 smoke (push) Successful in 8s
PostgreSQL Compatibility Matrix / PG18 smoke (push) Successful in 8s
Proxy Profile Validation / validate (push) Successful in 3s
Python Dependency Security / pip-audit (block high/critical) (push) Successful in 26s

This commit introduces a GitHub Actions workflow for running E2E API smoke tests on main branches and pull requests. It includes a test suite covering authentication, CRUD operations, metrics access, and alerts status. The README is updated with instructions for running the tests locally.
This commit is contained in:
2026-02-15 19:44:33 +01:00
parent 597579376f
commit 3e317abda8
3 changed files with 248 additions and 0 deletions

View File

@@ -0,0 +1,153 @@
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"