github-such-software/adaptive-multi-protocol-gateway
1
0
Fork 0
mirror of https://github.com/Such-Software/adaptive-multi-protocol-gateway.git synced 2026-10-17 07:31:32 +02:00
Code Issues Packages Projects Releases Wiki Activity

Add transport status and doctor preflight

Browse Source
This commit is contained in:
jwinterm
2026-07-06 04:47:48 -04:00
parent 2e9307ba26
commit 27f3805c19
11 changed files with 999 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+3
View File
@@ -101,6 +101,8 @@ The current implementation is dependency-free Python:
```sh
python3 -m ampg --config examples/wownero.gateway.toml plan
python3 -m ampg --config examples/wownero.gateway.toml plan --write-artifacts
python3 -m ampg --config examples/wownero.gateway.toml status
python3 -m ampg --config examples/wownero.gateway.toml doctor
python3 -m ampg --config examples/wownero.gateway.toml build
python3 -m ampg --config examples/wownero.gateway.toml manifest
python3 -m ampg --config examples/wownero.gateway.toml preview endpoints
@@ -127,6 +129,7 @@ passed. They are reviewable snippets, not installed daemon config.
- [ ] Pick source kind: `static-html`, `markdown`, or `ssg-output`.
- [ ] Select protocols and daemon policy.
- [ ] Run `ampg plan`; review output roots, ports, and config changes.
- [ ] Run `ampg status` or `ampg doctor`; review daemon ownership decisions.
- [ ] Run `ampg build`; inspect generated variants.
- [ ] Run AMPB fixture checks against generated manifests.
- [ ] Run `ampg apply`; publish through adopted or managed daemons.
ampg/cli.py
+108
View File
@@ -10,6 +10,7 @@ from .config import load_config
from .docsgen import generate_docs
from .manifest import write_fixture_manifests
from .plan import plan_gateway, write_plan_artifacts
from .platforms import PLATFORM_NAMES, platform_by_name
from .preview import PreviewServers, preview_endpoints, write_preview_fixture_manifests
from .route_manifest import (
load_route_manifest,
@@ -17,6 +18,7 @@ from .route_manifest import (
validate_route_manifest,
)
from .route_policy import RouteExposure, RouteIssue, route_exposures, route_issues
from .status import DoctorIssue, TransportStatus, doctor_gateway, gateway_status
def main(argv: list[str] | None = None) -> int:
@@ -34,6 +36,24 @@ def main(argv: list[str] | None = None) -> int:
action="store_true",
help="Write generated config snippets to the configured plan root.",
)
status_parser = subcommands.add_parser(
"status",
help="Show enabled transport daemon decisions.",
)
status_parser.add_argument(
"--platform",
choices=PLATFORM_NAMES,
help="Override platform detection for dry-run checks.",
)
doctor_parser = subcommands.add_parser(
"doctor",
help="Check source, renderer, route, and daemon readiness.",
)
doctor_parser.add_argument(
"--platform",
choices=PLATFORM_NAMES,
help="Override platform detection for dry-run checks.",
)
subcommands.add_parser("build", help="Build enabled protocol outputs.")
subcommands.add_parser("manifest", help="Write AMPB fixture manifests.")
preview_parser = subcommands.add_parser("preview", help="Preview generated outputs locally.")
@@ -108,6 +128,10 @@ def main(argv: list[str] | None = None) -> int:
config = load_config(args.config)
if args.command == "plan":
return _cmd_plan(config, write_artifacts=args.write_artifacts)
if args.command == "status":
return _cmd_status(config, args)
if args.command == "doctor":
return _cmd_doctor(config, args)
if args.command == "build":
return _cmd_build(config)
if args.command == "manifest":
@@ -124,6 +148,41 @@ def main(argv: list[str] | None = None) -> int:
return 1
def _cmd_status(config, args) -> int:
statuses = gateway_status(config, platform_provider=_platform_override(args))
for status in statuses:
_print_transport_status(status)
print(
"AMPG_STATUS_SUMMARY "
f"sites={len(config.sites)} "
f"protocols={len(statuses)} "
f"ok={sum(1 for status in statuses if status.status == 'ok')} "
f"warnings={sum(1 for status in statuses if status.status == 'warn')} "
f"errors={sum(1 for status in statuses if status.status == 'error')}"
)
return 0
def _cmd_doctor(config, args) -> int:
platform_provider = _platform_override(args)
statuses = gateway_status(config, platform_provider=platform_provider)
issues = doctor_gateway(config, platform_provider=platform_provider, statuses=statuses)
for status in statuses:
_print_transport_status(status)
for issue in issues:
_print_doctor_issue(issue)
errors = sum(1 for issue in issues if issue.severity == "error")
warnings = sum(1 for issue in issues if issue.severity == "warning")
print(
"AMPG_DOCTOR_SUMMARY "
f"sites={len(config.sites)} "
f"checks={len(statuses) + len(issues)} "
f"warnings={warnings} "
f"errors={errors}"
)
return 1 if errors else 0
def _cmd_plan(config, *, write_artifacts: bool) -> int:
for line in plan_gateway(config):
artifact_text = ",".join(str(path) for path in line.artifacts) if line.artifacts else "-"
@@ -176,6 +235,41 @@ def _cmd_build(config) -> int:
return 0
def _print_transport_status(status: TransportStatus) -> None:
executable_path = status.executable_path or "-"
print(
"AMPG_STATUS "
f"site={status.site_id} "
f"protocol={status.protocol} "
f"renderer={status.renderer} "
f"daemon={status.daemon} "
f"policy={status.daemon_policy} "
f"platform={status.platform} "
f"supervisor={status.supervisor} "
f"adapter={status.adapter} "
f"backend={status.backend} "
f"installed={_bool(status.installed)} "
f"running={_bool(status.running)} "
f"adoptable={_bool(status.adoptable)} "
f"manageable={_bool(status.manageable)} "
f"executable={executable_path} "
f"status={status.status} "
f"action={status.action} "
f"message=\"{_quote(status.message)}\""
)
def _print_doctor_issue(issue: DoctorIssue) -> None:
print(
"AMPG_DOCTOR "
f"site={issue.site_id} "
f"protocol={issue.protocol} "
f"severity={issue.severity} "
f"code={issue.code} "
f"message=\"{_quote(issue.message)}\""
)
def _cmd_manifest(config) -> int:
for manifest in write_fixture_manifests(config):
print(
@@ -366,3 +460,17 @@ def _cmd_docs(args) -> int:
print(f"AMPG_DOCS status=ok mode={mode} changed={changed_text}")
return 0
return 1
def _platform_override(args):
if getattr(args, "platform", None):
return platform_by_name(args.platform)
return None
def _bool(value: bool) -> str:
return str(value).lower()
def _quote(value: str) -> str:
return value.replace("\\", "\\\\").replace('"', '\\"')
ampg/metadata.py
+11
View File
@@ -465,4 +465,15 @@ DAEMON_ADAPTERS = (
"IPFS is content-addressed distribution, not an anonymity layer.",
),
),
DaemonAdapter(
daemon=DEFAULT_DAEMONS["reticulum"],
protocols=("reticulum",),
default_policy=DEFAULT_POLICIES["reticulum"],
generated_artifacts=("Reticulum page-service plan",),
notes=(
"Reticulum output is planned as small page-service content.",
"Reticulum is resilient/private routing, not an anonymity layer.",
"Physical interfaces may require explicit operator setup.",
),
),
)
ampg/platforms.py
+151
View File
@@ -0,0 +1,151 @@
from __future__ import annotations
from collections.abc import Callable, Mapping
from dataclasses import dataclass
from pathlib import Path
import os
import platform
@dataclass(frozen=True)
class PlatformProvider:
name: str
process_supervisor: str
can_manage_daemons: bool
can_write_system_config: bool
state_root: Path
notes: tuple[str, ...] = ()
PLATFORM_NAMES = (
"android-termux",
"linux-systemd",
"linux-user",
"macos-launchd",
"unknown",
)
def detect_platform(
env: Mapping[str, str] | None = None,
*,
system_name: str | None = None,
path_exists: Callable[[Path], bool] | None = None,
) -> PlatformProvider:
environ = env if env is not None else os.environ
system = system_name or platform.system()
exists = path_exists or Path.exists
if _looks_like_termux(environ):
prefix = Path(environ.get("PREFIX", "/data/data/com.termux/files/usr"))
return PlatformProvider(
name="android-termux",
process_supervisor="termux-services",
can_manage_daemons=True,
can_write_system_config=False,
state_root=prefix / "var/lib/ampg",
notes=(
"User-space daemons only.",
"Public ports and battery/background policy remain device-owned.",
),
)
if system == "Linux":
if exists(Path("/run/systemd/system")):
return PlatformProvider(
name="linux-systemd",
process_supervisor="systemd",
can_manage_daemons=True,
can_write_system_config=True,
state_root=Path("/var/lib/ampg"),
notes=("System service management is available after operator approval.",),
)
return PlatformProvider(
name="linux-user",
process_supervisor="user-process",
can_manage_daemons=True,
can_write_system_config=False,
state_root=_home(environ) / ".local/state/ampg",
notes=("Runs AMPG-owned daemons in user space.",),
)
if system == "Darwin":
return PlatformProvider(
name="macos-launchd",
process_supervisor="launchd-user",
can_manage_daemons=True,
can_write_system_config=False,
state_root=_home(environ) / "Library/Application Support/AMPG",
notes=("Uses user LaunchAgents or foreground processes.",),
)
return PlatformProvider(
name="unknown",
process_supervisor="manual",
can_manage_daemons=False,
can_write_system_config=False,
state_root=_home(environ) / ".ampg",
notes=("AMPG can render and plan, but daemon management is disabled.",),
)
def platform_by_name(name: str) -> PlatformProvider:
if name == "android-termux":
return PlatformProvider(
name=name,
process_supervisor="termux-services",
can_manage_daemons=True,
can_write_system_config=False,
state_root=Path("/data/data/com.termux/files/usr/var/lib/ampg"),
notes=("User-space daemons only.",),
)
if name == "linux-systemd":
return PlatformProvider(
name=name,
process_supervisor="systemd",
can_manage_daemons=True,
can_write_system_config=True,
state_root=Path("/var/lib/ampg"),
notes=("System service management is available after operator approval.",),
)
if name == "linux-user":
return PlatformProvider(
name=name,
process_supervisor="user-process",
can_manage_daemons=True,
can_write_system_config=False,
state_root=Path("~/.local/state/ampg").expanduser(),
notes=("Runs AMPG-owned daemons in user space.",),
)
if name == "macos-launchd":
return PlatformProvider(
name=name,
process_supervisor="launchd-user",
can_manage_daemons=True,
can_write_system_config=False,
state_root=Path("~/Library/Application Support/AMPG").expanduser(),
notes=("Uses user LaunchAgents or foreground processes.",),
)
if name == "unknown":
return PlatformProvider(
name=name,
process_supervisor="manual",
can_manage_daemons=False,
can_write_system_config=False,
state_root=Path("~/.ampg").expanduser(),
notes=("Daemon management is disabled.",),
)
raise ValueError(f"unknown platform provider {name!r}")
def _looks_like_termux(env: Mapping[str, str]) -> bool:
prefix = env.get("PREFIX", "")
return (
"com.termux" in prefix
or bool(env.get("TERMUX_VERSION"))
or (bool(env.get("ANDROID_ROOT")) and "com.termux" in env.get("HOME", ""))
)
def _home(env: Mapping[str, str]) -> Path:
return Path(env.get("HOME", "~")).expanduser()
ampg/status.py
+290
View File
@@ -0,0 +1,290 @@
from __future__ import annotations
from collections.abc import Callable
from dataclasses import dataclass
import shutil
import subprocess
from .config import GatewayConfig, ProtocolConfig, SiteConfig
from .platforms import PlatformProvider, detect_platform
from .route_policy import route_issues
from .transports import (
TransportAdapter,
fallback_transport_adapter,
transport_adapter,
)
VALID_DAEMON_POLICIES = {"external", "adopt", "manage", "auto"}
SUPPORTED_RENDERERS = {"clearnet", "privacy-html", "gemtext"}
@dataclass(frozen=True)
class DaemonProbe:
installed: bool
running: bool
executable_path: str | None = None
@dataclass(frozen=True)
class TransportStatus:
site_id: str
protocol: str
renderer: str
daemon: str
daemon_policy: str
platform: str
supervisor: str
adapter: str
backend: str
installed: bool
running: bool
adoptable: bool
manageable: bool
executable_path: str | None
status: str
action: str
message: str
@dataclass(frozen=True)
class DoctorIssue:
site_id: str
severity: str
code: str
message: str
protocol: str = "-"
DaemonProbeFunc = Callable[[TransportAdapter], DaemonProbe]
def gateway_status(
config: GatewayConfig,
*,
platform_provider: PlatformProvider | None = None,
daemon_probe: DaemonProbeFunc | None = None,
) -> list[TransportStatus]:
provider = platform_provider or detect_platform()
probe = daemon_probe or probe_daemon
statuses: list[TransportStatus] = []
for site in config.sites:
for protocol in site.protocols.values():
if not protocol.enabled:
continue
adapter = transport_adapter(protocol.name, protocol.daemon)
adapter = adapter or fallback_transport_adapter(protocol.name, protocol.daemon)
daemon = probe(adapter)
statuses.append(_transport_status(site, protocol, adapter, provider, daemon))
return statuses
def doctor_gateway(
config: GatewayConfig,
*,
platform_provider: PlatformProvider | None = None,
daemon_probe: DaemonProbeFunc | None = None,
statuses: list[TransportStatus] | None = None,
) -> list[DoctorIssue]:
if statuses is None:
statuses = gateway_status(
config,
platform_provider=platform_provider,
daemon_probe=daemon_probe,
)
issues: list[DoctorIssue] = []
for site in config.sites:
if not site.source.path.exists():
issues.append(
DoctorIssue(
site_id=site.id,
severity="error",
code="source-missing",
message=f"source path does not exist: {site.source.path}",
)
)
elif not site.source.path.is_dir():
issues.append(
DoctorIssue(
site_id=site.id,
severity="error",
code="source-not-directory",
message=f"source path is not a directory: {site.source.path}",
)
)
for protocol in site.protocols.values():
if not protocol.enabled:
continue
if protocol.renderer not in SUPPORTED_RENDERERS:
issues.append(
DoctorIssue(
site_id=site.id,
protocol=protocol.name,
severity="error",
code="unsupported-renderer",
message=f"renderer {protocol.renderer!r} is not implemented",
)
)
output_root = site.outputs.root / protocol.name
if not (output_root / ".ampg-output").exists():
issues.append(
DoctorIssue(
site_id=site.id,
protocol=protocol.name,
severity="warning",
code="output-missing",
message=f"build output is missing: {output_root}",
)
)
for status in statuses:
if status.status == "error":
issues.append(
DoctorIssue(
site_id=status.site_id,
protocol=status.protocol,
severity="error",
code="daemon-status",
message=status.message,
)
)
elif status.status == "warn":
issues.append(
DoctorIssue(
site_id=status.site_id,
protocol=status.protocol,
severity="warning",
code="daemon-status",
message=status.message,
)
)
for issue in route_issues(config):
issues.append(
DoctorIssue(
site_id=issue.site_id,
protocol="-",
severity="error",
code=issue.code,
message=issue.message,
)
)
return issues
def probe_daemon(adapter: TransportAdapter) -> DaemonProbe:
executable_path = shutil.which(adapter.executable)
running = _process_running(adapter.executable)
return DaemonProbe(
installed=bool(executable_path),
running=running,
executable_path=executable_path,
)
def _transport_status(
site: SiteConfig,
protocol: ProtocolConfig,
adapter: TransportAdapter,
provider: PlatformProvider,
daemon: DaemonProbe,
) -> TransportStatus:
adapter_known = adapter.backend != "unknown"
adoptable = adapter.adopt_supported and daemon.running
manageable = adapter.managed_supported and provider.can_manage_daemons
policy = protocol.daemon_policy
if policy not in VALID_DAEMON_POLICIES:
status = "error"
action = "unavailable"
message = f"unknown daemon policy {policy!r}"
elif policy == "external":
status = "ok" if adapter_known else "warn"
action = "render-only"
message = (
"render outputs and review generated snippets"
if adapter_known
else "no registered daemon adapter; render output only"
)
elif not adapter_known:
status = "error"
action = "unavailable"
message = f"no registered adapter for {protocol.name}/{protocol.daemon}"
elif policy == "adopt":
if adoptable:
status = "ok"
action = "adopt-existing"
message = "existing daemon appears to be running"
elif daemon.installed:
status = "error"
action = "unavailable"
message = f"{adapter.executable} is installed but not running"
else:
status = "error"
action = "unavailable"
message = f"{adapter.executable} is not installed or not on PATH"
elif policy == "manage":
if manageable:
status = "ok"
action = "manage-owned"
message = f"AMPG can manage {protocol.name} with {provider.name}"
else:
status = "error"
action = "unavailable"
message = f"{provider.name} cannot manage {protocol.name}/{protocol.daemon}"
else:
if adoptable:
status = "ok"
action = "adopt-existing"
message = "existing daemon appears to be running"
elif manageable:
status = "ok"
action = "manage-owned"
if daemon.installed:
message = f"{adapter.executable} is installed but not running; AMPG would manage it"
else:
message = f"{adapter.executable} is missing; AMPG would prepare managed setup"
else:
status = "error"
action = "unavailable"
message = f"no running daemon and {provider.name} cannot manage {protocol.name}/{protocol.daemon}"
return TransportStatus(
site_id=site.id,
protocol=protocol.name,
renderer=protocol.renderer,
daemon=protocol.daemon,
daemon_policy=policy,
platform=provider.name,
supervisor=provider.process_supervisor,
adapter=f"{adapter.protocol}/{adapter.daemon}",
backend=adapter.backend,
installed=daemon.installed,
running=daemon.running,
adoptable=adoptable,
manageable=manageable,
executable_path=daemon.executable_path,
status=status,
action=action,
message=message,
)
def _process_running(executable: str) -> bool:
pgrep = shutil.which("pgrep")
if not pgrep:
return False
try:
result = subprocess.run(
[pgrep, "-x", executable],
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
timeout=1,
check=False,
)
except (OSError, subprocess.SubprocessError):
return False
return result.returncode == 0
ampg/transports.py
+125
View File
@@ -0,0 +1,125 @@
from __future__ import annotations
from dataclasses import dataclass
@dataclass(frozen=True)
class TransportAdapter:
protocol: str
daemon: str
executable: str
backend: str
adopt_supported: bool
managed_supported: bool
default_ports: tuple[int, ...] = ()
generated_artifacts: tuple[str, ...] = ()
notes: tuple[str, ...] = ()
TRANSPORT_ADAPTERS = (
TransportAdapter(
protocol="clearnet",
daemon="nginx",
executable="nginx",
backend="http",
adopt_supported=True,
managed_supported=True,
default_ports=(80, 443),
generated_artifacts=("nginx server block snippet",),
notes=(
"Clearnet ingress is usually adopted because TLS and public vhost policy are operator-owned.",
),
),
TransportAdapter(
protocol="tor",
daemon="tor",
executable="tor",
backend="hidden-service",
adopt_supported=True,
managed_supported=True,
default_ports=(18080,),
generated_artifacts=("torrc hidden-service snippet", "loopback HTTP server block"),
notes=(
"Adopted Tor keeps existing HiddenServiceDir material.",
"Managed Tor stores onion service keys under AMPG state.",
),
),
TransportAdapter(
protocol="tor",
daemon="arti",
executable="arti",
backend="hidden-service",
adopt_supported=False,
managed_supported=False,
default_ports=(18080,),
generated_artifacts=("future Arti onion-service config",),
notes=("Modeled as a future backend until the AMPG Arti adapter is implemented.",),
),
TransportAdapter(
protocol="i2p",
daemon="i2pd",
executable="i2pd",
backend="server-tunnel",
adopt_supported=True,
managed_supported=True,
default_ports=(18081,),
generated_artifacts=("i2pd server tunnel snippet", "loopback HTTP server block"),
notes=("Destination keys are transport identity material and must be preserved.",),
),
TransportAdapter(
protocol="gemini",
daemon="agate",
executable="agate",
backend="gemini-server",
adopt_supported=True,
managed_supported=True,
default_ports=(1965,),
generated_artifacts=("Agate plan values",),
notes=("Gemini serves generated Gemtext output directly.",),
),
TransportAdapter(
protocol="ipfs",
daemon="ipfs",
executable="ipfs",
backend="content-addressed",
adopt_supported=True,
managed_supported=True,
default_ports=(5001, 8080),
generated_artifacts=("pin/add plan",),
notes=("IPFS is content-addressed distribution, not an anonymity layer.",),
),
TransportAdapter(
protocol="reticulum",
daemon="rnsd",
executable="rnsd",
backend="reticulum-page-service",
adopt_supported=True,
managed_supported=True,
default_ports=(),
generated_artifacts=("Reticulum page-service plan",),
notes=(
"Reticulum is resilient/private routing, not an anonymity layer.",
"Physical interfaces may require operator setup outside AMPG.",
),
),
)
def transport_adapter(protocol: str, daemon: str) -> TransportAdapter | None:
for adapter in TRANSPORT_ADAPTERS:
if adapter.protocol == protocol and adapter.daemon == daemon:
return adapter
return None
def fallback_transport_adapter(protocol: str, daemon: str) -> TransportAdapter:
return TransportAdapter(
protocol=protocol,
daemon=daemon,
executable=daemon,
backend="unknown",
adopt_supported=False,
managed_supported=False,
generated_artifacts=(),
notes=("No AMPG adapter is registered for this protocol/daemon pair.",),
)
docs/architecture.md
+16
View File
@@ -91,6 +91,22 @@ Daemon adapters provide a common interface:
Adapters must be conservative. They do not modify non-AMPG config without showing a
plan first, and managed config lives under AMPG-owned state directories.
## Platform providers
Platform providers decide what daemon management means on the current host. AMPG detects
systemd Linux, user-space Linux, macOS user launchd, Termux-style Android, and unknown
platforms. A provider can allow managed daemons without allowing system config writes.
Operators can override detection for dry runs:
```sh
python3 -m ampg --config gateway.toml doctor --platform linux-systemd
python3 -m ampg --config gateway.toml doctor --platform android-termux
```
This keeps the same gateway config portable across a VPS, laptop, old phone, or manual
render-only environment while making daemon ownership decisions explicit.
## Interaction boundary
v1 is static-first, but the architecture reserves a path for interactive applications.
docs/daemon-management.md
+26 -2
View File
@@ -31,13 +31,37 @@ For each enabled protocol, AMPG checks:
3. Common system services and ports.
4. Known config paths for supported daemons.
The result is shown in `ampg plan`.
The result is shown in `ampg status`. `ampg plan` shows rendered outputs and reviewable
config artifacts.
## Preflight commands
```sh
python3 -m ampg --config gateway.toml status
python3 -m ampg --config gateway.toml doctor
python3 -m ampg --config gateway.toml doctor --platform android-termux
```
`status` prints one row per enabled protocol with the selected platform provider,
adapter, installed/running daemon probe, action, and policy result.
`doctor` checks source paths, renderer support, route exposure, output readiness, and
daemon policy feasibility. It exits nonzero only for errors; missing build output is a
warning so operators can run it before the first build.
Platform providers describe how AMPG may supervise managed daemons:
- `linux-systemd`: system service management after approval.
- `linux-user`: user-space foreground or user service management.
- `macos-launchd`: user LaunchAgents or foreground processes.
- `android-termux`: Termux-style user-space daemons for mobile/server experiments.
- `unknown`: render and plan only; daemon management is disabled.
## Adapter notes
### Clearnet
Default daemon: Caddy.
Default daemon: nginx.
AMPG should usually adopt clearnet ingress instead of managing it. Existing TLS,
firewall, and reverse proxy rules are operator-owned. Managed clearnet is useful for
docs/generated/daemon-adapters.md
+11
View File
@@ -54,3 +54,14 @@ Notes:
Notes:
- IPFS output is static web content for a local gateway or later pinning.
- IPFS is content-addressed distribution, not an anonymity layer.
## rnsd
- Protocols: reticulum
- Default policy: `auto`
- Generated artifacts: Reticulum page-service plan
Notes:
- Reticulum output is planned as small page-service content.
- Reticulum is resilient/private routing, not an anonymity layer.
- Physical interfaces may require explicit operator setup.
tests/test_platforms.py
+56
View File
@@ -0,0 +1,56 @@
import unittest
from pathlib import Path
from ampg.platforms import detect_platform, platform_by_name
class PlatformsTest(unittest.TestCase):
def test_detects_termux_before_generic_linux(self):
provider = detect_platform(
{
"PREFIX": "/data/data/com.termux/files/usr",
"HOME": "/data/data/com.termux/files/home",
},
system_name="Linux",
path_exists=lambda path: path == Path("/run/systemd/system"),
)
self.assertEqual("android-termux", provider.name)
self.assertTrue(provider.can_manage_daemons)
self.assertFalse(provider.can_write_system_config)
self.assertEqual(
Path("/data/data/com.termux/files/usr/var/lib/ampg"),
provider.state_root,
)
def test_detects_systemd_linux(self):
provider = detect_platform(
{"HOME": "/home/operator"},
system_name="Linux",
path_exists=lambda path: path == Path("/run/systemd/system"),
)
self.assertEqual("linux-systemd", provider.name)
self.assertEqual("systemd", provider.process_supervisor)
self.assertTrue(provider.can_write_system_config)
def test_detects_user_space_linux_without_systemd(self):
provider = detect_platform(
{"HOME": "/home/operator"},
system_name="Linux",
path_exists=lambda path: False,
)
self.assertEqual("linux-user", provider.name)
self.assertEqual(Path("/home/operator/.local/state/ampg"), provider.state_root)
self.assertFalse(provider.can_write_system_config)
def test_named_platforms_are_available_for_dry_runs(self):
self.assertEqual("macos-launchd", platform_by_name("macos-launchd").name)
self.assertEqual("unknown", platform_by_name("unknown").name)
with self.assertRaisesRegex(ValueError, "unknown platform provider"):
platform_by_name("mainframe")
if __name__ == "__main__":
unittest.main()
tests/test_status.py
+202
View File
@@ -0,0 +1,202 @@
import contextlib
import io
import tempfile
import unittest
from pathlib import Path
from ampg.cli import main
from ampg.config import load_config
from ampg.platforms import platform_by_name
from ampg.status import DaemonProbe, doctor_gateway, gateway_status
def _probe(*, installed: bool = False, running: bool = False):
def probe(adapter):
executable_path = f"/usr/bin/{adapter.executable}" if installed else None
return DaemonProbe(
installed=installed,
running=running,
executable_path=executable_path,
)
return probe
def _write_config(root: Path, protocol_block: str, *, source_exists: bool = True) -> Path:
source = root / "site"
if source_exists:
source.mkdir()
(source / "index.html").write_text("<h1>Hello</h1>", encoding="utf-8")
config_path = root / "gateway.toml"
config_path.write_text(
f"""
[[site]]
id = "example"
domain = "example.test"
[site.source]
kind = "static-html"
path = "./site"
[site.outputs]
root = "./out"
plan_root = "./plan"
{protocol_block}
""",
encoding="utf-8",
)
return config_path
class StatusTest(unittest.TestCase):
def test_auto_policy_adopts_running_daemon(self):
with tempfile.TemporaryDirectory() as tmp:
config_path = _write_config(
Path(tmp),
"""
[site.protocols.tor]
enabled = true
renderer = "privacy-html"
daemon = "tor"
daemon_policy = "auto"
""",
)
config = load_config(config_path)
statuses = gateway_status(
config,
platform_provider=platform_by_name("linux-user"),
daemon_probe=_probe(installed=True, running=True),
)
self.assertEqual(1, len(statuses))
self.assertEqual("ok", statuses[0].status)
self.assertEqual("adopt-existing", statuses[0].action)
self.assertTrue(statuses[0].adoptable)
def test_auto_policy_manages_missing_daemon_on_manageable_platform(self):
with tempfile.TemporaryDirectory() as tmp:
config_path = _write_config(
Path(tmp),
"""
[site.protocols.i2p]
enabled = true
renderer = "privacy-html"
daemon = "i2pd"
daemon_policy = "auto"
""",
)
config = load_config(config_path)
statuses = gateway_status(
config,
platform_provider=platform_by_name("android-termux"),
daemon_probe=_probe(installed=False, running=False),
)
self.assertEqual("ok", statuses[0].status)
self.assertEqual("manage-owned", statuses[0].action)
self.assertTrue(statuses[0].manageable)
def test_adopt_policy_errors_when_daemon_is_missing(self):
with tempfile.TemporaryDirectory() as tmp:
config_path = _write_config(
Path(tmp),
"""
[site.protocols.clearnet]
enabled = true
renderer = "clearnet"
daemon = "nginx"
daemon_policy = "adopt"
""",
)
config = load_config(config_path)
statuses = gateway_status(
config,
platform_provider=platform_by_name("linux-user"),
daemon_probe=_probe(installed=False, running=False),
)
self.assertEqual("error", statuses[0].status)
self.assertEqual("unavailable", statuses[0].action)
self.assertIn("nginx is not installed", statuses[0].message)
def test_unknown_adapter_errors_unless_external(self):
with tempfile.TemporaryDirectory() as tmp:
config_path = _write_config(
Path(tmp),
"""
[site.protocols.hyperweb]
enabled = true
renderer = "clearnet"
daemon = "hyperd"
daemon_policy = "auto"
""",
)
config = load_config(config_path)
statuses = gateway_status(
config,
platform_provider=platform_by_name("linux-user"),
daemon_probe=_probe(installed=False, running=False),
)
self.assertEqual("error", statuses[0].status)
self.assertEqual("unknown", statuses[0].backend)
self.assertIn("no registered adapter", statuses[0].message)
def test_doctor_reports_source_renderer_and_status_issues(self):
with tempfile.TemporaryDirectory() as tmp:
config_path = _write_config(
Path(tmp),
"""
[site.protocols.tor]
enabled = true
renderer = "micron"
daemon = "tor"
daemon_policy = "adopt"
""",
source_exists=False,
)
config = load_config(config_path)
issues = doctor_gateway(
config,
platform_provider=platform_by_name("linux-user"),
daemon_probe=_probe(installed=False, running=False),
)
codes = {issue.code for issue in issues}
self.assertIn("source-missing", codes)
self.assertIn("unsupported-renderer", codes)
self.assertIn("daemon-status", codes)
self.assertGreaterEqual(sum(1 for issue in issues if issue.severity == "error"), 3)
def test_doctor_cli_allows_warnings_without_errors(self):
with tempfile.TemporaryDirectory() as tmp:
config_path = _write_config(
Path(tmp),
"""
[site.protocols.clearnet]
enabled = true
renderer = "clearnet"
daemon = "nginx"
daemon_policy = "external"
""",
)
stdout = io.StringIO()
with contextlib.redirect_stdout(stdout):
status = main(["--config", str(config_path), "doctor", "--platform", "unknown"])
self.assertEqual(0, status)
output = stdout.getvalue()
self.assertIn("AMPG_DOCTOR_SUMMARY", output)
self.assertIn("warnings=1", output)
self.assertIn("errors=0", output)
if __name__ == "__main__":
unittest.main()