test: baseline test suite — 86 tests covering auth, CRUD, settings, chat, terraform, reports, terminal

This commit is contained in:
nogueiraguh
2026-04-03 17:31:39 -03:00
parent 856e3b6674
commit 2e74d316ae
13 changed files with 1126 additions and 0 deletions

View File

93
backend/tests/conftest.py Normal file
View File

@@ -0,0 +1,93 @@
"""
Shared fixtures for the OCI CIS Agent test suite.
Uses a temporary SQLite database and the real FastAPI app.
"""
import os
import sys
import tempfile
import pytest
from httpx import AsyncClient, ASGITransport
# Ensure backend is importable
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
# Use temp dir for all data (DB, configs, reports, etc.)
_tmpdir = tempfile.mkdtemp(prefix="cisagent_test_")
os.environ["DATA_DIR"] = _tmpdir
os.environ["APP_SECRET"] = "a" * 128 # valid 64-byte hex
@pytest.fixture(scope="session")
def anyio_backend():
return "asyncio"
@pytest.fixture(scope="session")
def app():
"""Create the FastAPI app with a fresh DB, manually init since ASGI transport skips lifespan."""
import app as app_module
# Disable rate limiting for tests
app_module._LOGIN_MAX = 999999
app_module.init_db()
return app_module.app
@pytest.fixture(scope="session")
async def client(app):
"""Async HTTP client for the test session."""
transport = ASGITransport(app=app)
async with AsyncClient(transport=transport, base_url="http://test") as c:
yield c
@pytest.fixture(scope="session")
async def admin_token(client):
"""Login as admin and return JWT token."""
r = await client.post("/api/auth/login", json={"username": "admin", "password": "admin123"})
assert r.status_code == 200, f"Admin login failed: {r.text}"
return r.json()["token"]
@pytest.fixture(scope="session")
def admin_headers(admin_token):
"""Auth headers for admin user."""
return {"Authorization": f"Bearer {admin_token}"}
@pytest.fixture(scope="session")
async def user_token(client, admin_headers):
"""Create a regular user and return their JWT token."""
r = await client.post("/api/auth/register", json={
"first_name": "Test", "last_name": "User",
"username": "testuser", "email": "test@test.com",
"password": "Test1234!", "role": "user"
}, headers=admin_headers)
assert r.status_code == 200, f"User creation failed: {r.text}"
r2 = await client.post("/api/auth/login", json={"username": "testuser", "password": "Test1234!"})
assert r2.status_code == 200
return r2.json()["token"]
@pytest.fixture(scope="session")
def user_headers(user_token):
"""Auth headers for regular user."""
return {"Authorization": f"Bearer {user_token}"}
@pytest.fixture(scope="session")
async def viewer_token(client, admin_headers):
"""Create a viewer user and return their JWT token."""
r = await client.post("/api/auth/register", json={
"first_name": "View", "last_name": "Only",
"username": "viewer", "email": "viewer@test.com",
"password": "View1234!", "role": "viewer"
}, headers=admin_headers)
assert r.status_code == 200
r2 = await client.post("/api/auth/login", json={"username": "viewer", "password": "View1234!"})
assert r2.status_code == 200
return r2.json()["token"]
@pytest.fixture(scope="session")
def viewer_headers(viewer_token):
return {"Authorization": f"Bearer {viewer_token}"}

View File

