"""Linux host regressions: reject incomplete capture manifests and stale readiness."""
import importlib.util
import hashlib
import io
import json
import os
from pathlib import Path
import sys
import tempfile
import types
import unittest
from unittest.mock import patch


@unittest.skipUnless(sys.platform.startswith('linux'), 'host requires Linux flock')
class HostLeaseTests(unittest.TestCase):
    @classmethod
    def setUpClass(cls):
        configured_source = os.environ.get('VDI_NATIVE_CURSOR_HOST')
        source = Path(configured_source) if configured_source else (
            Path(__file__).resolve().parents[2] / 'terminal/native-cursor/host.py')
        spec = importlib.util.spec_from_file_location('cursor_host', source)
        cls.host = importlib.util.module_from_spec(spec)
        spec.loader.exec_module(cls.host)

    def test_display_restart_cannot_reuse_previous_ready_or_receipt(self):
        h = self.host
        lease = 'a' * 32
        with tempfile.TemporaryDirectory() as directory:
            run = Path(directory)
            (run / 'ready').write_text('b' * 32)
            (run / 'receipt').write_text('nvfbc-v1\n')
            calls = []

            def systemctl(*args):
                calls.append(args)
                if args == ('start', h.GUARD):
                    self.assertFalse((run / 'ready').exists())
                    self.assertFalse((run / 'receipt').exists())
                    self.assertEqual((run / 'lease').read_text(), lease)
                    (run / 'ready').write_text(lease)
                return ''

            cursor = types.SimpleNamespace(size=(2560, 1440), read=lambda: {})
            hello = json.dumps(dict(version=1, lease=lease, size=[2560, 1440]))
            stdin = types.SimpleNamespace(buffer=io.BytesIO((hello + '\nstop\n').encode()))
            stdout = io.StringIO()
            with patch.object(h, 'RUN', run), patch.object(h, 'systemctl', systemctl), \
                 patch.object(h, 'XCursor', lambda: cursor), patch.object(h.sys, 'stdin', stdin), \
                 patch.object(h.sys, 'stdout', stdout), \
                 patch.object(h.select, 'select', lambda *args: ([stdin], [], [])):
                h.stream()
            self.assertEqual(json.loads(stdout.getvalue())['lease'], lease)
            self.assertEqual(calls, [('stop', h.GUARD), ('start', h.GUARD), ('stop', h.GUARD)])
            self.assertFalse((run / 'lease').exists())

    def test_incomplete_manifest_rejected_before_stopping_capture(self):
        h = self.host
        with tempfile.TemporaryDirectory() as directory:
            root = Path(directory)
            (root / 'lease').write_text('a' * 32)
            (root / 'manifest.json').write_text(json.dumps({'sunshine': '0' * 64}))
            with patch.object(h, 'RUN', root), patch.object(h, 'ROOT', root), \
                 patch.object(h, 'systemctl') as systemctl:
                with self.assertRaisesRegex(RuntimeError, 'incomplete pinned capture manifest'):
                    h.guard()
                systemctl.assert_not_called()

    def test_software_sprite_guard_stops_capture_before_unhide_on_failure(self):
        h = self.host
        with tempfile.TemporaryDirectory() as directory:
            root = Path(directory)
            (root / 'lease').write_text('a' * 32)
            manifest = {}
            for name in ('sunshine', 'capture-no-cursor.so'):
                (root / name).write_bytes(name.encode())
                manifest[name] = hashlib.sha256(name.encode()).hexdigest()
            (root / 'manifest.json').write_text(json.dumps(manifest))
            (root / 'hide-server-cursor').touch()
            events = []

            class CursorOwner:
                def __enter__(self):
                    return self

                def hide_server_cursor(self):
                    events.append('hide')

                def __exit__(self, *_):
                    events.append('restore')

            def capture(lease):
                events.append('capture')
                raise RuntimeError('producer failed')

            with patch.object(h, 'ROOT', root), patch.object(h, 'RUN', root), \
                 patch.object(h, 'XCursor', CursorOwner), \
                 patch.object(h, 'guard_capture', capture), \
                 patch.object(h, 'systemctl', lambda *args: events.append(args)):
                with self.assertRaisesRegex(RuntimeError, 'producer failed'):
                    h.guard()
            self.assertEqual(events, ['hide', 'capture', ('stop', h.SERVICE), 'restore'])


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