"""The web app's library, player and HTTP API, against the GRBL simulator.
Run from the software folder:  python3 -m unittest discover -s tests -v"""
import json
import os
import socket
import sys
import tempfile
import threading
import time
import unittest

os.environ['SANDTABLE_QUIET'] = '1'      # never pop up desktop notifications from the tests
import urllib.request

sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))
sys.path.insert(0, os.path.dirname(__file__))

from sandtable import config, patterns, thr
from sandtable.grbl import Controller
from sandtable.ui import Panel
from sandtable.web import Library, Player, serve
from fakegrbl import FakeGrbl

CFG = config.load('/nonexistent.toml')


def make_library(tmp):
    root = os.path.join(tmp, 'patterns')
    os.makedirs(os.path.join(root, 'spirals'))
    os.makedirs(os.path.join(root, 'flowers'))
    thr.save(os.path.join(root, 'spirals', 'tiny.thr'), patterns.spiral(loops=2))
    thr.save(os.path.join(root, 'flowers', 'rose.thr'), patterns.rose(petals=3))
    return Library(CFG, root=root, cache=os.path.join(tmp, 'cache'), removed=os.path.join(tmp, 'removed'))


def make_controller(speedup):
    fake = FakeGrbl(speedup=speedup, carriage_mm=30.0, switch_at=2.0)
    c = Controller(fake, 'grbl', log=lambda *a: None)
    c.wake(wait=0.3)
    return fake, c


def wait_for(cond, timeout=30):
    t_end = time.monotonic() + timeout
    while time.monotonic() < t_end:
        if cond():
            return True
        time.sleep(0.05)
    return False


class TestLibrary(unittest.TestCase):
    def setUp(self):
        self.tmp = tempfile.TemporaryDirectory()
        self.lib = make_library(self.tmp.name)

    def tearDown(self):
        self.tmp.cleanup()

    def test_items_and_thumbs(self):
        items = self.lib.items()
        self.assertEqual([d['id'] for d in items], ['flowers/rose.thr', 'spirals/tiny.thr'])
        png = self.lib.thumb('spirals/tiny.thr')
        self.assertTrue(os.path.getsize(png) > 1000)
        self.assertGreater(self.lib.items()[1]['est'], 0)      # running time known once the thumbnail is made

    def test_thumb_key_follows_contents_not_dates(self):
        k1 = self.lib._key('spirals/tiny.thr')
        p = self.lib.path('spirals/tiny.thr')
        os.utime(p, (1, 1))                                   # same contents, other date
        self.assertEqual(self.lib._key('spirals/tiny.thr'), k1)
        with open(p, 'a') as f:
            f.write('0 0\n')
        self.assertNotEqual(self.lib._key('spirals/tiny.thr'), k1)

    def test_refuses_paths_outside(self):
        for bad in ('../cache/index.json', '../../tiny.thr', '/etc/passwd', 'spirals/../../x.thr', 'spirals/nope.thr'):
            with self.assertRaises(KeyError):
                self.lib.path(bad)

    def test_upload_and_remove(self):
        ident = self.lib.add('My Design (1).thr', '# sandify\n0 0\n3.14 1\n')
        self.assertEqual(ident, 'uploads/My-Design-1.thr')
        self.assertEqual(self.lib.add('My Design (1).thr', '0 0\n1 1\n'), 'uploads/My-Design-1-2.thr')
        with self.assertRaises(ValueError):
            self.lib.add('junk.thr', 'hello\nworld\n')
        self.lib.remove(ident)
        self.assertNotIn(ident, [d['id'] for d in self.lib.items()])
        self.assertTrue(os.path.exists(os.path.join(self.tmp.name, 'removed', ident)))


