alex/headquarter
1
1
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

test: add comprehensive tests for tool workshop functionality

Browse Source 
Backend tests:
- Unit tests for docker_build service (successful/failed builds, context, paths)
- Unit tests for readiness_probe service (success, timeout, retries, edge cases)
- Integration tests for config_folders API (CRUD + project overrides)
- Integration tests for tool_types API with new fields
- Integration tests for tool_configs API with new fields

Frontend tests:
- ToolWorkshopPage component tests (all 3 tabs, create/edit/delete)
- API client tests for tool_types and config_folders

Fixes:
- Add field_validator import to tool_configs.py
- Add JSON import to tool_config model
- Update frontend test button names to match UI (Create Tool Type, Add Config, Create Folder)

Quality gates: backend unit tests passing (23/23)
This commit is contained in:
Fusion
2026-05-22 19:45:10 +02:00
parent 8dd350286e
commit dacf105200
13 changed files with 2621 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/api/src/api/tool_configs.py
+1 -1
View File
@@ -4,7 +4,7 @@ import logging
import uuid
from fastapi import APIRouter, Depends, HTTPException, status
from pydantic import BaseModel, Field
from pydantic import BaseModel, Field, field_validator
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
apps/api/src/models/tool_config.py
+1 -1
View File
@@ -1,7 +1,7 @@
import uuid
from typing import TYPE_CHECKING
from sqlalchemy import ForeignKey, String, Text
from sqlalchemy import ForeignKey, JSON, String, Text
from sqlalchemy import Uuid as UUID
from sqlalchemy.orm import Mapped, mapped_column, relationship
apps/api/tests/integration/test_config_folders_api.py
+546
View File
@@ -0,0 +1,546 @@
import uuid
from datetime import UTC, datetime, timedelta
import asyncio
import pytest
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
def test_list_config_folders_requires_authentication() -> None:
_prepare_test_db()
app = _load_app()
client = TestClient(app)
response = client.get("/config-folders")
assert response.status_code == 401
@pytest.mark.integration
def test_list_config_folders_returns_user_folders() -> None:
_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
data = response.json()
assert len(data) == 1
assert data[0]["name"] == "my-dotfiles"
assert data[0]["files"] == {".zshrc": "export ZSH=\"$HOME/.oh-my-zsh\""}
assert data[0]["is_active"] == True
@pytest.mark.integration
def test_list_config_folders_only_returns_own_folders() -> None:
_prepare_test_db()
user1_id = "11111111-1111-1111-1111-111111111111"
user2_id = "22222222-2222-2222-2222-222222222222"
_insert_user(user1_id)
_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",
"files": {
".zshrc": "export ZSH=\"$HOME/.oh-my-zsh\"",
".gitconfig": "[user]\\nname = Test User",
},
}
response = client.post("/config-folders", json=payload)
assert response.status_code == 201
data = response.json()
assert data["name"] == "my-dotfiles"
assert data["description"] == "My personal configuration files"
assert data["mount_path"] == "/home/user"
assert data["files"] == {
".zshrc": "export ZSH=\"$HOME/.oh-my-zsh\"",
".gitconfig": "[user]\\nname = Test User",
}
assert data["is_active"] == True
assert "id" in data
@pytest.mark.integration
def test_create_config_folder_duplicate_name() -> None:
_prepare_test_db()
user_id = "11111111-1111-1111-1111-111111111111"
_insert_user(user_id)
_insert_config_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",
"files": {},
}
response = client.post("/config-folders", json=payload)
assert response.status_code == 409
@pytest.mark.integration
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
payload = {
"name": "too-large",
"mount_path": "/home/user",
"files": {
"large.txt": large_content,
},
}
response = client.post("/config-folders", json=payload)
assert response.status_code == 422
@pytest.mark.integration
def test_create_config_folder_path_traversal_attack() -> 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": "attack",
"mount_path": "/home/user",
"files": {
"../../../etc/passwd": "root:x:0:0",
},
}
response = client.post("/config-folders", json=payload)
assert response.status_code == 422
@pytest.mark.integration
def test_get_config_folder_by_id() -> 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,
"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": {},
}
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"},
)
app = _load_app()
client = TestClient(app)
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
data = response.json()
assert data["project_overrides"][project_id]["mount_path"] == "/new"
assert data["project_overrides"][project_id]["files"] == {".zshrc": "new"}
@pytest.mark.integration
def test_delete_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")
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": {}},
)
# Delete it
response = client.delete(f"/config-folders/{folder_id}/overrides/{project_id}")
assert response.status_code == 204
# Verify it's gone
get_response = client.get(f"/config-folders/{folder_id}")
data = get_response.json()
assert project_id not in data.get("project_overrides", {})
apps/api/tests/integration/test_projects_api.py
+1 -1
View File
@@ -6,7 +6,7 @@ import pytest
from sqlalchemy import text
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker
from src.auth.jwt_service import mint_access_token
from src.auth.session import mint_access_token
from src.config import Settings, build_database_url
from src.models import Base
from src.models.project import Project
apps/api/tests/integration/test_tool_configs_api_extended.py
+381
View File
@@ -0,0 +1,381 @@
import uuid
from datetime import UTC, datetime, timedelta
import asyncio
import pytest
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
def test_create_tool_config_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, "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": "advanced-config",
"value": "test-value",
"config_type": "env",
"port_override": 9090,
"start_command": "python app.py --port 9090",
"working_directory": "/app/src",
"environment_variables": {"DEBUG": "true", "LOG_LEVEL": "debug"},
"volumes": [
{"source": "dotfiles", "target": "/home/user/.config", "type": "config_folder"},
],
}
response = client.post("/tool-configs", json=payload)
assert response.status_code == 201
data = response.json()
assert data["key"] == "advanced-config"
assert data["port_override"] == 9090
assert data["start_command"] == "python app.py --port 9090"
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["environment_variables"] == {"NODE_ENV": "production"}
assert len(data["volumes"]) == 1
@pytest.mark.integration
def test_list_tool_configs_returns_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,
"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"}],
)
app = _load_app()
client = TestClient(app)
client.cookies.set("session", _mint_token(user_id))
response = client.get("/tool-configs")
assert response.status_code == 200
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",
}
response = client.post("/tool-configs", json=payload)
assert response.status_code == 201
data = response.json()
assert data["key"] == "simple-config"
# New fields should have default values
assert data["port_override"] is None
assert data["start_command"] is None
assert data["working_directory"] is None
assert data["environment_variables"] == {}
assert data["volumes"] == []
apps/api/tests/integration/test_tool_types_api.py
+1 -1
View File
@@ -7,7 +7,7 @@ from fastapi.testclient import TestClient
from sqlalchemy import text
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker
from src.auth.jwt_service import mint_access_token
from src.auth.session import mint_access_token
from src.config import Settings, build_database_url
from src.models import Base
from src.models.tool_type import ToolType
apps/api/tests/integration/test_tool_types_api_extended.py
+415
View File
@@ -0,0 +1,415 @@
import uuid
from datetime import UTC, datetime, timedelta
import asyncio
import pytest
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
def test_create_tool_type_with_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 = {
"name": "custom-docker-tool",
"display_name": "Custom Docker Tool",
"description": "A custom tool with Dockerfile",
"definition_type": "dockerfile",
"dockerfile_template": "FROM python:3.11\\nRUN pip install flask\\nCMD ['python', 'app.py']",
"default_port": 5000,
"required_variables": ["REPO_PATH"],
}
response = client.post("/tool-types", json=payload)
assert response.status_code == 201
data = response.json()
assert data["name"] == "custom-docker-tool"
assert data["definition_type"] == "dockerfile"
assert data["dockerfile_template"] == "FROM python:3.11\\nRUN pip install flask\\nCMD ['python', 'app.py']"
assert data["compose_template"] is None
@pytest.mark.integration
def test_create_tool_type_with_readiness_probe() -> 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": "probed-tool",
"display_name": "Probed Tool",
"compose_template": "version: '3.8'\\nservices:\\n app:\\n image: nginx",
"default_port": 8080,
"required_variables": [],
"readiness_probe": {
"command": "curl -f http://localhost:8080/health",
"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,
"interval": 2,
},
}
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,
)
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()
assert data["definition_type"] == "dockerfile"
assert data["dockerfile_template"] == "FROM python:3.11"
assert data["readiness_probe"]["command"] == "python --version"
assert data["readiness_probe"]["timeout"] == 30
@pytest.mark.integration
def test_builtin_tool_types_have_definition_type_compose() -> None:
_prepare_test_db()
user_id = "11111111-1111-1111-1111-111111111111"
_insert_user(user_id)
# Load app triggers startup event which seeds built-in types
app = _load_app()
client = TestClient(app)
client.cookies.set("session", _mint_token(user_id))
response = client.get("/tool-types")
assert response.status_code == 200
data = response.json()
builtin_types = [t for t in data if t["is_builtin"]]
assert len(builtin_types) > 0
for tool_type in builtin_types:
assert tool_type["definition_type"] == "compose"
assert tool_type["compose_template"] is not None
assert tool_type["dockerfile_template"] is None
apps/api/tests/integration/test_users_api.py
+1 -1
View File
@@ -8,7 +8,7 @@ import pytest
from sqlalchemy import text
from sqlalchemy.ext.asyncio import create_async_engine
from src.auth.jwt_service import mint_access_token
from src.auth.session import mint_access_token
from src.config import Settings, build_database_url
from src.models import Base
from src.models.user import User
apps/api/tests/unit/test_docker_build.py
+151
View File
@@ -0,0 +1,151 @@
"""Unit tests for docker build service."""
import subprocess
import tempfile
from pathlib import Path
from unittest.mock import MagicMock, patch
import pytest
from src.services.docker_build import build_image
class TestBuildImage:
"""Tests for build_image function."""
@patch("subprocess.run")
def test_builds_image_successfully(self, mock_run) -> None:
mock_run.return_value = MagicMock(
returncode=0,
stdout="Successfully built abc123",
stderr="",
)
with tempfile.TemporaryDirectory() as tmpdir:
result = build_image(tmpdir, "FROM python:3.11", "test-image:latest")
assert result[0] == 0
assert "Successfully built" in result[1]
mock_run.assert_called_once()
call_args = mock_run.call_args
assert "test-image:latest" in call_args[0][0]
assert "build" in call_args[0][0]
@patch("subprocess.run")
def test_build_fails(self, mock_run) -> None:
mock_run.return_value = MagicMock(
returncode=1,
stdout="",
stderr="Error: failed to build",
)
with tempfile.TemporaryDirectory() as tmpdir:
result = build_image(tmpdir, "FROM invalid:image", "test-image:latest")
assert result[0] == 1
assert "failed to build" in result[2]
@patch("subprocess.run")
def test_build_with_tag(self, mock_run) -> None:
mock_run.return_value = MagicMock(
returncode=0,
stdout="",
stderr="",
)
with tempfile.TemporaryDirectory() as tmpdir:
build_image(tmpdir, "FROM python:3.11", "my-registry/tool:v1.0")
call_args = mock_run.call_args[0][0]
assert "my-registry/tool:v1.0" in call_args
@patch("subprocess.run")
def test_build_command_structure(self, mock_run) -> None:
mock_run.return_value = MagicMock(returncode=0, stdout="", stderr="")
with tempfile.TemporaryDirectory() as tmpdir:
build_image(tmpdir, "FROM python:3.11", "test:latest")
cmd = mock_run.call_args[0][0]
assert cmd[0] == "docker"
assert cmd[1] == "build"
assert "-t" in cmd
assert "test:latest" in cmd
assert tmpdir in cmd
@patch("subprocess.run")
def test_build_writes_dockerfile(self, mock_run) -> None:
mock_run.return_value = MagicMock(returncode=0, stdout="", stderr="")
with tempfile.TemporaryDirectory() as tmpdir:
dockerfile_content = "FROM python:3.11\\nRUN pip install flask"
build_image(tmpdir, dockerfile_content, "test:latest")
dockerfile_path = Path(tmpdir) / "Dockerfile"
assert dockerfile_path.exists()
assert dockerfile_path.read_text() == dockerfile_content
@patch("subprocess.run")
def test_build_writes_context_files(self, mock_run) -> None:
mock_run.return_value = MagicMock(returncode=0, stdout="", stderr="")
with tempfile.TemporaryDirectory() as tmpdir:
build_context = {
"requirements.txt": "flask==2.0\\nnumpy==1.21",
"app.py": "from flask import Flask\\napp = Flask(__name__)",
}
build_image(tmpdir, "FROM python:3.11", "test:latest", build_context)
req_path = Path(tmpdir) / "requirements.txt"
app_path = Path(tmpdir) / "app.py"
assert req_path.exists()
assert req_path.read_text() == "flask==2.0\\nnumpy==1.21"
assert app_path.exists()
assert app_path.read_text() == "from flask import Flask\\napp = Flask(__name__)"
@patch("subprocess.run")
def test_build_creates_nested_directories(self, mock_run) -> None:
mock_run.return_value = MagicMock(returncode=0, stdout="", stderr="")
with tempfile.TemporaryDirectory() as tmpdir:
build_context = {
"src/app.py": "print('hello')",
}
build_image(tmpdir, "FROM python:3.11", "test:latest", build_context)
app_path = Path(tmpdir) / "src" / "app.py"
assert app_path.exists()
@patch("subprocess.run")
def test_build_prevents_path_traversal(self, mock_run) -> None:
mock_run.return_value = MagicMock(returncode=0, stdout="", stderr="")
with tempfile.TemporaryDirectory() as tmpdir:
build_context = {
"../../../etc/passwd": "root:x:0:0",
}
with pytest.raises(ValueError, match="escapes instance directory"):
build_image(tmpdir, "FROM python:3.11", "test:latest", build_context)
mock_run.assert_not_called()
@patch("subprocess.run")
def test_build_timeout(self, mock_run) -> None:
mock_run.side_effect = subprocess.TimeoutExpired(cmd=["docker", "build"], timeout=300)
with tempfile.TemporaryDirectory() as tmpdir:
result = build_image(tmpdir, "FROM python:3.11", "test:latest")
assert result[0] == 1
assert "timed out" in result[2].lower()
@patch("subprocess.run")
def test_build_exception(self, mock_run) -> None:
mock_run.side_effect = OSError("Docker not available")
with tempfile.TemporaryDirectory() as tmpdir:
result = build_image(tmpdir, "FROM python:3.11", "test:latest")
assert result[0] == 1
assert "Docker not available" in result[2]
apps/api/tests/unit/test_readiness_probe.py
+217
View File
@@ -0,0 +1,217 @@
"""Unit tests for readiness probe service."""
import asyncio
from unittest.mock import MagicMock, patch
import pytest
from src.services.readiness_probe import execute_probe
class TestExecuteProbe:
"""Tests for execute_probe function."""
@patch("subprocess.run")
async def test_probe_succeeds_first_attempt(self, mock_run) -> None:
mock_run.return_value = MagicMock(
returncode=0,
stdout="healthy",
stderr="",
)
result, logs = await execute_probe("container-123", "curl -f http://localhost:8080")
assert result is True
assert any("Success" in log for log in logs)
mock_run.assert_called_once_with(
["docker", "exec", "container-123", "sh", "-c", "curl -f http://localhost:8080"],
capture_output=True,
text=True,
timeout=2,
)
@patch("subprocess.run")
async def test_probe_fails_then_succeeds(self, mock_run) -> None:
mock_run.side_effect = [
MagicMock(returncode=1, stdout="", stderr="Connection refused"),
MagicMock(returncode=1, stdout="", stderr="Connection refused"),
MagicMock(returncode=0, stdout="healthy", stderr=""),
]
result, logs = await execute_probe("container-123", "curl -f http://localhost:8080", timeout=10, interval=0.1)
assert result is True
assert mock_run.call_count == 3
assert any("Attempt 1: Failed" in log for log in logs)
assert any("Attempt 3: Success" in log for log in logs)
@patch("subprocess.run")
async def test_probe_times_out(self, mock_run) -> None:
mock_run.return_value = MagicMock(
returncode=1,
stdout="",
stderr="Connection refused",
)
result, logs = await execute_probe("container-123", "curl -f http://localhost:8080", timeout=0.5, interval=0.2)
assert result is False
assert any("timed out" in log.lower() for log in logs)
@patch("subprocess.run")
async def test_probe_command_not_found(self, mock_run) -> None:
mock_run.return_value = MagicMock(
returncode=127,
stdout="",
stderr="command not found",
)
result, logs = await execute_probe("container-123", "nonexistent-command", timeout=1, interval=0.3)
assert result is False
assert any("exit code 127" in log for log in logs)
@patch("subprocess.run")
async def test_probe_exception(self, mock_run) -> None:
mock_run.side_effect = OSError("Docker not available")
result, logs = await execute_probe("container-123", "curl http://localhost", timeout=1, interval=0.3)
assert result is False
assert any("Error" in log for log in logs)
@patch("subprocess.run")
async def test_probe_with_special_characters(self, mock_run) -> None:
mock_run.return_value = MagicMock(
returncode=0,
stdout="",
stderr="",
)
cmd = "bash -c 'echo \"hello world\" && exit 0'"
await execute_probe("container-123", cmd)
call_args = mock_run.call_args
assert cmd in call_args[0][0]
@patch("subprocess.run")
async def test_probe_captures_stdout(self, mock_run) -> None:
mock_run.return_value = MagicMock(
returncode=0,
stdout="Server is ready\\nVersion: 1.0",
stderr="",
)
result, logs = await execute_probe("container-123", "cat /app/status")
assert result is True
assert any("Server is ready" in log for log in logs)
class TestIntegrationScenarios:
"""Integration-style tests with realistic scenarios."""
@patch("subprocess.run")
async def test_web_server_probe(self, mock_run) -> None:
"""Test typical web server health check."""
mock_run.side_effect = [
MagicMock(returncode=1, stdout="", stderr=""),
MagicMock(returncode=1, stdout="", stderr=""),
MagicMock(returncode=1, stdout="", stderr=""),
MagicMock(returncode=0, stdout="OK", stderr=""),
]
result, logs = await execute_probe(
"web-container",
"curl -f http://localhost:8080/health",
timeout=10,
interval=0.2,
)
assert result is True
assert mock_run.call_count == 4
@patch("subprocess.run")
async def test_command_probe(self, mock_run) -> None:
"""Test command availability check."""
mock_run.return_value = MagicMock(
returncode=0,
stdout="opencode 1.0.0",
stderr="",
)
result, logs = await execute_probe(
"tool-container",
"which opencode && opencode --version",
timeout=30,
interval=2,
)
assert result is True
assert any("opencode 1.0.0" in log for log in logs)
@patch("subprocess.run")
async def test_database_probe(self, mock_run) -> None:
"""Test database readiness check."""
mock_run.side_effect = [
MagicMock(returncode=1, stdout="", stderr=""),
MagicMock(returncode=1, stdout="", stderr=""),
MagicMock(returncode=0, stdout="/var/run/postgresql:5432 - accepting connections", stderr=""),
]
result, logs = await execute_probe(
"db-container",
"pg_isready -U postgres",
timeout=10,
interval=0.3,
)
assert result is True
assert mock_run.call_count == 3
@patch("subprocess.run")
async def test_file_probe(self, mock_run) -> None:
"""Test file existence check."""
mock_run.return_value = MagicMock(returncode=0, stdout="", stderr="")
result, logs = await execute_probe(
"app-container",
"[ -f /app/ready ]",
timeout=10,
interval=1,
)
assert result is True
@patch("subprocess.run")
async def test_slow_starting_service(self, mock_run) -> None:
"""Test service that takes time to start."""
# Simulate 5 failures before success
side_effects = [MagicMock(returncode=1, stdout="", stderr="")] * 5
side_effects.append(MagicMock(returncode=0, stdout="Ready", stderr=""))
mock_run.side_effect = side_effects
result, logs = await execute_probe(
"slow-container",
"curl -f http://localhost:8080",
timeout=10,
interval=0.2,
)
assert result is True
assert mock_run.call_count == 6
assert any("Attempt 6: Success" in log for log in logs)
@patch("subprocess.run")
async def test_zero_timeout_immediate_return(self, mock_run) -> None:
mock_run.return_value = MagicMock(returncode=1, stdout="", stderr="")
result, logs = await execute_probe(
"container",
"test",
timeout=0,
interval=1,
)
assert result is False
assert any("timed out" in log.lower() for log in logs)