"""Disposable UF2/USB fixtures. Never opens a host serial/block device."""
import errno
import hashlib
import os
from pathlib import Path
import struct
import sys
import tempfile
from types import SimpleNamespace
import unittest
from unittest.mock import patch

sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
import radio_uf2 as uf2
import c01_radio_owner as owner


def firmware(blocks=2):
    result = bytearray()
    for number in range(blocks):
        header = struct.pack('<8I', 0x0A324655, 0x9E5D5157, 0x2000,
                             0x26000 + number * 256, 256, number, blocks, 0xADA52840)
        result += header + bytes([number % 256]) * 256 + bytes(220) + struct.pack('<I', 0x0AB16F30)
    return bytes(result)


class FirmwareTests(unittest.TestCase):
    def test_application_only_image(self):
        raw = firmware()
        value = uf2.validate_image(raw)
        self.assertEqual(value['sha256'], hashlib.sha256(raw).hexdigest())
        self.assertEqual(value['blocks'], 2)
        # Block ordering is not an authority; complete distinct regions are.
        self.assertEqual(uf2.validate_image(raw[512:] + raw[:512])['blocks'], 2)

    def test_corruption_family_reserved_regions_and_missing_blocks(self):
        for offset, value in [(0, 0), (4, 0), (8, 0), (8, 0x2001), (28, 0),
                              (12, 0), (12, 0xF4000), (12, 0x26001), (16, 512),
                              (20, 2), (24, 3), (508, 0), (512 + 20, 0),
                              (512 + 12, 0x26000), (512 + 12, 0x28000)]:
            with self.subTest(offset=offset, value=value):
                raw = bytearray(firmware())
                struct.pack_into('<I', raw, offset, value)
                with self.assertRaises(ValueError):
                    uf2.validate_image(bytes(raw))
        for raw in (b'', firmware()[:-1], firmware()[512:], b'x' * (uf2.MAX_IMAGE + 512)):
            with self.subTest(size=len(raw)), self.assertRaises(ValueError):
                uf2.validate_image(raw)

    def test_bootloader_identity_and_softdevice(self):
        board = b'UF2 Bootloader 0.9.2\r\nBoard-ID: WisBlock-RAK4631-Board\r\n'
        uf2.validate_board_info(board)
        uf2.validate_board_info(board + b'SoftDevice: S140 6.1.1\n')
        for bad in (board.replace(b'RAK4631', b'RAK11300'), board * 2,
                    board + b'SoftDevice: S140 7.3.0\n', board + b'\xff', b'x' * 4097):
            with self.subTest(raw=bad[:60]), self.assertRaises((ValueError, UnicodeError)):
                uf2.validate_board_info(bad)

    def test_copy_short_writes_and_receipt_is_not_verification(self):
        with tempfile.TemporaryFile() as source, tempfile.TemporaryFile() as target:
            source.write(firmware()); source.flush()
            original = os.write
            with patch.object(uf2.os, 'write', side_effect=lambda fd, data: original(fd, data[:37])):
                receipt = uf2.copy_image(source.fileno(), target.fileno())
            target.seek(0)
            self.assertEqual(target.read(), firmware())
            self.assertEqual(receipt['state'], 'written-awaiting-device-verification')
            self.assertNotIn('verified', receipt)

    def test_failure_never_returns_a_receipt(self):
        for failure in (lambda *_: 0, OSError(errno.ENODEV, 'USB disconnected')):
            with tempfile.TemporaryFile() as source, tempfile.TemporaryFile() as target:
                source.write(firmware()); source.flush()
                with patch.object(uf2.os, 'write', side_effect=failure), self.assertRaises((ValueError, OSError)):
                    uf2.copy_image(source.fileno(), target.fileno())
        with tempfile.TemporaryFile() as source, tempfile.TemporaryFile() as target:
            source.write(firmware()); source.flush()
            with patch.object(uf2.os, 'fsync', side_effect=OSError(errno.EIO, 'flush failed')), self.assertRaises(OSError):
                uf2.copy_image(source.fileno(), target.fileno())

    def test_bad_image_and_existing_destination_are_untouched(self):
        with tempfile.TemporaryFile() as source, tempfile.TemporaryFile() as target:
            source.write(b'invalid'); source.flush()
            with patch.object(uf2.os, 'write') as write, self.assertRaises(ValueError):
                uf2.copy_image(source.fileno(), target.fileno())
            write.assert_not_called()
        with tempfile.TemporaryFile() as source, tempfile.TemporaryFile() as target:
            source.write(firmware()); source.flush()
            target.write(b'operator data'); target.flush()
            with self.assertRaises(ValueError):
                uf2.copy_image(source.fileno(), target.fileno())
            target.seek(0)
            self.assertEqual(target.read(), b'operator data')


