SILVER/NODEDC_MISSION_CORE
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

feat(local-service): allow exact worktree migration

Browse Source
This commit is contained in:
DCCONSTRUCTIONS
2026-08-30 23:23:36 +03:00
parent 2a5763d3fb
commit 177be8869f
5 changed files with 421 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_local_service_launchd.py
+214 -12
View File
@@ -3,7 +3,32 @@ from __future__ import annotations
import plistlib
from pathlib import Path
from k1link.local_service_launchd import plan_mission_core_launch_agent
import pytest
from k1link.local_service_launchd import (
MissionCoreLaunchAgentError,
plan_mission_core_launch_agent,
)
def _write_agent(
*,
path: Path,
repository: Path,
uv_entrypoint: Path,
environment: dict[str, str] | None = None,
) -> None:
path.write_bytes(
plistlib.dumps(
{
"Label": "com.nodedc.mission-core.local",
"ProgramArguments": [str(uv_entrypoint), "run", "k1link", "serve"],
"WorkingDirectory": str(repository),
"EnvironmentVariables": environment or {"PATH": "/usr/bin:/bin"},
}
)
)
path.chmod(0o600)
def test_launch_agent_plan_disables_sync_and_enables_watchdog(tmp_path: Path) -> None:
@@ -13,17 +38,7 @@ def test_launch_agent_plan_disables_sync_and_enables_watchdog(tmp_path: Path) ->
uv_entrypoint.write_text("#!/bin/sh\n")
uv_entrypoint.chmod(0o700)
agent = tmp_path / "agent.plist"
agent.write_bytes(
plistlib.dumps(
{
"Label": "com.nodedc.mission-core.local",
"ProgramArguments": [str(uv_entrypoint), "run", "k1link", "serve"],
"WorkingDirectory": str(repository),
"EnvironmentVariables": {"PATH": "/usr/bin:/bin"},
}
)
)
agent.chmod(0o600)
_write_agent(path=agent, repository=repository, uv_entrypoint=uv_entrypoint)
plan = plan_mission_core_launch_agent(
repository_root=repository,
@@ -44,3 +59,190 @@ def test_launch_agent_plan_disables_sync_and_enables_watchdog(tmp_path: Path) ->
assert desired["AbandonProcessGroup"] is False
assert desired["ExitTimeOut"] == 20
assert plan.current_sha256 != plan.desired_sha256
assert plan.current_working_directory == repository
assert plan.desired_working_directory == repository
assert plan.preserved_data_directory is None
assert "MISSIONCORE_DATA_DIR" not in desired["EnvironmentVariables"]
assert plan.to_dict()["changes"]["repository_migration"] is False
assert plan.to_dict()["changes"]["data_directory_preserved"] is False
def test_launch_agent_plan_rejects_cross_repository_migration_by_default(
tmp_path: Path,
) -> None:
current_repository = tmp_path / "current-repo"
current_repository.mkdir()
desired_repository = tmp_path / "desired-repo"
desired_repository.mkdir()
uv_entrypoint = tmp_path / "uv"
uv_entrypoint.write_text("#!/bin/sh\n")
uv_entrypoint.chmod(0o700)
agent = tmp_path / "agent.plist"
_write_agent(
path=agent,
repository=current_repository,
uv_entrypoint=uv_entrypoint,
)
with pytest.raises(
MissionCoreLaunchAgentError,
match="current launch agent targets another repository",
):
plan_mission_core_launch_agent(
repository_root=desired_repository,
agent_path=agent,
)
def test_launch_agent_plan_binds_explicit_cross_repository_migration(
tmp_path: Path,
) -> None:
current_repository = tmp_path / "current-repo"
current_repository.mkdir()
desired_repository = tmp_path / "desired-repo"
desired_repository.mkdir()
other_repository = tmp_path / "other-repo"
other_repository.mkdir()
data_directory = current_repository / ".runtime" / "mission-core"
data_directory.mkdir(parents=True, mode=0o700)
uv_entrypoint = tmp_path / "uv"
uv_entrypoint.write_text("#!/bin/sh\n")
uv_entrypoint.chmod(0o700)
agent = tmp_path / "agent.plist"
_write_agent(
path=agent,
repository=current_repository,
uv_entrypoint=uv_entrypoint,
)
with pytest.raises(
MissionCoreLaunchAgentError,
match="does not target the expected current repository",
):
plan_mission_core_launch_agent(
repository_root=desired_repository,
agent_path=agent,
expected_current_repository_root=other_repository,
)
plan = plan_mission_core_launch_agent(
repository_root=desired_repository,
agent_path=agent,
expected_current_repository_root=current_repository,
)
desired = plistlib.loads(plan.desired_payload)
document = plan.to_dict()
assert plan.current_working_directory == current_repository
assert plan.desired_working_directory == desired_repository
assert plan.preserved_data_directory == data_directory
assert desired["WorkingDirectory"] == str(desired_repository)
assert desired["EnvironmentVariables"]["MISSIONCORE_DATA_DIR"] == str(data_directory)
assert document["current_working_directory"] == str(current_repository)
assert document["desired_working_directory"] == str(desired_repository)
assert document["preserved_data_directory"] == str(data_directory)
assert document["changes"]["repository_migration"] is True
assert document["changes"]["data_directory_preserved"] is True
assert document["changes"]["preserved_data_directory"] == str(data_directory)
def test_launch_agent_migration_retains_existing_private_data_directory(
tmp_path: Path,
) -> None:
current_repository = tmp_path / "current-repo"
current_repository.mkdir()
desired_repository = tmp_path / "desired-repo"
desired_repository.mkdir()
configured_data_directory = tmp_path / "canonical-data"
configured_data_directory.mkdir(mode=0o700)
uv_entrypoint = tmp_path / "uv"
uv_entrypoint.write_text("#!/bin/sh\n")
uv_entrypoint.chmod(0o700)
agent = tmp_path / "agent.plist"
_write_agent(
path=agent,
repository=current_repository,
uv_entrypoint=uv_entrypoint,
environment={
"PATH": "/usr/bin:/bin",
"MISSIONCORE_DATA_DIR": str(configured_data_directory),
},
)
plan = plan_mission_core_launch_agent(
repository_root=desired_repository,
agent_path=agent,
expected_current_repository_root=current_repository,
)
desired = plistlib.loads(plan.desired_payload)
assert plan.preserved_data_directory == configured_data_directory
assert (
desired["EnvironmentVariables"]["MISSIONCORE_DATA_DIR"]
== str(configured_data_directory)
)
def test_launch_agent_migration_rejects_symlinked_data_directory(tmp_path: Path) -> None:
current_repository = tmp_path / "current-repo"
current_repository.mkdir()
desired_repository = tmp_path / "desired-repo"
desired_repository.mkdir()
private_directory = tmp_path / "private-data"
private_directory.mkdir(mode=0o700)
symlinked_directory = tmp_path / "data-link"
symlinked_directory.symlink_to(private_directory, target_is_directory=True)
uv_entrypoint = tmp_path / "uv"
uv_entrypoint.write_text("#!/bin/sh\n")
uv_entrypoint.chmod(0o700)
agent = tmp_path / "agent.plist"
_write_agent(
path=agent,
repository=current_repository,
uv_entrypoint=uv_entrypoint,
environment={
"PATH": "/usr/bin:/bin",
"MISSIONCORE_DATA_DIR": str(symlinked_directory),
},
)
with pytest.raises(
MissionCoreLaunchAgentError,
match="data directory is not a private canonical directory",
):
plan_mission_core_launch_agent(
repository_root=desired_repository,
agent_path=agent,
expected_current_repository_root=current_repository,
)
def test_launch_agent_migration_rejects_nonprivate_default_data_directory(
tmp_path: Path,
) -> None:
current_repository = tmp_path / "current-repo"
current_repository.mkdir()
desired_repository = tmp_path / "desired-repo"
desired_repository.mkdir()
data_directory = current_repository / ".runtime" / "mission-core"
data_directory.mkdir(parents=True, mode=0o755)
data_directory.chmod(0o755)
uv_entrypoint = tmp_path / "uv"
uv_entrypoint.write_text("#!/bin/sh\n")
uv_entrypoint.chmod(0o700)
agent = tmp_path / "agent.plist"
_write_agent(
path=agent,
repository=current_repository,
uv_entrypoint=uv_entrypoint,
)
with pytest.raises(
MissionCoreLaunchAgentError,
match="data directory is not a private canonical directory",
):
plan_mission_core_launch_agent(
repository_root=desired_repository,
agent_path=agent,
expected_current_repository_root=current_repository,
)
tests/test_manage_mission_core_launch_agent.py
+54
View File
@@ -1,7 +1,10 @@
from __future__ import annotations
import importlib.util
import json
import plistlib
import subprocess
import sys
from pathlib import Path
from typing import Any
@@ -15,6 +18,57 @@ manager = importlib.util.module_from_spec(_SPEC)
_SPEC.loader.exec_module(manager)
def test_plan_cli_authorizes_only_the_declared_current_repository(
monkeypatch: Any,
capsys: Any,
tmp_path: Path,
) -> None:
current_repository = tmp_path / "current-repo"
current_repository.mkdir()
desired_repository = tmp_path / "desired-repo"
desired_repository.mkdir()
data_directory = current_repository / ".runtime" / "mission-core"
data_directory.mkdir(parents=True, mode=0o700)
uv_entrypoint = tmp_path / "uv"
uv_entrypoint.write_text("#!/bin/sh\n")
uv_entrypoint.chmod(0o700)
agent = tmp_path / "agent.plist"
agent.write_bytes(
plistlib.dumps(
{
"Label": "com.nodedc.mission-core.local",
"ProgramArguments": [str(uv_entrypoint), "run", "k1link", "serve"],
"WorkingDirectory": str(current_repository),
"EnvironmentVariables": {"PATH": "/usr/bin:/bin"},
}
)
)
agent.chmod(0o600)
monkeypatch.setattr(
sys,
"argv",
[
str(_SCRIPT),
"plan",
"--repository-root",
str(desired_repository),
"--expected-current-repository-root",
str(current_repository),
"--agent-path",
str(agent),
],
)
assert manager.main() == 0
document = json.loads(capsys.readouterr().out)
assert document["current_working_directory"] == str(current_repository)
assert document["desired_working_directory"] == str(desired_repository)
assert document["preserved_data_directory"] == str(data_directory)
assert document["changes"]["repository_migration"] is True
assert document["changes"]["data_directory_preserved"] is True
def test_reload_waits_for_launchd_transition_before_bootstrap(
monkeypatch: Any,
tmp_path: Path,