"""
Протокол датчика в xstem-scope.py — на тех же примерах, что и tests/protocol.test.js.
Железо не нужно.  python3 -m unittest discover -s tests -v
"""

import importlib.util
import json
import os
import sys
import unittest

HERE = os.path.dirname(os.path.abspath(__file__))
APP = os.path.join(HERE, '..', 'app')

sys.dont_write_bytecode = True
_spec = importlib.util.spec_from_file_location('xstem_scope', os.path.join(APP, 'xstem-scope.py'))
xs = importlib.util.module_from_spec(_spec)
_spec.loader.exec_module(xs)

with open(os.path.join(HERE, 'vectors.json'), encoding='utf-8') as f:
    V = json.load(f)


class FrameParserTest(unittest.TestCase):
    def test_frames_any_chunking(self):
        v = V['frames']
        data = bytes.fromhex(v['stream_hex'])
        # Порт отдаёт поток кусками произвольной длины: результат не должен от этого зависеть
        for size in (1, 3, 7, 64, len(data)):
            with self.subTest(chunk=size):
                p = xs.FrameParser(v['packet'])
                got = []
                for i in range(0, len(data), size):
                    got += [[fr[0], fr[1]] for fr in p.feed(data[i:i + size])]
                self.assertEqual(got, v['frames'])


class RepliesTest(unittest.TestCase):
    def test_info(self):
        for case in V['info']:
            with self.subTest(case['name']):
                self.assertEqual(xs.parse_info(bytes.fromhex(case['reply_hex'])), case['expect'])

    def test_clean(self):
        for case in V['clean']:
            with self.subTest(case['name']):
                self.assertEqual(xs.clean(bytes.fromhex(case['reply_hex'])), case['expect'])


class CommandsTest(unittest.TestCase):
    def test_commands(self):
        for case in V['commands']:
            with self.subTest(case['name']):
                period, packet, cmds = xs.commands(case['settings'])
                self.assertEqual({'period': period, 'packet': packet, 'list': cmds}, case['expect'])


class StubScope:
    """Вместо датчика: программа отвечает на запросы, как с настоящим."""
    info = {'model': 'B202', 'firmware': '3.02', 'serial': '00417'}
    sensor = type('S', (), {'port': '/dev/ttyUSB0'})()
    packet, state, settings, frame_id, problem = 3072, 'stopped', None, 0, None

    def __init__(self):
        self.started = []

    def start(self, s):
        self.started.append(s)
        self.settings = s

    def stop(self):
        pass


class HttpTest(unittest.TestCase):
    """Любой сайт, открытый у ученика, не должен управлять датчиком через localhost."""

    def setUp(self):
        import threading
        from http.server import ThreadingHTTPServer
        self.scope = StubScope()
        self.server = ThreadingHTTPServer(('127.0.0.1', 0), xs.make_handler(self.scope))
        self.port = self.server.server_address[1]
        threading.Thread(target=self.server.serve_forever, daemon=True).start()

    def tearDown(self):
        self.server.shutdown()
        self.server.server_close()

    def post(self, ctype, host=None):
        import http.client
        c = http.client.HTTPConnection('127.0.0.1', self.port, timeout=5)
        headers = {'Content-Type': ctype}
        if host:
            headers['Host'] = host
        c.request('POST', '/api/start', body=json.dumps(V['commands'][0]['settings']), headers=headers)
        r = c.getresponse()
        r.read()
        c.close()
        return r.status

    def test_page_can_start(self):
        self.assertEqual(self.post('application/json', f'localhost:{self.port}'), 200)
        self.assertEqual(self.post('application/json; charset=utf-8', f'127.0.0.1:{self.port}'), 200)
        self.assertEqual(len(self.scope.started), 2)

    def test_foreign_site_cannot(self):
        # «Простой» запрос без OPTIONS, который может послать любая страница
        self.assertEqual(self.post('text/plain', f'localhost:{self.port}'), 403)
        # Подмена DNS: чужое имя указывает на 127.0.0.1
        self.assertEqual(self.post('application/json', f'evil.example:{self.port}'), 403)
        self.assertEqual(self.scope.started, [])

    def test_unplugged_board_answers(self):
        # ESP32 переставили в другой разъём: программа держит мёртвый порт
        class Gone:
            port = '/dev/ttyUSB1'

            def send(self, cmd):
                raise OSError(5, 'Input/output error')
        import http.client
        self.server.RequestHandlerClass = xs.make_handler(self.scope, Gone())
        c = http.client.HTTPConnection('127.0.0.1', self.port, timeout=5)
        c.request('POST', '/api/gen', body='{"cmd": "?"}',
                  headers={'Content-Type': 'application/json', 'Host': f'localhost:{self.port}'})
        r = c.getresponse()
        self.assertEqual(r.status, 503)
        self.assertIn('ESP32', json.loads(r.read())['error'])

    def test_device_file_served(self):
        import urllib.request
        with urllib.request.urlopen(f'http://127.0.0.1:{self.port}/b202.json') as r:
            self.assertEqual(json.load(r)['model'], 'B202')


