Paul Payne
00e8d58c6a
behavior (tool/daemon/frontend) is now a derived display label computed from a program's deployments — path->tool, static->frontend, process->daemon (behavior_for_runner). load_config populates it so every reader gets the live value; _spec_to_yaml_dict excludes it so it is never written to disk. _summary_from_service derives it from the runner instead of hardcoding daemon. Fixtures gain path/python deployments so tool/daemon behaviors derive.
196 lines
6.8 KiB
Python
196 lines
6.8 KiB
Python
"""Test fixtures for castle-api."""
|
|
|
|
from collections.abc import Generator
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
import yaml
|
|
from fastapi.testclient import TestClient
|
|
|
|
import castle_api.config as api_config
|
|
from castle_api.main import app
|
|
from castle_core.registry import (
|
|
Deployment,
|
|
NodeConfig,
|
|
NodeRegistry,
|
|
save_registry,
|
|
)
|
|
|
|
|
|
def _write_castle_config(root: Path, config: dict) -> None:
|
|
"""Scatter a nested castle config dict into the directory-per-resource layout."""
|
|
globals_data = {k: v for k, v in config.items() if k in ("gateway", "repo")}
|
|
(root / "castle.yaml").write_text(yaml.dump(globals_data, default_flow_style=False))
|
|
for section in ("programs", "services", "jobs"):
|
|
entries = config.get(section) or {}
|
|
if not entries:
|
|
continue
|
|
section_dir = root / section
|
|
section_dir.mkdir(parents=True, exist_ok=True)
|
|
for name, spec in entries.items():
|
|
(section_dir / f"{name}.yaml").write_text(
|
|
yaml.dump(spec, default_flow_style=False)
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def castle_root(tmp_path: Path) -> Generator[Path, None, None]:
|
|
"""Create a temporary castle root with directory-per-resource config."""
|
|
config = {
|
|
"gateway": {"port": 9000},
|
|
"programs": {
|
|
"test-tool": {
|
|
"description": "Test tool",
|
|
"source": "test-tool",
|
|
"system_dependencies": ["pandoc"],
|
|
},
|
|
"test-tool-2": {
|
|
"description": "Another test tool",
|
|
"source": "test-tool-2",
|
|
"version": "2.0.0",
|
|
},
|
|
"wired-in": {
|
|
"description": "Adopted repo, no stack",
|
|
"source": "wired-in",
|
|
"repo": "https://github.com/someone/wired-in.git",
|
|
"commands": {
|
|
"lint": [["make", "lint"]],
|
|
"test": [["make", "test"]],
|
|
"run": [["./bin/wired-in"]],
|
|
},
|
|
},
|
|
},
|
|
"services": {
|
|
# Path deployments — behavior "tool" derives from the `path` runner
|
|
# (behavior is derived from deployments, never stored).
|
|
"test-tool": {"program": "test-tool", "run": {"runner": "path"}},
|
|
"test-tool-2": {"program": "test-tool-2", "run": {"runner": "path"}},
|
|
"wired-in": {"program": "wired-in", "run": {"runner": "path"}},
|
|
"test-svc": {
|
|
"program": "test-svc-comp",
|
|
"description": "Test service",
|
|
"run": {
|
|
"runner": "python",
|
|
"program": "test-svc",
|
|
},
|
|
"expose": {
|
|
"http": {
|
|
"internal": {"port": 19000},
|
|
"health_path": "/health",
|
|
}
|
|
},
|
|
"proxy": True,
|
|
"manage": {"systemd": {}},
|
|
},
|
|
},
|
|
"jobs": {
|
|
"test-job": {
|
|
"description": "Test job",
|
|
"run": {
|
|
"runner": "command",
|
|
"argv": ["test-job"],
|
|
},
|
|
"schedule": "0 2 * * *",
|
|
},
|
|
},
|
|
}
|
|
_write_castle_config(tmp_path, config)
|
|
yield tmp_path
|
|
|
|
|
|
@pytest.fixture
|
|
def registry_path(tmp_path: Path, castle_root: Path) -> Generator[Path, None, None]:
|
|
"""Create a temporary registry.yaml and patch the module to use it."""
|
|
reg_path = tmp_path / "registry.yaml"
|
|
registry = NodeRegistry(
|
|
node=NodeConfig(
|
|
hostname="test-node",
|
|
castle_root=str(castle_root),
|
|
gateway_port=9000,
|
|
),
|
|
deployed={
|
|
"test-svc": Deployment(
|
|
runner="python",
|
|
run_cmd=["uv", "run", "test-svc"],
|
|
env={
|
|
"TEST_SVC_PORT": "19000",
|
|
"TEST_SVC_DATA_DIR": "/home/user/.castle/data/test-svc",
|
|
},
|
|
description="Test service",
|
|
behavior="daemon",
|
|
port=19000,
|
|
health_path="/health",
|
|
subdomain="test-svc",
|
|
managed=True,
|
|
),
|
|
},
|
|
)
|
|
save_registry(registry, reg_path)
|
|
|
|
# Patch the registry path and helper functions
|
|
import castle_core.registry as reg_mod
|
|
import castle_api.routes as routes_mod
|
|
import castle_api.services as services_mod
|
|
import castle_api.nodes as nodes_mod
|
|
import castle_api.stream as stream_mod
|
|
import castle_api.config_editor as config_editor_mod
|
|
|
|
original_path = reg_mod.REGISTRY_PATH
|
|
reg_mod.REGISTRY_PATH = reg_path
|
|
|
|
def _get_registry() -> NodeRegistry:
|
|
from castle_core.registry import load_registry
|
|
|
|
return load_registry(reg_path)
|
|
|
|
def _get_castle_root() -> Path | None:
|
|
return castle_root
|
|
|
|
# Save originals and patch everywhere these are imported
|
|
originals = {
|
|
"api_config.get_registry": api_config.get_registry,
|
|
"api_config.get_castle_root": api_config.get_castle_root,
|
|
"routes.get_registry": routes_mod.get_registry,
|
|
"routes.get_castle_root": routes_mod.get_castle_root,
|
|
"services.get_registry": services_mod.get_registry,
|
|
"services.get_castle_root": services_mod.get_castle_root,
|
|
"nodes.get_registry": nodes_mod.get_registry,
|
|
"stream.get_registry": stream_mod.get_registry,
|
|
"config_editor.get_registry": config_editor_mod.get_registry,
|
|
"config_editor.get_castle_root": config_editor_mod.get_castle_root,
|
|
}
|
|
|
|
for mod in [
|
|
api_config,
|
|
routes_mod,
|
|
services_mod,
|
|
nodes_mod,
|
|
stream_mod,
|
|
config_editor_mod,
|
|
]:
|
|
if hasattr(mod, "get_registry"):
|
|
mod.get_registry = _get_registry
|
|
if hasattr(mod, "get_castle_root"):
|
|
mod.get_castle_root = _get_castle_root
|
|
|
|
yield reg_path
|
|
|
|
reg_mod.REGISTRY_PATH = original_path
|
|
api_config.get_registry = originals["api_config.get_registry"]
|
|
api_config.get_castle_root = originals["api_config.get_castle_root"]
|
|
routes_mod.get_registry = originals["routes.get_registry"]
|
|
routes_mod.get_castle_root = originals["routes.get_castle_root"]
|
|
services_mod.get_registry = originals["services.get_registry"]
|
|
services_mod.get_castle_root = originals["services.get_castle_root"]
|
|
nodes_mod.get_registry = originals["nodes.get_registry"]
|
|
stream_mod.get_registry = originals["stream.get_registry"]
|
|
config_editor_mod.get_registry = originals["config_editor.get_registry"]
|
|
config_editor_mod.get_castle_root = originals["config_editor.get_castle_root"]
|
|
|
|
|
|
@pytest.fixture
|
|
def client(registry_path: Path) -> Generator[TestClient, None, None]:
|
|
"""Create a test client with temporary registry."""
|
|
with TestClient(app) as client:
|
|
yield client
|