#!/usr/bin/python3
"""Real PID1/netns fixture, only inside the dedicated disposable container."""
import hashlib
import ast
import importlib.util
import json
import os
from pathlib import Path
import subprocess
import time


def run(*argv, **kwargs):
    return subprocess.run(argv, check=True, capture_output=True, text=True, timeout=20, **kwargs).stdout.strip()


assert os.environ.get('RUNTIME_UPLINK_DISPOSABLE') == '1'
assert Path('/run/systemd/container').exists()
assert run('hostname') == 'runtime-uplink-fixture'
SOURCE = Path('/opt/uplink')
helper = SOURCE / 'runtime_uplink.py'
spec = importlib.util.spec_from_file_location('uplink', helper)
uplink = importlib.util.module_from_spec(spec)
spec.loader.exec_module(uplink)


def net(*args):
    return run('ip', 'netns', 'exec', 'archipelago-runtime', 'ip', *args)


def state(unit, property):
    return run('systemctl', 'show', unit, '-p', property, '--value')


def until(test):
    deadline = time.monotonic() + 15
    while not test():
        assert time.monotonic() < deadline, 'fixture deadline'
        time.sleep(.1)


run('ip', 'netns', 'add', 'archipelago-runtime')
# A misleading wg interface with an address and carrier is NOT a default.
net('link', 'add', 'wg0', 'type', 'dummy')
net('addr', 'add', '10.44.0.1/16', 'dev', 'wg0')
net('link', 'set', 'wg0', 'up')
net('link', 'set', 'lo', 'up')
runtime = Path('/run/uplink-runtime')
runtime.mkdir()
sentinel = runtime / 'wallet-sentinel'
sentinel.write_bytes(b'disposable retained data: never reset')
baseline = hashlib.sha256(sentinel.read_bytes()).hexdigest()
ns = Path('/run/netns/archipelago-runtime')
ns_before = (ns.stat().st_dev, ns.stat().st_ino)

# Render the actual C01 bootstrap template, changing only paths and the Podman
# command to a sentinel. The barrier, ordering, timeout and restart behavior are
# the production directives. No Podman/app operation occurs in this fixture.
template = (SOURCE / 'archipelago-runtime-podman.service').read_text()
template = template.replace('/usr/libexec/runtime_uplink.py', str(helper)).replace('%t', str(runtime))
template = template.replace('ExecStart=/usr/bin/podman info --format={{.Store.GraphRoot}}',
                            'ExecStart=/usr/bin/touch /run/bootstrap-ran')
template = template.replace('[Service]', '[Service]\nNetworkNamespacePath=/run/netns/archipelago-runtime')
Path('/etc/systemd/system/uplink-bootstrap.service').write_text(template)
# Project the exact user-manager preflight directives into a PID1-owned fixture
# unit. This tests the gate before manager exec, not a full C01 migration.
manager = (SOURCE / '81-c01-runtime-access.conf.in').read_text()
directives = '\n'.join(line for line in manager.splitlines() if line.startswith(('ExecCondition=', 'NetworkNamespacePath=', 'TimeoutStartSec=')))
directives = directives.replace('/usr/libexec/runtime_uplink.py', str(helper)).replace('/run/user/1000', str(runtime))
Path('/etc/systemd/system/uplink-manager.service').write_text(
    '[Service]\nType=oneshot\nRemainAfterExit=yes\nExecStart=/usr/bin/touch /run/manager-ran\n' + directives + '\n')
modules = next(ast.literal_eval(node.value) for node in ast.parse((SOURCE / 'provision-d09-lifecycle.py').read_text()).body
               if isinstance(node, ast.Assign) and any(isinstance(t, ast.Name) and t.id == 'MODULES' for t in node.targets))
assert 'runtime_uplink.py' in modules
assert '--c01' in (SOURCE / 'archipelago-runtime-worker.service').read_text()
assert 'ExecCondition=' in (SOURCE / '79-runtime-uplink.conf').read_text()
Path('/etc/systemd/system/uplink-stopped.service').write_text('[Service]\nExecStart=/usr/bin/touch /run/stopped-app-ran\n')
run('systemctl', 'daemon-reload')
run('systemctl', 'start', '--no-block', 'uplink-bootstrap.service')
run('systemctl', 'start', '--no-block', 'uplink-manager.service')
until(lambda: state('uplink-bootstrap.service', 'SubState') in ('condition', 'start-pre'))
time.sleep(1.5)
assert not Path('/run/bootstrap-ran').exists()
assert not Path('/run/manager-ran').exists()
# Actual kernel default, deliberately another nonstandard name. No WAN packets.
net('link', 'add', 'wg-real-default', 'type', 'dummy')
net('addr', 'add', '192.0.2.2/24', 'dev', 'wg-real-default')
net('link', 'set', 'wg-real-default', 'up')
net('route', 'add', 'default', 'via', '192.0.2.1', 'dev', 'wg-real-default')
until(lambda: state('uplink-bootstrap.service', 'ActiveState') == 'active')
assert Path('/run/bootstrap-ran').exists()
until(lambda: Path('/run/manager-ran').exists())
assert state('uplink-bootstrap.service', 'NRestarts') == '0'
# Exercise the actual unprivileged USER-unit namespace lookup permissions too.
run('ip', 'netns', 'exec', 'archipelago-runtime', 'runuser', '-u', 'nobody', '--',
    'python3', '-I', '-S', str(helper), '--c01', '--runtime-dir', str(runtime), '--timeout=5')

