import json
import os
import socket
import subprocess
import sys
import tempfile
import threading
import unittest
from pathlib import Path
from unittest.mock import patch

from hydra.profiles import Profile, save_profile
from hydra.runtime import Runtime, RuntimeError as VMError, build_qemu_argv
from test_qmp import QMPServer, greeting, send


# External protocol fixtures isolate lifecycle tests from user VMs and real disks.
FAKE_QEMU = r'''#!/usr/bin/env python3
import json, os, socket, sys
args = sys.argv[1:]
def val(flag): return args[args.index(flag)+1]
if os.environ.get('HYDRA_TEST_FAIL'):
    print('intentional boot failure', file=sys.stderr)
    sys.exit(1)
path = val('-chardev').split('path=', 1)[1].split(',server=', 1)[0].replace(',,', ',')
pid = os.fork()
if pid:
    sys.exit(0)
os.setsid()
with open(val('-pidfile'), 'w') as f: f.write(str(os.getpid()))
s = socket.socket(socket.AF_UNIX)
s.bind(path)
s.listen(8)
alive = True
while alive:
    c, _ = s.accept()
    try:
        with c:
            c.sendall(b'{"QMP":{"version":{"qemu":{"major":10,"minor":0,"micro":0},"package":""},"capabilities":[]}}\n')
            with c.makefile('rb') as stream:
                for line in stream:
                    cmd = json.loads(line)
                    name = cmd['execute']
                    result = {'query-name': {'name': val('-name')}, 'query-uuid': {'UUID': val('-uuid')},
                              'query-status': {'status': 'running', 'running': True, 'singlestep': False}}.get(name, {})
                    c.sendall(json.dumps({'return': result, 'id': cmd['id']}).encode()+b'\n')
                    if name == 'quit':
                        alive = False
                        break
    except (BrokenPipeError, ConnectionResetError):
        pass
s.close()
os.unlink(path)
'''

FAKE_IMG = r'''#!/usr/bin/env python3
import sys
with open(sys.argv[-2], 'xb') as f: f.write(b'new qcow2 fixture')
'''


