Private
Public Access
91 lines
4.8 KiB
Python
91 lines
4.8 KiB
Python
"""Separate operating pages and independent controls; no real hardware."""
|
|
import asyncio
|
|
from io import BytesIO
|
|
from pathlib import Path
|
|
import tempfile
|
|
import time
|
|
import unittest
|
|
|
|
import httpx
|
|
from PIL import Image
|
|
from halloween.web import create_app
|
|
|
|
|
|
class PagesTest(unittest.TestCase):
|
|
def test_pages_contain_only_their_own_controls_and_pollers(self):
|
|
async def scenario():
|
|
with tempfile.TemporaryDirectory() as folder:
|
|
app=create_app(demo=True,state_dir=Path(folder))
|
|
async with httpx.AsyncClient(transport=httpx.ASGITransport(app),base_url='http://local') as c:
|
|
root=(await c.get('/')).text
|
|
grave=(await c.get('/grab')).text
|
|
camera=(await c.get('/kamera')).text
|
|
self.assertEqual(root,grave)
|
|
for page in (grave,camera):
|
|
self.assertIn('href="/grab"',page)
|
|
self.assertIn('href="/kamera"',page)
|
|
self.assertIn('Alles stoppen',page)
|
|
self.assertIn('id="events-list"',grave)
|
|
self.assertNotIn('id="follow-panel"',grave)
|
|
self.assertNotIn('/api/follow',grave)
|
|
self.assertIn('id="setup-start"',camera)
|
|
self.assertIn('src="/stream.mjpg"',camera)
|
|
self.assertNotIn('id="detection-enabled"',camera)
|
|
for endpoint in ('/api/events','/api/profiles','/api/detection'):
|
|
self.assertNotIn(endpoint,camera)
|
|
asyncio.run(scenario())
|
|
|
|
def test_preview_and_grave_controls_are_independent_hardware_is_interlocked(self):
|
|
async def scenario():
|
|
with tempfile.TemporaryDirectory() as folder:
|
|
app=create_app(demo=True,state_dir=Path(folder))
|
|
jpeg=BytesIO(); Image.new('RGB',(64,48),'green').save(jpeg,'JPEG')
|
|
class Camera:
|
|
demo=False
|
|
error=None
|
|
def latest_timed(self): return 1,jpeg.getvalue(),time.monotonic()
|
|
class Driver:
|
|
closed=False
|
|
def move(self,*args): pass
|
|
def close(self): self.closed=True
|
|
driver=Driver()
|
|
app.state.follow.camera=Camera()
|
|
app.state.follow.factory=lambda:driver
|
|
h={'X-Halloween-UI':'1'}
|
|
async with httpx.AsyncClient(transport=httpx.ASGITransport(app),base_url='http://local') as c:
|
|
cfg=app.state.detection.get_config(); cfg['enabled']=True
|
|
await c.post('/api/detection/config',headers=h,json=cfg)
|
|
start=await c.post('/api/follow/start',headers=h,json={})
|
|
self.assertEqual(start.status_code,200)
|
|
self.assertFalse(app.state.controller.stopped)
|
|
self.assertTrue(app.state.detection.get_config()['enabled'])
|
|
await c.post('/api/follow/stop',headers=h,json={})
|
|
blocked=await c.post('/api/follow/start',headers=h,json={'hardware':True,'acknowledged':True})
|
|
self.assertEqual(blocked.status_code,409)
|
|
self.assertFalse(app.state.follow.active)
|
|
cfg['enabled']=False
|
|
await c.post('/api/detection/config',headers=h,json=cfg)
|
|
started=await c.post('/api/follow/start',headers=h,json={'hardware':True,'acknowledged':True,'operation':'setup'})
|
|
self.assertEqual(started.status_code,200)
|
|
session=started.json()['session']
|
|
self.assertFalse(app.state.controller.stopped)
|
|
self.assertEqual((await c.post('/api/grave/stop')).status_code,403)
|
|
await c.post('/api/grave/stop',headers=h)
|
|
self.assertTrue(app.state.controller.stopped)
|
|
self.assertTrue(app.state.follow.active)
|
|
await c.post('/api/grave/reset',headers=h)
|
|
app.state.follow.heartbeat(session)
|
|
self.assertFalse(driver.closed)
|
|
await c.post('/api/mode',headers=h,json={'mode':'sanft'})
|
|
self.assertTrue(app.state.follow.active)
|
|
self.assertEqual(app.state.controller.mode,'sanft')
|
|
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)
|
|
self.assertTrue(driver.closed)
|
|
self.assertTrue(app.state.controller.stopped)
|
|
asyncio.run(scenario())
|
|
|
|
if __name__=='__main__': unittest.main()
|