# Retain a distinct real nsfs object with a misleading link but no default.
run('ip', 'netns', 'add', 'retained-rootless')
run('ip', '-n', 'retained-rootless', 'link', 'add', 'wg0', 'type', 'dummy')
run('ip', '-n', 'retained-rootless', 'addr', 'add', '10.44.0.1/16', 'dev', 'wg0')
run('ip', '-n', 'retained-rootless', 'link', 'set', 'wg0', 'up')
retained = runtime / 'containers/networks/rootless-netns/rootless-netns'
retained.parent.mkdir(parents=True)
retained.touch()
run('mount', '--bind', '/run/netns/retained-rootless', str(retained))
pin = os.open(retained, os.O_RDONLY)
retained_id = uplink.namespace_id(pin)
info = retained.parent / 'info.json'
info.write_text('{"IPAddresses":["192.0.2.2"]}')
before_info = info.read_bytes()
before_routes = run('ip', '-j', '-n', 'retained-rootless', 'route', 'show', 'table', 'all')
run('systemctl', 'stop', 'uplink-bootstrap.service')
Path('/run/bootstrap-ran').unlink()
run('systemctl', 'start', 'uplink-bootstrap.service')
assert state('uplink-bootstrap.service', 'ActiveState') == 'inactive'
assert 'retained rootless namespace requires owner reconciliation' in run('journalctl', '-u', 'uplink-bootstrap.service', '--no-pager')
assert not Path('/run/bootstrap-ran').exists()
assert uplink.namespace_id(pin) == retained_id == (retained.stat().st_dev, retained.stat().st_ino)
assert before_routes == run('ip', '-j', '-n', 'retained-rootless', 'route', 'show', 'table', 'all')
assert info.read_bytes() == before_info
assert hashlib.sha256(sentinel.read_bytes()).hexdigest() == baseline
assert (ns.stat().st_dev, ns.stat().st_ino) == ns_before
assert state('uplink-stopped.service', 'ActiveState') == 'inactive'
assert not Path('/run/stopped-app-ran').exists()
# Simulate the repaired network but stale Podman cache: still refuse, preserve.
info.write_text('{"IPAddresses":["10.44.0.1"]}')
before_info = info.read_bytes()
run('ip', '-n', 'retained-rootless', 'addr', 'add', '192.0.2.2/24', 'dev', 'wg0')
run('ip', '-n', 'retained-rootless', 'route', 'add', 'default', 'via', '192.0.2.1', 'dev', 'wg0')
run('systemctl', 'start', 'uplink-bootstrap.service')
assert state('uplink-bootstrap.service', 'ActiveState') == 'inactive'
assert 'address cache differs from uplink' in run('journalctl', '-u', 'uplink-bootstrap.service', '--no-pager')
assert info.read_bytes() == before_info and uplink.namespace_id(pin) == retained_id
# Only the fixture owner prepares consistent metadata; the helper never does.
info.write_text('{"IPAddresses":["192.0.2.2"]}')
run('systemctl', 'start', 'uplink-bootstrap.service')
assert state('uplink-bootstrap.service', 'ActiveState') == 'active'
assert uplink.namespace_id(pin) == retained_id
os.close(pin)
# Host namespace cannot impersonate C01 readiness even with an appointed path.
rejected = subprocess.run(['python3', '-I', '-S', str(helper), '--c01', '--runtime-dir', str(runtime), '--timeout=1'], capture_output=True, text=True)
assert rejected.returncode == 1 and 'outside the appointed' in rejected.stderr
# The actual legacy manager condition must skip an unavailable-uplink activation
# even with Restart=always. Outer fixture namespace deliberately has no default.
legacy = (SOURCE / '79-legacy-manager-uplink.conf').read_text()
legacy = legacy.replace('/opt/archipelago/scripts/runtime_uplink.py', str(helper))
legacy = legacy.replace('--runtime-dir=/run/user/1000', '--runtime-dir=/run/legacy-empty --timeout=1')
Path('/etc/systemd/system/uplink-legacy.service').write_text(
    '[Service]\nExecStart=/usr/bin/touch /run/legacy-ran\nRestart=always\n' + legacy)
run('systemctl', 'daemon-reload')
run('systemctl', 'start', 'uplink-legacy.service')
time.sleep(2)
assert state('uplink-legacy.service', 'ActiveState') == 'inactive'
assert not Path('/run/legacy-ran').exists()
assert state('uplink-legacy.service', 'NRestarts') == '0'
print(json.dumps({'boot_order_no_default_blocked': True, 'real_default_releases_bootstrap': True,
                  'user_manager_preflight_ordered': True, 'provisioning_module_present': True,
                  'unprivileged_namespace_gate': True,
                  'misleading_interface_names_ignored': True, 'retained_broken_refused_preserved': True,
                  'healthy_retained_namespace_preserved': True, 'stale_cache_refused_preserved': True,
                  'stopped_app_preserved': True,
                  'host_cannot_satisfy_c01': True, 'legacy_deadline_no_restart_loop': True,
                  'daemon_restarts': 0}, sort_keys=True))