feat: add bootstrap endpoint + deploy skill for AI agents
- POST /auth/bootstrap — creates first admin, self-deactivates after - da setup bootstrap — CLI command for agent-driven setup - da setup verify — structured health check (JSON output for agents) - cli/skills/deploy.md — complete deployment guide for AI agents - 6 bootstrap tests including full agent deployment flow simulation - 156 total tests passing
This commit is contained in:
ZdenekSrotyr
parent
a74f69d6b1
commit
bca5e91826
4 changed files with 518 additions and 51 deletions
|
|
@ -1,4 +1,6 @@
|
|||
"""Auth endpoints — login, token generation."""
|
||||
"""Auth endpoints — login, token generation, bootstrap."""
|
||||
|
||||
import uuid
|
||||
|
||||
from fastapi import APIRouter, Depends, HTTPException
|
||||
from pydantic import BaseModel
|
||||
|
|
@ -25,6 +27,12 @@ class TokenResponse(BaseModel):
|
|||
role: str
|
||||
|
||||
|
||||
class BootstrapRequest(BaseModel):
|
||||
email: str
|
||||
name: str = ""
|
||||
password: str = ""
|
||||
|
||||
|
||||
@router.post("/token", response_model=TokenResponse)
|
||||
async def create_token(
|
||||
request: TokenRequest,
|
||||
|
|
@ -36,8 +44,15 @@ async def create_token(
|
|||
if not user:
|
||||
raise HTTPException(status_code=401, detail="User not found")
|
||||
|
||||
# TODO: In production, verify password_hash with argon2
|
||||
# For greenfield demo, we issue tokens to any registered user
|
||||
# If user has password_hash, verify it
|
||||
if user.get("password_hash") and request.password:
|
||||
try:
|
||||
from argon2 import PasswordHasher
|
||||
ph = PasswordHasher()
|
||||
ph.verify(user["password_hash"], request.password)
|
||||
except Exception:
|
||||
raise HTTPException(status_code=401, detail="Invalid password")
|
||||
|
||||
token = create_access_token(
|
||||
user_id=user["id"],
|
||||
email=user["email"],
|
||||
|
|
@ -49,3 +64,49 @@ async def create_token(
|
|||
email=user["email"],
|
||||
role=user["role"],
|
||||
)
|
||||
|
||||
|
||||
@router.post("/bootstrap", response_model=TokenResponse)
|
||||
async def bootstrap(
|
||||
request: BootstrapRequest,
|
||||
conn: duckdb.DuckDBPyConnection = Depends(_get_db),
|
||||
):
|
||||
"""Create the first admin user. Only works when no users exist.
|
||||
|
||||
This endpoint allows an AI agent to bootstrap a fresh instance
|
||||
without needing docker exec or SSH. It automatically deactivates
|
||||
after the first user is created.
|
||||
"""
|
||||
repo = UserRepository(conn)
|
||||
existing = repo.list_all()
|
||||
if existing:
|
||||
raise HTTPException(
|
||||
status_code=403,
|
||||
detail=f"Bootstrap disabled — {len(existing)} users already exist. Use /auth/token to login.",
|
||||
)
|
||||
|
||||
user_id = str(uuid.uuid4())
|
||||
password_hash = None
|
||||
if request.password:
|
||||
try:
|
||||
from argon2 import PasswordHasher
|
||||
password_hash = PasswordHasher().hash(request.password)
|
||||
except ImportError:
|
||||
import hashlib
|
||||
password_hash = hashlib.sha256(request.password.encode()).hexdigest()
|
||||
|
||||
repo.create(
|
||||
id=user_id,
|
||||
email=request.email,
|
||||
name=request.name or request.email.split("@")[0],
|
||||
role="admin",
|
||||
password_hash=password_hash,
|
||||
)
|
||||
|
||||
token = create_access_token(user_id=user_id, email=request.email, role="admin")
|
||||
return TokenResponse(
|
||||
access_token=token,
|
||||
user_id=user_id,
|
||||
email=request.email,
|
||||
role="admin",
|
||||
)
|
||||
|
|
|
|||
|
|
@ -1,4 +1,4 @@
|
|||
"""Setup commands — da setup init/test-connection/deploy/first-sync/verify."""
|
||||
"""Setup commands — da setup init/bootstrap/test-connection/first-sync/verify."""
|
||||
|
||||
import json
|
||||
import os
|
||||
|
|
@ -14,9 +14,8 @@ setup_app = typer.Typer(help="Instance setup (guided by AI agent)")
|
|||
def setup_init(
|
||||
server: str = typer.Option("http://localhost:8000", help="Server URL"),
|
||||
):
|
||||
"""Initialize a new instance configuration."""
|
||||
"""Initialize CLI config to point at a server."""
|
||||
typer.echo(f"Server: {server}")
|
||||
typer.echo("Creating config directory...")
|
||||
|
||||
from cli.config import _config_dir
|
||||
config_dir = _config_dir()
|
||||
|
|
@ -26,9 +25,50 @@ def setup_init(
|
|||
config = {"server": server}
|
||||
config_file.write_text(yaml.dump(config))
|
||||
typer.echo(f"Config saved to {config_file}")
|
||||
|
||||
os.environ["DA_SERVER"] = server
|
||||
typer.echo("\nNext: da setup bootstrap --email admin@company.com")
|
||||
|
||||
|
||||
@setup_app.command("bootstrap")
|
||||
def bootstrap(
|
||||
email: str = typer.Argument(..., help="Admin email"),
|
||||
name: str = typer.Option("", help="Display name"),
|
||||
password: str = typer.Option("", help="Optional password"),
|
||||
server: str = typer.Option(None, help="Server URL override"),
|
||||
):
|
||||
"""Create the first admin user on a fresh instance.
|
||||
|
||||
Only works when the database has zero users.
|
||||
After this, use 'da login' for normal auth.
|
||||
"""
|
||||
if server:
|
||||
os.environ["DA_SERVER"] = server
|
||||
|
||||
typer.echo("Bootstrapping first admin user...")
|
||||
try:
|
||||
resp = api_post("/auth/bootstrap", json={
|
||||
"email": email,
|
||||
"name": name or email.split("@")[0],
|
||||
"password": password,
|
||||
})
|
||||
if resp.status_code == 200:
|
||||
data = resp.json()
|
||||
# Save token automatically
|
||||
from cli.config import save_token
|
||||
save_token(data["access_token"], data["email"], data["role"])
|
||||
typer.echo(f"Admin user created: {data['email']}")
|
||||
typer.echo(f"Token saved — you are now logged in as admin.")
|
||||
typer.echo("\nNext: da setup test-connection")
|
||||
elif resp.status_code == 403:
|
||||
typer.echo(f"Bootstrap disabled: {resp.json().get('detail', '')}")
|
||||
typer.echo("Users already exist. Use: da login --email your@email.com")
|
||||
else:
|
||||
typer.echo(f"Failed: {resp.text}", err=True)
|
||||
raise typer.Exit(1)
|
||||
except Exception as e:
|
||||
typer.echo(f"Connection error: {e}", err=True)
|
||||
typer.echo("Is the server running? Check: docker compose ps")
|
||||
raise typer.Exit(1)
|
||||
|
||||
|
||||
@setup_app.command("test-connection")
|
||||
|
|
@ -41,6 +81,11 @@ def test_connection():
|
|||
typer.echo(f" Server: {health.get('status', 'unknown')}")
|
||||
for svc, info in health.get("services", {}).items():
|
||||
typer.echo(f" {svc}: {info.get('status', '?')}")
|
||||
|
||||
if health.get("status") == "healthy":
|
||||
typer.echo("\nServer is healthy.")
|
||||
else:
|
||||
typer.echo("\nServer has issues. Check: da diagnose --json")
|
||||
except Exception as e:
|
||||
typer.echo(f" FAILED: {e}", err=True)
|
||||
raise typer.Exit(1)
|
||||
|
|
@ -55,77 +100,130 @@ def first_sync():
|
|||
try:
|
||||
resp = api_post("/api/sync/trigger")
|
||||
if resp.status_code == 200:
|
||||
typer.echo(f" {resp.json().get('message', 'Sync triggered')}")
|
||||
data = resp.json()
|
||||
typer.echo(f" Status: {data.get('status', '?')}")
|
||||
typer.echo(f" {data.get('message', '')}")
|
||||
elif resp.status_code == 403:
|
||||
typer.echo(" Permission denied. Are you logged in as admin?")
|
||||
typer.echo(" Run: da login --email admin@company.com")
|
||||
raise typer.Exit(1)
|
||||
else:
|
||||
typer.echo(f" Failed: {resp.text}", err=True)
|
||||
raise typer.Exit(1)
|
||||
except Exception as e:
|
||||
typer.echo(f" Error: {e}", err=True)
|
||||
raise typer.Exit(1)
|
||||
|
||||
typer.echo("\nNext: da setup verify")
|
||||
typer.echo("\nWait for sync to complete, then: da setup verify")
|
||||
|
||||
|
||||
@setup_app.command("verify")
|
||||
def verify():
|
||||
"""Verify the instance is working end-to-end."""
|
||||
typer.echo("Running verification checks...")
|
||||
def verify(as_json: bool = typer.Option(False, "--json", help="Output as JSON")):
|
||||
"""Verify the instance is working end-to-end.
|
||||
|
||||
Checks: server health → auth → data sync → manifest → query capability.
|
||||
Returns structured report for AI agents.
|
||||
"""
|
||||
checks = []
|
||||
|
||||
# 1. Health
|
||||
# 1. Server reachable
|
||||
try:
|
||||
resp = api_get("/api/health")
|
||||
h = resp.json()
|
||||
checks.append(("Health", h.get("status") == "healthy", h.get("status")))
|
||||
checks.append({
|
||||
"name": "server",
|
||||
"status": "pass" if h.get("status") == "healthy" else "warn",
|
||||
"detail": h.get("status"),
|
||||
})
|
||||
except Exception as e:
|
||||
checks.append(("Health", False, str(e)))
|
||||
checks.append({"name": "server", "status": "fail", "detail": str(e)})
|
||||
_report(checks, as_json)
|
||||
return
|
||||
|
||||
# 2. Data
|
||||
# 2. Auth works (token valid)
|
||||
from cli.config import get_token
|
||||
token = get_token()
|
||||
if token:
|
||||
try:
|
||||
resp = api_get("/api/sync/manifest")
|
||||
if resp.status_code == 200:
|
||||
checks.append({"name": "auth", "status": "pass", "detail": "token valid"})
|
||||
else:
|
||||
checks.append({"name": "auth", "status": "fail", "detail": f"HTTP {resp.status_code}"})
|
||||
except Exception as e:
|
||||
checks.append({"name": "auth", "status": "fail", "detail": str(e)})
|
||||
else:
|
||||
checks.append({"name": "auth", "status": "fail", "detail": "no token — run: da login"})
|
||||
|
||||
# 3. Data available
|
||||
try:
|
||||
resp = api_get("/api/sync/manifest")
|
||||
m = resp.json()
|
||||
table_count = len(m.get("tables", {}))
|
||||
checks.append(("Data", table_count > 0, f"{table_count} tables"))
|
||||
total_rows = sum(t.get("rows", 0) for t in m.get("tables", {}).values())
|
||||
if table_count > 0:
|
||||
checks.append({"name": "data", "status": "pass", "detail": f"{table_count} tables, {total_rows:,} rows"})
|
||||
else:
|
||||
checks.append({"name": "data", "status": "warn", "detail": "0 tables — run: da setup first-sync"})
|
||||
except Exception as e:
|
||||
checks.append(("Data", False, str(e)))
|
||||
checks.append({"name": "data", "status": "fail", "detail": str(e)})
|
||||
|
||||
# 3. Users
|
||||
# 4. Users exist
|
||||
try:
|
||||
resp = api_get("/api/users")
|
||||
if resp.status_code == 200:
|
||||
count = len(resp.json())
|
||||
checks.append(("Users", count > 0, f"{count} users"))
|
||||
checks.append({"name": "users", "status": "pass", "detail": f"{count} users"})
|
||||
elif resp.status_code == 403:
|
||||
checks.append({"name": "users", "status": "pass", "detail": "exists (need admin for count)"})
|
||||
else:
|
||||
checks.append(("Users", True, "requires admin token"))
|
||||
checks.append({"name": "users", "status": "warn", "detail": f"HTTP {resp.status_code}"})
|
||||
except Exception as e:
|
||||
checks.append(("Users", False, str(e)))
|
||||
checks.append({"name": "users", "status": "fail", "detail": str(e)})
|
||||
|
||||
for name, ok, detail in checks:
|
||||
status = "PASS" if ok else "FAIL"
|
||||
typer.echo(f" [{status}] {name}: {detail}")
|
||||
|
||||
all_ok = all(ok for _, ok, _ in checks)
|
||||
if all_ok:
|
||||
typer.echo("\nAll checks passed! Instance is ready.")
|
||||
else:
|
||||
typer.echo("\nSome checks failed. Review above.")
|
||||
raise typer.Exit(1)
|
||||
|
||||
|
||||
@setup_app.command("add-user")
|
||||
def add_first_user(
|
||||
email: str = typer.Argument(..., help="Admin email"),
|
||||
name: str = typer.Option("", help="Display name"),
|
||||
):
|
||||
"""Add the first admin user to the instance."""
|
||||
resp = api_post("/api/users", json={
|
||||
"email": email,
|
||||
"name": name or email.split("@")[0],
|
||||
"role": "admin",
|
||||
# 5. Web UI accessible
|
||||
try:
|
||||
resp = api_get("/login")
|
||||
checks.append({
|
||||
"name": "web_ui",
|
||||
"status": "pass" if resp.status_code == 200 else "fail",
|
||||
"detail": f"HTTP {resp.status_code}, {len(resp.content)} bytes",
|
||||
})
|
||||
if resp.status_code == 201:
|
||||
typer.echo(f"Admin user created: {email}")
|
||||
elif resp.status_code == 409:
|
||||
typer.echo(f"User {email} already exists.")
|
||||
except Exception as e:
|
||||
checks.append({"name": "web_ui", "status": "fail", "detail": str(e)})
|
||||
|
||||
# 6. Swagger docs
|
||||
try:
|
||||
resp = api_get("/docs")
|
||||
checks.append({
|
||||
"name": "api_docs",
|
||||
"status": "pass" if resp.status_code == 200 else "fail",
|
||||
"detail": f"HTTP {resp.status_code}",
|
||||
})
|
||||
except Exception as e:
|
||||
checks.append({"name": "api_docs", "status": "fail", "detail": str(e)})
|
||||
|
||||
_report(checks, as_json)
|
||||
|
||||
|
||||
def _report(checks: list, as_json: bool):
|
||||
all_pass = all(c["status"] == "pass" for c in checks)
|
||||
has_fail = any(c["status"] == "fail" for c in checks)
|
||||
|
||||
if as_json:
|
||||
typer.echo(json.dumps({
|
||||
"overall": "pass" if all_pass else ("fail" if has_fail else "warn"),
|
||||
"checks": checks,
|
||||
}, indent=2))
|
||||
else:
|
||||
typer.echo(f"Failed: {resp.text}", err=True)
|
||||
for c in checks:
|
||||
icon = {"pass": "OK", "fail": "FAIL", "warn": "WARN"}[c["status"]]
|
||||
typer.echo(f" [{icon:4s}] {c['name']}: {c['detail']}")
|
||||
typer.echo("")
|
||||
if all_pass:
|
||||
typer.echo("All checks passed! Instance is ready.")
|
||||
elif has_fail:
|
||||
typer.echo("Some checks FAILED. See above for details.")
|
||||
raise typer.Exit(1)
|
||||
else:
|
||||
typer.echo("Instance is running but some items need attention.")
|
||||
|
|
|
|||
188
cli/skills/deploy.md
Normal file
188
cli/skills/deploy.md
Normal file
|
|
@ -0,0 +1,188 @@
|
|||
# Deploy — Complete server deployment guide for AI agents
|
||||
|
||||
## Prerequisites
|
||||
|
||||
You need:
|
||||
- A Linux server with SSH access (Ubuntu 22.04+ recommended)
|
||||
- Docker + Docker Compose installed on the server
|
||||
- A domain pointing to the server IP (optional but recommended for SSL)
|
||||
- Keboola Storage Token + Stack URL + Project ID (for data source)
|
||||
|
||||
## Step-by-step deployment
|
||||
|
||||
### 1. Connect to server
|
||||
|
||||
```bash
|
||||
ssh user@your-server-ip
|
||||
```
|
||||
|
||||
### 2. Clone the repository
|
||||
|
||||
```bash
|
||||
git clone https://github.com/padak/tmp_oss.git /opt/data-analyst
|
||||
cd /opt/data-analyst
|
||||
git checkout feature/v2-fastapi-duckdb-docker-cli
|
||||
```
|
||||
|
||||
### 3. Create .env file
|
||||
|
||||
```bash
|
||||
cp config/.env.template .env
|
||||
```
|
||||
|
||||
Edit `.env` with these REQUIRED values:
|
||||
```
|
||||
JWT_SECRET_KEY=<random 32+ char string>
|
||||
DATA_DIR=/data
|
||||
DATA_SOURCE=keboola
|
||||
KEBOOLA_STORAGE_TOKEN=<your token>
|
||||
KEBOOLA_STACK_URL=https://connection.keboola.com
|
||||
KEBOOLA_PROJECT_ID=<your project id>
|
||||
```
|
||||
|
||||
Generate a random JWT secret:
|
||||
```bash
|
||||
python3 -c "import secrets; print(secrets.token_urlsafe(32))"
|
||||
```
|
||||
|
||||
### 4. Start Docker
|
||||
|
||||
```bash
|
||||
docker compose up -d
|
||||
```
|
||||
|
||||
Wait for health check:
|
||||
```bash
|
||||
sleep 5
|
||||
curl http://localhost:8000/api/health
|
||||
```
|
||||
|
||||
Expected: `{"status": "healthy", ...}`
|
||||
|
||||
### 5. Bootstrap first admin user
|
||||
|
||||
From your LOCAL machine (not the server):
|
||||
|
||||
```bash
|
||||
da setup init --server http://SERVER_IP:8000
|
||||
da setup bootstrap admin@company.com
|
||||
```
|
||||
|
||||
Or directly via curl:
|
||||
```bash
|
||||
curl -X POST http://SERVER_IP:8000/auth/bootstrap \
|
||||
-H "Content-Type: application/json" \
|
||||
-d '{"email": "admin@company.com", "name": "Admin"}'
|
||||
```
|
||||
|
||||
This returns a JWT token. Save it.
|
||||
|
||||
### 6. Trigger first data sync
|
||||
|
||||
```bash
|
||||
da setup first-sync
|
||||
```
|
||||
|
||||
Or via curl:
|
||||
```bash
|
||||
curl -X POST http://SERVER_IP:8000/api/sync/trigger \
|
||||
-H "Authorization: Bearer <TOKEN>"
|
||||
```
|
||||
|
||||
### 7. Verify everything works
|
||||
|
||||
```bash
|
||||
da setup verify --json
|
||||
```
|
||||
|
||||
Expected: all checks PASS.
|
||||
|
||||
### 8. Add more users
|
||||
|
||||
```bash
|
||||
da admin add-user analyst@company.com --role analyst
|
||||
```
|
||||
|
||||
## Troubleshooting
|
||||
|
||||
### Server not reachable
|
||||
```bash
|
||||
# Check containers
|
||||
docker compose ps
|
||||
|
||||
# Check logs
|
||||
docker compose logs app --tail 50
|
||||
|
||||
# Restart
|
||||
docker compose restart app
|
||||
```
|
||||
|
||||
### Data sync fails
|
||||
```bash
|
||||
# Check logs
|
||||
docker compose logs app --tail 50 | grep -i error
|
||||
|
||||
# Common issues:
|
||||
# - KEBOOLA_STORAGE_TOKEN expired → get new token
|
||||
# - KEBOOLA_STACK_URL wrong → check URL
|
||||
# - data_description.md missing → needs table configuration
|
||||
```
|
||||
|
||||
### No tables after sync
|
||||
```bash
|
||||
# Check if data_description.md exists
|
||||
docker compose exec app ls /data/docs/data_description.md
|
||||
|
||||
# Check sync state
|
||||
curl -H "Authorization: Bearer <TOKEN>" http://SERVER_IP:8000/api/sync/manifest
|
||||
```
|
||||
|
||||
### Out of memory
|
||||
```bash
|
||||
# Check memory
|
||||
docker stats --no-stream
|
||||
|
||||
# Increase Docker memory limit in docker-compose.yml:
|
||||
# services.app.deploy.resources.limits.memory: 4g
|
||||
```
|
||||
|
||||
## For Google OAuth (optional)
|
||||
|
||||
1. Go to https://console.cloud.google.com/apis/credentials
|
||||
2. Create OAuth 2.0 Client ID (Web application)
|
||||
3. Add authorized redirect URI: `https://your-domain.com/auth/google/callback`
|
||||
4. Add to .env:
|
||||
```
|
||||
GOOGLE_CLIENT_ID=<client id>
|
||||
GOOGLE_CLIENT_SECRET=<client secret>
|
||||
```
|
||||
5. Restart: `docker compose restart app`
|
||||
|
||||
## For SSL (optional)
|
||||
|
||||
Use Caddy as reverse proxy (simplest auto-SSL):
|
||||
```bash
|
||||
# Install Caddy
|
||||
apt install caddy
|
||||
|
||||
# /etc/caddy/Caddyfile
|
||||
your-domain.com {
|
||||
reverse_proxy localhost:8000
|
||||
}
|
||||
|
||||
# Restart
|
||||
systemctl restart caddy
|
||||
```
|
||||
|
||||
## Agent deployment flow (automated)
|
||||
|
||||
An AI agent should follow this sequence:
|
||||
1. `ssh user@server` — connect
|
||||
2. Clone repo, create .env
|
||||
3. `docker compose up -d` — start
|
||||
4. `curl /api/health` — verify server
|
||||
5. `curl -X POST /auth/bootstrap` — create admin
|
||||
6. `curl -X POST /api/sync/trigger` — sync data
|
||||
7. Wait 30-60 seconds
|
||||
8. `curl /api/sync/manifest` — verify data
|
||||
9. Report success with URL + admin credentials
|
||||
120
tests/test_bootstrap.py
Normal file
120
tests/test_bootstrap.py
Normal file
|
|
@ -0,0 +1,120 @@
|
|||
"""Tests for bootstrap endpoint — first admin user creation."""
|
||||
|
||||
import os
|
||||
import pytest
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def fresh_client(tmp_path):
|
||||
"""Client with EMPTY database — no users."""
|
||||
os.environ["DATA_DIR"] = str(tmp_path)
|
||||
os.environ["JWT_SECRET_KEY"] = "test-secret-32chars-minimum!!!!!"
|
||||
from app.main import create_app
|
||||
app = create_app()
|
||||
return TestClient(app)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def seeded_client(tmp_path):
|
||||
"""Client with one existing user."""
|
||||
os.environ["DATA_DIR"] = str(tmp_path)
|
||||
os.environ["JWT_SECRET_KEY"] = "test-secret-32chars-minimum!!!!!"
|
||||
from app.main import create_app
|
||||
from src.db import get_system_db
|
||||
from src.repositories.users import UserRepository
|
||||
conn = get_system_db()
|
||||
UserRepository(conn).create(id="existing", email="existing@test.com", name="E", role="admin")
|
||||
conn.close()
|
||||
return TestClient(create_app())
|
||||
|
||||
|
||||
class TestBootstrap:
|
||||
def test_bootstrap_on_empty_db(self, fresh_client):
|
||||
"""First call creates admin and returns token."""
|
||||
resp = fresh_client.post("/auth/bootstrap", json={
|
||||
"email": "admin@test.com",
|
||||
"name": "Admin",
|
||||
})
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert data["email"] == "admin@test.com"
|
||||
assert data["role"] == "admin"
|
||||
assert "access_token" in data
|
||||
|
||||
def test_bootstrap_with_password(self, fresh_client):
|
||||
"""Bootstrap with password sets password hash."""
|
||||
resp = fresh_client.post("/auth/bootstrap", json={
|
||||
"email": "admin@test.com",
|
||||
"password": "securepass123",
|
||||
})
|
||||
assert resp.status_code == 200
|
||||
|
||||
# Token works
|
||||
token = resp.json()["access_token"]
|
||||
resp2 = fresh_client.get("/api/health")
|
||||
assert resp2.status_code == 200
|
||||
|
||||
def test_bootstrap_disabled_when_users_exist(self, seeded_client):
|
||||
"""Bootstrap fails with 403 when users already exist."""
|
||||
resp = seeded_client.post("/auth/bootstrap", json={
|
||||
"email": "hacker@evil.com",
|
||||
})
|
||||
assert resp.status_code == 403
|
||||
assert "already exist" in resp.json()["detail"]
|
||||
|
||||
def test_bootstrap_then_login(self, fresh_client):
|
||||
"""After bootstrap, normal /auth/token login works."""
|
||||
# Bootstrap
|
||||
fresh_client.post("/auth/bootstrap", json={
|
||||
"email": "admin@test.com",
|
||||
})
|
||||
|
||||
# Normal login
|
||||
resp = fresh_client.post("/auth/token", json={
|
||||
"email": "admin@test.com",
|
||||
})
|
||||
assert resp.status_code == 200
|
||||
assert resp.json()["role"] == "admin"
|
||||
|
||||
def test_bootstrap_second_call_fails(self, fresh_client):
|
||||
"""Second bootstrap call fails — endpoint self-deactivates."""
|
||||
fresh_client.post("/auth/bootstrap", json={"email": "admin@test.com"})
|
||||
|
||||
resp = fresh_client.post("/auth/bootstrap", json={"email": "second@test.com"})
|
||||
assert resp.status_code == 403
|
||||
|
||||
def test_full_agent_flow(self, fresh_client):
|
||||
"""Simulate full AI agent deployment flow."""
|
||||
# 1. Health check (no auth)
|
||||
resp = fresh_client.get("/api/health")
|
||||
assert resp.status_code == 200
|
||||
assert resp.json()["status"] == "healthy"
|
||||
|
||||
# 2. Bootstrap admin
|
||||
resp = fresh_client.post("/auth/bootstrap", json={
|
||||
"email": "agent@company.com", "name": "AI Agent",
|
||||
})
|
||||
assert resp.status_code == 200
|
||||
token = resp.json()["access_token"]
|
||||
headers = {"Authorization": f"Bearer {token}"}
|
||||
|
||||
# 3. Check manifest (empty, no data yet)
|
||||
resp = fresh_client.get("/api/sync/manifest", headers=headers)
|
||||
assert resp.status_code == 200
|
||||
assert len(resp.json()["tables"]) == 0
|
||||
|
||||
# 4. List users
|
||||
resp = fresh_client.get("/api/users", headers=headers)
|
||||
assert resp.status_code == 200
|
||||
assert len(resp.json()) == 1
|
||||
|
||||
# 5. Add analyst user
|
||||
resp = fresh_client.post("/api/users", json={
|
||||
"email": "analyst@company.com", "name": "Analyst",
|
||||
}, headers=headers)
|
||||
assert resp.status_code == 201
|
||||
|
||||
# 6. Verify
|
||||
resp = fresh_client.get("/api/health")
|
||||
assert resp.json()["services"]["users"]["count"] == 2
|
||||
Loading…
Reference in a new issue