Fusion
f33a563003
- Add ToolManifest Pydantic models with validators for ports, mounts, health checks, and traefik config - Implement in-memory ToolRegistry with YAML loading and built-in manifest scanning - Add FastAPI CRUD routes for listing, retrieving, and creating tool manifests - Include built-in manifests for runfusion and code-server - Harden web Dockerfile with unprivileged nginx and port 8080 - Add tool manifest specification documentation and architecture updates Fusion-Task-Id: FN-003
160 lines
4.8 KiB
Python
160 lines
4.8 KiB
Python
"""Tests for the in-memory tool manifest registry."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from pydantic import ValidationError
|
|
|
|
from app.tools.models import PortConfig, ToolManifest, TraefikConfig
|
|
from app.tools.registry import ToolRegistry
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Fixtures
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@pytest.fixture
|
|
def registry() -> ToolRegistry:
|
|
return ToolRegistry()
|
|
|
|
|
|
@pytest.fixture
|
|
def sample_manifest() -> ToolManifest:
|
|
return ToolManifest(
|
|
id="test-tool",
|
|
name="Test Tool",
|
|
image="test:latest",
|
|
ports=[PortConfig(container_port=8080, primary=True)],
|
|
traefik=TraefikConfig(enabled=False),
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Register / get round-trip
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_register_and_get(registry: ToolRegistry, sample_manifest: ToolManifest) -> None:
|
|
registry.register(sample_manifest)
|
|
retrieved = registry.get("test-tool")
|
|
assert retrieved is not None
|
|
assert retrieved.id == "test-tool"
|
|
|
|
|
|
def test_get_missing_returns_none(registry: ToolRegistry) -> None:
|
|
assert registry.get("missing") is None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# List
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_list_returns_all(registry: ToolRegistry) -> None:
|
|
m1 = ToolManifest(
|
|
id="tool-a",
|
|
name="Tool A",
|
|
image="a:latest",
|
|
ports=[PortConfig(container_port=8080, primary=True)],
|
|
traefik=TraefikConfig(enabled=False),
|
|
)
|
|
m2 = ToolManifest(
|
|
id="tool-b",
|
|
name="Tool B",
|
|
image="b:latest",
|
|
ports=[PortConfig(container_port=3000, primary=True)],
|
|
traefik=TraefikConfig(enabled=False),
|
|
)
|
|
registry.register(m1)
|
|
registry.register(m2)
|
|
assert len(registry.list()) == 2
|
|
ids = {m.id for m in registry.list()}
|
|
assert ids == {"tool-a", "tool-b"}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Overwrite behavior
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_register_overwrites_existing(
|
|
registry: ToolRegistry,
|
|
sample_manifest: ToolManifest,
|
|
) -> None:
|
|
registry.register(sample_manifest)
|
|
updated = ToolManifest(
|
|
id="test-tool",
|
|
name="Updated Tool",
|
|
image="updated:latest",
|
|
ports=[PortConfig(container_port=8080, primary=True)],
|
|
traefik=TraefikConfig(enabled=False),
|
|
)
|
|
registry.register(updated)
|
|
retrieved = registry.get("test-tool")
|
|
assert retrieved is not None
|
|
assert retrieved.name == "Updated Tool"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Remove
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_remove_returns_manifest(registry: ToolRegistry, sample_manifest: ToolManifest) -> None:
|
|
registry.register(sample_manifest)
|
|
removed = registry.remove("test-tool")
|
|
assert removed is not None
|
|
assert removed.id == "test-tool"
|
|
assert registry.get("test-tool") is None
|
|
|
|
|
|
def test_remove_missing_returns_none(registry: ToolRegistry) -> None:
|
|
assert registry.remove("missing") is None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Load file
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_load_valid_yaml_file(registry: ToolRegistry, tmp_path: Path) -> None:
|
|
yaml_path = tmp_path / "my-tool.yml"
|
|
yaml_path.write_text(
|
|
"""
|
|
id: my-tool
|
|
name: My Tool
|
|
image: my-tool:latest
|
|
ports:
|
|
- container_port: 8080
|
|
primary: true
|
|
traefik:
|
|
enabled: false
|
|
""",
|
|
encoding="utf-8",
|
|
)
|
|
manifest = registry.load_file(yaml_path)
|
|
assert manifest.id == "my-tool"
|
|
assert manifest.name == "My Tool"
|
|
assert manifest.ports[0].container_port == 8080
|
|
|
|
|
|
def test_load_invalid_yaml_raises(registry: ToolRegistry, tmp_path: Path) -> None:
|
|
yaml_path = tmp_path / "bad-tool.yml"
|
|
yaml_path.write_text(
|
|
"""
|
|
id: BAD ID
|
|
name: Bad Tool
|
|
image: bad:latest
|
|
""",
|
|
encoding="utf-8",
|
|
)
|
|
with pytest.raises(ValidationError):
|
|
registry.load_file(yaml_path)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Load builtin manifests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_load_builtin_manifests(registry: ToolRegistry) -> None:
|
|
registry.load_builtin_manifests()
|
|
# Built-in manifests from Step 4 may not exist yet in isolation,
|
|
# but the method should not raise regardless of directory contents.
|
|
assert isinstance(registry.list(), list)
|