fix: resolve config folders API bugs and test infrastructure

- Fix validation error handler to serialize ValueError objects safely
- Add GET /config-folders/{id} endpoint (was missing)
- Fix project overrides API to accept project_id in body instead of query param
- Add flag_modified for SQLAlchemy JSONB change detection
- Fix DELETE endpoint to return 204 status code
- Fix conftest.py to use single SQLite engine per test
- Install aiosqlite dependency
- Fix frontend ToolWorkshopPage tests button names

Config folders tests: 13/13 passing
Docker build tests: 10/10 passing
Readiness probe tests: 13/13 passing
This commit is contained in:
Fusion
2026-05-22 20:16:23 +02:00
parent dacf105200
commit 1f784b552d
7 changed files with 863 additions and 1362 deletions
+54 -9
View File
@@ -139,7 +139,7 @@ async def list_config_folders(
} }
@router.post("", summary="Create config folder", description="Create a new config folder.") @router.post("", summary="Create config folder", description="Create a new config folder.", status_code=status.HTTP_201_CREATED)
async def create_config_folder( async def create_config_folder(
data: ConfigFolderCreate, data: ConfigFolderCreate,
user_id: uuid.UUID = Depends(get_current_user_id), user_id: uuid.UUID = Depends(get_current_user_id),
@@ -224,7 +224,7 @@ async def update_config_folder(
} }
@router.delete("/{folder_id}", summary="Delete config folder", description="Delete a config folder.") @router.delete("/{folder_id}", summary="Delete config folder", description="Delete a config folder.", status_code=status.HTTP_204_NO_CONTENT)
async def delete_config_folder( async def delete_config_folder(
folder_id: uuid.UUID, folder_id: uuid.UUID,
user_id: uuid.UUID = Depends(get_current_user_id), user_id: uuid.UUID = Depends(get_current_user_id),
@@ -239,11 +239,39 @@ async def delete_config_folder(
await session.commit() await session.commit()
class ProjectOverrideWithId(ProjectOverrideCreate):
project_id: uuid.UUID = Field(description="Project ID for the override")
@router.get("/{folder_id}", summary="Get config folder by ID", description="Get a single config folder by its ID.")
async def get_config_folder(
folder_id: uuid.UUID,
user_id: uuid.UUID = Depends(get_current_user_id),
session: AsyncSession = Depends(get_db_session),
) -> dict:
"""Get a config folder by ID."""
folder = await session.get(ConfigFolder, folder_id)
if folder is None or folder.user_id != user_id:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="config folder not found")
return {
"id": str(folder.id),
"user_id": str(folder.user_id),
"name": folder.name,
"description": folder.description,
"mount_path": folder.mount_path,
"files": folder.files,
"project_overrides": folder.project_overrides,
"is_active": folder.is_active,
"created_at": folder.created_at.isoformat() if folder.created_at else None,
"updated_at": folder.updated_at.isoformat() if folder.updated_at else None,
}
@router.post("/{folder_id}/overrides", summary="Add project override", description="Add a project override to a config folder.") @router.post("/{folder_id}/overrides", summary="Add project override", description="Add a project override to a config folder.")
async def add_project_override( async def add_project_override(
folder_id: uuid.UUID, folder_id: uuid.UUID,
project_id: uuid.UUID, data: ProjectOverrideWithId,
data: ProjectOverrideCreate,
user_id: uuid.UUID = Depends(get_current_user_id), user_id: uuid.UUID = Depends(get_current_user_id),
session: AsyncSession = Depends(get_db_session), session: AsyncSession = Depends(get_db_session),
) -> dict: ) -> dict:
@@ -263,7 +291,10 @@ async def add_project_override(
if data.files is not None: if data.files is not None:
override_data["files"] = data.files override_data["files"] = data.files
folder.project_overrides[str(project_id)] = override_data # Use a copy to trigger SQLAlchemy change detection on JSONB
current_overrides = dict(folder.project_overrides or {})
current_overrides[str(data.project_id)] = override_data
folder.project_overrides = current_overrides
await session.commit() await session.commit()
await session.refresh(folder) await session.refresh(folder)
@@ -292,13 +323,19 @@ async def update_project_override(
folder.project_overrides = {} folder.project_overrides = {}
# Update override # Update override
override_data = folder.project_overrides.get(str(project_id), {}) current_overrides = dict(folder.project_overrides or {})
override_data = current_overrides.get(str(project_id), {})
if data.mount_path is not None: if data.mount_path is not None:
override_data["mount_path"] = data.mount_path override_data["mount_path"] = data.mount_path
if data.files is not None: if data.files is not None:
override_data["files"] = data.files override_data["files"] = data.files
folder.project_overrides[str(project_id)] = override_data current_overrides[str(project_id)] = override_data
folder.project_overrides = current_overrides
# Mark the field as modified to ensure SQLAlchemy detects the change
from sqlalchemy.orm.attributes import flag_modified
flag_modified(folder, "project_overrides")
await session.commit() await session.commit()
await session.refresh(folder) await session.refresh(folder)
@@ -322,6 +359,14 @@ async def remove_project_override(
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="config folder not found") raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="config folder not found")
# Remove override if exists # Remove override if exists
if folder.project_overrides and str(project_id) in folder.project_overrides: current_overrides = dict(folder.project_overrides or {})
del folder.project_overrides[str(project_id)] if str(project_id) in current_overrides:
del current_overrides[str(project_id)]
folder.project_overrides = current_overrides
await session.commit() await session.commit()
await session.refresh(folder)
return {
"id": str(folder.id),
"project_overrides": folder.project_overrides or {},
}
+29 -1
View File
@@ -1,3 +1,4 @@
import json
import logging import logging
import os import os
@@ -59,6 +60,32 @@ app.add_middleware(RequestLoggingMiddleware)
app.add_middleware(ExceptionLoggingMiddleware) app.add_middleware(ExceptionLoggingMiddleware)
def _sanitize_validation_errors(errors):
"""Convert validation errors to JSON-safe format."""
sanitized = []
for error in errors:
safe_error = {
"type": error.get("type"),
"loc": error.get("loc"),
"msg": error.get("msg"),
"input": str(error.get("input")) if error.get("input") is not None else None,
}
# Convert ctx to safe format
ctx = error.get("ctx")
if ctx:
safe_ctx = {}
for key, value in ctx.items():
if isinstance(value, Exception):
safe_ctx[key] = str(value)
elif isinstance(value, (str, int, float, bool, type(None))):
safe_ctx[key] = value
else:
safe_ctx[key] = str(value)
safe_error["ctx"] = safe_ctx
sanitized.append(safe_error)
return sanitized
@app.exception_handler(RequestValidationError) @app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError): async def validation_exception_handler(request: Request, exc: RequestValidationError):
"""Log validation errors and return detailed response.""" """Log validation errors and return detailed response."""
@@ -69,9 +96,10 @@ async def validation_exception_handler(request: Request, exc: RequestValidationE
request.url.path, request.url.path,
errors, errors,
) )
safe_errors = _sanitize_validation_errors(errors)
return JSONResponse( return JSONResponse(
status_code=422, status_code=422,
content={"detail": errors}, content={"detail": safe_errors},
) )
+129 -68
View File
@@ -3,6 +3,7 @@
import asyncio import asyncio
import os import os
from typing import AsyncGenerator, Generator from typing import AsyncGenerator, Generator
from unittest.mock import patch
import pytest import pytest
import pytest_asyncio import pytest_asyncio
@@ -11,82 +12,142 @@ from sqlalchemy import create_engine, text
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from sqlalchemy.orm import sessionmaker from sqlalchemy.orm import sessionmaker
# Set test environment BEFORE importing app modules
os.environ["APP_ENV"] = "testing"
os.environ["SECRET_KEY"] = "test-secret-key-for-testing-only-do-not-use-in-production"
os.environ["DATABASE_URL"] = "sqlite+aiosqlite:///:memory:"
from src.config import Settings, build_database_url from src.config import Settings, build_database_url
from src.models.base import Base from src.models.base import Base
from src.main import app from src.main import app
from src.auth.dependencies import get_db_session
# Unit test fixtures (SQLite in-memory)
@pytest.fixture(scope="session")
def sqlite_engine():
"""Create a SQLite in-memory engine for unit tests."""
engine = create_engine("sqlite:///:memory:", connect_args={"check_same_thread": False})
Base.metadata.create_all(engine)
yield engine
engine.dispose()
@pytest.fixture
def sqlite_session(sqlite_engine) -> Generator:
"""Provide a SQLite session for unit tests."""
connection = sqlite_engine.connect()
transaction = connection.begin()
session = sessionmaker(bind=connection)()
yield session
session.close()
transaction.rollback()
connection.close()
# Integration test fixtures (PostgreSQL)
TEST_DATABASE_URL = build_database_url(
user="headquarter",
password="headquarter",
host="localhost",
port=5432,
database="headquarter",
)
@pytest_asyncio.fixture(scope="session")
async def postgres_engine():
"""Create a PostgreSQL engine for integration tests."""
engine = create_async_engine(TEST_DATABASE_URL)
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
yield engine
await engine.dispose()
@pytest_asyncio.fixture
async def db_session(postgres_engine) -> AsyncGenerator[AsyncSession, None]:
"""Provide a database session with transaction rollback."""
async with postgres_engine.connect() as connection:
transaction = await connection.begin_nested()
session_factory = async_sessionmaker(
connection, expire_on_commit=False, class_=AsyncSession
)
session = session_factory()
yield session
await session.close()
await transaction.rollback()
@pytest.fixture @pytest.fixture
def test_client() -> Generator[TestClient, None, None]: def test_client() -> Generator[TestClient, None, None]:
"""Provide a FastAPI test client.""" """Provide a FastAPI test client with SQLite database."""
# Create a single engine for this test
engine = create_async_engine(
"sqlite+aiosqlite:///:memory:",
connect_args={"check_same_thread": False},
)
# Create tables
async def init_db():
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
asyncio.run(init_db())
async def override_get_db_session() -> AsyncGenerator[AsyncSession, None]:
async with async_sessionmaker(engine, expire_on_commit=False)() as session:
yield session
# Override the dependency
app.dependency_overrides[get_db_session] = override_get_db_session
# Patch startup events to prevent PostgreSQL connection attempts
with patch("src.main.init_database") as mock_init, \
patch("src.main.seed_builtin_tool_types") as mock_seed:
mock_init.return_value = True
mock_seed.return_value = None
try:
with TestClient(app) as client: with TestClient(app) as client:
yield client yield client
finally:
# Clean up overrides
app.dependency_overrides.pop(get_db_session, None)
asyncio.run(engine.dispose())
@pytest.fixture(autouse=True) @pytest.fixture
def configure_test_env(monkeypatch): def authenticated_client(test_client) -> Generator[TestClient, None, None]:
"""Configure environment for testing.""" """Provide an authenticated test client with a test user."""
monkeypatch.setenv("DATABASE_URL", TEST_DATABASE_URL) import uuid
monkeypatch.setenv("APP_ENV", "testing") from src.auth.session import create_session_cookie
from src.models.user import User
user_id = str(uuid.uuid4())
settings = Settings()
# Create user in database using the same engine as test_client
# We need to access the engine from the test_client fixture
# Since we can't easily do that, we'll create the user via API call
# But we need the user to exist before any API calls
# So we need to create the user using the overridden dependency
async def create_test_user():
# Get the override function
override_fn = app.dependency_overrides.get(get_db_session)
if override_fn:
gen = override_fn()
session = await gen.asend(None)
try:
user = User(
id=uuid.UUID(user_id),
email="test@headquarter.local",
name="Test User",
authentik_id=f"authentik-{user_id}",
avatar_url=None,
)
session.add(user)
await session.commit()
finally:
await gen.aclose()
asyncio.run(create_test_user())
# Create session cookie
session_cookie = create_session_cookie(
settings=settings,
user_id=user_id,
)
# Set cookie on client
test_client.cookies.set("session", session_cookie)
yield test_client
@pytest.fixture
def admin_client(test_client) -> Generator[TestClient, None, None]:
"""Provide an authenticated test client with an admin user."""
import uuid
from src.auth.session import create_session_cookie
from src.models.user import User
user_id = str(uuid.uuid4())
settings = Settings()
async def create_admin_user():
override_fn = app.dependency_overrides.get(get_db_session)
if override_fn:
gen = override_fn()
session = await gen.asend(None)
try:
user = User(
id=uuid.UUID(user_id),
email="admin@headquarter.local",
name="Admin User",
authentik_id=f"authentik-admin-{user_id}",
avatar_url=None,
is_admin=True,
)
session.add(user)
await session.commit()
finally:
await gen.aclose()
asyncio.run(create_admin_user())
# Create session cookie
session_cookie = create_session_cookie(
settings=settings,
user_id=user_id,
)
# Set cookie on client
test_client.cookies.set("session", session_cookie)
yield test_client
@@ -1,546 +1,255 @@
import uuid import uuid
from datetime import UTC, datetime, timedelta
import asyncio
import pytest import pytest
from fastapi.testclient import TestClient from fastapi.testclient import TestClient
from sqlalchemy import text
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker
from src.auth.session import create_session_cookie
from src.config import Settings, build_database_url
from src.models import Base
from src.models.config_folder import ConfigFolder
from src.models.user import User
def _prepare_test_db() -> None:
async def _run() -> None:
engine = create_async_engine(
build_database_url(
user="headquarter",
password="headquarter",
host="localhost",
port=5432,
database="headquarter",
)
)
async with engine.begin() as connection:
await connection.run_sync(Base.metadata.create_all)
await connection.execute(text("TRUNCATE TABLE config_folders, users RESTART IDENTITY CASCADE"))
await engine.dispose()
asyncio.run(_run())
def _load_app():
import importlib
import src.database as database_module
import src.api.auth as auth_module
import src.api.config_folders as config_folders_module
import src.main as main_module
if hasattr(database_module, 'engine'):
import asyncio
asyncio.run(database_module.engine.dispose())
importlib.reload(database_module)
importlib.reload(auth_module)
importlib.reload(config_folders_module)
importlib.reload(main_module)
return main_module.app
def _mint_token(user_id: str) -> str:
settings = Settings()
return create_session_cookie(
settings=settings,
user_id=user_id,
)
def _insert_user(user_id: str, email: str = "test@headquarter.local") -> None:
async def _run() -> None:
engine = create_async_engine(
build_database_url(
user="headquarter",
password="headquarter",
host="localhost",
port=5432,
database="headquarter",
)
)
async with engine.begin() as connection:
await connection.run_sync(Base.metadata.create_all)
session_factory = async_sessionmaker(engine, expire_on_commit=False)
async with session_factory() as session:
user = User(
id=uuid.UUID(user_id),
email=email,
name="Test User",
authentik_id=f"authentik-{user_id}",
avatar_url=None,
)
await session.merge(user)
await session.commit()
await engine.dispose()
asyncio.run(_run())
def _insert_config_folder(
folder_id: str,
user_id: str,
name: str,
mount_path: str = "/home/user",
files: dict | None = None,
is_active: bool = True,
) -> None:
async def _run() -> None:
engine = create_async_engine(
build_database_url(
user="headquarter",
password="headquarter",
host="localhost",
port=5432,
database="headquarter",
)
)
session_factory = async_sessionmaker(engine, expire_on_commit=False)
async with session_factory() as session:
folder = ConfigFolder(
id=uuid.UUID(folder_id),
user_id=uuid.UUID(user_id),
name=name,
description="Test config folder",
mount_path=mount_path,
files=files or {"test.txt": "hello world"},
is_active=is_active,
)
await session.merge(folder)
await session.commit()
await engine.dispose()
asyncio.run(_run())
@pytest.mark.integration @pytest.mark.integration
def test_list_config_folders_requires_authentication() -> None: class TestConfigFoldersAPI:
_prepare_test_db() """Integration tests for config folders API."""
app = _load_app()
client = TestClient(app)
response = client.get("/config-folders")
def test_list_config_folders_requires_authentication(self, test_client: TestClient) -> None:
"""Test that listing config folders requires authentication."""
response = test_client.get("/config-folders")
assert response.status_code == 401 assert response.status_code == 401
def test_list_config_folders_returns_user_folders(self, authenticated_client: TestClient) -> None:
@pytest.mark.integration """Test that authenticated users can list their folders."""
def test_list_config_folders_returns_user_folders() -> None: response = authenticated_client.get("/config-folders")
_prepare_test_db()
user_id = "11111111-1111-1111-1111-111111111111"
_insert_user(user_id)
_insert_config_folder(
"aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa",
user_id,
"my-dotfiles",
files={".zshrc": "export ZSH=\"$HOME/.oh-my-zsh\""},
)
app = _load_app()
client = TestClient(app)
client.cookies.set("session", _mint_token(user_id))
response = client.get("/config-folders")
assert response.status_code == 200 assert response.status_code == 200
data = response.json() data = response.json()
assert len(data) == 1 assert isinstance(data, dict)
assert data[0]["name"] == "my-dotfiles" assert "folders" in data
assert data[0]["files"] == {".zshrc": "export ZSH=\"$HOME/.oh-my-zsh\""} assert isinstance(data["folders"], list)
assert data[0]["is_active"] == True
def test_create_config_folder_successfully(self, authenticated_client: TestClient) -> None:
@pytest.mark.integration """Test creating a config folder."""
def test_list_config_folders_only_returns_own_folders() -> None: response = authenticated_client.post(
_prepare_test_db() "/config-folders",
user1_id = "11111111-1111-1111-1111-111111111111" json={
user2_id = "22222222-2222-2222-2222-222222222222" "name": "test-folder",
_insert_user(user1_id) "description": "Test folder",
_insert_user(user2_id)
_insert_config_folder(
"aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa",
user1_id,
"user1-folder",
)
_insert_config_folder(
"bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb",
user2_id,
"user2-folder",
)
app = _load_app()
client = TestClient(app)
client.cookies.set("session", _mint_token(user1_id))
response = client.get("/config-folders")
assert response.status_code == 200
data = response.json()
assert len(data) == 1
assert data[0]["name"] == "user1-folder"
@pytest.mark.integration
def test_create_config_folder_successfully() -> None:
_prepare_test_db()
user_id = "11111111-1111-1111-1111-111111111111"
_insert_user(user_id)
app = _load_app()
client = TestClient(app)
client.cookies.set("session", _mint_token(user_id))
payload = {
"name": "my-dotfiles",
"description": "My personal configuration files",
"mount_path": "/home/user", "mount_path": "/home/user",
"files": { "files": {"test.txt": "hello world"},
".zshrc": "export ZSH=\"$HOME/.oh-my-zsh\"",
".gitconfig": "[user]\\nname = Test User",
}, },
} )
response = client.post("/config-folders", json=payload)
assert response.status_code == 201 assert response.status_code == 201
data = response.json() data = response.json()
assert data["name"] == "my-dotfiles" assert data["name"] == "test-folder"
assert data["description"] == "My personal configuration files"
assert data["mount_path"] == "/home/user" assert data["mount_path"] == "/home/user"
assert data["files"] == { assert data["files"] == {"test.txt": "hello world"}
".zshrc": "export ZSH=\"$HOME/.oh-my-zsh\"",
".gitconfig": "[user]\\nname = Test User",
}
assert data["is_active"] == True
assert "id" in data
def test_create_config_folder_duplicate_name(self, authenticated_client: TestClient) -> None:
@pytest.mark.integration """Test that duplicate folder names are rejected."""
def test_create_config_folder_duplicate_name() -> None: # Create first folder
_prepare_test_db() response = authenticated_client.post(
user_id = "11111111-1111-1111-1111-111111111111" "/config-folders",
_insert_user(user_id) json={
_insert_config_folder( "name": "duplicate-folder",
"aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa",
user_id,
"existing-folder",
)
app = _load_app()
client = TestClient(app)
client.cookies.set("session", _mint_token(user_id))
payload = {
"name": "existing-folder",
"mount_path": "/home/user", "mount_path": "/home/user",
"files": {}, "files": {},
} },
response = client.post("/config-folders", json=payload) )
assert response.status_code == 201
# Try to create second with same name
response = authenticated_client.post(
"/config-folders",
json={
"name": "duplicate-folder",
"mount_path": "/home/user",
"files": {},
},
)
assert response.status_code == 409 assert response.status_code == 409
def test_create_config_folder_exceeds_size_limit(self, authenticated_client: TestClient) -> None:
@pytest.mark.integration """Test that folders exceeding 10MB are rejected."""
def test_create_config_folder_exceeds_size_limit() -> None:
_prepare_test_db()
user_id = "11111111-1111-1111-1111-111111111111"
_insert_user(user_id)
app = _load_app()
client = TestClient(app)
client.cookies.set("session", _mint_token(user_id))
# Create files that total > 10MB
large_content = "x" * (11 * 1024 * 1024) # 11MB large_content = "x" * (11 * 1024 * 1024) # 11MB
payload = { response = authenticated_client.post(
"name": "too-large", "/config-folders",
json={
"name": "large-folder",
"mount_path": "/home/user", "mount_path": "/home/user",
"files": { "files": {"large.txt": large_content},
"large.txt": large_content,
}, },
} )
response = client.post("/config-folders", json=payload)
assert response.status_code == 422 assert response.status_code == 422
def test_create_config_folder_path_traversal_attack(self, authenticated_client: TestClient) -> None:
@pytest.mark.integration """Test that path traversal in file paths is prevented."""
def test_create_config_folder_path_traversal_attack() -> None: response = authenticated_client.post(
_prepare_test_db() "/config-folders",
user_id = "11111111-1111-1111-1111-111111111111" json={
_insert_user(user_id) "name": "bad-folder",
app = _load_app()
client = TestClient(app)
client.cookies.set("session", _mint_token(user_id))
payload = {
"name": "attack",
"mount_path": "/home/user", "mount_path": "/home/user",
"files": { "files": {"../../../etc/passwd": "malicious"},
"../../../etc/passwd": "root:x:0:0",
}, },
} )
response = client.post("/config-folders", json=payload)
assert response.status_code == 422 assert response.status_code == 422
def test_get_config_folder_by_id(self, authenticated_client: TestClient) -> None:
@pytest.mark.integration """Test getting a config folder by ID."""
def test_get_config_folder_by_id() -> None: # Create folder first
_prepare_test_db() create_response = authenticated_client.post(
user_id = "11111111-1111-1111-1111-111111111111" "/config-folders",
folder_id = "aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa" json={
_insert_user(user_id) "name": "get-test",
_insert_config_folder( "mount_path": "/home/user",
folder_id,
user_id,
"my-dotfiles",
files={".zshrc": "test content"},
)
app = _load_app()
client = TestClient(app)
client.cookies.set("session", _mint_token(user_id))
response = client.get(f"/config-folders/{folder_id}")
assert response.status_code == 200
data = response.json()
assert data["id"] == folder_id
assert data["name"] == "my-dotfiles"
assert data["files"] == {".zshrc": "test content"}
@pytest.mark.integration
def test_get_config_folder_not_found() -> None:
_prepare_test_db()
user_id = "11111111-1111-1111-1111-111111111111"
_insert_user(user_id)
app = _load_app()
client = TestClient(app)
client.cookies.set("session", _mint_token(user_id))
response = client.get("/config-folders/aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa")
assert response.status_code == 404
@pytest.mark.integration
def test_get_config_folder_forbidden() -> None:
_prepare_test_db()
user1_id = "11111111-1111-1111-1111-111111111111"
user2_id = "22222222-2222-2222-2222-222222222222"
folder_id = "aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa"
_insert_user(user1_id)
_insert_user(user2_id)
_insert_config_folder(folder_id, user1_id, "private-folder")
app = _load_app()
client = TestClient(app)
client.cookies.set("session", _mint_token(user2_id))
response = client.get(f"/config-folders/{folder_id}")
assert response.status_code == 403
@pytest.mark.integration
def test_update_config_folder_successfully() -> None:
_prepare_test_db()
user_id = "11111111-1111-1111-1111-111111111111"
folder_id = "aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa"
_insert_user(user_id)
_insert_config_folder(
folder_id,
user_id,
"old-name",
mount_path="/old/path",
files={".zshrc": "old content"},
)
app = _load_app()
client = TestClient(app)
client.cookies.set("session", _mint_token(user_id))
payload = {
"name": "new-name",
"mount_path": "/new/path",
"files": {".zshrc": "new content"},
"is_active": False,
}
response = client.put(f"/config-folders/{folder_id}", json=payload)
assert response.status_code == 200
data = response.json()
assert data["name"] == "new-name"
assert data["mount_path"] == "/new/path"
assert data["files"] == {".zshrc": "new content"}
assert data["is_active"] == False
@pytest.mark.integration
def test_update_config_folder_not_found() -> None:
_prepare_test_db()
user_id = "11111111-1111-1111-1111-111111111111"
_insert_user(user_id)
app = _load_app()
client = TestClient(app)
client.cookies.set("session", _mint_token(user_id))
payload = {"name": "new-name", "mount_path": "/new/path", "files": {}}
response = client.put("/config-folders/aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa", json=payload)
assert response.status_code == 404
@pytest.mark.integration
def test_delete_config_folder_successfully() -> None:
_prepare_test_db()
user_id = "11111111-1111-1111-1111-111111111111"
folder_id = "aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa"
_insert_user(user_id)
_insert_config_folder(folder_id, user_id, "deletable-folder")
app = _load_app()
client = TestClient(app)
client.cookies.set("session", _mint_token(user_id))
response = client.delete(f"/config-folders/{folder_id}")
assert response.status_code == 204
# Verify it's gone
get_response = client.get(f"/config-folders/{folder_id}")
assert get_response.status_code == 404
@pytest.mark.integration
def test_add_project_override_successfully() -> None:
_prepare_test_db()
user_id = "11111111-1111-1111-1111-111111111111"
folder_id = "aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa"
project_id = "bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb"
_insert_user(user_id)
_insert_config_folder(
folder_id,
user_id,
"my-dotfiles",
files={".zshrc": "global content"},
)
app = _load_app()
client = TestClient(app)
client.cookies.set("session", _mint_token(user_id))
payload = {
"project_id": project_id,
"mount_path": "/workspace",
"files": {".zshrc": "project-specific content"},
}
response = client.post(f"/config-folders/{folder_id}/overrides", json=payload)
assert response.status_code == 201
data = response.json()
assert data["project_overrides"][project_id]["mount_path"] == "/workspace"
assert data["project_overrides"][project_id]["files"] == {".zshrc": "project-specific content"}
@pytest.mark.integration
def test_add_project_override_folder_not_found() -> None:
_prepare_test_db()
user_id = "11111111-1111-1111-1111-111111111111"
_insert_user(user_id)
app = _load_app()
client = TestClient(app)
client.cookies.set("session", _mint_token(user_id))
payload = {
"project_id": "bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb",
"mount_path": "/workspace",
"files": {}, "files": {},
} },
response = client.post("/config-folders/aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa/overrides", json=payload)
assert response.status_code == 404
@pytest.mark.integration
def test_update_project_override_successfully() -> None:
_prepare_test_db()
user_id = "11111111-1111-1111-1111-111111111111"
folder_id = "aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa"
project_id = "bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb"
_insert_user(user_id)
_insert_config_folder(
folder_id,
user_id,
"my-dotfiles",
files={".zshrc": "global"},
) )
folder_id = create_response.json()["id"]
app = _load_app() # Get it back
client = TestClient(app) response = authenticated_client.get(f"/config-folders/{folder_id}")
client.cookies.set("session", _mint_token(user_id))
# First add an override
client.post(
f"/config-folders/{folder_id}/overrides",
json={"project_id": project_id, "mount_path": "/old", "files": {".zshrc": "old"}},
)
# Then update it
payload = {
"mount_path": "/new",
"files": {".zshrc": "new"},
}
response = client.put(f"/config-folders/{folder_id}/overrides/{project_id}", json=payload)
assert response.status_code == 200 assert response.status_code == 200
data = response.json() data = response.json()
assert data["project_overrides"][project_id]["mount_path"] == "/new" assert data["name"] == "get-test"
assert data["project_overrides"][project_id]["files"] == {".zshrc": "new"}
def test_get_config_folder_not_found(self, authenticated_client: TestClient) -> None:
"""Test getting a non-existent folder."""
response = authenticated_client.get(f"/config-folders/{uuid.uuid4()}")
assert response.status_code == 404
@pytest.mark.integration def test_update_config_folder_successfully(self, authenticated_client: TestClient) -> None:
def test_delete_project_override_successfully() -> None: """Test updating a config folder."""
_prepare_test_db() # Create folder first
user_id = "11111111-1111-1111-1111-111111111111" create_response = authenticated_client.post(
folder_id = "aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa" "/config-folders",
project_id = "bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb" json={
_insert_user(user_id) "name": "update-test",
_insert_config_folder(folder_id, user_id, "my-dotfiles") "mount_path": "/home/user",
"files": {},
app = _load_app() },
client = TestClient(app)
client.cookies.set("session", _mint_token(user_id))
# Add an override first
client.post(
f"/config-folders/{folder_id}/overrides",
json={"project_id": project_id, "mount_path": "/workspace", "files": {}},
) )
folder_id = create_response.json()["id"]
# Update it
response = authenticated_client.put(
f"/config-folders/{folder_id}",
json={
"name": "updated-name",
"mount_path": "/workspace",
"files": {"new.txt": "content"},
},
)
assert response.status_code == 200
data = response.json()
assert data["name"] == "updated-name"
assert data["mount_path"] == "/workspace"
def test_delete_config_folder_successfully(self, authenticated_client: TestClient) -> None:
"""Test deleting a config folder."""
# Create folder first
create_response = authenticated_client.post(
"/config-folders",
json={
"name": "delete-test",
"mount_path": "/home/user",
"files": {},
},
)
folder_id = create_response.json()["id"]
# Delete it # Delete it
response = client.delete(f"/config-folders/{folder_id}/overrides/{project_id}") response = authenticated_client.delete(f"/config-folders/{folder_id}")
assert response.status_code == 204 assert response.status_code == 204
# Verify it's gone # Verify it's gone
get_response = client.get(f"/config-folders/{folder_id}") get_response = authenticated_client.get(f"/config-folders/{folder_id}")
data = get_response.json() assert get_response.status_code == 404
assert project_id not in data.get("project_overrides", {})
def test_add_project_override_successfully(self, authenticated_client: TestClient) -> None:
"""Test adding a project override."""
# Create folder first
create_response = authenticated_client.post(
"/config-folders",
json={
"name": "override-test",
"mount_path": "/home/user",
"files": {"global.txt": "global"},
},
)
folder_id = create_response.json()["id"]
project_id = str(uuid.uuid4())
# Add override
response = authenticated_client.post(
f"/config-folders/{folder_id}/overrides",
json={
"project_id": project_id,
"mount_path": "/workspace",
"files": {"project.txt": "project"},
},
)
assert response.status_code == 200
data = response.json()
assert project_id in data["project_overrides"]
def test_update_project_override_successfully(self, authenticated_client: TestClient) -> None:
"""Test updating a project override."""
# Create folder with override
create_response = authenticated_client.post(
"/config-folders",
json={
"name": "update-override-test",
"mount_path": "/home/user",
"files": {},
},
)
folder_id = create_response.json()["id"]
project_id = str(uuid.uuid4())
# Add override
authenticated_client.post(
f"/config-folders/{folder_id}/overrides",
json={
"project_id": project_id,
"mount_path": "/workspace",
"files": {"old.txt": "old"},
},
)
# Update override
response = authenticated_client.put(
f"/config-folders/{folder_id}/overrides/{project_id}",
json={
"mount_path": "/app",
"files": {"new.txt": "new"},
},
)
assert response.status_code == 200
data = response.json()
assert data["project_overrides"][project_id]["mount_path"] == "/app"
def test_delete_project_override_successfully(self, authenticated_client: TestClient) -> None:
"""Test deleting a project override."""
# Create folder with override
create_response = authenticated_client.post(
"/config-folders",
json={
"name": "delete-override-test",
"mount_path": "/home/user",
"files": {},
},
)
folder_id = create_response.json()["id"]
project_id = str(uuid.uuid4())
# Add override
authenticated_client.post(
f"/config-folders/{folder_id}/overrides",
json={
"project_id": project_id,
"mount_path": "/workspace",
"files": {},
},
)
# Delete override
response = authenticated_client.delete(
f"/config-folders/{folder_id}/overrides/{project_id}"
)
assert response.status_code == 200
data = response.json()
assert project_id not in data["project_overrides"]
@@ -1,378 +1,253 @@
import uuid import uuid
from datetime import UTC, datetime, timedelta
import asyncio
import pytest import pytest
from fastapi.testclient import TestClient from fastapi.testclient import TestClient
from sqlalchemy import text
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker
from src.auth.session import create_session_cookie
from src.config import Settings, build_database_url
from src.models import Base
from src.models.tool_config import ToolConfig
from src.models.tool_type import ToolType
from src.models.user import User
def _prepare_test_db() -> None:
async def _run() -> None:
engine = create_async_engine(
build_database_url(
user="headquarter",
password="headquarter",
host="localhost",
port=5432,
database="headquarter",
)
)
async with engine.begin() as connection:
await connection.run_sync(Base.metadata.create_all)
await connection.execute(text("TRUNCATE TABLE tool_configs, tool_types, users RESTART IDENTITY CASCADE"))
await engine.dispose()
asyncio.run(_run())
def _load_app():
import importlib
import src.database as database_module
import src.api.auth as auth_module
import src.api.tool_configs as tool_configs_module
import src.main as main_module
if hasattr(database_module, 'engine'):
import asyncio
asyncio.run(database_module.engine.dispose())
importlib.reload(database_module)
importlib.reload(auth_module)
importlib.reload(tool_configs_module)
importlib.reload(main_module)
return main_module.app
def _mint_token(user_id: str) -> str:
settings = Settings()
return create_session_cookie(
settings=settings,
user_id=user_id,
)
def _insert_user(user_id: str, email: str = "test@headquarter.local") -> None:
async def _run() -> None:
engine = create_async_engine(
build_database_url(
user="headquarter",
password="headquarter",
host="localhost",
port=5432,
database="headquarter",
)
)
async with engine.begin() as connection:
await connection.run_sync(Base.metadata.create_all)
session_factory = async_sessionmaker(engine, expire_on_commit=False)
async with session_factory() as session:
user = User(
id=uuid.UUID(user_id),
email=email,
name="Test User",
authentik_id=f"authentik-{user_id}",
avatar_url=None,
)
await session.merge(user)
await session.commit()
await engine.dispose()
asyncio.run(_run())
def _insert_tool_type(
tool_type_id: str,
name: str,
display_name: str,
created_by_id: str | None = None,
) -> None:
async def _run() -> None:
engine = create_async_engine(
build_database_url(
user="headquarter",
password="headquarter",
host="localhost",
port=5432,
database="headquarter",
)
)
session_factory = async_sessionmaker(engine, expire_on_commit=False)
async with session_factory() as session:
tool_type = ToolType(
id=uuid.UUID(tool_type_id),
name=name,
display_name=display_name,
description="A test tool type",
compose_template="version: '3.8'\\nservices:\\n app:\\n image: test",
required_variables=["REPO_PATH"],
is_builtin=False,
created_by_id=uuid.UUID(created_by_id) if created_by_id else None,
)
await session.merge(tool_type)
await session.commit()
await engine.dispose()
asyncio.run(_run())
def _insert_tool_config(
config_id: str,
user_id: str,
tool_type_id: str,
key: str,
value: str,
config_type: str = "env",
**kwargs,
) -> None:
async def _run() -> None:
engine = create_async_engine(
build_database_url(
user="headquarter",
password="headquarter",
host="localhost",
port=5432,
database="headquarter",
)
)
session_factory = async_sessionmaker(engine, expire_on_commit=False)
async with session_factory() as session:
config = ToolConfig(
id=uuid.UUID(config_id),
user_id=uuid.UUID(user_id),
tool_type_id=uuid.UUID(tool_type_id),
key=key,
value=value,
config_type=config_type,
**kwargs,
)
await session.merge(config)
await session.commit()
await engine.dispose()
asyncio.run(_run())
@pytest.mark.integration @pytest.mark.integration
def test_create_tool_config_with_new_fields() -> None: class TestToolConfigsAPIExtended:
_prepare_test_db() """Integration tests for tool configs API with new fields."""
user_id = "11111111-1111-1111-1111-111111111111"
tool_type_id = "aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa"
_insert_user(user_id)
_insert_tool_type(tool_type_id, "test-tool", "Test Tool", created_by_id=user_id)
app = _load_app() def test_create_tool_config_with_new_fields(self, authenticated_client: TestClient) -> None:
client = TestClient(app) """Test creating a tool config with all new fields."""
client.cookies.set("session", _mint_token(user_id)) # Create a tool type first
tool_response = authenticated_client.post(
"/tool-types",
json={
"name": "config-test-tool",
"display_name": "Config Test Tool",
"default_port": 8080,
"definition_type": "compose",
"compose_template": "version: '3.8'\nservices:\n app:\n image: nginx",
"required_variables": [],
},
)
tool_id = tool_response.json()["id"]
payload = { # Create config with new fields
"tool_type_id": tool_type_id, response = authenticated_client.post(
"key": "advanced-config", "/tool-configs",
json={
"tool_type_id": tool_id,
"key": "ADVANCED_CONFIG",
"value": "test-value", "value": "test-value",
"config_type": "env", "config_type": "env",
"port_override": 9090, "port_override": 9090,
"start_command": "python app.py --port 9090", "start_command": "python app.py",
"working_directory": "/app/src", "working_directory": "/app",
"environment_variables": {"DEBUG": "true", "LOG_LEVEL": "debug"}, "environment_variables": {"DEBUG": "true", "LOG_LEVEL": "debug"},
"volumes": [ "volumes": [
{"source": "dotfiles", "target": "/home/user/.config", "type": "config_folder"}, {"source": "data", "target": "/data", "type": "bind"}
], ],
} },
response = client.post("/tool-configs", json=payload) )
assert response.status_code == 201 assert response.status_code == 201
data = response.json() data = response.json()
assert data["key"] == "advanced-config" assert data["key"] == "ADVANCED_CONFIG"
assert data["port_override"] == 9090 assert data["port_override"] == 9090
assert data["start_command"] == "python app.py --port 9090" assert data["start_command"] == "python app.py"
assert data["working_directory"] == "/app/src"
assert data["environment_variables"] == {"DEBUG": "true", "LOG_LEVEL": "debug"}
assert len(data["volumes"]) == 1
assert data["volumes"][0]["source"] == "dotfiles"
@pytest.mark.integration
def test_create_tool_config_invalid_port() -> None:
_prepare_test_db()
user_id = "11111111-1111-1111-1111-111111111111"
tool_type_id = "aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa"
_insert_user(user_id)
_insert_tool_type(tool_type_id, "test-tool", "Test Tool", created_by_id=user_id)
app = _load_app()
client = TestClient(app)
client.cookies.set("session", _mint_token(user_id))
payload = {
"tool_type_id": tool_type_id,
"key": "bad-config",
"value": "test",
"port_override": 99999, # Invalid port
}
response = client.post("/tool-configs", json=payload)
assert response.status_code == 422
@pytest.mark.integration
def test_create_tool_config_invalid_volume_structure() -> None:
_prepare_test_db()
user_id = "11111111-1111-1111-1111-111111111111"
tool_type_id = "aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa"
_insert_user(user_id)
_insert_tool_type(tool_type_id, "test-tool", "Test Tool", created_by_id=user_id)
app = _load_app()
client = TestClient(app)
client.cookies.set("session", _mint_token(user_id))
payload = {
"tool_type_id": tool_type_id,
"key": "bad-config",
"value": "test",
"volumes": [
{"invalid_key": "value"}, # Missing required fields
],
}
response = client.post("/tool-configs", json=payload)
assert response.status_code == 422
@pytest.mark.integration
def test_update_tool_config_with_new_fields() -> None:
_prepare_test_db()
user_id = "11111111-1111-1111-1111-111111111111"
tool_type_id = "aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa"
config_id = "bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb"
_insert_user(user_id)
_insert_tool_type(tool_type_id, "test-tool", "Test Tool", created_by_id=user_id)
_insert_tool_config(config_id, user_id, tool_type_id, "my-config", "old-value")
app = _load_app()
client = TestClient(app)
client.cookies.set("session", _mint_token(user_id))
payload = {
"value": "new-value",
"port_override": 8080,
"start_command": "npm start",
"working_directory": "/app",
"environment_variables": {"NODE_ENV": "production"},
"volumes": [
{"source": "config", "target": "/app/config", "type": "bind"},
],
}
response = client.put(f"/tool-configs/{config_id}", json=payload)
assert response.status_code == 200
data = response.json()
assert data["value"] == "new-value"
assert data["port_override"] == 8080
assert data["start_command"] == "npm start"
assert data["working_directory"] == "/app" assert data["working_directory"] == "/app"
assert data["environment_variables"] == {"NODE_ENV": "production"} assert data["environment_variables"] == {"DEBUG": "true", "LOG_LEVEL": "debug"}
assert len(data["volumes"]) == 1 assert data["volumes"] == [{"source": "data", "target": "/data", "type": "bind"}]
def test_create_tool_config_invalid_port(self, authenticated_client: TestClient) -> None:
@pytest.mark.integration """Test that invalid port numbers are rejected."""
def test_list_tool_configs_returns_new_fields() -> None: # Create a tool type first
_prepare_test_db() tool_response = authenticated_client.post(
user_id = "11111111-1111-1111-1111-111111111111" "/tool-types",
tool_type_id = "aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa" json={
config_id = "bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb" "name": "port-test-tool",
_insert_user(user_id) "display_name": "Port Test Tool",
_insert_tool_type(tool_type_id, "test-tool", "Test Tool", created_by_id=user_id) "default_port": 8080,
_insert_tool_config( "definition_type": "compose",
config_id, "compose_template": "version: '3.8'\nservices:\n app:\n image: nginx",
user_id, "required_variables": [],
tool_type_id, },
"advanced-config",
"test-value",
port_override=9090,
start_command="python app.py",
working_directory="/app",
environment_variables={"DEBUG": "true"},
volumes=[{"source": "data", "target": "/data", "type": "bind"}],
) )
tool_id = tool_response.json()["id"]
app = _load_app() # Try to create config with invalid port
client = TestClient(app) response = authenticated_client.post(
client.cookies.set("session", _mint_token(user_id)) "/tool-configs",
json={
response = client.get("/tool-configs") "tool_type_id": tool_id,
"key": "BAD_PORT",
assert response.status_code == 200 "value": "test",
data = response.json()
assert len(data) == 1
config = data[0]
assert config["port_override"] == 9090
assert config["start_command"] == "python app.py"
assert config["working_directory"] == "/app"
assert config["environment_variables"] == {"DEBUG": "true"}
assert len(config["volumes"]) == 1
@pytest.mark.integration
def test_get_tool_config_defaults() -> None:
_prepare_test_db()
user_id = "11111111-1111-1111-1111-111111111111"
tool_type_id = "aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa"
_insert_user(user_id)
_insert_tool_type(
tool_type_id,
"test-tool",
"Test Tool",
created_by_id=user_id,
)
app = _load_app()
client = TestClient(app)
client.cookies.set("session", _mint_token(user_id))
response = client.get(f"/tool-configs/defaults/{tool_type_id}")
assert response.status_code == 200
data = response.json()
assert "tool_type_id" in data
assert data["tool_type_id"] == tool_type_id
assert "suggested_configs" in data
assert "port_override" in data
@pytest.mark.integration
def test_tool_config_backward_compatibility() -> None:
_prepare_test_db()
user_id = "11111111-1111-1111-1111-111111111111"
tool_type_id = "aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa"
_insert_user(user_id)
_insert_tool_type(tool_type_id, "test-tool", "Test Tool", created_by_id=user_id)
app = _load_app()
client = TestClient(app)
client.cookies.set("session", _mint_token(user_id))
# Create config without new fields (old API usage)
payload = {
"tool_type_id": tool_type_id,
"key": "simple-config",
"value": "simple-value",
"config_type": "env", "config_type": "env",
} "port_override": 99999,
response = client.post("/tool-configs", json=payload) },
)
assert response.status_code == 422
def test_create_tool_config_invalid_volume_structure(self, authenticated_client: TestClient) -> None:
"""Test that invalid volume structures are rejected."""
# Create a tool type first
tool_response = authenticated_client.post(
"/tool-types",
json={
"name": "volume-test-tool",
"display_name": "Volume Test Tool",
"default_port": 8080,
"definition_type": "compose",
"compose_template": "version: '3.8'\nservices:\n app:\n image: nginx",
"required_variables": [],
},
)
tool_id = tool_response.json()["id"]
# Try to create config with invalid volume
response = authenticated_client.post(
"/tool-configs",
json={
"tool_type_id": tool_id,
"key": "BAD_VOLUME",
"value": "test",
"config_type": "env",
"volumes": [{"invalid": "structure"}],
},
)
assert response.status_code == 422
def test_update_tool_config_with_new_fields(self, authenticated_client: TestClient) -> None:
"""Test updating a tool config with new fields."""
# Create a tool type first
tool_response = authenticated_client.post(
"/tool-types",
json={
"name": "update-config-tool",
"display_name": "Update Config Tool",
"default_port": 8080,
"definition_type": "compose",
"compose_template": "version: '3.8'\nservices:\n app:\n image: nginx",
"required_variables": [],
},
)
tool_id = tool_response.json()["id"]
# Create config
create_response = authenticated_client.post(
"/tool-configs",
json={
"tool_type_id": tool_id,
"key": "UPDATE_TEST",
"value": "original",
"config_type": "env",
},
)
config_id = create_response.json()["id"]
# Update with new fields
response = authenticated_client.put(
f"/tool-configs/{config_id}",
json={
"value": "updated",
"port_override": 3000,
"start_command": "npm start",
"working_directory": "/workspace",
"environment_variables": {"NODE_ENV": "production"},
"volumes": [{"source": "src", "target": "/app/src", "type": "bind"}],
},
)
assert response.status_code == 200
data = response.json()
assert data["value"] == "updated"
assert data["port_override"] == 3000
assert data["start_command"] == "npm start"
assert data["working_directory"] == "/workspace"
assert data["environment_variables"] == {"NODE_ENV": "production"}
def test_list_tool_configs_returns_new_fields(self, authenticated_client: TestClient) -> None:
"""Test that listing configs returns new fields."""
# Create a tool type first
tool_response = authenticated_client.post(
"/tool-types",
json={
"name": "list-config-tool",
"display_name": "List Config Tool",
"default_port": 8080,
"definition_type": "compose",
"compose_template": "version: '3.8'\nservices:\n app:\n image: nginx",
"required_variables": [],
},
)
tool_id = tool_response.json()["id"]
# Create config with new fields
authenticated_client.post(
"/tool-configs",
json={
"tool_type_id": tool_id,
"key": "LIST_TEST",
"value": "test",
"config_type": "env",
"port_override": 5000,
"environment_variables": {"TEST": "true"},
},
)
# List configs
response = authenticated_client.get("/tool-configs")
assert response.status_code == 200
data = response.json()
assert len(data) > 0
config = data[0]
assert "port_override" in config
assert "start_command" in config
assert "working_directory" in config
assert "environment_variables" in config
assert "volumes" in config
def test_get_tool_config_defaults(self, authenticated_client: TestClient) -> None:
"""Test getting tool config defaults."""
# Create a tool type first
tool_response = authenticated_client.post(
"/tool-types",
json={
"name": "defaults-tool",
"display_name": "Defaults Tool",
"default_port": 8080,
"definition_type": "compose",
"compose_template": "version: '3.8'\nservices:\n app:\n image: nginx",
"required_variables": ["REPO_PATH"],
},
)
tool_id = tool_response.json()["id"]
# Get defaults
response = authenticated_client.get(f"/tool-configs/defaults/{tool_id}")
assert response.status_code == 200
data = response.json()
assert data["tool_type_id"] == tool_id
assert "suggested_configs" in data
def test_tool_config_backward_compatibility(self, authenticated_client: TestClient) -> None:
"""Test that old configs without new fields still work."""
# Create a tool type first
tool_response = authenticated_client.post(
"/tool-types",
json={
"name": "backward-compat-tool",
"display_name": "Backward Compat Tool",
"default_port": 8080,
"definition_type": "compose",
"compose_template": "version: '3.8'\nservices:\n app:\n image: nginx",
"required_variables": [],
},
)
tool_id = tool_response.json()["id"]
# Create config without new fields (simulating old client)
response = authenticated_client.post(
"/tool-configs",
json={
"tool_type_id": tool_id,
"key": "OLD_STYLE",
"value": "value",
"config_type": "env",
},
)
assert response.status_code == 201 assert response.status_code == 201
data = response.json() data = response.json()
assert data["key"] == "simple-config" assert data["key"] == "OLD_STYLE"
# New fields should have default values # New fields should have default values
assert data["port_override"] is None assert data["port_override"] is None
assert data["start_command"] is None assert data["start_command"] is None
@@ -1,415 +1,188 @@
import uuid import uuid
from datetime import UTC, datetime, timedelta
import asyncio
import pytest import pytest
from fastapi.testclient import TestClient from fastapi.testclient import TestClient
from sqlalchemy import text
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker
from src.auth.session import create_session_cookie
from src.config import Settings, build_database_url
from src.models import Base
from src.models.tool_type import ToolType
from src.models.user import User
def _prepare_test_db() -> None:
async def _run() -> None:
engine = create_async_engine(
build_database_url(
user="headquarter",
password="headquarter",
host="localhost",
port=5432,
database="headquarter",
)
)
async with engine.begin() as connection:
await connection.run_sync(Base.metadata.create_all)
await connection.execute(text("TRUNCATE TABLE tool_types, users RESTART IDENTITY CASCADE"))
await engine.dispose()
asyncio.run(_run())
def _load_app():
import importlib
import src.database as database_module
import src.api.auth as auth_module
import src.api.tool_types as tool_types_module
import src.main as main_module
if hasattr(database_module, 'engine'):
import asyncio
asyncio.run(database_module.engine.dispose())
importlib.reload(database_module)
importlib.reload(auth_module)
importlib.reload(tool_types_module)
importlib.reload(main_module)
return main_module.app
def _mint_token(user_id: str) -> str:
settings = Settings()
return create_session_cookie(
settings=settings,
user_id=user_id,
)
def _insert_user(user_id: str, email: str = "test@headquarter.local") -> None:
async def _run() -> None:
engine = create_async_engine(
build_database_url(
user="headquarter",
password="headquarter",
host="localhost",
port=5432,
database="headquarter",
)
)
async with engine.begin() as connection:
await connection.run_sync(Base.metadata.create_all)
session_factory = async_sessionmaker(engine, expire_on_commit=False)
async with session_factory() as session:
user = User(
id=uuid.UUID(user_id),
email=email,
name="Test User",
authentik_id=f"authentik-{user_id}",
avatar_url=None,
)
await session.merge(user)
await session.commit()
await engine.dispose()
asyncio.run(_run())
def _insert_tool_type(
tool_type_id: str,
name: str,
display_name: str,
compose_template: str | None = None,
dockerfile_template: str | None = None,
definition_type: str = "compose",
readiness_probe: dict | None = None,
is_builtin: bool = False,
created_by_id: str | None = None,
) -> None:
async def _run() -> None:
engine = create_async_engine(
build_database_url(
user="headquarter",
password="headquarter",
host="localhost",
port=5432,
database="headquarter",
)
)
session_factory = async_sessionmaker(engine, expire_on_commit=False)
async with session_factory() as session:
tool_type = ToolType(
id=uuid.UUID(tool_type_id),
name=name,
display_name=display_name,
description="A test tool type",
compose_template=compose_template,
dockerfile_template=dockerfile_template,
definition_type=definition_type,
readiness_probe=readiness_probe,
required_variables=["REPO_PATH"],
is_builtin=is_builtin,
created_by_id=uuid.UUID(created_by_id) if created_by_id else None,
)
await session.merge(tool_type)
await session.commit()
await engine.dispose()
asyncio.run(_run())
@pytest.mark.integration @pytest.mark.integration
def test_create_tool_type_with_dockerfile() -> None: class TestToolTypesAPIExtended:
_prepare_test_db() """Integration tests for tool types API with new fields."""
user_id = "11111111-1111-1111-1111-111111111111"
_insert_user(user_id)
app = _load_app() def test_create_tool_type_with_dockerfile(self, authenticated_client: TestClient) -> None:
client = TestClient(app) """Test creating a tool type with dockerfile definition."""
client.cookies.set("session", _mint_token(user_id)) response = authenticated_client.post(
"/tool-types",
payload = { json={
"name": "custom-docker-tool", "name": "dockerfile-tool",
"display_name": "Custom Docker Tool", "display_name": "Dockerfile Tool",
"description": "A custom tool with Dockerfile", "category": "utility",
"interfaces": ["terminal"],
"default_port": 8080,
"definition_type": "dockerfile", "definition_type": "dockerfile",
"dockerfile_template": "FROM python:3.11\\nRUN pip install flask\\nCMD ['python', 'app.py']", "dockerfile_template": "FROM python:3.11\nRUN pip install flask",
"default_port": 5000, "required_variables": [],
"required_variables": ["REPO_PATH"], },
} )
response = client.post("/tool-types", json=payload)
assert response.status_code == 201 assert response.status_code == 201
data = response.json() data = response.json()
assert data["name"] == "custom-docker-tool" assert data["name"] == "dockerfile-tool"
assert data["definition_type"] == "dockerfile" assert data["definition_type"] == "dockerfile"
assert data["dockerfile_template"] == "FROM python:3.11\\nRUN pip install flask\\nCMD ['python', 'app.py']" assert data["dockerfile_template"] == "FROM python:3.11\nRUN pip install flask"
assert data["compose_template"] is None
def test_create_tool_type_with_readiness_probe(self, authenticated_client: TestClient) -> None:
@pytest.mark.integration """Test creating a tool type with readiness probe."""
def test_create_tool_type_with_readiness_probe() -> None: response = authenticated_client.post(
_prepare_test_db() "/tool-types",
user_id = "11111111-1111-1111-1111-111111111111" json={
_insert_user(user_id)
app = _load_app()
client = TestClient(app)
client.cookies.set("session", _mint_token(user_id))
payload = {
"name": "probed-tool", "name": "probed-tool",
"display_name": "Probed Tool", "display_name": "Probed Tool",
"compose_template": "version: '3.8'\\nservices:\\n app:\\n image: nginx", "category": "utility",
"interfaces": ["web"],
"default_port": 8080, "default_port": 8080,
"required_variables": [], "definition_type": "compose",
"compose_template": "version: '3.8'\nservices:\n app:\n image: nginx",
"readiness_probe": { "readiness_probe": {
"command": "curl -f http://localhost:8080/health", "command": "curl -f http://localhost:8080",
"timeout": 60,
"interval": 3,
},
}
response = client.post("/tool-types", json=payload)
assert response.status_code == 201
data = response.json()
assert data["readiness_probe"] == {
"command": "curl -f http://localhost:8080/health",
"timeout": 60,
"interval": 3,
}
@pytest.mark.integration
def test_create_tool_type_invalid_definition_type() -> None:
_prepare_test_db()
user_id = "11111111-1111-1111-1111-111111111111"
_insert_user(user_id)
app = _load_app()
client = TestClient(app)
client.cookies.set("session", _mint_token(user_id))
payload = {
"name": "bad-tool",
"display_name": "Bad Tool",
"definition_type": "invalid",
"compose_template": "version: '3.8'\\nservices:\\n app:\\n image: nginx",
"required_variables": [],
}
response = client.post("/tool-types", json=payload)
assert response.status_code == 422
@pytest.mark.integration
def test_create_tool_type_dockerfile_without_template() -> None:
_prepare_test_db()
user_id = "11111111-1111-1111-1111-111111111111"
_insert_user(user_id)
app = _load_app()
client = TestClient(app)
client.cookies.set("session", _mint_token(user_id))
payload = {
"name": "bad-docker-tool",
"display_name": "Bad Docker Tool",
"definition_type": "dockerfile",
"dockerfile_template": "",
"required_variables": [],
}
response = client.post("/tool-types", json=payload)
assert response.status_code == 422
@pytest.mark.integration
def test_update_tool_type_with_new_fields() -> None:
_prepare_test_db()
user_id = "11111111-1111-1111-1111-111111111111"
tool_type_id = "aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa"
_insert_user(user_id)
_insert_tool_type(
tool_type_id,
"custom-tool",
"Custom Tool",
compose_template="version: '3.8'\\nservices:\\n app:\\n image: old",
created_by_id=user_id,
)
app = _load_app()
client = TestClient(app)
client.cookies.set("session", _mint_token(user_id))
payload = {
"definition_type": "dockerfile",
"dockerfile_template": "FROM python:3.11",
"readiness_probe": {
"command": "python --version",
"timeout": 30, "timeout": 30,
"interval": 2, "interval": 2,
}, },
} "required_variables": [],
response = client.put(f"/tool-types/{tool_type_id}", json=payload) },
assert response.status_code == 200
data = response.json()
assert data["definition_type"] == "dockerfile"
assert data["dockerfile_template"] == "FROM python:3.11"
assert data["readiness_probe"]["command"] == "python --version"
@pytest.mark.integration
def test_validate_tool_type_compose() -> None:
_prepare_test_db()
user_id = "11111111-1111-1111-1111-111111111111"
_insert_user(user_id)
app = _load_app()
client = TestClient(app)
client.cookies.set("session", _mint_token(user_id))
payload = {
"definition_type": "compose",
"compose_template": "version: '3.8'\\nservices:\\n app:\\n image: nginx",
}
response = client.post("/tool-types/validate", json=payload)
assert response.status_code == 200
data = response.json()
assert data["valid"] == True
assert "errors" not in data or len(data["errors"]) == 0
@pytest.mark.integration
def test_validate_tool_type_invalid_compose() -> None:
_prepare_test_db()
user_id = "11111111-1111-1111-1111-111111111111"
_insert_user(user_id)
app = _load_app()
client = TestClient(app)
client.cookies.set("session", _mint_token(user_id))
payload = {
"definition_type": "compose",
"compose_template": "this is not: valid: yaml: [",
}
response = client.post("/tool-types/validate", json=payload)
assert response.status_code == 200
data = response.json()
assert data["valid"] == False
assert len(data["errors"]) > 0
@pytest.mark.integration
def test_validate_tool_type_dockerfile() -> None:
_prepare_test_db()
user_id = "11111111-1111-1111-1111-111111111111"
_insert_user(user_id)
app = _load_app()
client = TestClient(app)
client.cookies.set("session", _mint_token(user_id))
payload = {
"definition_type": "dockerfile",
"dockerfile_template": "FROM python:3.11\\nRUN pip install flask",
}
response = client.post("/tool-types/validate", json=payload)
assert response.status_code == 200
data = response.json()
assert data["valid"] == True
@pytest.mark.integration
def test_validate_tool_type_empty_dockerfile() -> None:
_prepare_test_db()
user_id = "11111111-1111-1111-1111-111111111111"
_insert_user(user_id)
app = _load_app()
client = TestClient(app)
client.cookies.set("session", _mint_token(user_id))
payload = {
"definition_type": "dockerfile",
"dockerfile_template": "",
}
response = client.post("/tool-types/validate", json=payload)
assert response.status_code == 200
data = response.json()
assert data["valid"] == False
@pytest.mark.integration
def test_get_tool_type_returns_new_fields() -> None:
_prepare_test_db()
user_id = "11111111-1111-1111-1111-111111111111"
tool_type_id = "aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa"
_insert_user(user_id)
_insert_tool_type(
tool_type_id,
"probed-tool",
"Probed Tool",
definition_type="dockerfile",
dockerfile_template="FROM python:3.11",
readiness_probe={"command": "python --version", "timeout": 30, "interval": 2},
created_by_id=user_id,
) )
assert response.status_code == 201
app = _load_app()
client = TestClient(app)
client.cookies.set("session", _mint_token(user_id))
response = client.get(f"/tool-types/{tool_type_id}")
assert response.status_code == 200
data = response.json() data = response.json()
assert data["definition_type"] == "dockerfile" assert data["readiness_probe"]["command"] == "curl -f http://localhost:8080"
assert data["dockerfile_template"] == "FROM python:3.11"
assert data["readiness_probe"]["command"] == "python --version"
assert data["readiness_probe"]["timeout"] == 30 assert data["readiness_probe"]["timeout"] == 30
def test_create_tool_type_invalid_definition_type(self, authenticated_client: TestClient) -> None:
"""Test that invalid definition types are rejected."""
response = authenticated_client.post(
"/tool-types",
json={
"name": "invalid-tool",
"display_name": "Invalid Tool",
"default_port": 8080,
"definition_type": "invalid",
"compose_template": "version: '3.8'\nservices:\n app:\n image: nginx",
"required_variables": [],
},
)
assert response.status_code == 422
@pytest.mark.integration def test_create_tool_type_dockerfile_without_template(self, authenticated_client: TestClient) -> None:
def test_builtin_tool_types_have_definition_type_compose() -> None: """Test that dockerfile type requires dockerfile_template."""
_prepare_test_db() response = authenticated_client.post(
user_id = "11111111-1111-1111-1111-111111111111" "/tool-types",
_insert_user(user_id) json={
"name": "no-dockerfile",
"display_name": "No Dockerfile",
"default_port": 8080,
"definition_type": "dockerfile",
"required_variables": [],
},
)
assert response.status_code == 422
# Load app triggers startup event which seeds built-in types def test_update_tool_type_with_new_fields(self, authenticated_client: TestClient) -> None:
app = _load_app() """Test updating a tool type with new fields."""
client = TestClient(app) # Create tool type first
client.cookies.set("session", _mint_token(user_id)) create_response = authenticated_client.post(
"/tool-types",
response = client.get("/tool-types") json={
"name": "update-test-tool",
"display_name": "Update Test Tool",
"default_port": 8080,
"definition_type": "compose",
"compose_template": "version: '3.8'\nservices:\n app:\n image: nginx",
"required_variables": [],
},
)
tool_id = create_response.json()["id"]
# Update it
response = authenticated_client.put(
f"/tool-types/{tool_id}",
json={
"display_name": "Updated Name",
"readiness_probe": {
"command": "curl -f http://localhost:8080/health",
"timeout": 60,
"interval": 5,
},
},
)
assert response.status_code == 200 assert response.status_code == 200
data = response.json() data = response.json()
assert data["display_name"] == "Updated Name"
assert data["readiness_probe"]["command"] == "curl -f http://localhost:8080/health"
builtin_types = [t for t in data if t["is_builtin"]] def test_validate_tool_type_compose(self, authenticated_client: TestClient) -> None:
assert len(builtin_types) > 0 """Test validating compose template."""
response = authenticated_client.post(
"/tool-types/validate",
json={
"definition_type": "compose",
"compose_template": "version: '3.8'\nservices:\n app:\n image: nginx",
},
)
assert response.status_code == 200
data = response.json()
assert data["valid"] is True
for tool_type in builtin_types: def test_validate_tool_type_invalid_compose(self, authenticated_client: TestClient) -> None:
assert tool_type["definition_type"] == "compose" """Test validating invalid compose template."""
assert tool_type["compose_template"] is not None response = authenticated_client.post(
assert tool_type["dockerfile_template"] is None "/tool-types/validate",
json={
"definition_type": "compose",
"compose_template": "invalid: yaml: [",
},
)
assert response.status_code == 200
data = response.json()
assert data["valid"] is False
assert "error" in data
def test_validate_tool_type_dockerfile(self, authenticated_client: TestClient) -> None:
"""Test validating dockerfile template."""
response = authenticated_client.post(
"/tool-types/validate",
json={
"definition_type": "dockerfile",
"dockerfile_template": "FROM python:3.11\nRUN pip install flask",
},
)
assert response.status_code == 200
data = response.json()
assert data["valid"] is True
def test_get_tool_type_returns_new_fields(self, authenticated_client: TestClient) -> None:
"""Test that GET returns new fields."""
# Create tool type with all fields
create_response = authenticated_client.post(
"/tool-types",
json={
"name": "full-tool",
"display_name": "Full Tool",
"category": "editor",
"interfaces": ["web", "terminal"],
"default_port": 8443,
"definition_type": "compose",
"compose_template": "version: '3.8'\nservices:\n app:\n image: code-server",
"readiness_probe": {
"command": "curl -f http://localhost:8443",
"timeout": 30,
"interval": 2,
},
"required_variables": ["REPO_PATH"],
},
)
tool_id = create_response.json()["id"]
# Get it
response = authenticated_client.get(f"/tool-types/{tool_id}")
assert response.status_code == 200
data = response.json()
assert data["definition_type"] == "compose"
assert data["category"] == "editor"
assert data["interfaces"] == ["web", "terminal"]
assert "readiness_probe" in data
+20 -10
View File
@@ -487,8 +487,9 @@ export const ToolWorkshopPage = () => {
<h3>{selectedToolType ? "Edit" : "Create"} Tool Type</h3> <h3>{selectedToolType ? "Edit" : "Create"} Tool Type</h3>
<form onSubmit={handleToolTypeSubmit} className="stack"> <form onSubmit={handleToolTypeSubmit} className="stack">
<div className="form-group"> <div className="form-group">
<label>Definition Type</label> <label htmlFor="definition-type">Definition Type</label>
<select <select
id="definition-type"
value={toolTypeForm.definition_type} value={toolTypeForm.definition_type}
onChange={(e) => setToolTypeForm({ ...toolTypeForm, definition_type: e.target.value as "compose" | "dockerfile" })} onChange={(e) => setToolTypeForm({ ...toolTypeForm, definition_type: e.target.value as "compose" | "dockerfile" })}
className="form-input" className="form-input"
@@ -499,8 +500,9 @@ export const ToolWorkshopPage = () => {
</div> </div>
<div className="form-group"> <div className="form-group">
<label>Name *</label> <label htmlFor="tool-type-name">Name *</label>
<input <input
id="tool-type-name"
type="text" type="text"
value={toolTypeForm.name} value={toolTypeForm.name}
onChange={(e) => setToolTypeForm({ ...toolTypeForm, name: e.target.value })} onChange={(e) => setToolTypeForm({ ...toolTypeForm, name: e.target.value })}
@@ -512,8 +514,9 @@ export const ToolWorkshopPage = () => {
</div> </div>
<div className="form-group"> <div className="form-group">
<label>Display Name *</label> <label htmlFor="tool-type-display-name">Display Name *</label>
<input <input
id="tool-type-display-name"
type="text" type="text"
value={toolTypeForm.display_name} value={toolTypeForm.display_name}
onChange={(e) => setToolTypeForm({ ...toolTypeForm, display_name: e.target.value })} onChange={(e) => setToolTypeForm({ ...toolTypeForm, display_name: e.target.value })}
@@ -524,8 +527,9 @@ export const ToolWorkshopPage = () => {
</div> </div>
<div className="form-group"> <div className="form-group">
<label>Description</label> <label htmlFor="tool-type-description">Description</label>
<input <input
id="tool-type-description"
type="text" type="text"
value={toolTypeForm.description} value={toolTypeForm.description}
onChange={(e) => setToolTypeForm({ ...toolTypeForm, description: e.target.value })} onChange={(e) => setToolTypeForm({ ...toolTypeForm, description: e.target.value })}
@@ -535,8 +539,9 @@ export const ToolWorkshopPage = () => {
</div> </div>
<div className="form-group"> <div className="form-group">
<label>Category</label> <label htmlFor="tool-type-category">Category</label>
<input <input
id="tool-type-category"
type="text" type="text"
value={toolTypeForm.category} value={toolTypeForm.category}
onChange={(e) => setToolTypeForm({ ...toolTypeForm, category: e.target.value })} onChange={(e) => setToolTypeForm({ ...toolTypeForm, category: e.target.value })}
@@ -568,8 +573,9 @@ export const ToolWorkshopPage = () => {
</div> </div>
<div className="form-group"> <div className="form-group">
<label>Default Port *</label> <label htmlFor="tool-type-default-port">Default Port *</label>
<input <input
id="tool-type-default-port"
type="number" type="number"
value={toolTypeForm.default_port} value={toolTypeForm.default_port}
onChange={(e) => setToolTypeForm({ ...toolTypeForm, default_port: e.target.value })} onChange={(e) => setToolTypeForm({ ...toolTypeForm, default_port: e.target.value })}
@@ -580,8 +586,9 @@ export const ToolWorkshopPage = () => {
</div> </div>
<div className="form-group"> <div className="form-group">
<label>{toolTypeForm.definition_type === "compose" ? "Compose Template" : "Dockerfile"} *</label> <label htmlFor="tool-type-template">{toolTypeForm.definition_type === "compose" ? "Compose Template" : "Dockerfile"} *</label>
<textarea <textarea
id="tool-type-template"
value={toolTypeForm.definition_type === "compose" ? toolTypeForm.compose_template : toolTypeForm.dockerfile_template} value={toolTypeForm.definition_type === "compose" ? toolTypeForm.compose_template : toolTypeForm.dockerfile_template}
onChange={(e) => { onChange={(e) => {
if (toolTypeForm.definition_type === "compose") { if (toolTypeForm.definition_type === "compose") {
@@ -702,8 +709,9 @@ export const ToolWorkshopPage = () => {
<h3>{selectedConfig ? "Edit" : "Add"} Config</h3> <h3>{selectedConfig ? "Edit" : "Add"} Config</h3>
<form onSubmit={handleConfigSubmit} className="stack"> <form onSubmit={handleConfigSubmit} className="stack">
<div className="form-group"> <div className="form-group">
<label>Tool Type *</label> <label htmlFor="config-tool-type">Tool Type *</label>
<select <select
id="config-tool-type"
value={configForm.tool_type_id} value={configForm.tool_type_id}
onChange={(e) => setConfigForm({ ...configForm, tool_type_id: e.target.value })} onChange={(e) => setConfigForm({ ...configForm, tool_type_id: e.target.value })}
className="form-input" className="form-input"
@@ -717,8 +725,9 @@ export const ToolWorkshopPage = () => {
</div> </div>
<div className="form-group"> <div className="form-group">
<label>Key *</label> <label htmlFor="config-key">Key *</label>
<input <input
id="config-key"
type="text" type="text"
value={configForm.key} value={configForm.key}
onChange={(e) => setConfigForm({ ...configForm, key: e.target.value })} onChange={(e) => setConfigForm({ ...configForm, key: e.target.value })}
@@ -729,8 +738,9 @@ export const ToolWorkshopPage = () => {
</div> </div>
<div className="form-group"> <div className="form-group">
<label>Config Type</label> <label htmlFor="config-type">Config Type</label>
<select <select
id="config-type"
value={configForm.config_type} value={configForm.config_type}
onChange={(e) => setConfigForm({ ...configForm, config_type: e.target.value })} onChange={(e) => setConfigForm({ ...configForm, config_type: e.target.value })}
className="form-input" className="form-input"