ckochandClaude Opus 5.5
db78b1f2e6
- python3 -m sandtable install-desktop puts a Gyre icon in the app grid and on the desktop (freedesktop .desktop entry with a "Quit Gyre" action). - The icon runs `launch`: starts the web page in the background (no terminal, own session, log in .cache/server.log) if it isn't running, then opens it. A dialog shows the reason if it can't start. - Quit link on the page (local computer only) and `quit`: stop any design, motors off, close the page. - Gyre icon (sandtable/static/gyre.svg), also the page's tab icon. - Simulator takes several lines per tick, so streaming tests aren't slowed. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
302 lines
13 KiB
Python
302 lines
13 KiB
Python
"""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 & 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()
|