class FakeSensor:
    """Датчик без железа: на «Пуск» шлёт по кругу поток из vectors.json; unplug() — выдернуть."""

    def __init__(self, serial='00417'):
        import threading
        self.port, self.serial, self.lock = '/dev/ttyUSB0', serial, threading.Lock()
        self.plugged, self.closed, self.configured, self.calls = True, False, [], []
        self.stream = bytes.fromhex(V['frames']['stream_hex'])

    def info(self):
        return {'model': 'B202', 'firmware': '3.02', 'date': '1.12.2011', 'channels': 2,
                'serial': self.serial, 'baud': 921600}

    def command(self, body):
        return '3072' if body == 'PACKET_SIZE=?' else ''

    def configure(self, s):
        self.configured.append(s)
        self.calls.append('configure')
        return xs.commands(s)[:2]

    def start(self):
        self.calls.append('start')

    def stop(self):
        self.calls.append('stop')
        if not self.plugged:
            raise OSError(5, 'Input/output error')

    def read_chunk(self):
        import time
        time.sleep(0.01)
        if not self.plugged:
            raise OSError(5, 'Input/output error')
        return self.stream

    def alive(self):
        return self.plugged

    def close(self):
        self.closed = True

    def unplug(self):
        self.plugged = False


class ReconnectTest(unittest.TestCase):
    """Датчик выдернули посреди опыта: всё остановлено, вставили — «Пуск» работает без перезапуска программы."""

    SETTINGS = dict(V['commands'][0]['settings'], packet=V['frames']['packet'])

    def make(self, first):
        self.queue = []
        scope = xs.Scope(first, reopen=lambda: self.queue.pop(0) if self.queue else None, watch_s=0.05, boot_s=0.1)
        scope.watch()
        return scope

    def wait(self, cond, timeout=3):
        import time
        end = time.time() + timeout
        while time.time() < end:
            if cond():
                return True
            time.sleep(0.02)
        return False

    def test_unplug_during_measurement(self):
        first, again = FakeSensor(), FakeSensor()
        scope = self.make(first)
        scope.start(self.SETTINGS)
        self.assertTrue(self.wait(lambda: scope.state == 'running'))
        first.unplug()
        self.assertTrue(self.wait(lambda: scope.state == 'nosensor'), scope.state)
        self.assertTrue(first.closed)
        self.assertIn('«Пуск»', scope.problem)
        with self.assertRaises(xs.SensorMissing):
            scope.start(self.SETTINGS)
        self.queue.append(again)
        # Вставили: сам опыт не продолжается, ждём «Пуск»
        self.assertTrue(self.wait(lambda: scope.state == 'stopped'), scope.state)
        self.assertIsNone(scope.problem)
        self.assertEqual(again.configured, [])
        frames = scope.frame_id
        scope.start(self.SETTINGS)
        self.assertTrue(self.wait(lambda: scope.state == 'running' and scope.frame_id > frames), scope.state)
        scope.stop()

    def test_other_sensor_gets_its_number(self):
        first, other = FakeSensor('00417'), FakeSensor('00500')
        scope = self.make(first)
        first.unplug()
        self.assertTrue(self.wait(lambda: scope.state == 'nosensor'))
        self.queue.append(other)
        # Номер нового датчика — страница возьмёт его калибровку
        self.assertTrue(self.wait(lambda: scope.info['serial'] == '00500' and scope.state == 'stopped'), scope.state)


