docs: sync tool-types-definition specs and mark tasks complete
- Mark manual testing tasks as complete in tool-types-definition - Sync tool-types-definition spec to main specs directory
This commit is contained in:
Fusion
@@ -0,0 +1,494 @@
|
||||
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.jwt_service import mint_access_token
|
||||
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, git_repositories, ssh_keys, projects, 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
|
||||
|
||||
# Dispose old engine connections before reload to prevent pool exhaustion
|
||||
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 mint_access_token(
|
||||
settings=settings,
|
||||
subject=user_id,
|
||||
email="test@headquarter.local",
|
||||
name="Test User",
|
||||
expires_at=datetime.now(UTC) + timedelta(minutes=15),
|
||||
)
|
||||
|
||||
|
||||
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,
|
||||
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,
|
||||
required_variables=["REPO_PATH", "TOOL_NAME"],
|
||||
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_list_tool_types_requires_authentication() -> None:
|
||||
_prepare_test_db()
|
||||
app = _load_app()
|
||||
client = TestClient(app)
|
||||
|
||||
response = client.get("/tool-types")
|
||||
|
||||
assert response.status_code == 401
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
def test_list_tool_types_returns_all_types() -> None:
|
||||
_prepare_test_db()
|
||||
user_id = "11111111-1111-1111-1111-111111111111"
|
||||
_insert_user(user_id)
|
||||
_insert_tool_type(
|
||||
"aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa",
|
||||
"custom-tool",
|
||||
"Custom Tool",
|
||||
"version: '3.8'\nservices:\n app:\n image: custom",
|
||||
created_by_id=user_id,
|
||||
)
|
||||
|
||||
app = _load_app()
|
||||
client = TestClient(app)
|
||||
client.cookies.set("access_token", _mint_token(user_id))
|
||||
|
||||
response = client.get("/tool-types")
|
||||
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert len(data) >= 1
|
||||
custom_tool = next((t for t in data if t["name"] == "custom-tool"), None)
|
||||
assert custom_tool is not None
|
||||
assert custom_tool["display_name"] == "Custom Tool"
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
def test_get_tool_type_by_id() -> 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",
|
||||
"version: '3.8'\nservices:\n app:\n image: custom",
|
||||
created_by_id=user_id,
|
||||
)
|
||||
|
||||
app = _load_app()
|
||||
client = TestClient(app)
|
||||
client.cookies.set("access_token", _mint_token(user_id))
|
||||
|
||||
response = client.get(f"/tool-types/{tool_type_id}")
|
||||
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["id"] == tool_type_id
|
||||
assert data["name"] == "custom-tool"
|
||||
assert data["display_name"] == "Custom Tool"
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
def test_get_tool_type_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("access_token", _mint_token(user_id))
|
||||
|
||||
response = client.get("/tool-types/aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa")
|
||||
|
||||
assert response.status_code == 404
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
def test_create_tool_type_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("access_token", _mint_token(user_id))
|
||||
|
||||
payload = {
|
||||
"name": "my-custom-tool",
|
||||
"display_name": "My Custom Tool",
|
||||
"description": "A custom development tool",
|
||||
"compose_template": "version: '3.8'\nservices:\n app:\n image: custom:latest",
|
||||
"required_variables": ["REPO_PATH"],
|
||||
}
|
||||
response = client.post("/tool-types", json=payload)
|
||||
|
||||
assert response.status_code == 201
|
||||
data = response.json()
|
||||
assert data["name"] == "my-custom-tool"
|
||||
assert data["display_name"] == "My Custom Tool"
|
||||
assert data["description"] == "A custom development tool"
|
||||
assert data["is_builtin"] == False
|
||||
assert data["created_by_id"] == user_id
|
||||
assert "id" in data
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
def test_create_tool_type_duplicate_name() -> None:
|
||||
_prepare_test_db()
|
||||
user_id = "11111111-1111-1111-1111-111111111111"
|
||||
_insert_user(user_id)
|
||||
_insert_tool_type(
|
||||
"aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa",
|
||||
"existing-tool",
|
||||
"Existing Tool",
|
||||
"version: '3.8'\nservices:\n app:\n image: existing",
|
||||
created_by_id=user_id,
|
||||
)
|
||||
|
||||
app = _load_app()
|
||||
client = TestClient(app)
|
||||
client.cookies.set("access_token", _mint_token(user_id))
|
||||
|
||||
payload = {
|
||||
"name": "existing-tool",
|
||||
"display_name": "Existing Tool",
|
||||
"compose_template": "version: '3.8'\nservices:\n app:\n image: custom",
|
||||
"required_variables": [],
|
||||
}
|
||||
response = client.post("/tool-types", json=payload)
|
||||
|
||||
assert response.status_code == 409
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
def test_create_tool_type_invalid_yaml() -> None:
|
||||
_prepare_test_db()
|
||||
user_id = "11111111-1111-1111-1111-111111111111"
|
||||
_insert_user(user_id)
|
||||
|
||||
app = _load_app()
|
||||
client = TestClient(app)
|
||||
client.cookies.set("access_token", _mint_token(user_id))
|
||||
|
||||
payload = {
|
||||
"name": "bad-tool",
|
||||
"display_name": "Bad Tool",
|
||||
"compose_template": "this is not: valid: yaml: [",
|
||||
"required_variables": [],
|
||||
}
|
||||
response = client.post("/tool-types", json=payload)
|
||||
|
||||
assert response.status_code == 422
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
def test_create_tool_type_missing_services() -> None:
|
||||
_prepare_test_db()
|
||||
user_id = "11111111-1111-1111-1111-111111111111"
|
||||
_insert_user(user_id)
|
||||
|
||||
app = _load_app()
|
||||
client = TestClient(app)
|
||||
client.cookies.set("access_token", _mint_token(user_id))
|
||||
|
||||
payload = {
|
||||
"name": "bad-tool",
|
||||
"display_name": "Bad Tool",
|
||||
"compose_template": "version: '3.8'\ninvalid_key: value",
|
||||
"required_variables": [],
|
||||
}
|
||||
response = client.post("/tool-types", json=payload)
|
||||
|
||||
assert response.status_code == 422
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
def test_create_tool_type_missing_required_variable() -> None:
|
||||
_prepare_test_db()
|
||||
user_id = "11111111-1111-1111-1111-111111111111"
|
||||
_insert_user(user_id)
|
||||
|
||||
app = _load_app()
|
||||
client = TestClient(app)
|
||||
client.cookies.set("access_token", _mint_token(user_id))
|
||||
|
||||
payload = {
|
||||
"name": "bad-tool",
|
||||
"display_name": "Bad Tool",
|
||||
"compose_template": "version: '3.8'\nservices:\n app:\n image: custom",
|
||||
"required_variables": ["MISSING_VAR"],
|
||||
}
|
||||
response = client.post("/tool-types", json=payload)
|
||||
|
||||
assert response.status_code == 422
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
def test_update_tool_type_successfully() -> 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",
|
||||
"version: '3.8'\nservices:\n app:\n image: custom",
|
||||
created_by_id=user_id,
|
||||
)
|
||||
|
||||
app = _load_app()
|
||||
client = TestClient(app)
|
||||
client.cookies.set("access_token", _mint_token(user_id))
|
||||
|
||||
payload = {
|
||||
"display_name": "Updated Custom Tool",
|
||||
"description": "Updated description",
|
||||
}
|
||||
response = client.put(f"/tool-types/{tool_type_id}", json=payload)
|
||||
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["display_name"] == "Updated Custom Tool"
|
||||
assert data["description"] == "Updated description"
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
def test_update_tool_type_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("access_token", _mint_token(user_id))
|
||||
|
||||
payload = {"display_name": "Updated"}
|
||||
response = client.put("/tool-types/aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa", json=payload)
|
||||
|
||||
assert response.status_code == 404
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
def test_update_builtin_tool_type_fails() -> 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,
|
||||
"builtin-tool",
|
||||
"Built-in Tool",
|
||||
"version: '3.8'\nservices:\n app:\n image: builtin",
|
||||
is_builtin=True,
|
||||
)
|
||||
|
||||
app = _load_app()
|
||||
client = TestClient(app)
|
||||
client.cookies.set("access_token", _mint_token(user_id))
|
||||
|
||||
payload = {"display_name": "Updated"}
|
||||
response = client.put(f"/tool-types/{tool_type_id}", json=payload)
|
||||
|
||||
assert response.status_code == 403
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
def test_delete_tool_type_successfully() -> 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,
|
||||
"deletable-tool",
|
||||
"Deletable Tool",
|
||||
"version: '3.8'\nservices:\n app:\n image: custom",
|
||||
created_by_id=user_id,
|
||||
)
|
||||
|
||||
app = _load_app()
|
||||
client = TestClient(app)
|
||||
client.cookies.set("access_token", _mint_token(user_id))
|
||||
|
||||
response = client.delete(f"/tool-types/{tool_type_id}")
|
||||
|
||||
assert response.status_code == 204
|
||||
|
||||
# Verify it's gone
|
||||
get_response = client.get(f"/tool-types/{tool_type_id}")
|
||||
assert get_response.status_code == 404
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
def test_delete_tool_type_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("access_token", _mint_token(user_id))
|
||||
|
||||
response = client.delete("/tool-types/aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa")
|
||||
|
||||
assert response.status_code == 404
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
def test_delete_builtin_tool_type_fails() -> 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,
|
||||
"builtin-tool",
|
||||
"Built-in Tool",
|
||||
"version: '3.8'\nservices:\n app:\n image: builtin",
|
||||
is_builtin=True,
|
||||
)
|
||||
|
||||
app = _load_app()
|
||||
client = TestClient(app)
|
||||
client.cookies.set("access_token", _mint_token(user_id))
|
||||
|
||||
response = client.delete(f"/tool-types/{tool_type_id}")
|
||||
|
||||
assert response.status_code == 403
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
def test_builtin_tool_types_seeded_on_startup() -> 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("access_token", _mint_token(user_id))
|
||||
|
||||
response = client.get("/tool-types")
|
||||
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
|
||||
# Check that built-in types exist
|
||||
builtin_names = [t["name"] for t in data if t["is_builtin"]]
|
||||
assert "code-server" in builtin_names
|
||||
assert "jupyter-notebook" in builtin_names
|
||||
|
||||
# Verify built-in types have correct attributes
|
||||
code_server = next((t for t in data if t["name"] == "code-server"), None)
|
||||
assert code_server is not None
|
||||
assert code_server["display_name"] == "VS Code Server"
|
||||
assert "services" in code_server["compose_template"]
|
||||
assert code_server["required_variables"] == ["REPO_PATH", "TOOL_NAME"]
|
||||
Reference in New Issue
Block a user