@@ -0,0 +1,111 @@
"""
Tests for ADB (Autonomous Database) Config CRUD operations,
including vector table management.
"""
import pytest
_state: dict = {}
@pytest.mark.anyio
async def test_create_adb_config(client, admin_headers):
"""POST /api/adb/config — create via multipart form."""
r = await client.post(
"/api/adb/config",
data={
"config_name": "TestADB",
"dsn": "test_high",
"username": "testuser",
"password": "testpass",
},
headers=admin_headers,
)
assert r.status_code == 200, r.text
body = r.json()
assert "id" in body
_state["vid"] = body["id"]
@pytest.mark.anyio
async def test_list_adb_configs(client, admin_headers):
"""GET /api/adb/configs — list must include created config."""
r = await client.get("/api/adb/configs", headers=admin_headers)
assert r.status_code == 200
configs = r.json()
assert isinstance(configs, list)
assert any(c["id"] == _state["vid"] for c in configs)
@pytest.mark.anyio
async def test_add_vector_table(client, admin_headers):
"""POST /api/adb/{vid}/tables — add a vector table."""
vid = _state["vid"]
r = await client.post(
f"/api/adb/{vid}/tables",
json={"table_name": "TEST_TABLE"},
headers=admin_headers,
)
assert r.status_code == 200, r.text
body = r.json()
assert "id" in body
_state["tid"] = body["id"]
@pytest.mark.anyio
async def test_list_vector_tables(client, admin_headers):
"""GET /api/adb/{vid}/tables — list must include created table."""
vid = _state["vid"]
r = await client.get(f"/api/adb/{vid}/tables", headers=admin_headers)
assert r.status_code == 200
tables = r.json()
assert isinstance(tables, list)
assert any(t["id"] == _state["tid"] for t in tables)
@pytest.mark.anyio
async def test_update_vector_table(client, admin_headers):
"""PUT /api/adb/{vid}/tables/{tid} — update description."""
vid = _state["vid"]
tid = _state["tid"]
r = await client.put(
f"/api/adb/{vid}/tables/{tid}",
json={"description": "updated"},
headers=admin_headers,
)
assert r.status_code == 200, r.text
@pytest.mark.anyio
async def test_delete_vector_table(client, admin_headers):
"""DELETE /api/adb/{vid}/tables/{tid} — remove table."""
vid = _state["vid"]
tid = _state["tid"]
r = await client.delete(
f"/api/adb/{vid}/tables/{tid}",
headers=admin_headers,
)
assert r.status_code == 200, r.text
@pytest.mark.anyio
async def test_update_adb_config(client, admin_headers):
"""PUT /api/adb/configs/{vid} — update config_name via multipart form."""
vid = _state["vid"]
r = await client.put(
f"/api/adb/configs/{vid}",
data={
"config_name": "UpdatedADB",
"dsn": "test_high",
"username": "testuser",
},
headers=admin_headers,
)
assert r.status_code == 200, r.text
@pytest.mark.anyio
async def test_delete_adb_config(client, admin_headers):
"""DELETE /api/adb/configs/{vid} — remove config."""
vid = _state["vid"]
r = await client.delete(f"/api/adb/configs/{vid}", headers=admin_headers)
assert r.status_code == 200, r.text

223
backend/tests/test_auth.py Normal file
View File

