feat: add bootstrap domain (models, setup modules, planning)

- Profile, BootstrapAction, BootstrapPlan models
- Setup modules: hostname, locale, shell, ssh-keygen, runcmd
- Bootstrap planning with ordered phases and env validation

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
2026-03-16 04:56:46 +02:00
parent 31d7583b9a
commit 5f1ee18cb4
6 changed files with 404 additions and 0 deletions

View File

View File

@@ -0,0 +1,44 @@
"""Bootstrap domain models."""
from dataclasses import dataclass, field
from typing import Any, Optional
@dataclass(frozen=True)
class Profile:
"""A bootstrap profile from the manifest."""
name: str
os: str
arch: Optional[str]
hostname: Optional[str]
locale: Optional[str]
shell: Optional[str]
ssh_keys: list[dict[str, str]]
runcmd: list[str]
packages: list[Any] # Raw entries, resolved later
env_required: list[str]
@dataclass(frozen=True)
class BootstrapAction:
"""A single action in a bootstrap plan."""
phase: str # "validate" | "setup" | "packages" | "shell" | "dotfiles"
description: str
commands: list[str]
needs_sudo: bool = False
def __str__(self) -> str:
sudo = " (sudo)" if self.needs_sudo else ""
return f"[{self.phase}] {self.description}{sudo}"
@dataclass(frozen=True)
class BootstrapPlan:
"""Complete bootstrap plan."""
profile: str
actions: list[BootstrapAction]
packages_to_install: list[Any] # PackageDef list
@property
def total_steps(self) -> int:
return len(self.actions)

View File

@@ -0,0 +1,86 @@
"""Bootstrap setup modules -- each produces shell commands."""
from dataclasses import dataclass
from typing import Optional
class SetupModule:
"""Base for setup modules."""
def describe(self) -> str:
raise NotImplementedError
def plan(self) -> list[str]:
raise NotImplementedError
@dataclass(frozen=True)
class HostnameModule(SetupModule):
hostname: str
def describe(self) -> str:
return f"Set hostname to {self.hostname}"
def plan(self) -> list[str]:
return [
f"sudo hostnamectl set-hostname {self.hostname}",
f"echo '127.0.1.1 {self.hostname}' | sudo tee -a /etc/hosts",
]
@dataclass(frozen=True)
class LocaleModule(SetupModule):
locale: str
def describe(self) -> str:
return f"Set locale to {self.locale}"
def plan(self) -> list[str]:
return [
f"sudo locale-gen {self.locale}",
f"sudo update-locale LANG={self.locale}",
]
@dataclass(frozen=True)
class ShellModule(SetupModule):
shell: str
def describe(self) -> str:
return f"Install and configure shell: {self.shell}"
def plan(self) -> list[str]:
shell_path = f"/usr/bin/{self.shell}"
return [
f"sudo chsh -s {shell_path} $USER",
]
@dataclass(frozen=True)
class SSHKeygenModule(SetupModule):
keys: list[dict[str, str]]
def describe(self) -> str:
return f"Generate {len(self.keys)} SSH key(s)"
def plan(self) -> list[str]:
commands: list[str] = []
for key in self.keys:
path = key.get("path", "~/.ssh/id_ed25519")
key_type = key.get("type", "ed25519")
comment = key.get("comment", "")
cmd = f'ssh-keygen -t {key_type} -f {path} -N ""'
if comment:
cmd += f' -C "{comment}"'
commands.append(cmd)
return commands
@dataclass(frozen=True)
class RuncmdModule(SetupModule):
commands: list[str]
def describe(self) -> str:
return f"Run {len(self.commands)} custom command(s)"
def plan(self) -> list[str]:
return list(self.commands)

View File

