import asyncio from io import BytesIO from pathlib import Path import tempfile import unittest import httpx from PIL import Image, ImageDraw from halloween.follow import DEFAULT, FollowService, PCA9685, locate, validate from halloween.web import create_app def frame(color='lime', double=False): image = Image.new('RGB', (640,480), '#181818') draw = ImageDraw.Draw(image) draw.ellipse((440,180,510,250), fill=color) if double: draw.ellipse((100,180,170,250), fill=color) data = BytesIO(); image.save(data, 'JPEG'); return data.getvalue() class Camera: demo = False error = None def __init__(self, clock): self.clock, self.jpeg, self.captured = clock, frame(), clock() def latest_timed(self): return 1, self.jpeg, self.captured class Driver: def __init__(self): self.moves, self.closed = [], False def move(self, pan, tilt): self.moves.append((pan,tilt)) def close(self): self.closed = True class FollowTest(unittest.TestCase): def setUp(self): self.folder = tempfile.TemporaryDirectory(); self.addCleanup(self.folder.cleanup) self.now = 10. self.camera = Camera(lambda:self.now) self.drivers = [] def factory(): driver = Driver(); self.drivers.append(driver); return driver self.follow = FollowService(self.camera, Path(self.folder.name), factory, lambda:self.now) def start(self): return self.follow.begin(True, True) def step(self, target): self.now += .2; self.camera.captured = self.now self.follow.heartbeat(self.follow.session) self.follow.process(2,self.camera.jpeg,self.now,self.follow.revision,target) def test_marker_and_ambiguity(self): target = locate(frame(), DEFAULT) self.assertGreater(target['x'], .7) self.assertIsNone(locate(frame(double=True), DEFAULT)) self.assertIsNone(locate(frame('gray'), DEFAULT)) self.assertIsNone(locate(frame('red'), DEFAULT)) self.assertIsNotNone(locate(frame('red'), dict(DEFAULT,hue=0))) def test_similar_coloured_paper_does_not_hide_saturated_marker(self): image = Image.new('RGB', (640,480), '#181818') draw = ImageDraw.Draw(image) draw.rectangle((200,0,560,330), fill=(125,158,204)) # bluish paper, hue ~152, S ~95 draw.rectangle((250,250,520,330), fill=(120,153,198)) # slightly stronger paper shade draw.ellipse((300,30,350,80), fill=(68,73,177)) data = BytesIO(); image.save(data, 'JPEG'); data = data.getvalue() for tolerance in (10, 18, 35): target = locate(data, dict(DEFAULT, hue=165, tolerance=tolerance)) self.assertIsNotNone(target) self.assertLess(target['y'], .2) def test_photo_selection_and_no_persistent_image(self): shot = self.follow.capture() config = self.follow.select(shot['token'],.74,.45) self.assertLess(abs(config['hue']-85),3) self.assertEqual([p.name for p in Path(self.folder.name).iterdir()], ['follow.json']) with self.assertRaises(ValueError): self.follow.select('old',.74,.45) with self.assertRaises(ValueError): self.follow.select(shot['token'],float('nan'),.5) with self.assertRaises(ValueError): self.follow.select(shot['token'],.2,.2) def test_no_hardware_without_explicit_start_and_no_restart_resume(self): self.follow.begin(False) self.step({'x':.9,'y':.9}) self.assertEqual(self.drivers, []) self.follow.halt() with self.assertRaises(ValueError): self.follow.begin(True,False) self.assertFalse(FollowService(self.camera,Path(self.folder.name)).active) self.assertEqual(self.drivers, []) def test_rate_limits_reverse_deadband_and_stop(self): self.start(); driver=self.drivers[0] self.step({'x':.9,'y':.5}) self.assertAlmostEqual(self.follow.pan,1524) self.assertEqual(self.follow.tilt,1500) self.assertLessEqual(abs(driver.moves[-1][0]-driver.moves[-2][0]),40) for _ in range(30): self.step({'x':1.,'y':1.}) self.assertEqual(self.follow.pan,1600) self.follow.halt(); self.assertTrue(driver.closed) count=len(driver.moves) self.follow.process(99,self.camera.jpeg,self.now,self.follow.revision,{'x':0,'y':0}) self.assertEqual(len(driver.moves),count) self.follow.configure(dict(DEFAULT,pan_reverse=True)) self.start(); self.step({'x':.9,'y':.5}) self.assertLess(self.follow.pan,1500) def test_legacy_speed_migration_preserves_endpoints(self): import json legacy = dict(DEFAULT, pan_min=1234, pan_max=1789) del legacy['follow_speed'] Path(self.folder.name, 'follow.json').write_text(json.dumps(legacy)) restored = FollowService(self.camera, Path(self.folder.name)) self.assertEqual(restored.config, dict(legacy, follow_speed=80)) ancient = {k:v for k,v in legacy.items() if '_setup_' not in k} self.assertEqual(validate(ancient), restored.config) for speed in (39, 801, float('nan'), True): with self.assertRaises(ValueError): validate(dict(DEFAULT, follow_speed=speed)) def test_speed_is_time_based_and_delayed_frame_is_capped(self): config = dict(DEFAULT, pan_min=500, pan_max=2500, pan_setup_min=500, pan_setup_max=2500) for speed in (80, 400, 800): for fps in (15, 20): self.follow.halt() self.follow.configure(dict(config, follow_speed=speed)) self.start() for number in range(fps): self.now += 1/fps; self.camera.captured = self.now self.follow.heartbeat(self.follow.session) self.follow.process(number, self.camera.jpeg, self.now, self.follow.revision, {'x':1., 'y':.5}, 3.2) self.assertAlmostEqual(self.follow.pan, 1500+min(speed, 300)) self.assertEqual(self.follow.status()['fps'], fps) self.assertEqual(self.follow.status()['analysis_ms'], 3.2) before = self.follow.pan self.now += .5; self.camera.captured = self.now self.follow.process(99, self.camera.jpeg, self.now, self.follow.revision, {'x':1., 'y':.5}) self.assertAlmostEqual(self.follow.pan-before, min(speed, 300)*.1) self.follow.halt() self.assertEqual(self.follow.status()['fps'], 0) def test_search_is_opt_in_and_requires_prior_target_and_real_session(self): legacy = dict(DEFAULT); del legacy['search_enabled'] self.assertFalse(validate(legacy)['search_enabled']) with self.assertRaises(ValueError): validate(dict(DEFAULT, search_enabled=1)) self.follow.configure(dict(DEFAULT, search_enabled=True)) self.assertTrue(FollowService(self.camera,Path(self.folder.name)).config['search_enabled']) self.start() for _ in range(6): self.step(None) self.assertTrue(self.follow.active) # no blind search on start self.assertTrue(self.follow.waiting_for_target) self.follow.halt() self.follow.begin(False) self.step({'x':.7,'y':.7}) for _ in range(10): self.step(None) self.assertFalse(self.follow.status()['searching']) self.assertEqual(len(self.drivers),1) # preview creates no driver def test_search_sweep_limits_speed_and_finite_completion(self): for reverse in (False, True): self.follow.halt() self.follow.configure(dict(DEFAULT,search_enabled=True,pan_reverse=reverse, tilt_reverse=reverse,pan_min=1420,pan_max=1580,tilt_min=1430,tilt_max=1570, follow_speed=800)) self.start(); self.step({'x':.8,'y':.8}) driver=self.drivers[-1]; before=self.follow.pan self.step(None); self.step(None) self.assertEqual(self.follow.pan,before) # brief occlusion: hold for _ in range(6): self.step(None) self.assertTrue(self.follow.status()['searching']) self.assertGreater((self.follow.pan-before)*(-1 if reverse else 1),0) start=len(driver.moves)-1 for _ in range(200): if not self.follow.active: break self.step(None) self.assertTrue(self.follow.active) self.assertFalse(driver.closed) self.assertTrue(self.follow.waiting_for_target) for pan,tilt in driver.moves: self.assertTrue(1420 <= pan <= 1580) self.assertTrue(1430 <= tilt <= 1570) for a,b in zip(driver.moves[start:],driver.moves[start+1:]): self.assertLessEqual(abs(a[0]-b[0]),20.0001) self.assertLessEqual(abs(a[1]-b[1]),20.0001) self.assertEqual({p for p,t in driver.moves} & {1420,1580},{1420,1580}) self.assertEqual({t for p,t in driver.moves} & {1430,1570},{1430,1570}) def test_search_reacquisition_holds_then_resumes_and_new_loss_searches_again(self): self.follow.configure(dict(DEFAULT,search_enabled=True)) self.start(); self.step({'x':.7,'y':.5}) for _ in range(8): self.step(None) before=(self.follow.pan,self.follow.tilt) self.step({'x':.3,'y':.5}) self.assertEqual((self.follow.pan,self.follow.tilt),before) self.assertTrue(self.follow.status()['reacquiring']) self.step(None) # isolated detection must not end search self.assertFalse(self.follow.status()['reacquiring']) self.step({'x':.5,'y':.5}); self.step({'x':.5,'y':.5}) self.assertFalse(self.follow.status()['searching']) self.assertTrue(self.follow.active) for _ in range(8): self.step(None) self.assertTrue(self.follow.status()['searching']) self.follow.halt() self.assertFalse(self.follow.status()['searching']) self.start() self.assertIsNone(self.follow.last_target) def test_search_timeout_locked_axes_and_failure_watchdogs(self): self.follow.configure(dict(DEFAULT,search_enabled=True,follow_speed=40, pan_min=500,pan_max=2500,pan_setup_min=500,pan_setup_max=2500, tilt_min=1500,tilt_max=1500)) self.start(); self.step({'x':.9,'y':.8}) for _ in range(310): if not self.follow.active: break self.step(None) self.assertTrue(self.follow.active) self.assertTrue(self.follow.waiting_for_target) self.assertFalse(self.follow.status()['searching']) self.follow.halt() self.assertTrue(all(t==1500 for p,t in self.drivers[-1].moves)) for reason in ('camera','heartbeat','inference'): self.camera.captured=self.now self.start(); self.step({'x':.9,'y':.5}) for _ in range(8): self.step(None) self.assertTrue(self.follow.status()['searching']) self.now+=2 if reason!='camera': self.camera.captured=self.now if reason=='inference': self.follow.heartbeat(self.follow.session) self.follow.watchdog() self.assertFalse(self.follow.active,reason) self.assertTrue(self.drivers[-1].closed,reason) def test_missing_target_waits_indefinitely_and_reacquires_without_restarting(self): self.follow.configure(dict(DEFAULT,hue=179)) session=self.start() self.step({'x':.7,'y':.5}) driver=self.drivers[-1]; moves=len(driver.moves) for _ in range(900): self.step(None) # three minutes, healthy connection self.assertTrue(self.follow.active) self.assertEqual(self.follow.session,session) self.assertEqual(len(driver.moves),moves) self.assertTrue(self.follow.status()['waiting_for_target']) self.step({'x':.3,'y':.5}) self.assertEqual(len(driver.moves),moves) self.step({'x':.3,'y':.5}) self.assertFalse(self.follow.waiting_for_target) self.assertEqual(len(driver.moves),moves+1) self.assertEqual(self.follow.config['hue'],179) self.follow.halt() restored=FollowService(self.camera,Path(self.folder.name)) self.assertFalse(restored.active) self.assertEqual(restored.config['hue'],179) self.follow.configure(dict(self.follow.config,hue=0)) self.assertEqual(FollowService(self.camera,Path(self.folder.name)).config['hue'],0) def test_finished_search_waits_without_repeating_and_reacquires(self): self.follow.configure(dict(DEFAULT,search_enabled=True)) self.start(); self.step({'x':.7,'y':.5}) for _ in range(500): self.step(None) self.assertTrue(self.follow.waiting_for_target) count=len(self.drivers[-1].moves) for _ in range(100): self.step(None) self.assertEqual(len(self.drivers[-1].moves),count) self.step({'x':.3,'y':.5}); self.step({'x':.3,'y':.5}) self.assertFalse(self.follow.waiting_for_target) self.assertTrue(self.follow.active) self.assertGreater(len(self.drivers[-1].moves),count) self.now+=2; self.camera.captured=self.now self.follow.watchdog() self.assertFalse(self.follow.active) self.assertTrue(self.drivers[-1].closed) def test_start_uses_persisted_independent_endpoint_midpoints(self): config = dict(DEFAULT, pan_min=1000, pan_max=1800, pan_setup_min=500, pan_setup_max=2500, tilt_min=1700, tilt_max=2001, tilt_setup_max=2300, pan_reverse=True, tilt_reverse=True) self.follow.configure(config) for operation in ('setup', 'follow'): driver = Driver() service = FollowService(self.camera, Path(self.folder.name), lambda: driver, lambda: self.now) self.assertFalse(service.active) self.assertEqual(driver.moves, []) # restart does not move anything service.begin(True, True, operation) self.assertEqual(driver.moves, [(1400,1851)]) self.assertEqual(service.status()['setup_origin'], {'pan':1400,'tilt':1851}) service.halt() self.follow.configure(dict(config,pan_min=1200,pan_max=1200)) self.start() self.assertEqual(self.drivers[-1].moves, [(1200,1851)]) def test_remember_uses_fixed_session_origin_and_next_start_recalculates(self): self.follow.configure(dict(DEFAULT,pan_min=1700,pan_max=1800)) session = self.follow.begin(True,True,'setup') self.assertEqual(self.follow.pan,1750) for seq in range(1,5): self.follow.jog(session,'left',seq); self.step(None) self.follow.jog(session,None,5) count = len(self.drivers[-1].moves) with self.assertRaises(ValueError): self.follow.remember(session,'right') saved = self.follow.remember(session,'left') self.assertEqual(saved['pan_min'],1728) self.assertEqual(self.follow.status()['setup_origin']['pan'],1750) self.assertEqual(len(self.drivers[-1].moves),count) self.follow.halt(); self.start() self.assertEqual(self.drivers[-1].moves,[(1764,1500)]) def test_stale_results_cannot_move_new_session(self): self.start(); revision=self.follow.revision self.follow.halt(); self.start(); driver=self.drivers[-1] self.follow.process(2,self.camera.jpeg,self.now,revision,{'x':1,'y':1}) self.assertEqual(driver.moves,[(1500,1500)]) def test_lost_marker_heartbeat_and_camera_latch_off(self): for reason in ('heartbeat','camera','inference'): self.follow.halt(); self.camera.captured=self.now self.start(); driver=self.drivers[-1] if reason=='marker': for _ in range(6): self.step(None) else: self.now+=2 if reason!='camera': self.camera.captured=self.now if reason=='inference': self.follow.heartbeat(self.follow.session) self.follow.watchdog() self.assertFalse(self.follow.active, reason) self.assertTrue(driver.closed, reason) self.assertIsNotNone(self.follow.error) def test_invalid_limits_and_corrupt_saved_config(self): for config in (dict(DEFAULT,pan_min=1601), dict(DEFAULT,pan_max=2000), dict(DEFAULT,hue=float('inf')), dict(DEFAULT,pan_reverse='false')): with self.assertRaises(ValueError): validate(config) Path(self.folder.name,'follow.json').write_text('bad') follow=FollowService(self.camera,Path(self.folder.name)) with self.assertRaises(ValueError): follow.begin() def test_custom_setup_envelope_persistence_migration_and_follow_limits(self): old = {k:v for k,v in DEFAULT.items() if '_setup_' not in k} self.assertEqual(validate(old), DEFAULT) config = dict(DEFAULT, pan_setup_min=700, pan_setup_max=2300, tilt_setup_min=1000, tilt_setup_max=2000) self.follow.configure(config) self.assertEqual(FollowService(self.camera,Path(self.folder.name)).config, config) session = self.follow.begin(True,True,'setup') with self.assertRaises(ValueError): self.follow.configure(DEFAULT) for seq in range(1,130): self.follow.jog(session,'right',seq); self.step(None) self.assertEqual(self.follow.pan,2300) self.assertEqual(self.follow.tilt,1500) self.follow.jog(session,None,130) self.assertEqual(self.follow.remember(session,'right')['pan_max'],2290) for seq in range(131,350): self.follow.jog(session,'left',seq); self.step(None) self.assertEqual(self.follow.pan,700) self.follow.jog(session,None,350) self.assertEqual(self.follow.remember(session,'left')['pan_min'],710) self.follow.halt(); self.start() for _ in range(100): self.step({'x':1,'y':1}) self.assertEqual(self.follow.pan,2290) self.assertEqual(self.follow.tilt,1600) for bad in (dict(config,pan_setup_min=499), dict(config,tilt_setup_max=2501), dict(config,pan_min=600), dict(config,tilt_max=2100)): with self.assertRaises(ValueError): validate(bad) def test_close_error_is_visible_and_restart_is_explicit(self): self.start() def bad_close(): raise OSError('I2C ausgefallen') self.drivers[0].close=bad_close self.follow.halt() self.assertFalse(self.follow.active) self.assertIn('Servoversorgung ausschalten',self.follow.error) def test_failed_start_disables_channels_and_demo_rejects_hardware(self): driver=Driver() def bad_move(*args): raise OSError('I2C fehlgeschlagen') driver.move=bad_move self.follow.factory=lambda:driver with self.assertRaises(ValueError): self.start() self.assertTrue(driver.closed) self.assertFalse(self.follow.active) self.camera.demo=True with self.assertRaises(ValueError): self.start() def test_old_heartbeat_cannot_renew_new_session(self): old=self.start(); self.follow.halt(); current=self.start() self.assertNotEqual(old,current) with self.assertRaises(ValueError): self.follow.heartbeat(old) self.now+=2; self.camera.captured=self.now self.follow.watchdog() self.assertFalse(self.follow.active) def test_setup_hold_release_expiry_and_out_of_order_requests(self): session=self.follow.begin(True,True,'setup') driver=self.drivers[-1] self.follow.jog(session,'right',1) self.step(None) self.assertAlmostEqual(self.follow.pan,1508) self.step(None) # no renewal: 0.35 s motion lease has expired self.assertAlmostEqual(self.follow.pan,1508) self.follow.jog(session,'right',2) self.follow.jog(session,None,3) self.assertFalse(self.follow.jog(session,'right',2)['accepted']) self.step(None) self.assertAlmostEqual(self.follow.pan,1508) for _ in range(8): self.step(None) self.assertTrue(self.follow.active) # setup needs no colour marker self.follow.halt() with self.assertRaises(ValueError): self.follow.jog(session,'up',4) self.assertTrue(driver.closed) def test_setup_limits_reverse_and_remember_are_persistent(self): self.follow.configure(dict(DEFAULT,pan_min=1500,pan_max=1500,tilt_reverse=True)) session=self.follow.begin(True,True,'setup') for seq in range(1,5): self.follow.jog(session,'right',seq); self.step(None) self.assertGreater(self.follow.pan,1500) # setup intentionally exceeds saved limits with self.assertRaises(ValueError): self.follow.remember(session,'right') self.follow.jog(session,None,5) c=self.follow.remember(session,'right') self.assertEqual(c['pan_max'],1522) self.assertEqual(FollowService(self.camera,Path(self.folder.name)).config['pan_max'],1522) self.follow.jog(session,'up',6); self.step(None) self.assertGreater(self.follow.tilt,1500) for seq in range(7,80): self.follow.jog(session,'up',seq); self.step(None) self.assertEqual(self.follow.tilt,1900) self.follow.jog(session,None,80) self.assertEqual(self.follow.remember(session,'up')['tilt_max'],1890) def test_center_moves_slowly_to_current_saved_limits_without_saving(self): self.follow.configure(dict(DEFAULT,pan_min=1200,pan_max=1700, tilt_min=1400,tilt_max=1800)) session = self.follow.begin(True,True,'setup') for seq in range(1,11): self.follow.jog(session,'right',seq); self.step(None) self.follow.jog(session,None,11) self.follow.remember(session,'right') for seq in range(12,22): self.follow.jog(session,'up',seq); self.step(None) self.follow.jog(session,None,22) saved = self.follow.path.read_bytes() before = (self.follow.pan,self.follow.tilt) self.assertTrue(self.follow.center(session,23)['accepted']) self.assertEqual((self.follow.pan,self.follow.tilt),before) # no sudden jump with self.assertRaises(ValueError): self.follow.remember(session,'left') self.step(None) self.assertAlmostEqual(self.follow.pan,before[0]-8) self.assertAlmostEqual(self.follow.tilt,before[1]+8) for _ in range(40): self.step(None) self.assertEqual((self.follow.pan,self.follow.tilt),(1360,1600)) self.assertFalse(self.follow.status()['centering']) self.assertEqual(self.follow.path.read_bytes(),saved) self.assertEqual(self.follow.setup_origin,{'pan':1450,'tilt':1600}) def test_center_cancellation_stale_commands_watchdog_and_session(self): with self.assertRaises(ValueError): self.follow.center('none',1) session = self.start() with self.assertRaises(ValueError): self.follow.center(session,1) self.follow.halt(); session = self.follow.begin(True,True,'setup') with self.assertRaises(ValueError): self.follow.center('wrong',1) self.follow.jog(session,'right',1); self.step(None) self.follow.center(session,2) self.follow.jog(session,None,3) stopped = self.follow.pan self.assertFalse(self.follow.center(session,2)['accepted']) self.step(None); self.assertEqual(self.follow.pan,stopped) self.follow.center(session,4) self.follow.jog(session,'right',5); self.step(None) self.assertGreater(self.follow.pan,stopped) self.assertFalse(self.follow.status()['centering']) self.follow.center(session,6) self.now+=2; self.camera.captured=self.now self.follow.watchdog() self.assertFalse(self.follow.active) self.assertFalse(self.follow.status()['centering']) self.assertTrue(self.drivers[-1].closed) with self.assertRaises(ValueError): self.follow.center(session,7) def test_center_api_requires_header_and_setup_session(self): async def scenario(): app=create_app(demo=True,state_dir=Path(self.folder.name)) app.state.follow.camera=self.camera app.state.follow.clock=lambda:self.now h={'X-Halloween-UI':'1'} async with httpx.AsyncClient(transport=httpx.ASGITransport(app),base_url='http://local') as c: self.assertEqual((await c.post('/api/follow/center',json={})).status_code,403) self.assertEqual((await c.post('/api/follow/center',headers=h,json={})).status_code,422) start=await c.post('/api/follow/start',headers=h,json={'operation':'setup'}) session=start.json()['session'] result=await c.post('/api/follow/center',headers=h,json={'session':session,'sequence':1}) self.assertEqual(result.status_code,200) self.assertTrue(result.json()['accepted']) await c.post('/api/follow/stop',headers=h,json={}) self.assertFalse(app.state.follow.status()['centering']) asyncio.run(scenario()) def test_setup_cannot_be_controlled_by_wrong_session_or_follow_mode(self): session=self.start() with self.assertRaises(ValueError): self.follow.jog(session,'up',1) with self.assertRaises(ValueError): self.follow.remember(session,'up') self.follow.halt(); session=self.follow.begin(True,True,'setup') for sequence in (True,-1,1.5): with self.assertRaises(ValueError): self.follow.jog(session,'up',sequence) with self.assertRaises(ValueError): self.follow.jog('wrong','up',1) with self.assertRaises(ValueError): self.follow.remember('wrong','up') self.now+=2; self.camera.captured=self.now with self.assertRaises(ValueError): self.follow.jog(session,'up',1) self.assertFalse(self.follow.active) def test_register_protocol(self): class Bus: def __init__(self,n): self.writes=[]; self.closed=False def write_byte_data(self,a,r,v): self.writes.append((a,r,v)) def write_i2c_block_data(self,a,r,v): self.writes.append((a,r,v)) def close(self): self.closed=True driver=PCA9685(Bus) self.assertEqual(driver.bus.writes[:2],[(0x40,0x09,0x10),(0x40,0x0d,0x10)]) self.assertIn((0x40,0xfe,121),driver.bus.writes) driver.move(1500,1500) self.assertEqual(driver.bus.writes[-2:],[(0x40,6,[0,0,51,1]),(0x40,10,[0,0,51,1])]) driver.close(); self.assertTrue(driver.bus.closed) def test_api_guard_interlock_global_stop_and_reset(self): async def scenario(): app=create_app(demo=True,state_dir=Path(self.folder.name)) app.state.follow.camera=self.camera app.state.follow.clock=lambda:self.now app.state.follow.factory=lambda:Driver() app.state.camera.demo=True h={'X-Halloween-UI':'1'} async with httpx.AsyncClient(transport=httpx.ASGITransport(app),base_url='http://local') as c: self.assertEqual((await c.post('/api/follow/start',json={})).status_code,403) self.assertEqual((await c.post('/api/follow/start',headers=h,json={'hardware':True,'acknowledged':True})).status_code,200) cfg=app.state.detection.get_config(); cfg['enabled']=True self.assertEqual((await c.post('/api/detection/config',headers=h,json=cfg)).status_code,409) await c.post('/api/stop',headers=h) self.assertFalse(app.state.follow.active) await c.post('/api/follow/start',headers=h,json={'hardware':False}) await c.post('/api/reset',headers=h) self.assertFalse(app.state.follow.active) await c.post('/api/detection/config',headers=h,json=cfg) self.assertEqual((await c.post('/api/follow/start',headers=h,json={})).status_code,200) self.assertFalse(app.state.controller.stopped) asyncio.run(scenario()) if __name__=='__main__': unittest.main()