@@ -0,0 +1,223 @@
"""
Tests for authentication, registration, password change, logout, OIDC, and MFA endpoints.
"""
import uuid
import pytest
pytestmark = pytest.mark.anyio
# ---------------------------------------------------------------------------
# Helper: create a disposable user, login, return (user_id, token, headers)
# ---------------------------------------------------------------------------
async def _make_temp_user(client, admin_headers, *, password="Test1234!"):
"""Register a unique user and log them in. Returns (uid, token, headers)."""
uname = f"tmp_{uuid.uuid4().hex[:10]}"
r = await client.post("/api/auth/register", json={
"first_name": "Tmp", "last_name": "User",
"username": uname, "email": f"{uname}@test.com",
"password": password, "role": "user",
}, headers=admin_headers)
assert r.status_code == 200, r.text
uid = r.json()["id"]
r2 = await client.post("/api/auth/login", json={"username": uname, "password": password})
assert r2.status_code == 200, r2.text
token = r2.json()["token"]
return uid, token, {"Authorization": f"Bearer {token}"}
# ── Health ─────────────────────────────────────────────────────────────────────
async def test_health_check(client):
r = await client.get("/api/health")
assert r.status_code == 200
body = r.json()
assert body["status"] == "ok"
assert body["db_ok"] is True
# ── Login ──────────────────────────────────────────────────────────────────────
async def test_login_success(client):
r = await client.post("/api/auth/login", json={"username": "admin", "password": "admin123"})
assert r.status_code == 200
body = r.json()
assert "token" in body
assert "user" in body
async def test_login_wrong_password(client):
r = await client.post("/api/auth/login", json={"username": "admin", "password": "wrong"})
assert r.status_code == 401
async def test_login_nonexistent_user(client):
r = await client.post("/api/auth/login", json={"username": "nobody", "password": "x"})
assert r.status_code == 401
async def test_login_returns_must_change_password(client):
r = await client.post("/api/auth/login", json={"username": "admin", "password": "admin123"})
assert r.status_code == 200
body = r.json()
assert "must_change_password" in body
# ── Register ───────────────────────────────────────────────────────────────────
async def test_register_user_as_admin(client, admin_headers):
uname = f"regtest_{uuid.uuid4().hex[:8]}"
r = await client.post("/api/auth/register", json={
"first_name": "Reg", "last_name": "Test",
"username": uname, "email": f"{uname}@test.com",
"password": "Reg12345!", "role": "user",
}, headers=admin_headers)
assert r.status_code == 200
body = r.json()
assert body["username"] == uname
assert body["role"] == "user"
async def test_register_duplicate_username(client, admin_headers):
uname = f"dup_{uuid.uuid4().hex[:8]}"
payload = {
"first_name": "Dup", "last_name": "Test",
"username": uname, "email": f"{uname}@test.com",
"password": "Dup12345!", "role": "user",
}
r1 = await client.post("/api/auth/register", json=payload, headers=admin_headers)
assert r1.status_code == 200
r2 = await client.post("/api/auth/register", json=payload, headers=admin_headers)
assert r2.status_code == 400
async def test_register_as_non_admin(client, user_headers):
r = await client.post("/api/auth/register", json={
"first_name": "No", "last_name": "Way",
"username": "nope", "email": "nope@test.com",
"password": "Nope1234!", "role": "user",
}, headers=user_headers)
assert r.status_code == 403
async def test_register_invalid_role(client, admin_headers):
r = await client.post("/api/auth/register", json={
"first_name": "Bad", "last_name": "Role",
"username": f"badrole_{uuid.uuid4().hex[:8]}",
"email": "bad@test.com",
"password": "Bad12345!", "role": "superadmin",
}, headers=admin_headers)
assert r.status_code == 400
# ── Password change ────────────────────────────────────────────────────────────
async def test_change_password_success(client, admin_headers):
_, _, hdrs = await _make_temp_user(client, admin_headers)
r = await client.post("/api/auth/change-password", json={
"current_password": "Test1234!", "new_password": "NewPass1!",
}, headers=hdrs)
assert r.status_code == 200
async def test_change_password_wrong_current(client, admin_headers):
_, _, hdrs = await _make_temp_user(client, admin_headers)
r = await client.post("/api/auth/change-password", json={
"current_password": "WrongOld!", "new_password": "NewPass1!",
}, headers=hdrs)
assert r.status_code == 400
async def test_change_password_too_short(client, admin_headers):
_, _, hdrs = await _make_temp_user(client, admin_headers)
r = await client.post("/api/auth/change-password", json={
"current_password": "Test1234!", "new_password": "Ab1!",
}, headers=hdrs)
assert r.status_code == 400
async def test_change_password_no_uppercase(client, admin_headers):
_, _, hdrs = await _make_temp_user(client, admin_headers)
r = await client.post("/api/auth/change-password", json={
"current_password": "Test1234!", "new_password": "abcdefg1!",
}, headers=hdrs)
assert r.status_code == 400
async def test_change_password_no_lowercase(client, admin_headers):
_, _, hdrs = await _make_temp_user(client, admin_headers)
r = await client.post("/api/auth/change-password", json={
"current_password": "Test1234!", "new_password": "ABCDEFG1!",
}, headers=hdrs)
assert r.status_code == 400
async def test_change_password_no_number(client, admin_headers):
_, _, hdrs = await _make_temp_user(client, admin_headers)
r = await client.post("/api/auth/change-password", json={
"current_password": "Test1234!", "new_password": "Abcdefgh!",
}, headers=hdrs)
assert r.status_code == 400
async def test_change_password_no_special(client, admin_headers):
_, _, hdrs = await _make_temp_user(client, admin_headers)
r = await client.post("/api/auth/change-password", json={
"current_password": "Test1234!", "new_password": "Abcdefg1",
}, headers=hdrs)
assert r.status_code == 400
async def test_change_password_same_as_old(client, admin_headers):
_, _, hdrs = await _make_temp_user(client, admin_headers)
r = await client.post("/api/auth/change-password", json={
"current_password": "Test1234!", "new_password": "Test1234!",
}, headers=hdrs)
assert r.status_code == 400
# ── Logout ─────────────────────────────────────────────────────────────────────
async def test_logout(client, admin_headers):
_, _, hdrs = await _make_temp_user(client, admin_headers)
r = await client.post("/api/auth/logout", headers=hdrs)
assert r.status_code == 200
assert r.json().get("ok") is True
async def test_request_after_logout(client, admin_headers):
_, token, hdrs = await _make_temp_user(client, admin_headers)
await client.post("/api/auth/logout", headers=hdrs)
r = await client.get("/api/users/me", headers={"Authorization": f"Bearer {token}"})
assert r.status_code == 401
# ── OIDC ───────────────────────────────────────────────────────────────────────
async def test_oidc_config_public(client):
r = await client.get("/api/auth/oidc/config")
assert r.status_code == 200
assert "auth_mode" in r.json()
async def test_oidc_login_disabled(client):
r = await client.get("/api/auth/oidc/login", follow_redirects=False)
assert r.status_code == 400
# ── MFA ────────────────────────────────────────────────────────────────────────
async def test_mfa_setup(client, admin_headers):
_, _, hdrs = await _make_temp_user(client, admin_headers)
r = await client.post("/api/mfa/setup", headers=hdrs)
assert r.status_code == 200
body = r.json()
assert "secret" in body
assert "uri" in body
async def test_mfa_disable(client, admin_headers):
uid, _, _ = await _make_temp_user(client, admin_headers)
r = await client.post(f"/api/mfa/disable/{uid}", headers=admin_headers)
assert r.status_code == 200
assert r.json().get("ok") is True

