diff --git a/backend/tests/__init__.py b/backend/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/backend/tests/conftest.py b/backend/tests/conftest.py new file mode 100644 index 0000000..6c00f98 --- /dev/null +++ b/backend/tests/conftest.py @@ -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}"} diff --git a/backend/tests/test_adb_config.py b/backend/tests/test_adb_config.py new file mode 100644 index 0000000..14a2545 --- /dev/null +++ b/backend/tests/test_adb_config.py @@ -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 diff --git a/backend/tests/test_auth.py b/backend/tests/test_auth.py new file mode 100644 index 0000000..06864d5 --- /dev/null +++ b/backend/tests/test_auth.py @@ -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 diff --git a/backend/tests/test_chat.py b/backend/tests/test_chat.py new file mode 100644 index 0000000..b2a1b16 --- /dev/null +++ b/backend/tests/test_chat.py @@ -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 diff --git a/backend/tests/test_genai_config.py b/backend/tests/test_genai_config.py new file mode 100644 index 0000000..b22678d --- /dev/null +++ b/backend/tests/test_genai_config.py @@ -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 diff --git a/backend/tests/test_mcp_servers.py b/backend/tests/test_mcp_servers.py new file mode 100644 index 0000000..66f9b0b --- /dev/null +++ b/backend/tests/test_mcp_servers.py @@ -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 diff --git a/backend/tests/test_oci_config.py b/backend/tests/test_oci_config.py new file mode 100644 index 0000000..d2d4aa8 --- /dev/null +++ b/backend/tests/test_oci_config.py @@ -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) diff --git a/backend/tests/test_reports.py b/backend/tests/test_reports.py new file mode 100644 index 0000000..47590d9 --- /dev/null +++ b/backend/tests/test_reports.py @@ -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 diff --git a/backend/tests/test_settings.py b/backend/tests/test_settings.py new file mode 100644 index 0000000..93d577d --- /dev/null +++ b/backend/tests/test_settings.py @@ -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 diff --git a/backend/tests/test_terminal.py b/backend/tests/test_terminal.py new file mode 100644 index 0000000..cd02556 --- /dev/null +++ b/backend/tests/test_terminal.py @@ -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 diff --git a/backend/tests/test_terraform.py b/backend/tests/test_terraform.py new file mode 100644 index 0000000..3c0efc0 --- /dev/null +++ b/backend/tests/test_terraform.py @@ -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 diff --git a/backend/tests/test_users.py b/backend/tests/test_users.py new file mode 100644 index 0000000..a18b5cc --- /dev/null +++ b/backend/tests/test_users.py @@ -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"