144 lines
6.8 KiB
Python
144 lines
6.8 KiB
Python
import importlib.util
|
|
import json
|
|
import tempfile
|
|
import threading
|
|
import unittest
|
|
import urllib.request
|
|
import urllib.error
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(ROOT))
|
|
spec = importlib.util.spec_from_file_location('kiosk', ROOT / 'kiosk-server.py')
|
|
kiosk = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(kiosk)
|
|
|
|
|
|
class ControleTest(unittest.TestCase):
|
|
def test_tracking_controls_are_independent(self):
|
|
c = json.loads(self.request('/app/config')[1])
|
|
self.assertTrue(c.get('rastrearRosto'))
|
|
self.assertTrue(c.get('rastrearMaos'))
|
|
self.assertTrue(c.get('mostrarRastreamento'))
|
|
self.assertEqual(self.request('/app/config', {'seguirRastreamento': False, 'rastrearMaos': False})[0],200)
|
|
c = json.loads(self.request('/app/config')[1])
|
|
self.assertTrue(c['rastrearRosto'])
|
|
self.assertFalse(c['rastrearMaos'])
|
|
|
|
def test_sensor_and_camera_preferences(self):
|
|
c = json.loads(self.request('/app/config')[1])
|
|
self.assertTrue(c.get('sensorMic'))
|
|
self.assertTrue(c.get('sensorCamera'))
|
|
self.assertTrue(c.get('micSono'))
|
|
self.assertFalse(c.get('cameraSono', True))
|
|
self.assertEqual(c.get('videoResolucao'), 'auto')
|
|
self.assertEqual(self.request('/app/config', {'sensorMic': False, 'cameraSono': True,
|
|
'cameraLado': 'environment', 'videoResolucao': '2160', 'videoFPS': 60})[0], 200)
|
|
self.assertEqual(self.request('/app/config', {'cameraLado': 'invalida'})[0], 400)
|
|
self.assertEqual(self.request('/app/config', {'videoFPS': 999})[0], 400)
|
|
|
|
def setUp(self):
|
|
self.tmp = tempfile.TemporaryDirectory()
|
|
self.server = kiosk.Servidor(('127.0.0.1', 0), kiosk.Handler)
|
|
if hasattr(kiosk, 'ControlePC'):
|
|
self.server.controle = kiosk.ControlePC(Path(self.tmp.name) / 'app.json')
|
|
self.thread = threading.Thread(target=self.server.serve_forever, daemon=True)
|
|
self.thread.start()
|
|
self.url = 'http://127.0.0.1:%d' % self.server.server_port
|
|
kiosk.FILA.clear()
|
|
|
|
def tearDown(self):
|
|
self.server.shutdown()
|
|
self.server.server_close()
|
|
self.thread.join()
|
|
self.tmp.cleanup()
|
|
|
|
def request(self, route, data=None, kind='application/json'):
|
|
body = json.dumps(data).encode() if isinstance(data, dict) else data
|
|
req = urllib.request.Request(self.url + route, data=body, headers={'Content-Type': kind})
|
|
try:
|
|
with urllib.request.urlopen(req) as r:
|
|
return r.status, r.read()
|
|
except urllib.error.HTTPError as e:
|
|
return e.code, e.read()
|
|
|
|
def test_config_is_persisted_without_consuming_gestures(self):
|
|
kiosk.FILA.append('feliz')
|
|
code, body = self.request('/app/config', {'volume': 0.7, 'sompausa': 40})
|
|
self.assertEqual(code, 200)
|
|
self.assertEqual(json.loads(body)['config']['volume'], 0.7)
|
|
self.assertEqual(self.request('/app/config')[0], 200)
|
|
self.assertEqual(kiosk.FILA, ['feliz'])
|
|
self.assertEqual(json.loads((Path(self.tmp.name) / 'app.json').read_text())['sompausa'], 40)
|
|
|
|
def test_invalid_configuration_does_not_partially_apply(self):
|
|
code, _ = self.request('/app/config', {'volume': 0.8, 'sompausa': -2})
|
|
self.assertEqual(code, 400)
|
|
code, body = self.request('/app/config')
|
|
self.assertEqual(code, 200)
|
|
self.assertEqual(json.loads(body)['volume'], 0.35)
|
|
|
|
def test_telemetry_tracks_phone_not_panel(self):
|
|
self.assertEqual(self.request('/app/status', {'rosto': 'miudo', 'microfone': 0.42})[0], 200)
|
|
code, body = self.request('/app/status')
|
|
self.assertEqual(code, 200)
|
|
self.assertTrue(json.loads(body)['online'])
|
|
self.assertEqual(json.loads(body)['estado']['rosto'], 'miudo')
|
|
|
|
def test_frame_is_hidden_when_disabled_and_after_expiry(self):
|
|
jpeg = b'\xff\xd8fixture\xff\xd9'
|
|
self.assertEqual(self.request('/camera/frame', jpeg, 'image/jpeg')[0], 409)
|
|
self.request('/app/config', {'preview': True})
|
|
self.assertEqual(self.request('/camera/frame', jpeg, 'image/jpeg')[0], 200)
|
|
self.assertEqual(self.request('/camera/frame'), (200, jpeg))
|
|
self.server.controle.frame_at -= 4
|
|
self.assertEqual(self.request('/camera/frame')[0], 204)
|
|
self.request('/camera/frame', jpeg, 'image/jpeg')
|
|
self.request('/app/config', {'preview': False})
|
|
self.assertEqual(self.request('/camera/frame')[0], 204)
|
|
|
|
def test_rejects_non_jpeg_and_unknown_settings(self):
|
|
self.request('/app/config', {'preview': True})
|
|
self.assertEqual(self.request('/camera/frame', b'hello', 'image/jpeg')[0], 400)
|
|
self.assertEqual(self.request('/app/config', {'preview': 'false'})[0], 400)
|
|
self.assertEqual(self.request('/app/config', {'extra': 1})[0], 400)
|
|
|
|
def test_screen_power_settings_are_saved(self):
|
|
code, body = self.request('/app/config', {'economia': True, 'brilho': 0.65, 'brilhoRepouso': 0.06, 'repousoTela': 30})
|
|
self.assertEqual(code, 200)
|
|
self.assertEqual(json.loads(body)['config']['brilhoRepouso'], 0.06)
|
|
|
|
def test_server_does_not_expose_source_or_signing_key(self):
|
|
self.assertEqual(self.request('/controle_pc.py')[0], 404)
|
|
self.assertEqual(self.request('/android/keystore.jks')[0], 404)
|
|
|
|
def test_mouth_and_eyebrows_are_independent(self):
|
|
code, body = self.request('/app/config', {'boca': False, 'sobrancelhas': True})
|
|
self.assertEqual(code, 200)
|
|
self.assertFalse(json.loads(body)['config']['boca'])
|
|
self.assertTrue(json.loads(body)['config']['sobrancelhas'])
|
|
|
|
def test_rtc_signaling_keeps_messages_until_ack(self):
|
|
self.request('/app/config', {'preview': True})
|
|
code, body = self.request('/rtc/session', {'acao': 'abrir'})
|
|
self.assertEqual(code, 200)
|
|
sid = json.loads(body)['id']
|
|
code, _ = self.request('/rtc/send', {'id': sid, 'de': 'pc', 'mensagem': {'tipo': 'offer', 'sdp': 'example'}})
|
|
self.assertEqual(code, 200)
|
|
code, first = self.request('/rtc/poll?lado=app&desde=0')
|
|
self.assertEqual(code, 200)
|
|
self.assertEqual(json.loads(first)['mensagens'][0]['mensagem']['tipo'], 'offer')
|
|
self.assertEqual(self.request('/rtc/poll?lado=app&desde=0')[1], first)
|
|
self.assertEqual(json.loads(self.request('/rtc/poll?lado=pc&desde=0')[1])['mensagens'], [])
|
|
|
|
def test_rtc_old_session_cannot_inject_into_new_session(self):
|
|
self.request('/app/config', {'preview': True})
|
|
old = json.loads(self.request('/rtc/session', {'acao':'abrir'})[1])['id']
|
|
self.request('/rtc/session', {'acao':'abrir'})
|
|
self.assertEqual(self.request('/rtc/send', {'id':old, 'de':'app', 'mensagem':{'tipo':'answer','sdp':'old'}})[0], 409)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|