View File

@@ -0,0 +1,59 @@
"""Tests for chat session management endpoints."""
import pytest
@pytest.mark.anyio
async def test_list_sessions_empty(client, user_headers):
r = await client.get("/api/chat/sessions", headers=user_headers)
assert r.status_code == 200
body = r.json()
assert isinstance(body, list)
@pytest.mark.anyio
async def test_create_session_via_message(client, user_headers):
"""POST /api/chat with a fake genai config.
The endpoint should respond without a 500 — it may return 200 with a
message_id (background processing will fail) or 400/404 if the config
is not found.
"""
r = await client.post(
"/api/chat",
json={"message": "hello", "genai_config_id": "fake"},
headers=user_headers,
)
assert r.status_code != 500, f"Unexpected server error: {r.text}"
@pytest.mark.anyio
async def test_list_sessions(client, user_headers):
r = await client.get("/api/chat/sessions", headers=user_headers)
assert r.status_code == 200
assert isinstance(r.json(), list)
@pytest.mark.anyio
async def test_rename_session(client, user_headers):
"""Rename a session if one exists, otherwise skip gracefully."""
sessions = (await client.get("/api/chat/sessions", headers=user_headers)).json()
if not sessions:
pytest.skip("No chat session available to rename")
sid = sessions[0]["id"]
r = await client.put(
f"/api/chat/sessions/{sid}/title",
json={"title": "renamed"},
headers=user_headers,
)
assert r.status_code == 200
@pytest.mark.anyio
async def test_delete_session(client, user_headers):
"""Delete a session if one exists, otherwise skip gracefully."""
sessions = (await client.get("/api/chat/sessions", headers=user_headers)).json()
if not sessions:
pytest.skip("No chat session available to delete")
sid = sessions[0]["id"]
r = await client.delete(f"/api/chat/{sid}", headers=user_headers)
assert r.status_code == 200

View File