@@ -0,0 +1,116 @@
"""Bootstrap planning -- builds ordered action list."""
import os
from typing import Any, Optional
from flow.core.errors import ConfigError
from flow.domain.bootstrap.models import BootstrapAction, BootstrapPlan, Profile
from flow.domain.bootstrap.modules import (
HostnameModule,
LocaleModule,
RuncmdModule,
ShellModule,
SSHKeygenModule,
)
from flow.domain.packages.catalog import normalize_profile_entry, parse_catalog
from flow.domain.packages.models import PackageDef
from flow.domain.packages.resolution import resolve_spec
def parse_profile(name: str, raw: dict[str, Any]) -> Profile:
"""Parse a profile definition from manifest."""
return Profile(
name=name,
os=raw.get("os", "linux"),
arch=raw.get("arch"),
hostname=raw.get("hostname"),
locale=raw.get("locale"),
shell=raw.get("shell"),
ssh_keys=raw.get("ssh-keys") or raw.get("ssh_keys") or [],
runcmd=raw.get("runcmd") or [],
packages=raw.get("packages") or [],
env_required=raw.get("env-required") or raw.get("env_required") or [],
)
def plan_bootstrap(
profile: Profile,
manifest: dict[str, Any],
) -> BootstrapPlan:
"""Build a complete bootstrap plan from a profile."""
actions: list[BootstrapAction] = []
# Phase 1: Validate required env vars
missing = [v for v in profile.env_required if not os.environ.get(v)]
if missing:
raise ConfigError(
f"Missing required environment variables for profile '{profile.name}': "
+ ", ".join(missing)
)
# Phase 2: Setup modules
if profile.hostname:
m = HostnameModule(hostname=profile.hostname)
actions.append(BootstrapAction(
phase="setup", description=m.describe(),
commands=m.plan(), needs_sudo=True,
))
if profile.locale:
m = LocaleModule(locale=profile.locale)
actions.append(BootstrapAction(
phase="setup", description=m.describe(),
commands=m.plan(), needs_sudo=True,
))
if profile.ssh_keys:
m = SSHKeygenModule(keys=profile.ssh_keys)
actions.append(BootstrapAction(
phase="setup", description=m.describe(),
commands=m.plan(),
))
# Phase 3: Packages
catalog = parse_catalog(manifest)
packages_to_install: list[PackageDef] = []
for entry in profile.packages:
ref = normalize_profile_entry(entry)
pkg = resolve_spec(ref, catalog)
packages_to_install.append(pkg)
if packages_to_install:
pkg_names = [p.name for p in packages_to_install]
actions.append(BootstrapAction(
phase="packages",
description=f"Install {len(packages_to_install)} package(s): {', '.join(pkg_names[:5])}{'...' if len(pkg_names) > 5 else ''}",
commands=[], # Executed by PackageService
))
# Phase 4: Shell
if profile.shell:
m = ShellModule(shell=profile.shell)
actions.append(BootstrapAction(
phase="shell", description=m.describe(),
commands=m.plan(), needs_sudo=True,
))
# Phase 5: Custom commands
if profile.runcmd:
m = RuncmdModule(commands=profile.runcmd)
actions.append(BootstrapAction(
phase="setup", description=m.describe(),
commands=m.plan(),
))
# Phase 6: Dotfiles link
actions.append(BootstrapAction(
phase="dotfiles",
description="Link dotfiles",
commands=[], # Executed by DotfilesService
))
return BootstrapPlan(
profile=profile.name,
actions=actions,
packages_to_install=packages_to_install,
)

View File