class RuntimeTests(unittest.TestCase):
    def setUp(self):
        self.tmp = tempfile.TemporaryDirectory(prefix='hr-')
        self.addCleanup(self.tmp.cleanup)
        self.root = Path(self.tmp.name)
        self.runtime = Runtime(self.root / 'config space,comma', self.root / 'state space,comma',
                               self.root / 'run,comma', boot_timeout=1)
        self.iso = self.root / 'installer space,comma.iso'
        self.iso.touch()
        self.profile = Profile('guest', self.iso, graphics='software', disk_size='8M')
        self.runtime.add(self.profile)
        self.qemu = self.root / 'fake qemu'
        self.qemu.write_text(FAKE_QEMU)
        self.qemu.chmod(0o700)
        self.img = self.root / 'fake img'
        self.img.write_text(FAKE_IMG)
        self.img.chmod(0o700)
        env = patch.dict(os.environ, {'HYDRA_QEMU': str(self.qemu), 'HYDRA_QEMU_IMG': str(self.img)})
        env.start()
        self.addCleanup(env.stop)
        self.addCleanup(self.cleanup_guest)

    def cleanup_guest(self):
        try:
            if self.runtime.status('guest')['status'] == 'running':
                self.runtime.stop('guest', force=True)
        except VMError:
            pass

    def test_mac_persisted_distinct_and_stable_across_restarts(self):
        self.runtime.add(Profile('other', network='bridge'))
        other = self.runtime.profile('other')
        guest = self.runtime.profile('guest')
        self.assertIsNotNone(guest.mac)
        self.assertNotEqual(guest.mac, other.mac)
        octet = int(guest.mac.split(':')[0], 16)
        self.assertEqual(octet & 3, 2)  # locally administered, unicast
        def nic(profile, runtime):
            return next(arg for arg in build_qemu_argv(profile, runtime.paths(profile.name))
                        if arg.startswith('virtio-net-pci,'))
        expected = nic(guest, self.runtime)
        self.runtime.start('guest')
        self.runtime.stop('guest', force=True)
        self.runtime.start('guest')
        self.runtime.stop('guest', force=True)
        self.assertEqual(nic(self.runtime.profile('guest'), self.runtime), expected)
        changed_runtime = Runtime(self.runtime.config, self.runtime.state, self.root / 'different-run')
        self.assertEqual(nic(changed_runtime.profile('guest'), changed_runtime), expected)
        self.runtime.media('guest', None)
        self.assertEqual(self.runtime.profile('guest').mac, guest.mac)

    def test_legacy_profiles_gain_stable_owner_specific_argv_mac(self):
        file = self.runtime.config / 'legacy.toml'
        save_profile(file, Profile('legacy', network='bridge'))
        original = file.read_bytes()
        legacy = self.runtime.profile('legacy')
        self.assertIsNone(legacy.mac)
        paths = self.runtime.paths('legacy')
        def mac():
            return next(arg for arg in build_qemu_argv(legacy, paths) if arg.startswith('virtio-net-pci,'))
        with patch('os.getuid', return_value=1000):
            first = mac()
            self.assertEqual(mac(), first)
            self.assertNotEqual(first, next(arg for arg in build_qemu_argv(
                legacy, self.runtime.paths('other')) if arg.startswith('virtio-net-pci,')))
        with patch('os.getuid', return_value=1001):
            self.assertNotEqual(mac(), first)
        self.assertEqual(file.read_bytes(), original)

    def test_disk_boot_omits_media_and_requests_display(self):
        profile = Profile('guest', width=1920, height=1080, refresh_millihz=144000,
                          graphics='software')
        argv = build_qemu_argv(profile, self.runtime.paths('guest'))
        self.assertIn('virtio-vga,xres=1920,yres=1080,refresh_rate=144000', argv)
        self.assertNotIn('ide-cd,drive=iso', argv)
        self.assertIn('order=c', argv)
        devices = [json.loads(argv[i+1]) for i, item in enumerate(argv) if item == '-blockdev']
        self.assertEqual(len(devices), 2)
        stock = build_qemu_argv(Profile('guest', graphics='software'), self.runtime.paths('guest'))
        self.assertFalse(any('refresh_rate=' in arg for arg in stock))

    def test_missing_iso_is_error_at_start_without_creating_disk(self):
        self.iso.unlink()
        with self.assertRaisesRegex(VMError, 'ISO'):
            self.runtime.start('guest')
        self.assertFalse(self.runtime.paths('guest').disk.exists())
        self.assertEqual(self.runtime.status('guest')['status'], 'stopped')

    def test_media_changes_only_while_stopped_and_preserves_disk(self):
        paths = self.runtime.paths('guest')
        self.runtime.start('guest')
        disk = paths.disk.read_bytes()
        config = (self.runtime.config / 'guest.toml').read_bytes()
        for iso in (None, self.iso):
            with self.assertRaises(VMError):
                self.runtime.media('guest', iso)
        self.assertEqual((self.runtime.config / 'guest.toml').read_bytes(), config)
        self.iso.unlink()
        self.assertEqual(self.runtime.profile('guest').iso, self.iso)
        self.runtime.stop('guest', force=True)
        self.runtime.media('guest', None)
        self.assertIsNone(self.runtime.profile('guest').iso)
        self.runtime.start('guest')
        self.runtime.stop('guest', force=True)
        self.assertEqual(paths.disk.read_bytes(), disk)
        replacement = self.root / 'replacement.iso'
        replacement.touch()
        self.runtime.media('guest', replacement)
        self.assertEqual(self.runtime.profile('guest').iso, replacement)
        with self.assertRaisesRegex(VMError, 'ISO'):
            self.runtime.media('guest', self.iso)
        self.assertEqual(self.runtime.profile('guest').iso, replacement)

    def test_argv_is_private_and_comma_safe(self):
        paths = self.runtime.paths('guest')
        argv = build_qemu_argv(self.profile, paths)
        self.assertEqual(argv[0], str(self.qemu))
        blockdevs = [json.loads(argv[i+1]) for i, item in enumerate(argv) if item == '-blockdev']
        self.assertIn(str(paths.disk), [b.get('filename') for b in blockdevs])
        self.assertIn(str(self.iso), [b.get('filename') for b in blockdevs])
        spice = argv[argv.index('-spice')+1]
        self.assertIn('unix=on', spice)
        self.assertIn('disable-copy-paste=on', spice)
        self.assertIn('disable-agent-file-xfer=on', spice)
        self.assertIn('run,,comma', spice)
        self.assertIn('virtio-blk-pci,drive=disk', argv)
        self.assertTrue(any(arg.startswith('virtio-net-pci,netdev=net,mac=') for arg in argv))
        self.assertNotIn('-vnc', argv)
        self.assertNotIn('-virtfs', argv)

    def test_socket_removed_during_stat_is_safely_not_listening(self):
        paths = self.runtime.paths('guest')
        paths.qmp.parent.mkdir(parents=True)
        with socket.socket(socket.AF_UNIX) as fixture:
            fixture.bind(str(paths.qmp))
            original_lstat = Path.lstat
            def disappearing_socket(path):
                if path == paths.qmp:
                    path.unlink(missing_ok=True)
                return original_lstat(path)
            with patch.object(Path, 'lstat', disappearing_socket):
                self.assertFalse(self.runtime._listening(paths.qmp))

    def test_argv_provides_audio_and_manual_usb_channels(self):
        argv = build_qemu_argv(self.profile, self.runtime.paths('guest'))
        self.assertIn('spice,id=audio', argv)
        self.assertIn('hda-duplex,audiodev=audio', argv)
        for channel in range(3):
            self.assertIn(f'spicevmc,id=usbredir{channel},name=usbredir', argv)
            self.assertIn(f'usb-redir,chardev=usbredir{channel}', argv)
        self.assertFalse(any('usb-host' in argument for argument in argv))

    def test_metadata_symlink_does_not_overwrite_user_data(self):
        paths = self.runtime.paths('guest')
        self.runtime.prepare(paths)
        valuable = self.root / 'valuable data'
        valuable.write_text('preserve me')
        paths.metadata.symlink_to(valuable)
        self.runtime.start('guest')
        self.assertEqual(valuable.read_text(), 'preserve me')
        self.assertFalse(paths.metadata.is_symlink())

    def test_disk_symlink_never_changes_its_target(self):
        paths = self.runtime.paths('guest')
        self.runtime.prepare(paths)
        valuable = self.root / 'valuable disk'
        valuable.write_text('preserve me')
        paths.disk.symlink_to(valuable)
        with self.assertRaises(VMError):
            self.runtime.start('guest')
        self.assertEqual(valuable.read_text(), 'preserve me')

    def test_malformed_identity_record_is_safe_unknown(self):
        paths = self.runtime.paths('guest')
        self.runtime.prepare(paths)
        paths.metadata.write_text('[]')
        server = QMPServer(paths.qmp, greeting)
        self.addCleanup(server.close)
        self.assertEqual(self.runtime.status('guest')['status'], 'unknown')

    def test_foreign_qmp_name_blocks_control_even_when_uuid_matches(self):
        paths = self.runtime.paths('guest')
        self.runtime.prepare(paths)
        paths.metadata.write_text(json.dumps({'name': 'guest', 'uuid': paths.uuid}))
        commands = []
        def handler(conn):
            greeting(conn)
            for line in conn.makefile('rb'):
                req = json.loads(line)
                commands.append(req['execute'])
                result = {'query-name': {'name': 'other'}, 'query-uuid': {'UUID': paths.uuid}}.get(req['execute'], {})
                send(conn, {'return': result, 'id': req['id']})
        server = QMPServer(paths.qmp, handler)
        self.addCleanup(server.close)
        with self.assertRaises(VMError):
            self.runtime.stop('guest', force=True)
        self.assertNotIn('quit', commands)

    def test_reused_pid_without_qmp_never_controls_a_process(self):
        paths = self.runtime.paths('guest')
        self.runtime.prepare(paths)
        paths.pid.write_text(str(os.getpid()))
        self.assertEqual(self.runtime.status('guest')['status'], 'stopped')
        with patch('os.kill', side_effect=AssertionError('must not signal any PID')):
            with self.assertRaises(VMError):
                self.runtime.stop('guest', force=True)

    def test_malformed_query_result_is_safe_unknown(self):
        paths = self.runtime.paths('guest')
        self.runtime.prepare(paths)
        paths.metadata.write_text(json.dumps({'name': 'guest', 'uuid': paths.uuid}))
        def handler(conn):
            greeting(conn)
            for line in conn.makefile('rb'):
                req = json.loads(line)
                send(conn, {'return': [], 'id': req['id']})
        server = QMPServer(paths.qmp, handler)
        self.addCleanup(server.close)
        self.assertEqual(self.runtime.status('guest')['status'], 'unknown')

    def test_cli_qmp_failure_reports_error_without_traceback(self):
        paths = self.runtime.paths('guest')
        self.runtime.prepare(paths)
        paths.metadata.write_text(json.dumps({'name': 'guest', 'uuid': paths.uuid}))
        def handler(conn):
            greeting(conn)
            for line in conn.makefile('rb'):
                req = json.loads(line)
                result = {'query-name': {'name': 'guest'}, 'query-uuid': {'UUID': paths.uuid}}.get(req['execute'], {})
                if req['execute'] == 'system_powerdown':
                    send(conn, {'error': {'class': 'GenericError', 'desc': 'cannot power down'}, 'id': req['id']})
                else:
                    send(conn, {'return': result, 'id': req['id']})
        server = QMPServer(paths.qmp, handler)
        self.addCleanup(server.close)
        run = subprocess.run([sys.executable, 'bin/hydra', '--config', str(self.runtime.config),
                              '--state', str(self.runtime.state), '--runtime', str(self.runtime.runtime),
                              'shutdown', 'guest'], capture_output=True, text=True, timeout=4)
        self.assertEqual(run.returncode, 1)
        self.assertIn('cannot power down', run.stderr)
        self.assertNotIn('Traceback', run.stderr)

    def test_disk_and_profile_survive_shutdown_and_force(self):
        paths = self.runtime.paths('guest')
        paths.disk.parent.mkdir(parents=True, exist_ok=True)
        paths.disk.write_bytes(b'existing valuable disk')
        self.runtime.start('guest')
        self.assertEqual(self.runtime.status('guest')['status'], 'running')
        self.runtime.shutdown('guest')
        self.assertEqual(self.runtime.status('guest')['status'], 'running')
        with self.assertRaises(VMError):
            self.runtime.stop('guest', force=False)
        self.runtime.stop('guest', force=True)
        self.assertEqual(self.runtime.status('guest')['status'], 'stopped')
        self.assertEqual(paths.disk.read_bytes(), b'existing valuable disk')

    def test_simultaneous_start_only_launches_one_vm(self):
        barrier = threading.Barrier(2)
        results = []
        def start():
            barrier.wait()
            try:
                results.append(self.runtime.start('guest')['status'])
            except VMError as exc:
                results.append(str(exc))
        threads = [threading.Thread(target=start) for _ in range(2)]
        for thread in threads:
            thread.start()
        for thread in threads:
            thread.join(4)
            self.assertFalse(thread.is_alive())
        self.assertEqual(results.count('running'), 1)
        self.assertEqual(len(results), 2)
        self.assertEqual(self.runtime.paths('guest').disk.read_bytes(), b'new qcow2 fixture')

    def test_stale_socket_is_replaced_only_when_no_listener(self):
        paths = self.runtime.paths('guest')
        self.runtime.prepare(paths)
        stale = socket.socket(socket.AF_UNIX)
        stale.bind(str(paths.qmp))
        stale.close()
        self.runtime.start('guest')
        self.assertEqual(self.runtime.status('guest')['status'], 'running')

    def test_failed_boot_is_reported_and_disk_preserved(self):
        with patch.dict(os.environ, {'HYDRA_TEST_FAIL': '1'}), self.assertRaises(VMError):
            self.runtime.start('guest')
        self.assertEqual(self.runtime.status('guest')['status'], 'stopped')
        self.assertEqual(self.runtime.paths('guest').disk.read_bytes(), b'new qcow2 fixture')
        self.assertIn('intentional boot failure', self.runtime.paths('guest').log.read_text())

    def test_foreign_qmp_uuid_blocks_all_controls_and_restart(self):
        paths = self.runtime.paths('guest')
        self.runtime.prepare(paths)
        paths.metadata.write_text(json.dumps({'name': 'guest', 'uuid': paths.uuid}))
        paths.pid.write_text(str(os.getpid()))
        commands = []
        def handler(conn):
            greeting(conn)
            for line in conn.makefile('rb'):
                req = json.loads(line)
                commands.append(req['execute'])
                result = {'query-name': {'name': 'guest'}, 'query-uuid': {'UUID': 'foreign-uuid'}}.get(req['execute'], {})
                send(conn, {'return': result, 'id': req['id']})
        server = QMPServer(paths.qmp, handler)
        self.addCleanup(server.close)
        self.assertEqual(self.runtime.status('guest')['status'], 'unknown')
        for control in (lambda: self.runtime.shutdown('guest'), lambda: self.runtime.stop('guest', force=True),
                        lambda: self.runtime.start('guest')):
            with self.assertRaises(VMError):
                control()
        self.assertNotIn('quit', commands)
        self.assertNotIn('system_powerdown', commands)
        self.assertFalse(paths.disk.exists())

    def test_private_runtime_and_socket_length(self):
        self.runtime.prepare(self.runtime.paths('guest'))
        self.assertEqual(self.runtime.runtime.stat().st_mode & 0o777, 0o700)
        with self.assertRaises(VMError):
            Runtime(self.root / 'c', self.root / 's', self.root / ('r' * 100)).paths('guest')

    def test_cli_lifecycle_and_global_paths(self):
        base = [sys.executable, 'bin/hydra', '--config', str(self.runtime.config), '--state', str(self.runtime.state),
                '--runtime', str(self.runtime.runtime)]
        for args, expected in [(['list'], 'guest'), (['start', 'guest'], 'running'),
                               (['status', 'guest'], 'running'), (['stop', 'guest', '--force'], 'stopped')]:
            run = subprocess.run(base + args, text=True, capture_output=True, timeout=4)
            self.assertEqual(run.returncode, 0, run.stderr)
            self.assertIn(expected, run.stdout)


if __name__ == '__main__':
    unittest.main()