@@ -0,0 +1,103 @@
"""
Tests for GenAI Config CRUD operations.
"""
import io
import pytest
# Dummy PEM key for OCI config prerequisite
_DUMMY_PEM = b"""-----BEGIN RSA PRIVATE KEY-----
MIIBogIBAAJBALRiMLAHudeSA/x3hB2f+2NRkJmcOUNKwNuvTT0GKQFYSNDF0FNv
TESTDATAFORTHEMOCKPEMFILEXYZABC0123456789ABCDEF0123456789
-----END RSA PRIVATE KEY-----
"""
_state: dict = {}
@pytest.fixture(scope="module")
async def oci_config_id(client, admin_headers):
"""Create an OCI config that GenAI configs can reference."""
r = await client.post(
"/api/oci/config",
data={
"tenancy_name": "GenAI-OCI-Dep",
"tenancy_ocid": "ocid1.tenancy.oc1..genaidep",
"user_ocid": "ocid1.user.oc1..genaidep",
"fingerprint": "11:22:33:44",
"region": "us-ashburn-1",
"auth_type": "api_key",
},
files={"private_key": ("key.pem", io.BytesIO(_DUMMY_PEM), "application/x-pem-file")},
headers=admin_headers,
)
assert r.status_code == 200, r.text
cid = r.json()["id"]
yield cid
await client.delete(f"/api/oci/configs/{cid}", headers=admin_headers)
@pytest.mark.anyio
async def test_get_models(client, admin_headers):
"""GET /api/genai/models — returns models dict and regions list."""
r = await client.get("/api/genai/models", headers=admin_headers)
assert r.status_code == 200
body = r.json()
assert "models" in body and isinstance(body["models"], dict)
assert "regions" in body and isinstance(body["regions"], list)
@pytest.mark.anyio
async def test_create_genai_config(client, admin_headers, oci_config_id):
"""POST /api/genai/config — create via JSON body."""
r = await client.post(
"/api/genai/config",
json={
"name": "TestGenAI",
"oci_config_id": oci_config_id,
"model_id": "openai.gpt-4.1",
"compartment_id": "ocid1.compartment.oc1..test",
"genai_region": "us-ashburn-1",
},
headers=admin_headers,
)
assert r.status_code == 200, r.text
body = r.json()
assert "id" in body
_state["gid"] = body["id"]
@pytest.mark.anyio
async def test_list_genai_configs(client, admin_headers):
"""GET /api/genai/configs — list must include created config."""
r = await client.get("/api/genai/configs", headers=admin_headers)
assert r.status_code == 200
configs = r.json()
assert isinstance(configs, list)
assert any(c["id"] == _state["gid"] for c in configs)
@pytest.mark.anyio
async def test_update_genai_config(client, admin_headers, oci_config_id):
"""PUT /api/genai/configs/{gid} — update temperature."""
gid = _state["gid"]
r = await client.put(
f"/api/genai/configs/{gid}",
json={
"name": "TestGenAI",
"oci_config_id": oci_config_id,
"model_id": "openai.gpt-4.1",
"compartment_id": "ocid1.compartment.oc1..test",
"genai_region": "us-ashburn-1",
"temperature": 0.5,
},
headers=admin_headers,
)
assert r.status_code == 200, r.text
@pytest.mark.anyio
async def test_delete_genai_config(client, admin_headers):
"""DELETE /api/genai/configs/{gid} — remove config."""
gid = _state["gid"]
r = await client.delete(f"/api/genai/configs/{gid}", headers=admin_headers)
assert r.status_code == 200, r.text

View File

@@ -0,0 +1,68 @@
"""
Tests for MCP Server CRUD operations.
"""
import pytest
_state: dict = {}
@pytest.mark.anyio
async def test_create_mcp_server(client, admin_headers):
"""POST /api/mcp/servers — create via JSON body."""
r = await client.post(
"/api/mcp/servers",
json={
"name": "TestMCP",
"server_type": "stdio",
"command": "python3 test.py",
},
headers=admin_headers,
)
assert r.status_code == 200, r.text
body = r.json()
assert "id" in body
_state["mid"] = body["id"]
@pytest.mark.anyio
async def test_list_mcp_servers(client, admin_headers):
"""GET /api/mcp/servers — list must include created server."""
r = await client.get("/api/mcp/servers", headers=admin_headers)
assert r.status_code == 200
servers = r.json()
assert isinstance(servers, list)
assert any(s["id"] == _state["mid"] for s in servers)
@pytest.mark.anyio
async def test_update_mcp_server(client, admin_headers):
"""PUT /api/mcp/servers/{mid} — update name."""
mid = _state["mid"]
r = await client.put(
f"/api/mcp/servers/{mid}",
json={
"name": "UpdatedMCP",
"server_type": "stdio",
"command": "python3 test.py",
},
headers=admin_headers,
)
assert r.status_code == 200, r.text
@pytest.mark.anyio
async def test_toggle_mcp_server(client, admin_headers):
"""PUT /api/mcp/servers/{mid}/toggle — toggle active state."""
mid = _state["mid"]
r = await client.put(f"/api/mcp/servers/{mid}/toggle", headers=admin_headers)
assert r.status_code == 200, r.text
body = r.json()
assert "is_active" in body
@pytest.mark.anyio
async def test_delete_mcp_server(client, admin_headers):
"""DELETE /api/mcp/servers/{mid} — remove server."""
mid = _state["mid"]
r = await client.delete(f"/api/mcp/servers/{mid}", headers=admin_headers)
assert r.status_code == 200, r.text

View File