class ExperimentsTest(unittest.TestCase):
    """experiments.json правят руками: опыт с настройкой, которой нет на странице, тихо сломает урок."""

    def test_experiments(self):
        import re
        with open(os.path.join(APP, 'experiments.json'), encoding='utf-8') as f:
            exps = json.load(f)['experiments']
        with open(os.path.join(APP, 'index.html'), encoding='utf-8') as f:
            page = f.read()
        options = lambda sel: {int(v) for v in re.findall(r'<option value="(\d+)"', re.search(
            rf'<select id="{sel}">(.*?)</select>', page, re.S).group(1))}
        rolls, zooms = options('rollWin'), options('zoom')
        d = xs.DEVICE
        self.assertEqual(len({e['name'] for e in exps}), len(exps), 'названия опытов повторяются')
        self.assertNotIn('group', exps[0], 'первый опыт — без группы: его страница ставит при открытии')
        for e in exps:
            with self.subTest(e['name']):
                self.assertIn(e['period'], d['periods_us'])
                self.assertTrue(all(g in (0, 1, 2, 3) for g in e['gain']) and len(e['gain']) == 2)
                self.assertTrue(all(a in (0, 1) for a in e['ac']) and len(e['ac']) == 2)
                self.assertIn(e['trig'], ('N', '0', '1'))
                self.assertIn(e['falling'], (0, 1))
                self.assertIsInstance(e['level'], (int, float))
                self.assertIsInstance(e['hint'], str)
                self.assertIn(e.get('zoom', 1), zooms)
                if e.get('roll'):
                    self.assertIn(e['roll'], rolls)
                    self.assertGreaterEqual(e['period'], d['roll_min_period_us'], 'самописец на такой развёртке не включится')
                if e['trig'] != 'N':
                    volts = next(r['volts'] for r in d['ranges'] if r['gain'] == e['gain'][int(e['trig'])])
                    self.assertLessEqual(abs(e['level']), volts, 'уровень синхронизации за пределом канала')


class FilesTest(unittest.TestCase):
    """Файлы страницы перечислены в трёх местах: программа, установка, сайт. Забыть один — сломать класс."""

    def test_lists_agree(self):
        import re
        served = {name for name, _ in xs.STATIC.values()}
        for name in served:
            self.assertTrue(os.path.isfile(os.path.join(APP, name)), name)
        with open(os.path.join(HERE, '..', 'deploy', 'install.sh'), encoding='utf-8') as f:
            install = f.read()
        with open(os.path.join(HERE, '..', 'deploy', 'deploy.sh'), encoding='utf-8') as f:
            deploy = f.read()
        for name in served | {'xstem-scope.py'}:
            self.assertIn(f'$APP/{name}', install, f'install.sh не копирует {name}')
            self.assertIn(f'$APP/{name}', deploy, f'deploy.sh не раздаёт {name}')
        with open(os.path.join(HERE, '..', '..', '..', '..', 'hugo.yaml'), encoding='utf-8') as f:
            hugo = f.read()
        mount = re.search(r'source: firmware/Equipment/xSTEM-Scope/app.*?files:\n((?:\s+- "[^"]+"\n)+)', hugo, re.S)
        site = set(re.findall(r'"([^"]+)"', mount.group(1)))
        self.assertEqual(site, served, 'на сайт (hugo.yaml) монтируются не те файлы, что отдаёт программа')


class DeviceFileTest(unittest.TestCase):
    """b202.json правят руками (калибровка раз в год): ошибка в нём ломает страницу у всего класса."""

    def test_ranges_and_periods(self):
        d = xs.DEVICE
        self.assertEqual(sorted(r['gain'] for r in d['ranges']), [0, 1, 2, 3])
        self.assertEqual(d['periods_us'], sorted(set(d['periods_us'])))
        self.assertTrue(all(isinstance(p, int) and p > 0 for p in d['periods_us']))

    def test_calibration_entries(self):
        cal = xs.DEVICE['calibration']
        entries = {'default': cal['default'], **cal['sensors']}
        for serial, c in entries.items():
            with self.subTest(serial):
                if serial == 'default' or 'mv_per_code' in c:
                    mv = c['mv_per_code']
                    self.assertEqual(len(mv), 4)
                    self.assertTrue(all(isinstance(x, (int, float)) and x > 0 for x in mv))
                if serial == 'default' or 'zero_code' in c:
                    z = c['zero_code']
                    codes = [z] if isinstance(z, (int, float)) else [x for row in z for x in row]
                    if not isinstance(z, (int, float)):
                        self.assertEqual([len(row) for row in z], [4, 4])
                    self.assertTrue(all(1900 < x < 2200 for x in codes), 'ноль далеко от 2048')
                if serial != 'default':
                    self.assertIn('date', c, 'запишите дату калибровки')


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