#!/usr/bin/env python3
"""Exercise the actual CLI/system-bus/polkit boundary on the supplied ALT VM."""
import contextlib
import http.server
import json
import os
from pathlib import Path
import socket
import subprocess
import threading
import time

ROOT = Path('/var/tmp/roentgen-implementation')
BIN = str(ROOT / 'roentgen')
BODY = b'trojan://never-print-password@example.com:443?security=tls#First\n'


def command(*args, input=None, ok=True, user=None, env=None):
    argv = (["runuser", "-u", user, "--"] if user else []) + list(args)
    result = subprocess.run(argv, input=input, text=True, capture_output=True, timeout=40, env=env)
    if ok and result.returncode:
        raise AssertionError(f'{args[0:2]} failed: {result.stderr}')
    if not ok and not result.returncode:
        raise AssertionError(f'{args[0:2]} unexpectedly succeeded')
    return result


def cli(*args, **kwargs): return command(BIN, *args, **kwargs)


def status(): return json.loads(cli('status', '--json').stdout)


class Handler(http.server.BaseHTTPRequestHandler):
    def do_GET(self):
        self.send_response(200)
        self.send_header('Content-Type', 'text/plain')
        self.end_headers()
        self.wfile.write(BODY)
    def log_message(self, *args): pass


def main():
    global BODY
    assert os.geteuid() == 0 and socket.gethostname() == 'altctlrs'
    assert (ROOT / 'allow-network-tests').exists()
    # Test setup installs integration fixtures directly. This is deliberately not
    # an assertion that product install succeeds with the VM's unsupported core.
    Path('/etc/roentgen').mkdir(mode=0o700, exist_ok=True)
    Path('/var/lib/roentgen').mkdir(mode=0o700, exist_ok=True)
    assets = ROOT / 'assets' 
    destinations = {
        'roentgen.service': '/etc/systemd/system/roentgen.service',
        'roentgen-guard.service': '/etc/systemd/system/roentgen-guard.service',
        'roentgen-xray.service': '/etc/systemd/system/roentgen-xray.service',
        'org.roentgen.Controller1.conf': '/etc/dbus-1/system.d/org.roentgen.Controller1.conf',
        'org.roentgen.policy': '/usr/share/polkit-1/actions/org.roentgen.policy',
    }
    for source, destination in destinations.items():
        Path(destination).write_bytes((assets / source).read_bytes())
        Path(destination).chmod(0o644)
    target = Path('/usr/local/bin/roentgen')
    target.write_bytes(Path(BIN).read_bytes()); target.chmod(0o755)
    command('systemctl', 'daemon-reload')
    command('busctl', 'call', 'org.freedesktop.DBus', '/org/freedesktop/DBus', 'org.freedesktop.DBus', 'ReloadConfig')
    command('systemctl', 'start', 'roentgen-guard.service')
    command('systemctl', 'start', 'roentgen.service')
    server = http.server.ThreadingHTTPServer(('127.0.0.1', 0), Handler)
    threading.Thread(target=server.serve_forever, daemon=True).start()
    token_url = f'http://127.0.0.1:{server.server_port}/never-print-subscription-token'
    if 'Personal' in cli('subscription', 'list').stdout: cli('subscription', 'delete', 'Personal')
    results = []
    policy = Path('/etc/polkit-1/rules.d/00-roentgen-vm-test.rules')
    try:
        cli('subscription', 'add', '--name', 'Personal', input=token_url + '\n')
        before = status()
        assert before['selected_configuration'] is None and not before['requested_enabled']
        public = cli('subscription', 'list').stdout + cli('configuration', 'list').stdout + cli('status', '--json').stdout
        assert 'never-print-password' not in public and 'never-print-subscription-token' not in public
        results.append('real HTTP import remains disabled/unselected and public CLI output omits credentials')
        cli('enable', ok=False)
        assert status()['generation'] == before['generation']
        rows = command('busctl', 'call', 'org.roentgen.Controller1', '/org/roentgen/Controller1', 'org.roentgen.Controller1', 'Configurations', 's', '').stdout
        # busctl prints a quoted JSON string. Decode that string before its JSON payload.
        entries = json.loads(json.loads(rows[2:].strip()))
        selected = entries[0]['id']
        cli('configuration', 'select', selected)
        before = status()
        denied = cli('configuration', 'unset', ok=False, user='rustacean')
        assert status()['selected_configuration'] == selected and status()['generation'] == before['generation']
        assert 'authorization' in denied.stderr.lower() or 'accessdenied' in denied.stderr.lower()
        results.append('noninteractive non-root mutation is denied with durable state unchanged')
        policy.write_text('polkit.addRule(function(action, subject) { if (action.id == "org.roentgen.control" && subject.user == "rustacean") return polkit.Result.YES; });\n')
        policy.chmod(0o644)
        time.sleep(1)
        cli('configuration', 'unset', user='rustacean')
        assert status()['selected_configuration'] is None
        results.append('polkit site allow policy authorizes the actual ordinary bus caller')
        policy.unlink()
        time.sleep(1)
        cli('configuration', 'select', selected)
        BODY = b'trojan://never-print-password@example.com:443#Renamed\nvless://id@example.com:443#Unsupported\ntrojan://broken@host#Invalid\n'
        cli('subscription', 'update', 'Personal')
        listing = cli('configuration', 'list').stdout
        assert 'Renamed' in listing and 'unsupported' in listing and 'invalid' in listing
        assert status()['selected_configuration'] == selected
        results.append('mixed update retains ordered invalid/unsupported rows and normalized identity')
        BODY = b'<html>provider login page</html>'
        cli('subscription', 'update', 'Personal', ok=False)
        assert status()['selected_configuration'] == selected
        BODY = b''
        cli('subscription', 'update', 'Personal')
        assert status()['selected_configuration'] is None
        results.append('whole-document failure preserves selection while received empty list clears it')
        BODY = b'trojan://never-print-password@example.com:443#First\n'
        cli('subscription', 'update', '--all')
        output = command('busctl', 'call', 'org.roentgen.Controller1', '/org/roentgen/Controller1', 'org.roentgen.Controller1', 'Configurations', 's', '').stdout
        selected = json.loads(json.loads(output[2:].strip()))[0]['id']
        cli('configuration', 'select', selected)
        error = cli('enable', ok=False).stderr
        assert 'Unsupported Xray release' in error
        assert not status()['requested_enabled']
        results.append('unsupported packaged Xray activation is rejected before enabled intent changes')
        command('systemctl', 'restart', 'roentgen.service')
        assert status()['selected_configuration'] == selected
        results.append('service restart retains explicit selection and disabled intent')
        editor = ROOT / 'check-editor.py'
        editor.write_text('import os,pathlib,sys\np=pathlib.Path(sys.argv[-1])\nassert p.suffix==".txt" and (p.stat().st_mode & 0o777)==0o600\nassert p.stat().st_uid==os.getuid()\nassert sys.argv[1:3]==["--literal","two words"]\npathlib.Path("/var/tmp/roentgen-editor-path").write_text(str(p))\n')
        env = dict(os.environ, EDITOR=f'python3 {editor} --literal "two words"')
        cli('logs', '--editor', user='rustacean', env=env)
        snapshot = Path('/var/tmp/roentgen-editor-path').read_text()
        assert not Path(snapshot).exists()
        results.append('editor receives literal arguments and a user-owned private snapshot that is removed afterward')
        cli('subscription', 'delete', 'Personal')
        assert status()['selected_configuration'] is None
        results.append('subscription deletion clears selection without replacement')
        print(json.dumps({'passed': results}, indent=2))
    finally:
        server.shutdown()
        if policy.exists(): policy.unlink()
        command('systemctl', 'stop', 'roentgen.service', ok=True)


if __name__ == '__main__': main()