@@ -0,0 +1,101 @@
"""
Tests for OCI Config CRUD operations.
"""
import io
import pytest
# Dummy PEM key for uploads (valid RSA format header/footer, not a real key)
_DUMMY_PEM = b"""-----BEGIN RSA PRIVATE KEY-----
MIIBogIBAAJBALRiMLAHudeSA/x3hB2f+2NRkJmcOUNKwNuvTT0GKQFYSNDF0FNv
TESTDATAFORTHEMOCKPEMFILEXYZABC0123456789ABCDEF0123456789
-----END RSA PRIVATE KEY-----
"""
# Module-level storage for IDs across ordered tests
_state: dict = {}
@pytest.mark.anyio
async def test_create_oci_config(client, admin_headers):
"""POST /api/oci/config — create via multipart form."""
r = await client.post(
"/api/oci/config",
data={
"tenancy_name": "TestTenancy",
"tenancy_ocid": "ocid1.tenancy.oc1..test",
"user_ocid": "ocid1.user.oc1..test",
"fingerprint": "aa:bb:cc:dd",
"region": "us-ashburn-1",
"compartment_id": "ocid1.compartment.oc1..test",
"auth_type": "api_key",
},
files={"private_key": ("key.pem", io.BytesIO(_DUMMY_PEM), "application/x-pem-file")},
headers=admin_headers,
)
assert r.status_code == 200, r.text
body = r.json()
assert "id" in body
_state["cid"] = body["id"]
@pytest.mark.anyio
async def test_list_oci_configs(client, admin_headers):
"""GET /api/oci/configs — list must include the created config."""
r = await client.get("/api/oci/configs", headers=admin_headers)
assert r.status_code == 200
configs = r.json()
assert isinstance(configs, list)
assert any(c["id"] == _state["cid"] for c in configs)
@pytest.mark.anyio
async def test_update_oci_config(client, admin_headers):
"""PUT /api/oci/configs/{cid} — update tenancy_name via multipart form."""
cid = _state["cid"]
r = await client.put(
f"/api/oci/configs/{cid}",
data={
"tenancy_name": "Updated",
"region": "us-ashburn-1",
},
headers=admin_headers,
)
assert r.status_code == 200, r.text
@pytest.mark.anyio
async def test_delete_oci_config(client, admin_headers):
"""DELETE /api/oci/configs/{cid} — remove config."""
cid = _state["cid"]
r = await client.delete(f"/api/oci/configs/{cid}", headers=admin_headers)
assert r.status_code == 200, r.text
@pytest.mark.anyio
async def test_user_isolation(client, admin_headers, user_headers):
"""User A creates a config; user B cannot see it in their list."""
# Admin creates a config
r = await client.post(
"/api/oci/config",
data={
"tenancy_name": "IsolationTest",
"tenancy_ocid": "ocid1.tenancy.oc1..isolation",
"user_ocid": "ocid1.user.oc1..isolation",
"fingerprint": "ee:ff:00:11",
"region": "eu-frankfurt-1",
"auth_type": "api_key",
},
files={"private_key": ("key.pem", io.BytesIO(_DUMMY_PEM), "application/x-pem-file")},
headers=admin_headers,
)
assert r.status_code == 200
admin_cid = r.json()["id"]
# Regular user lists configs — should NOT see admin's config
r2 = await client.get("/api/oci/configs", headers=user_headers)
assert r2.status_code == 200
user_ids = [c["id"] for c in r2.json()]
assert admin_cid not in user_ids
# Cleanup
await client.delete(f"/api/oci/configs/{admin_cid}", headers=admin_headers)

View File

@@ -0,0 +1,17 @@
"""Tests for reports and CIS engine endpoints."""
import pytest
@pytest.mark.anyio
async def test_list_reports_empty(client, user_headers):
r = await client.get("/api/reports", headers=user_headers)
assert r.status_code == 200
assert isinstance(r.json(), list)
@pytest.mark.anyio
async def test_cis_engine_version(client, admin_headers):
r = await client.get("/api/cis-engine/version", headers=admin_headers)
assert r.status_code == 200
body = r.json()
assert "version" in body

View File

