carinha-control-panel/tests/test_controle_pc.py

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.limpar()

    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.adicionar('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.ordens(), ['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()