@@ -0,0 +1,73 @@
"""Tests for bootstrap setup modules."""
from flow.domain.bootstrap.modules import (
HostnameModule,
LocaleModule,
RuncmdModule,
ShellModule,
SSHKeygenModule,
)
class TestHostnameModule:
def test_describe(self):
m = HostnameModule(hostname="my-host")
assert isinstance(m.describe(), str)
assert "my-host" in m.describe()
def test_plan(self):
m = HostnameModule(hostname="my-host")
cmds = m.plan()
assert len(cmds) > 0
assert any("my-host" in c for c in cmds)
class TestLocaleModule:
def test_describe(self):
m = LocaleModule(locale="en_US.UTF-8")
assert isinstance(m.describe(), str)
assert "en_US.UTF-8" in m.describe()
def test_plan(self):
m = LocaleModule(locale="en_US.UTF-8")
cmds = m.plan()
assert len(cmds) > 0
class TestShellModule:
def test_describe(self):
m = ShellModule(shell="zsh")
assert isinstance(m.describe(), str)
assert "zsh" in m.describe()
def test_plan(self):
m = ShellModule(shell="zsh")
cmds = m.plan()
assert len(cmds) > 0
class TestSSHKeygenModule:
def test_describe(self):
m = SSHKeygenModule(keys=[{"path": "~/.ssh/id_ed25519"}])
assert isinstance(m.describe(), str)
assert "1" in m.describe()
def test_plan(self):
m = SSHKeygenModule(keys=[
{"path": "~/.ssh/id_ed25519", "type": "ed25519", "comment": "me@host"},
])
cmds = m.plan()
assert len(cmds) == 1
assert "ssh-keygen" in cmds[0]
class TestRuncmdModule:
def test_describe(self):
m = RuncmdModule(commands=["echo a", "echo b", "echo c"])
assert isinstance(m.describe(), str)
assert "3" in m.describe()
def test_plan(self):
m = RuncmdModule(commands=["echo hello"])
cmds = m.plan()
assert cmds == ["echo hello"]

View File

@@ -0,0 +1,85 @@
"""Tests for bootstrap planning."""
import pytest
from flow.core.errors import ConfigError
from flow.domain.bootstrap.models import BootstrapAction, Profile
from flow.domain.bootstrap.planning import parse_profile, plan_bootstrap
class TestParseProfile:
def test_basic(self):
raw = {
"os": "linux",
"hostname": "dev-box",
"shell": "zsh",
"packages": ["fd", "ripgrep"],
}
profile = parse_profile("work", raw)
assert profile.name == "work"
assert profile.os == "linux"
assert profile.hostname == "dev-box"
assert profile.shell == "zsh"
assert len(profile.packages) == 2
def test_defaults(self):
profile = parse_profile("minimal", {})
assert profile.os == "linux"
assert profile.hostname is None
assert profile.packages == []
def test_ssh_keys(self):
raw = {"ssh-keys": [{"path": "~/.ssh/id_ed25519", "type": "ed25519"}]}
profile = parse_profile("test", raw)
assert len(profile.ssh_keys) == 1
class TestPlanBootstrap:
def test_basic_plan(self):
profile = Profile(
name="test", os="linux", arch=None,
hostname="my-host", locale="en_US.UTF-8",
shell="zsh", ssh_keys=[], runcmd=[],
packages=["fd"], env_required=[],
)
manifest = {"packages": [{"name": "fd", "type": "pkg"}]}
plan = plan_bootstrap(profile, manifest)
assert plan.profile == "test"
assert plan.total_steps > 0
phases = [a.phase for a in plan.actions]
assert "setup" in phases
assert "packages" in phases
assert "dotfiles" in phases
def test_missing_env_raises(self, monkeypatch):
monkeypatch.delenv("REQUIRED_VAR", raising=False)
profile = Profile(
name="test", os="linux", arch=None,
hostname=None, locale=None, shell=None,
ssh_keys=[], runcmd=[], packages=[],
env_required=["REQUIRED_VAR"],
)
with pytest.raises(ConfigError, match="REQUIRED_VAR"):
plan_bootstrap(profile, {})
def test_runcmd_produces_action(self):
profile = Profile(
name="test", os="linux", arch=None,
hostname=None, locale=None, shell=None,
ssh_keys=[], runcmd=["echo hello", "echo world"],
packages=[], env_required=[],
)
plan = plan_bootstrap(profile, {})
runcmd_actions = [a for a in plan.actions if "custom command" in a.description.lower()]
assert len(runcmd_actions) == 1
def test_ssh_keys_action(self):
profile = Profile(
name="test", os="linux", arch=None,
hostname=None, locale=None, shell=None,
ssh_keys=[{"path": "~/.ssh/id", "type": "ed25519"}],
runcmd=[], packages=[], env_required=[],
)
plan = plan_bootstrap(profile, {})
ssh_actions = [a for a in plan.actions if "SSH" in a.description]
assert len(ssh_actions) == 1