@@ -0,0 +1,168 @@
"""Tests for settings, prompts, audit log, and config endpoints."""
import pytest
@pytest.mark.anyio
async def test_get_setting(client, admin_headers):
r = await client.get("/api/settings/auth_mode", headers=admin_headers)
assert r.status_code == 200
body = r.json()
assert "key" in body
assert "value" in body
@pytest.mark.anyio
async def test_put_setting(client, admin_headers):
r = await client.put(
"/api/settings/test_key",
json={"value": "test_value"},
headers=admin_headers,
)
assert r.status_code == 200
@pytest.mark.anyio
async def test_put_setting_non_admin(client, user_headers):
r = await client.put(
"/api/settings/test_key",
json={"value": "hack"},
headers=user_headers,
)
assert r.status_code == 403
@pytest.mark.anyio
async def test_sensitive_setting_blocked(client, user_headers):
r = await client.get("/api/settings/oidc_client_secret", headers=user_headers)
assert r.status_code == 403
@pytest.mark.anyio
async def test_sensitive_setting_admin(client, admin_headers):
r = await client.get("/api/settings/oidc_client_secret", headers=admin_headers)
assert r.status_code == 200
@pytest.mark.anyio
async def test_timezone_options(client, admin_headers):
r = await client.get("/api/settings/timezone/options", headers=admin_headers)
assert r.status_code == 200
body = r.json()
assert "timezones" in body
assert isinstance(body["timezones"], list)
@pytest.mark.anyio
async def test_get_system_prompts(client, admin_headers):
r = await client.get("/api/prompts/chat", headers=admin_headers)
assert r.status_code == 200
assert isinstance(r.json(), list)
@pytest.mark.anyio
async def test_create_system_prompt(client, admin_headers):
r = await client.post(
"/api/prompts",
json={
"agent": "chat",
"name": "test",
"content": "You are a test",
"is_active": False,
},
headers=admin_headers,
)
assert r.status_code == 200
@pytest.mark.anyio
async def test_update_system_prompt(client, admin_headers):
# Create a prompt first
r = await client.post(
"/api/prompts",
json={
"agent": "chat",
"name": "to_update",
"content": "original",
"is_active": False,
},
headers=admin_headers,
)
assert r.status_code == 200
pid = r.json().get("id")
assert pid is not None
r2 = await client.put(
f"/api/prompts/{pid}",
json={"content": "updated"},
headers=admin_headers,
)
assert r2.status_code == 200
@pytest.mark.anyio
async def test_delete_system_prompt(client, admin_headers):
# Create a prompt to delete
r = await client.post(
"/api/prompts",
json={
"agent": "chat",
"name": "to_delete",
"content": "temp",
"is_active": False,
},
headers=admin_headers,
)
assert r.status_code == 200
pid = r.json().get("id")
assert pid is not None
r2 = await client.delete(f"/api/prompts/{pid}", headers=admin_headers)
assert r2.status_code == 200
@pytest.mark.anyio
async def test_create_prompt_non_admin(client, user_headers):
r = await client.post(
"/api/prompts",
json={
"agent": "chat",
"name": "hacked",
"content": "nope",
"is_active": False,
},
headers=user_headers,
)
assert r.status_code == 403
@pytest.mark.anyio
async def test_audit_log(client, admin_headers):
r = await client.get("/api/audit-log", headers=admin_headers)
assert r.status_code == 200
assert isinstance(r.json(), list)
@pytest.mark.anyio
async def test_audit_log_user_sees_own(client, user_headers):
r = await client.get("/api/audit-log", headers=user_headers)
assert r.status_code == 200
@pytest.mark.anyio
async def test_config_logs(client, admin_headers):
r = await client.get("/api/config-logs", headers=admin_headers)
assert r.status_code == 200
assert isinstance(r.json(), list)
@pytest.mark.anyio
async def test_config_export(client, admin_headers):
r = await client.get("/api/config/export", headers=admin_headers)
assert r.status_code == 200
assert isinstance(r.json(), dict)
@pytest.mark.anyio
async def test_config_export_non_admin(client, user_headers):
r = await client.get("/api/config/export", headers=user_headers)
assert r.status_code == 403

View File

@@ -0,0 +1,17 @@
"""Tests for OCI CLI Terminal endpoints."""
import pytest
@pytest.mark.anyio
async def test_terminal_history_empty(client, user_headers):
r = await client.get("/api/terminal/history", headers=user_headers)
assert r.status_code == 200
assert isinstance(r.json(), list)
@pytest.mark.anyio
async def test_terminal_completions(client, user_headers):
r = await client.get(
"/api/terminal/completions", params={"prefix": "oci"}, headers=user_headers
)
assert r.status_code == 200

View File

