carinha-control-panel/tests/test_historico.py

"""Histórico de detecções: sensores liga/desliga, intervalo, pedido de foto e rotas HTTP."""
import http.client
import importlib.util
import json
import sys
import tempfile
import threading
import unittest
from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT))
from historico import Historico, MAX_FOTOS

spec = importlib.util.spec_from_file_location('kiosk_hist', ROOT / 'kiosk-server.py')
kiosk = importlib.util.module_from_spec(spec)
spec.loader.exec_module(kiosk)

JPEG = b'\xff\xd8' + b'foto' * 10 + b'\xff\xd9'


class HistoricoTest(unittest.TestCase):
    def setUp(self):
        self.tmp = tempfile.TemporaryDirectory()
        self.h = Historico(self.tmp.name, iniciar_pir=False)
        self.publicados = []
        self.h.publicar = lambda t, d: self.publicados.append((t, d))

    def tearDown(self):
        self.tmp.cleanup()

    def test_intervalo_e_sensor_desligado(self):
        self.assertIsNotNone(self.h.registrar('movimento', 'surpreso', 0.4, agora=100))
        self.assertIsNone(self.h.registrar('movimento', 'surpreso', 0.4, agora=105))   # dentro dos 15 s
        self.assertIsNotNone(self.h.registrar('som', 'surpreso', 0.9, agora=105))       # outro tipo vale
        self.h.config['pir'] = False
        self.assertIsNone(self.h.registrar('pir', agora=200))
        with self.assertRaises(ValueError):
            self.h.registrar('fumaça')
        self.assertEqual(self.h.listar()[1], 2)
        self.assertEqual(self.publicados[0][0], 'historico')

    def test_pedido_e_foto(self):
        ident = self.h.registrar('pir', 'presença', agora=10)
        self.assertEqual(self.h.pedido_atual(), ident)
        with self.assertRaises(ValueError):
            self.h.guardar_foto(ident, b'nao e jpeg')
        self.h.guardar_foto(ident, JPEG)
        self.assertIsNone(self.h.pedido_atual())
        self.assertTrue(self.h.evento(ident)['foto'])
        self.assertTrue((Path(self.tmp.name) / 'fotos' / f'evento-{ident}.jpg').is_file())

    def test_sem_foto_quando_desligado(self):
        self.h.config['foto'] = False
        self.h.registrar('pir', agora=10)
        self.assertIsNone(self.h.pedido_atual())

    def test_guarda_no_maximo_as_fotos_recentes(self):
        self.h.config['intervalo'] = 0
        ids = []
        for i in range(MAX_FOTOS + 3):
            ids.append(self.h.registrar('pir', agora=i))
            self.h.guardar_foto(ids[-1], JPEG)
        com_foto = [e for e in self.h.listar(500)[0] if e['foto']]
        self.assertEqual(len(com_foto), MAX_FOTOS)
        self.assertFalse(self.h.evento(ids[0])['foto'])
        self.assertEqual(len(list((Path(self.tmp.name) / 'fotos').glob('*.jpg'))), MAX_FOTOS)

    def test_apagar_e_limpar(self):
        a = self.h.registrar('pir', agora=1); self.h.guardar_foto(a, JPEG)
        b = self.h.registrar('som', agora=1)
        self.h.apagar(a)
        self.assertIsNone(self.h.evento(a))
        self.h.limpar()
        self.assertEqual(self.h.listar()[1], 0)
        self.assertEqual(list((Path(self.tmp.name) / 'fotos').glob('*.jpg')), [])


class HistoricoHttpTest(unittest.TestCase):
    def setUp(self):
        self.tmp = tempfile.TemporaryDirectory()
        self.server = kiosk.Servidor(('127.0.0.1', 0), kiosk.Handler)
        self.server.token = 'token-historico'
        self.server.historico = Historico(self.tmp.name, iniciar_pir=False)
        threading.Thread(target=self.server.serve_forever, daemon=True).start()

    def tearDown(self):
        self.server.shutdown(); self.server.server_close(); self.tmp.cleanup()

    def pedir(self, metodo, rota, corpo=None, tipo='application/json', token='token-historico'):
        c = http.client.HTTPConnection('127.0.0.1', self.server.server_port, timeout=5)
        h = {'Content-Type': tipo}
        if token:
            h['Authorization'] = 'Bearer ' + token
        if isinstance(corpo, dict):
            corpo = json.dumps(corpo)
        c.request(metodo, rota, body=corpo, headers=h)
        r = c.getresponse(); dados = r.read(); c.close()
        return r.status, dados

    def test_fluxo_completo(self):
        self.assertEqual(self.pedir('POST', '/historico/evento', {'tipo': 'pir'}, token=None)[0], 401)
        codigo, corpo = self.pedir('POST', '/historico/evento', {'tipo': 'movimento', 'evento': 'surpreso', 'nivel': 0.3})
        ident = json.loads(corpo)['id']
        self.assertEqual(json.loads(self.pedir('GET', '/historico/pedido')[1])['id'], ident)
        self.assertEqual(self.pedir('POST', f'/historico/foto?id={ident}', JPEG, 'image/jpeg')[0], 200)
        codigo, foto = self.pedir('GET', f'/historico/foto/evento-{ident}.jpg?t=token-historico', token=None)
        self.assertEqual((codigo, foto), (200, JPEG))
        self.assertEqual(self.pedir('GET', '/historico/foto/../historico.db')[0], 400)
        status = json.loads(self.pedir('GET', '/historico/status')[1])
        self.assertEqual(status['eventos'][0]['id'], ident)
        self.assertTrue(status['eventos'][0]['foto'])
        self.assertEqual(self.pedir('POST', '/historico/config', {'foto': False, 'intervalo': 30})[0], 200)
        self.assertEqual(self.pedir('POST', '/historico/config', {'intervalo': 9999})[0], 400)
        self.assertFalse(self.server.historico.config['foto'])


if __name__ == '__main__':
    unittest.main()