#!/usr/bin/env python3 """Real-HTTP smoke test against a running RAGCore app process. Runs *inside* the deployment network (e.g. the ``ci`` image on the Compose ``backend`` network) and talks plain HTTP to the live app (uvicorn), proving the process boundary end to end: * liveness/readiness endpoints and the OpenAPI version served; * authentication gating (401 on API routes without a token, login page on the Control Plane without a session); * a real bearer credential (provisioned through the real control-plane repositories) can call ``/v1/search`` and receives a well-formed response; * a real signed admin session (minted with the app's own signing key, the same mechanism the OIDC callback uses) renders every Control Plane screen: dashboard, sources, documents, jobs, system, query lab, applications, spaces, policies, profiles -- each 200 and carrying its expected markers. Usage: uv run python scripts/http_smoke.py --base-url http://app:8080 Exit code 0 when every check passed; non-zero otherwise. Nothing here is mocked; every failure prints the failing check. """ from __future__ import annotations import argparse import asyncio import hashlib import sys from datetime import UTC, datetime, timedelta from uuid import UUID, uuid4 import httpx from ragcore.application.auth import SignedTokenCodec, encode_session from ragcore.application.identity.credentials import ServiceAccountCredentialService from ragcore.config import get_settings from ragcore.domain.auth import Session from ragcore.domain.control import Role, Scope, ServiceAccount, Workspace from ragcore.infrastructure.database import ( create_database_engine, create_session_factory, session_scope, ) from ragcore.infrastructure.database.control import ControlPlaneRepository from ragcore.infrastructure.database.identity import PostgresCredentialRepository CHECKS: list[tuple[str, bool, str]] = [] def check(name: str, ok: bool, detail: str = "") -> None: CHECKS.append((name, ok, detail)) print(f"[{'OK' if ok else 'FAIL'}] {name}{(' -- ' + detail) if detail else ''}") async def provision(settings) -> tuple[UUID, UUID, str, Workspace]: # type: ignore[no-untyped-def] engine = create_database_engine(settings) factory = create_session_factory(engine) workspace_id = uuid4() workspace = Workspace( workspace_id, f"Smoke {workspace_id.hex[:6]}", f"smoke-{workspace_id.hex[:8]}" ) application = workspace.create_application( id=uuid4(), name="Smoke client", slug=f"smoke-client-{uuid4().hex[:8]}", scopes=frozenset({Scope.SEARCH, Scope.CONTEXT, Scope.SOURCES_SYNC}), ) space = workspace.create_knowledge_space( id=uuid4(), name="Smoke space", slug=f"smoke-space-{uuid4().hex[:8]}" ) account = ServiceAccount( id=uuid4(), workspace_id=workspace_id, application_id=application.id, name="Smoke service account", scopes=application.scopes, ) async with session_scope(factory) as session: repository = ControlPlaneRepository(session) await repository.create_workspace(workspace) await repository.create_space(space) await repository.create_application(application) await repository.create_service_account(account) await repository.create_grant( workspace.grant( id=uuid4(), application=application, knowledge_space=space, role=Role.EDITOR ) ) await session.commit() async def load(account_id: UUID) -> ServiceAccount | None: async with session_scope(factory) as session: return await ControlPlaneRepository(session).get_service_account_by_id(account_id) async with session_scope(factory) as session: service = ServiceAccountCredentialService( PostgresCredentialRepository(session), settings.api_token_pepper.get_secret_value(), service_account_loader=load, ) issued = await service.issue(account, scopes=account.scopes) await session.commit() await engine.dispose() return workspace_id, space.id, issued.token, workspace def mint_session_cookie(settings, workspace_id: UUID) -> str: # type: ignore[no-untyped-def] """The same signed-cookie session the OIDC callback issues (app signing key).""" codec = SignedTokenCodec( secret=hashlib.sha256(settings.secret_key.get_secret_value().encode()).digest() ) now = datetime.now(UTC) session = Session( id=uuid4(), subject=f"smoke-admin-{uuid4().hex[:8]}", workspace_id=workspace_id, roles=frozenset({Role.PLATFORM_ADMIN}), issued_at=now, expires_at=now + timedelta(minutes=30), csrf_token=uuid4().hex, ) return encode_session(session, codec=codec) async def main() -> int: parser = argparse.ArgumentParser() parser.add_argument("--base-url", required=True) args = parser.parse_args() settings = get_settings() base = args.base_url.rstrip("/") async with httpx.AsyncClient(base_url=base, timeout=30.0, follow_redirects=False) as http: live = await http.get("/health/live") check("GET /health/live is 200", live.status_code == 200, str(live.status_code)) ready = await http.get("/health/ready") check("GET /health/ready is 200", ready.status_code == 200, ready.text[:200]) openapi = await http.get("/openapi.json") version = ( openapi.json().get("info", {}).get("version") if openapi.status_code == 200 else None ) check("OpenAPI served with the release version", version == "1.2.0", str(version)) unauth = await http.post("/v1/search", json={"query": "x"}) check( "POST /v1/search without a token is 401", unauth.status_code == 401, str(unauth.status_code), ) anon_admin = await http.get("/admin/sources") check( "GET /admin/sources without a session is not served (401/redirect)", anon_admin.status_code in {401, 302, 303, 307}, str(anon_admin.status_code), ) login = await http.get("/login") check( "GET /login renders", login.status_code == 200 and "RAGCore" in login.text, str(login.status_code), ) workspace_id, space_id, token, _ = await provision(settings) search = await http.post( "/v1/search", json={"query": "smoke", "requested_space_ids": [str(space_id)]}, headers={"Authorization": f"Bearer {token}"}, ) body = ( search.json() if search.headers.get("content-type", "").startswith("application/json") else {} ) check( "POST /v1/search with a real bearer credential is 200 with the effective space", search.status_code == 200 and body.get("effective_space_ids") == [str(space_id)], f"{search.status_code} {str(body)[:160]}", ) cookie = mint_session_cookie(settings, workspace_id) http.cookies.set("ragcore_session", cookie) screens = { "/admin/dashboard": "Dashboard", "/admin/control/spaces": "Knowledge spaces", "/admin/sources": "Sources", "/admin/sources/new": "Add source", "/admin/documents": "Documents", "/admin/jobs": "Ingestion Jobs", "/admin/system": "System Health", "/admin/query-lab": "Query Lab", "/admin/control/applications": "Applications", "/admin/control/policies": "Policies", "/admin/profiles": "Profile", } for path, marker in screens.items(): response = await http.get(path) check( f"GET {path} renders with a real session", response.status_code == 200 and marker in response.text, f"{response.status_code}", ) nav = await http.get("/admin/dashboard") check( "grouped navigation present", all(label in nav.text for label in ("Knowledge", "Access", "Quality", "Operations")), ) created = await http.post( "/admin/sources", data={ "name": "Smoke runbooks", "kind": "folder", "space_id": str(space_id), "root": "/mnt/smoke-runbooks", "trust_level": "trusted_internal", }, ) check( "POST /admin/sources creates a folder source (303)", created.status_code == 303, str(created.status_code), ) if created.status_code == 303: detail = await http.get(created.headers["location"]) check( "source detail renders with configuration and sync action", detail.status_code == 200 and "/mnt/smoke-runbooks" in detail.text and "Sync now" in detail.text, str(detail.status_code), ) listing = await http.get("/admin/sources?q=smoke") check("sources list search finds it", "Smoke runbooks" in listing.text) source_path = created.headers["location"].split("?")[0] synced = await http.post(f"{source_path}/sync") check( "Sync now enqueues a sync_source job (303)", synced.status_code == 303, str(synced.status_code), ) jobs = await http.get("/admin/jobs?operation=sync_source") check("jobs screen lists the sync job", "Sync source" in jobs.text) lab = await http.post( "/admin/query-lab/run", data={"query": "smoke test query", "space_ids": [str(space_id)]}, ) check( "Query Lab run renders the inspector", lab.status_code == 200 and "Retrieval inspector" in lab.text, str(lab.status_code), ) failed = [name for name, ok, _ in CHECKS if not ok] print(f"\n{len(CHECKS) - len(failed)}/{len(CHECKS)} checks passed") return 1 if failed else 0 if __name__ == "__main__": sys.exit(asyncio.run(main()))