@@ -0,0 +1,77 @@
"""Tests for Terraform workspace endpoints."""
import pytest
_state = {}
@pytest.mark.anyio
async def test_list_workspaces_empty(client, admin_headers):
r = await client.get("/api/terraform/workspaces", headers=admin_headers)
assert r.status_code == 200
@pytest.mark.anyio
async def test_create_workspace(client, admin_headers):
r = await client.post(
"/api/terraform/workspaces",
json={
"session_id": "test-sess",
"oci_config_id": "fake",
"name": "test-ws",
"tf_code": "resource {}",
},
headers=admin_headers,
)
assert r.status_code == 200
body = r.json()
assert "id" in body
_state["wid"] = body["id"]
@pytest.mark.anyio
async def test_list_workspaces(client, admin_headers):
r = await client.get("/api/terraform/workspaces", headers=admin_headers)
assert r.status_code == 200
workspaces = r.json()
assert isinstance(workspaces, list)
# Workspace may not appear in list if session_id is fictional (JOIN with sessions)
# The important check is that the endpoint returns 200 with a list
assert isinstance(workspaces, list)
@pytest.mark.anyio
async def test_get_workspace(client, admin_headers):
wid = _state.get("wid")
assert wid, "Workspace not created"
r = await client.get(f"/api/terraform/workspaces/{wid}", headers=admin_headers)
assert r.status_code == 200
assert "tf_code" in r.json()
@pytest.mark.anyio
async def test_update_code(client, admin_headers):
wid = _state.get("wid")
assert wid, "Workspace not created"
r = await client.put(
f"/api/terraform/workspaces/{wid}/code",
json={"tf_code": "# updated"},
headers=admin_headers,
)
assert r.status_code == 200
@pytest.mark.anyio
async def test_workspace_status(client, admin_headers):
wid = _state.get("wid")
assert wid, "Workspace not created"
r = await client.get(f"/api/terraform/workspaces/{wid}/status", headers=admin_headers)
assert r.status_code == 200
assert "status" in r.json()
@pytest.mark.anyio
async def test_delete_workspace(client, admin_headers):
wid = _state.get("wid")
assert wid, "Workspace not created"
r = await client.delete(f"/api/terraform/workspaces/{wid}", headers=admin_headers)
assert r.status_code == 200

View File

@@ -0,0 +1,89 @@
"""
Tests for user management endpoints.
"""
import uuid
import pytest
pytestmark = pytest.mark.anyio
# ---------------------------------------------------------------------------
# Helper
# ---------------------------------------------------------------------------
async def _create_user(client, admin_headers, *, role="user"):
"""Register a unique user and return their id."""
uname = f"usr_{uuid.uuid4().hex[:10]}"
r = await client.post("/api/auth/register", json={
"first_name": "U", "last_name": "Test",
"username": uname, "email": f"{uname}@test.com",
"password": "Usr12345!", "role": role,
}, headers=admin_headers)
assert r.status_code == 200, r.text
return r.json()["id"]
# ── /api/users/me ──────────────────────────────────────────────────────────────
async def test_get_me(client, user_headers):
r = await client.get("/api/users/me", headers=user_headers)
assert r.status_code == 200
body = r.json()
for field in ("id", "username", "role", "auth_provider", "must_change_password"):
assert field in body, f"Missing field: {field}"
# ── /api/users (list) ─────────────────────────────────────────────────────────
async def test_list_users_admin(client, admin_headers):
r = await client.get("/api/users", headers=admin_headers)
assert r.status_code == 200
body = r.json()
assert isinstance(body, list)
assert len(body) > 0
assert "auth_provider" in body[0]
async def test_list_users_non_admin(client, user_headers):
r = await client.get("/api/users", headers=user_headers)
assert r.status_code == 403
# ── /api/users/{uid} (update) ─────────────────────────────────────────────────
async def test_update_user_role(client, admin_headers):
uid = await _create_user(client, admin_headers)
r = await client.put(f"/api/users/{uid}", json={"role": "viewer"}, headers=admin_headers)
assert r.status_code == 200
assert r.json().get("ok") is True
async def test_update_user_timezone(client, admin_headers):
uid = await _create_user(client, admin_headers)
r = await client.put(f"/api/users/{uid}", json={"timezone": "UTC"}, headers=admin_headers)
assert r.status_code == 200
assert r.json().get("ok") is True
# ── /api/users/{uid} (delete / deactivate) ────────────────────────────────────
async def test_delete_user(client, admin_headers):
uid = await _create_user(client, admin_headers)
r = await client.delete(f"/api/users/{uid}", headers=admin_headers)
assert r.status_code == 200
assert r.json().get("ok") is True
# ── /api/users/me/timezone ─────────────────────────────────────────────────────
async def test_get_user_timezone(client, user_headers):
r = await client.get("/api/users/me/timezone", headers=user_headers)
assert r.status_code == 200
assert "timezone" in r.json()
async def test_set_user_timezone(client, user_headers):
r = await client.put("/api/users/me/timezone", json={"timezone": "UTC"}, headers=user_headers)
assert r.status_code == 200
body = r.json()
assert body.get("ok") is True
assert body.get("timezone") == "UTC"