class TestPlayer(unittest.TestCase):
    def setUp(self):
        self.tmp = tempfile.TemporaryDirectory()
        self.lib = make_library(self.tmp.name)

    def tearDown(self):
        self.tmp.cleanup()

    def make(self, speedup, cfg=CFG):
        self.fake, self.c = make_controller(speedup)
        self.panel = Panel(self.c, cfg)
        self.alerts = []
        self.player = Player(self.panel, cfg, self.lib, alert=lambda t, m: self.alerts.append(m))
        self.player.speed = 40

    def test_plays_to_the_end(self):
        self.make(100)
        self.player.play(['spirals/tiny.thr'])
        self.assertTrue(wait_for(lambda: not self.player.running(), 60))
        self.assertEqual(self.player.state, 'idle')
        self.assertEqual(self.player.message, '')
        self.assertLessEqual(self.fake.max_rx, 128)
        self.assertTrue(any(l.startswith('$J=') for l in self.fake.lines))     # homed first
        # done: the motors are switched off, so the next design wakes them and homes again
        self.assertTrue(self.fake.sleeping)
        self.assertEqual(self.panel.status()['state'], 'Motors off')
        self.assertFalse(self.panel.homed)
        n_jogs = sum(1 for l in self.fake.lines if l.startswith('$J='))
        self.player.play(['spirals/tiny.thr'])
        self.assertTrue(wait_for(lambda: not self.player.running(), 60))
        self.assertEqual(self.player.message, '')
        self.assertGreater(sum(1 for l in self.fake.lines if l.startswith('$J=')), n_jogs)   # homed again
        self.assertTrue(self.fake.sleeping)
        self.fake.close()

    def test_stop_switches_the_motors_off(self):
        self.make(20)
        self.player.play(['flowers/rose.thr'])
        self.assertTrue(wait_for(lambda: self.player.now and self.player.now['frac'] > 0.02, 60))
        self.player.stop()
        self.assertEqual(self.player.state, 'idle')
        self.assertTrue(self.fake.sleeping)
        self.assertFalse(self.fake.alarm)               # stopped cleanly first, then slept
        self.assertFalse(self.panel.homed)
        self.fake.close()

    def test_stop_keeps_the_position(self):
        import copy
        cfg = copy.deepcopy(CFG)
        cfg['motion']['motors_off_when_done'] = False      # motors left on: no re-homing needed
        self.make(20, cfg)
        self.player.play(['flowers/rose.thr'])
        self.assertTrue(wait_for(lambda: self.player.now and self.player.now['frac'] > 0.02, 60))
        self.player.stop()
        self.assertFalse(self.player.running())
        self.assertEqual(self.player.state, 'idle')
        st = self.c.query()
        self.assertEqual(st.state, 'Idle')
        self.assertFalse(self.fake.alarm)
        real = self.fake.mpos[1] + self.fake.offset_true        # where the carriage really is
        # still homed: rho matches the carriage to within the simulator's homing overshoot. Without
        # putting the G92 offset back after the reset it would be ~30 mm out.
        self.assertAlmostEqual(st.wpos[1], real, delta=0.2)
        self.assertTrue(self.panel.homed)
        self.fake.close()

    def test_pause_resume_and_skip(self):
        self.make(20)
        self.player.play(['flowers/rose.thr', 'spirals/tiny.thr'])
        self.assertTrue(wait_for(lambda: self.player.now and self.player.now['name'] == 'rose' and self.player.now['frac'] > 0.01, 60))
        self.player.pause()
        self.assertEqual(self.player.state, 'paused')
        time.sleep(0.3)
        y = self.c.status.mpos[1]
        time.sleep(0.3)
        self.assertAlmostEqual(self.c.status.mpos[1], y, places=3)     # held still
        self.player.resume()
        self.player.skip()
        self.assertTrue(wait_for(lambda: self.player.now and self.player.now['name'] == 'tiny', 30))
        self.player.stop()
        self.assertEqual(self.player.state, 'idle')
        self.fake.close()

    def test_switch_trip_is_reported(self):
        self.make(20)
        self.player.play(['flowers/rose.thr'])
        self.assertTrue(wait_for(lambda: self.player.now and self.player.now['frac'] > 0.05, 60))
        with self.fake.lock:
            self.fake.switch_at = 1000.0                              # the switch closes everywhere
        self.assertTrue(wait_for(lambda: not self.player.running(), 30))
        self.assertIn('home switch', self.player.message)
        self.assertEqual(len(self.alerts), 1)
        self.assertFalse(self.panel.homed)
        self.fake.close()