class OwnerTests(unittest.TestCase):
    def setUp(self):
        master, slave = os.openpty()
        self.addCleanup(os.close, master); self.addCleanup(os.close, slave)
        self.radio = SimpleNamespace(fd=slave, path=os.ttyname(slave),
                                     expected=owner.identity(slave), board='rak4631')
        self.service = owner.Owner.__new__(owner.Owner)
        self.service.radio = self.radio
        self.service.uid = self.service.gid = 53002
        self.service.home = slave  # Job is replaced; no process consumes it.
        self.service.tools = {'esptool': 456, 'rnode': 789}
        self.service.groups = None
        self.service.custody = lambda: None
        self.service.enter = lambda: None
        self.service.admission_end = owner.now() + 2_000_000_000
        self.service.image = None
        self.args = {'action': 'write', 'board': 'rak4631', 'band': '1', 'nonce': 'ab' * 16}

    def load(self, raw):
        self.service.operation('image', {'size': len(raw), 'sha256': hashlib.sha256(raw).hexdigest()})
        self.addCleanup(self.service.clear_image)
        self.service.operation('image-chunk', {'data': raw.hex()})

    def test_owner_grants_only_sealed_image_and_fixed_destination(self):
        self.load(firmware())
        with tempfile.TemporaryFile() as target:
            def execute(groups, argv, fds, env, uid, gid, tool, *args):
                self.assertEqual(argv, ['/usr/local/bin/archy-esptool', '--archy-uf2-owned', self.args['nonce']])
                self.assertEqual(tool, 456)
                self.assertEqual(set(fds), {3, 4, 5, 6})
                import fcntl
                self.assertTrue(fcntl.fcntl(fds[5], fcntl.F_GET_SEALS) & fcntl.F_SEAL_WRITE)
                self.assertEqual(uf2.copy_image(fds[5], fds[6])['board'], 'rak4631')
            with patch.object(uf2, 'open_destination', return_value=os.dup(target.fileno())) as destination, patch.object(owner, 'Job', side_effect=execute):
                self.assertTrue(self.service.operation('run', self.args))
                destination.assert_called_once_with(self.radio.path, self.args['nonce'])
            target.seek(0); self.assertEqual(target.read(), firmware())

    def test_other_actions_wrong_board_and_malformed_image_do_not_touch_drive(self):
        self.load(b'invalid firmware')
        with patch.object(uf2, 'open_destination') as destination, patch.object(owner, 'Job') as job:
            for action in ('install', 'verify', 'write-slow', 'capabilities', 'write'):
                with self.subTest(action=action), self.assertRaises(ValueError):
                    self.service.operation('run', {**self.args, 'action': action})
            with self.assertRaises(ValueError):
                self.service.operation('run', {**self.args, 'board': 'heltecv4'})
            destination.assert_not_called(); job.assert_not_called()

    def test_replaced_serial_path_does_not_grant_volume(self):
        self.load(firmware())
        self.radio.expected[1] += 1
        with patch.object(uf2, 'open_destination') as destination, self.assertRaisesRegex(ValueError, 'appointed radio'):
            self.service.operation('run', self.args)
        destination.assert_not_called()


