wild-cloud/wild-pc
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor: Decouple roles.

Browse Source
This commit is contained in:
Paul Payne
2026-02-23 01:49:24 -08:00
parent 72d35f2641
commit eeaa5045d0
55 changed files with 2144 additions and 1276 deletions
Download Patch File Download Diff File Expand all files Collapse all files

107
core/src/castle_core/config.py
View File

@@ -9,7 +9,7 @@ from pathlib import Path
import yaml
from castle_core.manifest import ComponentManifest, Role
from castle_core.manifest import ComponentSpec, JobSpec, ServiceSpec
def find_castle_root() -> Path:
@@ -47,36 +47,30 @@ class CastleConfig:
root: Path
gateway: GatewayConfig
components: dict[str, ComponentManifest]
components: dict[str, ComponentSpec]
services: dict[str, ServiceSpec]
jobs: dict[str, JobSpec]
@property
def services(self) -> dict[str, ComponentManifest]:
"""Return components with the SERVICE role."""
return {k: v for k, v in self.components.items() if Role.SERVICE in v.roles}
@property
def tools(self) -> dict[str, ComponentManifest]:
"""Return components with the TOOL role."""
return {k: v for k, v in self.components.items() if Role.TOOL in v.roles}
@property
def workers(self) -> dict[str, ComponentManifest]:
"""Return components with the WORKER role."""
return {k: v for k, v in self.components.items() if Role.WORKER in v.roles}
@property
def managed(self) -> dict[str, ComponentManifest]:
"""Return components managed by systemd."""
def tools(self) -> dict[str, ComponentSpec]:
"""Return components that are tools (have install.path or tool spec)."""
return {
k: v
for k, v in self.components.items()
if v.manage and v.manage.systemd and v.manage.systemd.enable
if (v.install and v.install.path) or v.tool
}
@property
def frontends(self) -> dict[str, ComponentSpec]:
"""Return components that are frontends (have build outputs)."""
return {
k: v
for k, v in self.components.items()
if v.build and (v.build.outputs or v.build.commands)
}
def resolve_env_vars(
env: dict[str, str], manifest: ComponentManifest
) -> dict[str, str]:
def resolve_env_vars(env: dict[str, str]) -> dict[str, str]:
"""Resolve ${secret:NAME} references in env values."""
resolved = {}
for key, value in env.items():
@@ -100,11 +94,25 @@ def _read_secret(name: str) -> str:
return f"<MISSING_SECRET:{name}>"
def _parse_component(name: str, data: dict) -> ComponentManifest:
"""Parse a components: entry directly into a ComponentManifest."""
def _parse_component(name: str, data: dict) -> ComponentSpec:
"""Parse a components: entry into a ComponentSpec."""
data_copy = dict(data)
data_copy["id"] = name
return ComponentManifest.model_validate(data_copy)
return ComponentSpec.model_validate(data_copy)
def _parse_service(name: str, data: dict) -> ServiceSpec:
"""Parse a services: entry into a ServiceSpec."""
data_copy = dict(data)
data_copy["id"] = name
return ServiceSpec.model_validate(data_copy)
def _parse_job(name: str, data: dict) -> JobSpec:
"""Parse a jobs: entry into a JobSpec."""
data_copy = dict(data)
data_copy["id"] = name
return JobSpec.model_validate(data_copy)
def load_config(root: Path | None = None) -> CastleConfig:
@@ -122,11 +130,25 @@ def load_config(root: Path | None = None) -> CastleConfig:
gateway_data = data.get("gateway", {})
gateway = GatewayConfig(port=gateway_data.get("port", 9000))
components: dict[str, ComponentManifest] = {}
components: dict[str, ComponentSpec] = {}
for name, comp_data in data.get("components", {}).items():
components[name] = _parse_component(name, comp_data)
return CastleConfig(root=root, gateway=gateway, components=components)
services: dict[str, ServiceSpec] = {}
for name, svc_data in data.get("services", {}).items():
services[name] = _parse_service(name, svc_data)
jobs: dict[str, JobSpec] = {}
for name, job_data in data.get("jobs", {}).items():
jobs[name] = _parse_job(name, job_data)
return CastleConfig(
root=root,
gateway=gateway,
components=components,
services=services,
jobs=jobs,
)
def _clean_for_yaml(data: object, preserve_keys: set[str] | None = None) -> object:
@@ -165,11 +187,12 @@ _STRUCTURAL_KEYS = {
}
def _manifest_to_yaml_dict(manifest: ComponentManifest) -> dict:
"""Serialize a manifest to a YAML-friendly dict, preserving structural presence."""
full = manifest.model_dump(mode="json", exclude_none=True, exclude={"id", "roles"})
minimal = manifest.model_dump(
mode="json", exclude_none=True, exclude={"id", "roles"}, exclude_defaults=True
def _spec_to_yaml_dict(spec: ComponentSpec | ServiceSpec | JobSpec) -> dict:
"""Serialize a spec to a YAML-friendly dict, preserving structural presence."""
exclude_fields = {"id"}
full = spec.model_dump(mode="json", exclude_none=True, exclude=exclude_fields)
minimal = spec.model_dump(
mode="json", exclude_none=True, exclude=exclude_fields, exclude_defaults=True
)
def merge(full_val: object, min_val: object | None, key: str = "") -> object:
@@ -203,10 +226,22 @@ def _manifest_to_yaml_dict(manifest: ComponentManifest) -> dict:
def save_config(config: CastleConfig) -> None:
"""Save castle configuration to castle.yaml."""
data: dict = {"gateway": {"port": config.gateway.port}, "components": {}}
data: dict = {"gateway": {"port": config.gateway.port}}
for name, manifest in config.components.items():
data["components"][name] = _manifest_to_yaml_dict(manifest)
if config.components:
data["components"] = {}
for name, spec in config.components.items():
data["components"][name] = _spec_to_yaml_dict(spec)
if config.services:
data["services"] = {}
for name, spec in config.services.items():
data["services"][name] = _spec_to_yaml_dict(spec)
if config.jobs:
data["jobs"] = {}
for name, spec in config.jobs.items():
data["jobs"][name] = _spec_to_yaml_dict(spec)
config_path = config.root / "castle.yaml"
with open(config_path, "w") as f:

2
core/src/castle_core/generators/__init__.py
View File

@@ -8,7 +8,6 @@ from castle_core.generators.systemd import (
cron_to_oncalendar,
generate_timer,
generate_unit_from_deployed,
get_schedule_trigger,
timer_name,
unit_name,
)
@@ -19,7 +18,6 @@ __all__ = [
"generate_caddyfile_from_registry",
"generate_timer",
"generate_unit_from_deployed",
"get_schedule_trigger",
"timer_name",
"unit_name",
]

28
core/src/castle_core/generators/systemd.py
View File

@@ -5,7 +5,7 @@ from __future__ import annotations
import shutil
from pathlib import Path
from castle_core.manifest import ComponentManifest, RestartPolicy, SystemdSpec
from castle_core.manifest import RestartPolicy, SystemdSpec
from castle_core.registry import DeployedComponent
SYSTEMD_USER_DIR = Path.home() / ".config" / "systemd" / "user"
@@ -22,14 +22,6 @@ def timer_name(service_name: str) -> str:
return f"{UNIT_PREFIX}{service_name}.timer"
def get_schedule_trigger(manifest: ComponentManifest) -> object | None:
"""Return the schedule trigger if one exists, else None."""
for t in manifest.triggers:
if getattr(t, "type", None) == "schedule":
return t
return None
def cron_to_oncalendar(cron: str) -> str:
"""Best-effort conversion of cron expression to systemd OnCalendar.
@@ -142,17 +134,17 @@ WantedBy={wanted_by}
return unit
def generate_timer(name: str, manifest: ComponentManifest) -> str | None:
"""Generate a systemd timer unit if the component has a schedule trigger."""
trigger = get_schedule_trigger(manifest)
if trigger is None:
return None
description = manifest.description or name
def generate_timer(
name: str,
schedule: str,
description: str | None = None,
) -> str:
"""Generate a systemd timer unit from a cron schedule string."""
description = description or name
# Try to convert cron to OnCalendar, fall back to OnUnitActiveSec
on_calendar = cron_to_oncalendar(trigger.cron)
interval_sec = cron_to_interval_sec(trigger.cron)
on_calendar = cron_to_oncalendar(schedule)
interval_sec = cron_to_interval_sec(schedule)
timer_lines = ""
if on_calendar:

152
core/src/castle_core/manifest.py
View File

@@ -1,11 +1,11 @@
"""Castle component manifest — Pydantic models for the component registry."""
"""Castle manifest models — component specs, service specs, job specs."""
from __future__ import annotations
from enum import Enum
from typing import Annotated, Literal, Union
from pydantic import BaseModel, ConfigDict, Field, computed_field, model_validator
from pydantic import BaseModel, ConfigDict, Field, model_validator
EnvMap = dict[str, str]
@@ -22,16 +22,6 @@ class TLSMode(str, Enum):
LETSENCRYPT = "letsencrypt"
class Role(str, Enum):
TOOL = "tool"
SERVICE = "service"
WORKER = "worker"
FRONTEND = "frontend"
JOB = "job"
REMOTE = "remote"
CONTAINERIZED = "containerized"
# ---------------------
# Run specs (discriminated union)
# ---------------------
@@ -82,36 +72,6 @@ RunSpec = Annotated[
]
# ---------------------
# Triggers
# ---------------------
class TriggerManual(BaseModel):
type: Literal["manual"] = "manual"
class TriggerSchedule(BaseModel):
type: Literal["schedule"] = "schedule"
cron: str
timezone: str = "America/Los_Angeles"
class TriggerEvent(BaseModel):
type: Literal["event"] = "event"
source: str
topic: str | None = None
queue: str | None = None
class TriggerRequest(BaseModel):
type: Literal["request"] = "request"
protocol: Literal["http", "https", "grpc"] = "http"
TriggerSpec = Union[TriggerManual, TriggerSchedule, TriggerEvent, TriggerRequest]
# ---------------------
# Systemd management
# ---------------------
@@ -166,7 +126,6 @@ class ToolSpec(BaseModel):
model_config = ConfigDict(extra="ignore")
version: str = "1.0.0"
source: str | None = None
system_dependencies: list[str] = Field(default_factory=list)
@@ -238,90 +197,77 @@ class DefaultsSpec(BaseModel):
# ---------------------
# Component manifest
# Component spec — software identity
# ---------------------
class ComponentManifest(BaseModel):
class ComponentSpec(BaseModel):
"""Software catalog entry — what exists."""
id: str = ""
name: str | None = None
description: str | None = None
source: str | None = None
run: RunSpec | None = None
triggers: list[TriggerSpec] = Field(default_factory=list)
manage: ManageSpec | None = None
install: InstallSpec | None = None
tool: ToolSpec | None = None
expose: ExposeSpec | None = None
proxy: ProxySpec | None = None
build: BuildSpec | None = None
defaults: DefaultsSpec | None = None
provides: list[Capability] = Field(default_factory=list)
consumes: list[Capability] = Field(default_factory=list)
tags: list[str] = Field(default_factory=list)
@computed_field # type: ignore[prop-decorator]
@property
def roles(self) -> list[Role]:
roles: set[Role] = set()
if self.run:
if self.run.runner == "remote":
roles.add(Role.REMOTE)
if self.run.runner == "container":
roles.add(Role.CONTAINERIZED)
if self.install and self.install.path and self.install.path.enable:
roles.add(Role.TOOL)
if self.tool:
roles.add(Role.TOOL)
if self.expose and self.expose.http:
roles.add(Role.SERVICE)
if (
self.manage
and self.manage.systemd
and self.manage.systemd.enable
and not (self.expose and self.expose.http)
):
roles.add(Role.WORKER)
if self.build and (self.build.outputs or self.build.commands):
roles.add(Role.FRONTEND)
if any(getattr(t, "type", None) == "schedule" for t in self.triggers):
roles.add(Role.JOB)
if not roles:
roles.add(Role.TOOL)
return sorted(roles, key=lambda r: r.value)
@property
def source_dir(self) -> str | None:
"""Best-effort relative directory for this component's source.
Resolution order: source → tool.source (strip trailing /).
Returns None if no directory can be determined.
"""
"""Relative directory for this component's source, or None."""
if self.source:
return self.source.rstrip("/")
if self.tool and self.tool.source:
return self.tool.source.rstrip("/")
return None
# ---------------------
# Service spec — long-running daemon
# ---------------------
class ServiceSpec(BaseModel):
"""Long-running daemon deployment config."""
id: str = ""
component: str | None = None
description: str | None = None
run: RunSpec
expose: ExposeSpec | None = None
proxy: ProxySpec | None = None
manage: ManageSpec | None = None
defaults: DefaultsSpec | None = None
@model_validator(mode="after")
def _basic_consistency(self) -> ComponentManifest:
def _validate_consistency(self) -> ServiceSpec:
if self.manage and self.manage.systemd and self.manage.systemd.enable:
if self.run and self.run.runner == "remote":
if self.run.runner == "remote":
raise ValueError("manage.systemd cannot be enabled for runner=remote.")
return self
# ---------------------
# Job spec — scheduled task
# ---------------------
class JobSpec(BaseModel):
"""Scheduled task that runs periodically and exits."""
id: str = ""
component: str | None = None
description: str | None = None
run: RunSpec
schedule: str
timezone: str = "America/Los_Angeles"
manage: ManageSpec | None = None
defaults: DefaultsSpec | None = None

7
core/src/castle_core/registry.py
View File

@@ -35,7 +35,7 @@ class DeployedComponent:
run_cmd: list[str]
env: dict[str, str] = field(default_factory=dict)
description: str | None = None
roles: list[str] = field(default_factory=list)
category: str = "service"
port: int | None = None
health_path: str | None = None
proxy_path: str | None = None
@@ -82,7 +82,7 @@ def load_registry(path: Path | None = None) -> NodeRegistry:
run_cmd=comp_data.get("run_cmd", []),
env=comp_data.get("env", {}),
description=comp_data.get("description"),
roles=comp_data.get("roles", []),
category=comp_data.get("category", "service"),
port=comp_data.get("port"),
health_path=comp_data.get("health_path"),
proxy_path=comp_data.get("proxy_path"),
@@ -120,8 +120,7 @@ def save_registry(registry: NodeRegistry, path: Path | None = None) -> None:
entry["env"] = comp.env
if comp.description:
entry["description"] = comp.description
if comp.roles:
entry["roles"] = comp.roles
entry["category"] = comp.category
if comp.port is not None:
entry["port"] = comp.port
if comp.health_path:

22
core/tests/conftest.py
View File

@@ -16,9 +16,17 @@ def castle_root(tmp_path: Path) -> Generator[Path, None, None]:
config = {
"gateway": {"port": 18000},
"components": {
"test-tool": {
"description": "Test tool",
"install": {
"path": {"alias": "test-tool"},
},
},
},
"services": {
"test-svc": {
"component": "test-svc-comp",
"description": "Test service",
"source": "test-svc",
"run": {
"runner": "python",
"tool": "test-svc",
@@ -39,11 +47,15 @@ def castle_root(tmp_path: Path) -> Generator[Path, None, None]:
"systemd": {},
},
},
"test-tool": {
"description": "Test tool",
"install": {
"path": {"alias": "test-tool"},
},
"jobs": {
"test-job": {
"description": "Test job",
"run": {
"runner": "command",
"argv": ["test-job"],
},
"schedule": "0 2 * * *",
},
},
}

80
core/tests/test_config.py
View File

@@ -11,82 +11,64 @@ from castle_core.config import (
resolve_env_vars,
save_config,
)
from castle_core.manifest import ComponentManifest, Role
from castle_core.manifest import ComponentSpec, JobSpec, ServiceSpec
class TestLoadConfig:
"""Tests for loading castle.yaml."""
def test_load_basic(self, castle_root: Path) -> None:
"""Load a castle.yaml."""
"""Load a castle.yaml with three sections."""
config = load_config(castle_root)
assert isinstance(config, CastleConfig)
assert config.gateway.port == 18000
assert "test-svc" in config.components
assert "test-tool" in config.components
assert "test-svc" in config.services
assert "test-job" in config.jobs
def test_load_produces_manifests(self, castle_root: Path) -> None:
"""Components are ComponentManifest objects."""
def test_load_produces_typed_specs(self, castle_root: Path) -> None:
"""Each section produces the correct spec type."""
config = load_config(castle_root)
assert isinstance(config.components["test-svc"], ComponentManifest)
assert isinstance(config.components["test-tool"], ComponentManifest)
def test_service_roles(self, castle_root: Path) -> None:
"""Service with expose.http gets SERVICE role."""
config = load_config(castle_root)
svc = config.components["test-svc"]
assert Role.SERVICE in svc.roles
def test_tool_roles(self, castle_root: Path) -> None:
"""Tool with install.path gets TOOL role."""
config = load_config(castle_root)
tool = config.components["test-tool"]
assert Role.TOOL in tool.roles
assert isinstance(config.components["test-tool"], ComponentSpec)
assert isinstance(config.services["test-svc"], ServiceSpec)
assert isinstance(config.jobs["test-job"], JobSpec)
def test_service_expose(self, castle_root: Path) -> None:
"""Service has correct expose spec."""
config = load_config(castle_root)
svc = config.components["test-svc"]
svc = config.services["test-svc"]
assert svc.expose.http.internal.port == 19000
assert svc.expose.http.health_path == "/health"
def test_service_proxy(self, castle_root: Path) -> None:
"""Service has correct proxy spec."""
config = load_config(castle_root)
svc = config.components["test-svc"]
svc = config.services["test-svc"]
assert svc.proxy.caddy.path_prefix == "/test-svc"
def test_service_run_spec(self, castle_root: Path) -> None:
"""Service has correct RunSpec."""
config = load_config(castle_root)
svc = config.components["test-svc"]
svc = config.services["test-svc"]
assert svc.run.runner == "python"
assert svc.run.tool == "test-svc"
assert svc.source == "test-svc"
def test_tool_no_run(self, castle_root: Path) -> None:
"""Tool without run block has no run spec."""
def test_service_component_ref(self, castle_root: Path) -> None:
"""Service references a component."""
config = load_config(castle_root)
tool = config.components["test-tool"]
assert tool.run is None
svc = config.services["test-svc"]
assert svc.component == "test-svc-comp"
def test_services_property(self, castle_root: Path) -> None:
"""Services property filters to SERVICE role."""
def test_job_schedule(self, castle_root: Path) -> None:
"""Job has correct schedule."""
config = load_config(castle_root)
assert "test-svc" in config.services
assert "test-tool" not in config.services
job = config.jobs["test-job"]
assert job.schedule == "0 2 * * *"
def test_tools_property(self, castle_root: Path) -> None:
"""Tools property filters to TOOL role."""
"""Tools property filters to components with install.path or tool."""
config = load_config(castle_root)
assert "test-tool" in config.tools
assert "test-svc" not in config.tools
def test_managed_property(self, castle_root: Path) -> None:
"""Managed property returns systemd-managed components."""
config = load_config(castle_root)
assert "test-svc" in config.managed
assert "test-tool" not in config.managed
def test_missing_config_raises(self, tmp_path: Path) -> None:
"""Missing castle.yaml raises FileNotFoundError."""
@@ -105,11 +87,13 @@ class TestSaveConfig:
assert config2.gateway.port == config.gateway.port
assert set(config2.components.keys()) == set(config.components.keys())
assert set(config2.services.keys()) == set(config.services.keys())
assert set(config2.jobs.keys()) == set(config.jobs.keys())
def test_save_adds_component(self, castle_root: Path) -> None:
"""Adding a component and saving persists it."""
config = load_config(castle_root)
config.components["new-lib"] = ComponentManifest(
config.components["new-lib"] = ComponentSpec(
id="new-lib", description="A new library"
)
save_config(config)
@@ -123,7 +107,9 @@ class TestSaveConfig:
config = load_config(castle_root)
save_config(config)
config2 = load_config(castle_root)
assert "test-svc" in config2.managed
svc = config2.services["test-svc"]
assert svc.manage is not None
assert svc.manage.systemd is not None
class TestResolveEnvVars:
@@ -131,16 +117,14 @@ class TestResolveEnvVars:
def test_no_vars(self) -> None:
"""Plain values pass through unchanged."""
manifest = ComponentManifest(id="test")
env = {"MY_VAR": "plain_value"}
resolved = resolve_env_vars(env, manifest)
resolved = resolve_env_vars(env)
assert resolved["MY_VAR"] == "plain_value"
def test_unrecognized_vars_preserved(self) -> None:
"""Non-secret ${} references pass through unchanged."""
manifest = ComponentManifest(id="test")
env = {"MY_VAR": "${unknown_var}"}
resolved = resolve_env_vars(env, manifest)
resolved = resolve_env_vars(env)
assert resolved["MY_VAR"] == "${unknown_var}"
def test_resolve_secret(
@@ -152,9 +136,8 @@ class TestResolveEnvVars:
(secrets_dir / "API_KEY").write_text("my-secret-key\n")
monkeypatch.setattr("castle_core.config.SECRETS_DIR", secrets_dir)
manifest = ComponentManifest(id="test")
env = {"API_KEY": "${secret:API_KEY}"}
resolved = resolve_env_vars(env, manifest)
resolved = resolve_env_vars(env)
assert resolved["API_KEY"] == "my-secret-key"
def test_resolve_missing_secret(
@@ -165,7 +148,6 @@ class TestResolveEnvVars:
secrets_dir.mkdir()
monkeypatch.setattr("castle_core.config.SECRETS_DIR", secrets_dir)
manifest = ComponentManifest(id="test")
env = {"API_KEY": "${secret:NONEXISTENT}"}
resolved = resolve_env_vars(env, manifest)
resolved = resolve_env_vars(env)
assert resolved["API_KEY"] == "<MISSING_SECRET:NONEXISTENT>"

242
core/tests/test_manifest.py
View File

@@ -1,4 +1,4 @@
"""Tests for castle manifest — role derivation, validation."""
"""Tests for castle manifest models."""
from __future__ import annotations
@@ -6,171 +6,183 @@ import pytest
from castle_core.manifest import (
BuildSpec,
CaddySpec,
ComponentManifest,
ComponentSpec,
ExposeSpec,
HttpExposeSpec,
HttpInternal,
InstallSpec,
JobSpec,
ManageSpec,
PathInstallSpec,
ProxySpec,
Role,
RunCommand,
RunContainer,
RunPython,
RunRemote,
ServiceSpec,
SystemdSpec,
ToolSpec,
TriggerSchedule,
)
class TestRoleDerivation:
"""Tests for computed role derivation."""
class TestComponentSpec:
"""Tests for component (software catalog) model."""
def test_service_from_expose_http(self) -> None:
"""Component with expose.http gets SERVICE role."""
m = ComponentManifest(
id="svc",
run=RunPython(runner="python", tool="svc"),
expose=ExposeSpec(http=HttpExposeSpec(internal=HttpInternal(port=8000))),
def test_minimal(self) -> None:
"""Minimal component just needs an id."""
c = ComponentSpec(id="bare")
assert c.description is None
assert c.source is None
assert c.install is None
assert c.tool is None
assert c.build is None
def test_tool_component(self) -> None:
"""Component with tool and install specs."""
c = ComponentSpec(
id="my-tool",
description="A tool",
source="my-tool/",
tool=ToolSpec(),
install=InstallSpec(path=PathInstallSpec(alias="my-tool")),
)
assert Role.SERVICE in m.roles
assert c.source == "my-tool/"
assert c.install.path.alias == "my-tool"
def test_tool_from_install_path(self) -> None:
"""Component with install.path gets TOOL role."""
m = ComponentManifest(
id="mytool",
install=InstallSpec(path=PathInstallSpec(alias="mytool")),
)
assert Role.TOOL in m.roles
def test_worker_from_systemd_without_http(self) -> None:
"""Component managed by systemd but no HTTP gets WORKER role."""
m = ComponentManifest(
id="worker",
run=RunCommand(runner="command", argv=["worker-bin"]),
manage=ManageSpec(systemd=SystemdSpec()),
)
assert Role.WORKER in m.roles
assert Role.SERVICE not in m.roles
def test_container_role(self) -> None:
"""Container runner gets CONTAINERIZED role."""
m = ComponentManifest(
id="container",
run=RunContainer(runner="container", image="redis:7"),
)
assert Role.CONTAINERIZED in m.roles
def test_remote_role(self) -> None:
"""Remote runner gets REMOTE role."""
m = ComponentManifest(
id="remote",
run=RunRemote(runner="remote", base_url="http://example.com"),
)
assert Role.REMOTE in m.roles
def test_job_from_schedule_trigger(self) -> None:
"""Component with schedule trigger gets JOB role."""
m = ComponentManifest(
id="job",
run=RunCommand(runner="command", argv=["backup"]),
triggers=[TriggerSchedule(cron="0 * * * *")],
)
assert Role.JOB in m.roles
def test_frontend_from_build(self) -> None:
"""Component with build outputs gets FRONTEND role."""
m = ComponentManifest(
id="frontend",
run=RunCommand(runner="command", argv=["serve"]),
def test_frontend_component(self) -> None:
"""Component with build spec."""
c = ComponentSpec(
id="my-app",
build=BuildSpec(commands=[["pnpm", "build"]], outputs=["dist/"]),
)
assert Role.FRONTEND in m.roles
assert c.build.outputs == ["dist/"]
def test_tool_from_tool_spec(self) -> None:
"""Component with tool spec gets TOOL role."""
m = ComponentManifest(
id="docx2md",
tool=ToolSpec(source="docx2md/"),
)
assert Role.TOOL in m.roles
def test_source_dir_from_source(self) -> None:
"""source_dir uses source field."""
c = ComponentSpec(id="x", source="components/x/")
assert c.source_dir == "components/x"
def test_tool_spec_without_install(self) -> None:
"""Tool spec alone is enough for TOOL role, no install.path needed."""
m = ComponentManifest(
id="my-tool",
tool=ToolSpec(),
)
assert Role.TOOL in m.roles
def test_source_dir_none(self) -> None:
"""source_dir returns None when no source available."""
c = ComponentSpec(id="x")
assert c.source_dir is None
def test_fallback_to_tool(self) -> None:
"""Component with no indicators defaults to TOOL."""
m = ComponentManifest(id="bare")
assert m.roles == [Role.TOOL]
def test_multiple_roles(self) -> None:
"""Component can have multiple roles."""
m = ComponentManifest(
id="multi",
run=RunPython(runner="python", tool="multi"),
expose=ExposeSpec(http=HttpExposeSpec(internal=HttpInternal(port=8000))),
install=InstallSpec(path=PathInstallSpec(alias="multi")),
)
assert Role.SERVICE in m.roles
assert Role.TOOL in m.roles
class TestServiceSpec:
"""Tests for service (long-running daemon) model."""
def test_systemd_with_http_is_service_not_worker(self) -> None:
"""Systemd + HTTP = SERVICE, not WORKER."""
m = ComponentManifest(
def test_basic_service(self) -> None:
"""Service with run and expose."""
s = ServiceSpec(
id="svc",
run=RunPython(runner="python", tool="svc"),
expose=ExposeSpec(http=HttpExposeSpec(internal=HttpInternal(port=8000))),
)
assert s.run.runner == "python"
assert s.expose.http.internal.port == 8000
def test_service_with_component_ref(self) -> None:
"""Service can reference a component."""
s = ServiceSpec(
id="svc",
component="my-component",
run=RunPython(runner="python", tool="svc"),
)
assert s.component == "my-component"
def test_service_with_proxy(self) -> None:
"""Service with proxy spec."""
s = ServiceSpec(
id="svc",
run=RunPython(runner="python", tool="svc"),
proxy=ProxySpec(caddy=CaddySpec(path_prefix="/svc")),
)
assert s.proxy.caddy.path_prefix == "/svc"
def test_service_with_manage(self) -> None:
"""Service with systemd management."""
s = ServiceSpec(
id="svc",
run=RunCommand(runner="command", argv=["bin"]),
manage=ManageSpec(systemd=SystemdSpec()),
)
assert Role.SERVICE in m.roles
assert Role.WORKER not in m.roles
class TestConsistencyValidation:
"""Tests for model validation."""
assert s.manage.systemd.enable is True
def test_remote_with_systemd_raises(self) -> None:
"""Remote runner + systemd management is invalid."""
with pytest.raises(
ValueError, match="manage.systemd cannot be enabled for runner=remote"
):
ComponentManifest(
ServiceSpec(
id="bad",
run=RunRemote(runner="remote", base_url="http://example.com"),
manage=ManageSpec(systemd=SystemdSpec()),
)
def test_no_run_is_valid(self) -> None:
"""Component with no run spec is valid (registration-only)."""
m = ComponentManifest(id="reg-only", description="Just registered")
assert m.run is None
assert m.roles == [Role.TOOL]
def test_no_run_is_invalid(self) -> None:
"""Service requires a run spec."""
with pytest.raises(Exception):
ServiceSpec(id="bad")
class TestJobSpec:
"""Tests for job (scheduled task) model."""
def test_basic_job(self) -> None:
"""Job with run and schedule."""
j = JobSpec(
id="my-job",
run=RunCommand(runner="command", argv=["backup"]),
schedule="0 2 * * *",
)
assert j.schedule == "0 2 * * *"
assert j.timezone == "America/Los_Angeles"
def test_job_with_component_ref(self) -> None:
"""Job can reference a component."""
j = JobSpec(
id="sync",
component="protonmail",
run=RunCommand(runner="command", argv=["protonmail", "sync"]),
schedule="*/5 * * * *",
)
assert j.component == "protonmail"
def test_job_requires_schedule(self) -> None:
"""Job without schedule is invalid."""
with pytest.raises(Exception):
JobSpec(
id="bad",
run=RunCommand(runner="command", argv=["x"]),
)
def test_job_custom_timezone(self) -> None:
"""Job with custom timezone."""
j = JobSpec(
id="job",
run=RunCommand(runner="command", argv=["x"]),
schedule="0 0 * * *",
timezone="UTC",
)
assert j.timezone == "UTC"
class TestModelSerialization:
"""Tests for model_dump behavior."""
def test_dump_excludes_none(self) -> None:
def test_dump_component_excludes_none(self) -> None:
"""model_dump with exclude_none drops None fields."""
m = ComponentManifest(id="test", description="Test")
data = m.model_dump(exclude_none=True, exclude={"id", "roles"})
c = ComponentSpec(id="test", description="Test")
data = c.model_dump(exclude_none=True, exclude={"id"})
assert "description" in data
assert "run" not in data
assert "manage" not in data
assert "install" not in data
assert "tool" not in data
def test_dump_service(self) -> None:
"""Full service manifest serializes correctly."""
m = ComponentManifest(
"""Full service serializes correctly."""
s = ServiceSpec(
id="svc",
description="A service",
run=RunPython(runner="python", tool="svc", cwd="svc"),
run=RunPython(runner="python", tool="svc"),
expose=ExposeSpec(
http=HttpExposeSpec(
internal=HttpInternal(port=9001), health_path="/health"
@@ -179,7 +191,7 @@ class TestModelSerialization:
proxy=ProxySpec(caddy=CaddySpec(path_prefix="/svc")),
manage=ManageSpec(systemd=SystemdSpec()),
)
data = m.model_dump(exclude_none=True, exclude={"id", "roles"})
data = s.model_dump(exclude_none=True, exclude={"id"})
assert data["run"]["runner"] == "python"
assert data["expose"]["http"]["internal"]["port"] == 9001
assert data["proxy"]["caddy"]["path_prefix"] == "/svc"

23
core/tests/test_systemd.py
View File

@@ -3,6 +3,7 @@
from __future__ import annotations
from castle_core.generators.systemd import (
generate_timer,
generate_unit_from_deployed,
unit_name,
)
@@ -115,3 +116,25 @@ class TestUnitFromDeployed:
)
unit = generate_unit_from_deployed("my-svc", deployed)
assert "/data/repos/" not in unit
class TestGenerateTimer:
"""Tests for timer generation from schedule strings."""
def test_daily_timer(self) -> None:
"""Daily cron produces OnCalendar timer."""
timer = generate_timer("my-job", schedule="0 2 * * *", description="Nightly")
assert "Description=Castle timer: Nightly" in timer
assert "OnCalendar=*-*-* 02:00:00" in timer
assert "WantedBy=timers.target" in timer
def test_interval_timer(self) -> None:
"""*/N minute cron produces OnUnitActiveSec timer."""
timer = generate_timer("sync", schedule="*/5 * * * *")
assert "OnUnitActiveSec=300s" in timer
assert "OnBootSec=60" in timer
def test_fallback_description(self) -> None:
"""Timer uses name when no description given."""
timer = generate_timer("my-job", schedule="0 0 * * *")
assert "Description=Castle timer: my-job" in timer