class TestReconnect(unittest.TestCase):
    def setUp(self):
        self.tmp = tempfile.TemporaryDirectory()
        self.lib = make_library(self.tmp.name)

    def tearDown(self):
        self.tmp.cleanup()

    def test_cable_out_and_back_in(self):
        fake, c = make_controller(20)
        fresh = []

        def opener(cfg):                          # "plugging back in" gives a new simulated Uno
            if not fresh:
                raise OSError('the Uno is not connected: plug in its USB cable (it reconnects by itself)')
            return fresh[-1]
        panel = Panel(c, CFG, opener=opener)
        player = Player(panel, CFG, self.lib)
        player.play(['flowers/rose.thr'])
        self.assertTrue(wait_for(lambda: player.now and player.now['frac'] > 0.02, 60))
        fake.unplugged = True
        self.assertTrue(wait_for(lambda: not player.running(), 10))
        self.assertIn('USB connection', player.message)
        self.assertTrue(wait_for(lambda: not panel.connected, 5))
        self.assertFalse(panel.homed)
        self.assertTrue(wait_for(lambda: 'not connected' in panel.error, 5))      # trying to reconnect
        new = FakeGrbl(speedup=20, carriage_mm=30.0)
        fresh.append(new)
        self.assertTrue(wait_for(lambda: panel.connected, 10))
        self.assertEqual(panel.error, '')
        self.assertEqual(c.query().state, 'Idle')
        player.play(['spirals/tiny.thr'])            # homes again, then draws
        self.assertTrue(wait_for(lambda: player.now and player.now['name'] == 'tiny', 60))
        self.assertTrue(panel.homed)
        player.stop()
        self.assertTrue(new.sleeping)                 # motors off after the stop
        new.close()
        fake.close()


class TestQuit(unittest.TestCase):
    def test_quit_closes_the_page_and_rests_the_motors(self):
        with tempfile.TemporaryDirectory() as tmp:
            lib = make_library(tmp)
            fake, c = make_controller(100)
            s = socket.socket(); s.bind(('127.0.0.1', 0)); port = s.getsockname()[1]; s.close()
            th = threading.Thread(target=serve, args=(c, CFG), kwargs={'port': port, 'open_browser': False, 'library': lib},
                                  daemon=True)
            th.start()
            from sandtable import launcher
            self.assertTrue(wait_for(lambda: launcher.status(port) is not None, 10))
            self.assertTrue(launcher.status(port)['can_quit'])
            launcher.quit_gyre(port)
            th.join(10)
            self.assertFalse(th.is_alive())                  # the page has closed
            self.assertIsNone(launcher.status(port))
            self.assertTrue(fake.sleeping)                   # motors off
            fake.close()


class TestHttp(unittest.TestCase):
    @classmethod
    def setUpClass(cls):
        cls.tmp = tempfile.TemporaryDirectory()
        lib = make_library(cls.tmp.name)
        cls.fake, c = make_controller(100)
        s = socket.socket()
        s.bind(('127.0.0.1', 0))
        cls.port = s.getsockname()[1]
        s.close()
        threading.Thread(target=serve, args=(c, CFG), kwargs={'port': cls.port, 'open_browser': False, 'library': lib},
                         daemon=True).start()
        assert wait_for(lambda: cls._up(), 10)

    @classmethod
    def _up(cls):
        try:
            urllib.request.urlopen(f'http://127.0.0.1:{cls.port}/api/status', timeout=1)
            return True
        except OSError:
            return False

    def get(self, path):
        with urllib.request.urlopen(f'http://127.0.0.1:{self.port}{path}', timeout=10) as r:
            return r.read()

    def post(self, path, body=None, raw=None):
        data = raw if raw is not None else json.dumps(body or {}).encode()
        req = urllib.request.Request(f'http://127.0.0.1:{self.port}/api/{path}', data=data, method='POST')
        with urllib.request.urlopen(req, timeout=40) as r:
            return json.loads(r.read())

    def test_pages_and_play(self):
        self.assertIn(b'Gyre', self.get('/'))
        self.assertIn(b'jog &amp; calibrate', self.get('/jog'))
        lib = json.loads(self.get('/api/library'))
        self.assertEqual(len(lib), 2)
        self.assertTrue(self.get('/thumb?id=spirals/tiny.thr').startswith(b'\x89PNG'))
        r = self.post('play', {'ids': ['spirals/tiny.thr']})
        self.assertTrue(r['ok'], r)
        self.assertFalse(self.post('jog', {'axis': 'Y', 'd': 1})['ok'])      # refused while playing
        self.assertTrue(wait_for(lambda: json.loads(self.get('/api/status'))['player']['state'] == 'idle', 60))
        st = json.loads(self.get('/api/status'))
        self.assertEqual(st['machine']['state'], 'Motors off')
        self.assertEqual(st['player']['message'], '')
        self.assertFalse(self.post('play', {'ids': ['../../etc/passwd']})['ok'])
        up = self.post('upload?name=new.thr', raw=b'0 0\n6.28 1\n')
        self.assertEqual(up['msg'], 'uploads/new.thr')
        self.assertFalse(json.loads(self.get('/api/status'))['power_buttons'])      # off unless on the Pi
        self.assertFalse(self.post('poweroff')['ok'])


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