left4me/l4d2host/l4d2host/service_control.py
mwiegand 49992b3a26
refactor(repo): uv workspace + hatchling + layout restructure
Migrate from pip-install-e + setuptools to a uv workspace with a
committed uv.lock for deterministic deps. Switch both members to
hatchling, and move package sources into nested standard layout
(l4d2host/l4d2host/, l4d2web/l4d2web/) so builds work from a
read-only source tree — setuptools wrote egg-info to source under
the old layout, which broke uv sync on the root-owned /opt/left4me/src.

Local dev install: `pip install -e ./l4d2host -e ./l4d2web` -> `uv sync`.
.envrc switches from `layout python python3.13` to `use uv`. Python
pinned to 3.13 via .python-version.

l4d2web now declares its cross-dep on l4d2host explicitly via
[tool.uv.sources] (workspace = true). l4d2web/alembic.ini and
l4d2web/alembic/ stay at the project root (standard alembic layout).

Test fixes:
- tests/__init__.py added to both test dirs so pytest doesn't shadow
  l4d2host as a namespace package via outer-dir walk.
- 3 CWD-relative paths in tests (l4d2web/static/css/{tokens,layout}.css
  and js/sse.js) anchored to Path(__file__) so they survive layout
  changes.
- Two test_install.py tests now monkeypatch HOME to tmp_path so they
  stop silently mutating ~/.steam/sdk32 on every run.

628 tests pass under sandboxed `uv run pytest`.

Per docs/superpowers/plans/2026-05-15-uv-workspace-execution.md;
prereq for the ckn-bw bundle's uv-sync action (queued).

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
2026-05-15 22:04:29 +02:00

85 lines
2.5 KiB
Python

import subprocess
from typing import Callable, Iterator, Sequence
from l4d2host.process import CommandResult, run_command
SYSTEMCTL_HELPER = "/usr/local/libexec/left4me/left4me-systemctl"
JOURNALCTL_HELPER = "/usr/local/libexec/left4me/left4me-journalctl"
def systemctl_command(action: str, name: str) -> list[str]:
return ["sudo", "-n", SYSTEMCTL_HELPER, action, name]
def journalctl_command(name: str, lines: int = 200, follow: bool = True) -> list[str]:
follow_arg = "--follow" if follow else "--no-follow"
return ["sudo", "-n", JOURNALCTL_HELPER, name, "--lines", str(lines), follow_arg]
def enable_service(
name: str,
*,
on_stdout: Callable[[str], None] | None = None,
on_stderr: Callable[[str], None] | None = None,
passthrough: bool = False,
should_cancel: Callable[[], bool] | None = None,
) -> CommandResult:
return run_command(
systemctl_command("enable", name),
on_stdout=on_stdout,
on_stderr=on_stderr,
passthrough=passthrough,
should_cancel=should_cancel,
)
def disable_service(
name: str,
*,
on_stdout: Callable[[str], None] | None = None,
on_stderr: Callable[[str], None] | None = None,
passthrough: bool = False,
should_cancel: Callable[[], bool] | None = None,
) -> CommandResult:
return run_command(
systemctl_command("disable", name),
on_stdout=on_stdout,
on_stderr=on_stderr,
passthrough=passthrough,
should_cancel=should_cancel,
)
def show_service(name: str) -> CommandResult:
return run_command(systemctl_command("show", name))
def stream_command(cmd: Sequence[str]) -> Iterator[str]:
command = list(cmd)
proc = subprocess.Popen(
command,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
bufsize=1,
)
try:
if proc.stdout is None:
return
for raw in iter(proc.stdout.readline, ""):
yield raw.rstrip("\n")
returncode = proc.wait()
if returncode is None:
returncode = proc.poll()
stderr = proc.stderr.read() if proc.stderr is not None else ""
if returncode:
raise subprocess.CalledProcessError(returncode=returncode, cmd=command, stderr=stderr)
finally:
if proc.poll() is None:
proc.terminate()
proc.wait(timeout=2)
def stream_journal(name: str, *, lines: int = 200, follow: bool = True) -> Iterator[str]:
return stream_command(journalctl_command(name, lines=lines, follow=follow))