class VolumeTests(unittest.TestCase):
    def setUp(self):
        self.temp = tempfile.TemporaryDirectory()
        self.addCleanup(self.temp.cleanup)
        self.root = Path(self.temp.name)
        self.usb = self.root / 'devices/1-2'
        self.usb.mkdir(parents=True)
        for name, value in [('idVendor', '239a'), ('idProduct', '0029'), ('serial', 'board-unique')]:
            (self.usb / name).write_text(value + '\n')
        (self.usb / 'tty/ttyACM0').mkdir(parents=True)
        (self.usb / 'block/sda').mkdir(parents=True)
        tty = self.root / 'tty/ttyACM0'; tty.mkdir(parents=True)
        (tty / 'device').symlink_to(self.usb / 'tty/ttyACM0')
        (self.root / 'block').mkdir()
        (self.root / 'block/sda').symlink_to(self.usb / 'block/sda')
        self.volume = self.root / 'RAK volume'; self.volume.mkdir()
        device = self.volume.stat().st_dev
        self.number = f'{os.major(device)}:{os.minor(device)}'
        (self.usb / 'block/sda/dev').write_text(self.number)
        (self.volume / 'INFO_UF2.TXT').write_text('Board-ID: WisBlock-RAK4631-Board\nSoftDevice: S140 6.1.1\n')
        self.mountinfo = self.root / 'mountinfo'
        self.mountinfo.write_text(self.mount_line())
        mapping = {'/sys/class/tty': self.root / 'tty', '/sys/class/block': self.root / 'block',
                   '/proc/self/mountinfo': self.mountinfo}
        self.paths = patch.object(uf2, 'Path', side_effect=lambda value: mapping.get(str(value), Path(value)))
        self.paths.start(); self.addCleanup(self.paths.stop)

    def mount_line(self, number=None):
        path = str(self.volume).replace(' ', '\\040')
        return f'1 0 {number or self.number} / {path} rw - vfat /dev/sda rw\n'

    def test_same_usb_volume_only_and_exclusive_creation(self):
        nonce = 'ab' * 16
        fd = uf2.open_destination('/dev/ttyACM0', nonce)
        os.close(fd)
        with self.assertRaises(FileExistsError):
            uf2.open_destination('/dev/ttyACM0', nonce)
        self.assertTrue((self.volume / ('archy-' + nonce + '.uf2')).exists())

    def test_no_label_search_or_unrelated_volume_fallback(self):
        self.mountinfo.write_text(self.mount_line('999:999'))
        with self.assertRaisesRegex(ValueError, 'mount the RAK4631'):
            uf2.open_boot_volume('/dev/ttyACM0')

    def test_ambiguous_mount_wrong_board_and_symlink_are_refused(self):
        self.mountinfo.write_text(self.mount_line() * 2)
        with self.assertRaises(ValueError): uf2.open_boot_volume('/dev/ttyACM0')
        self.mountinfo.write_text(self.mount_line())
        info = self.volume / 'INFO_UF2.TXT'
        info.write_text('Board-ID: another-board\n')
        with self.assertRaises(ValueError): uf2.open_boot_volume('/dev/ttyACM0')
        other = self.root / 'elsewhere'; other.write_text('Board-ID: WisBlock-RAK4631-Board\n')
        info.unlink(); info.symlink_to(other)
        with self.assertRaises(OSError): uf2.open_boot_volume('/dev/ttyACM0')

    def test_usb_identity_change_is_refused(self):
        original = uf2.usb_identity
        initial = original(self.usb)
        with patch.object(uf2, 'usb_identity', side_effect=[initial, (*initial[:4], 'replacement')]), self.assertRaisesRegex(ValueError, 'changed'):
            uf2.open_boot_volume('/dev/ttyACM0')

    def test_only_native_bootloader_is_admitted(self):
        for path in ('/dev/ttyUSB0', '/dev/serial/by-id/a', '/tmp/ttyACM0'):
            with self.subTest(path=path), self.assertRaises(ValueError): uf2.open_boot_volume(path)
        (self.usb / 'idProduct').write_text('8029')
        with self.assertRaisesRegex(ValueError, 'bootloader mode'): uf2.open_boot_volume('/dev/ttyACM0')


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