"""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_guarda_so_os_ultimos_dias(self): agora = 10 * 86400 velho = self.h.registrar('pir', agora=agora - 4 * 86400) self.h.guardar_foto(velho, JPEG) novo = self.h.registrar('pir', agora=agora - 86400) self.assertEqual(self.h.limpar_antigos(agora), 1) # 3 dias por padrão self.assertIsNone(self.h.evento(velho)) self.assertIsNotNone(self.h.evento(novo)) self.assertFalse((Path(self.tmp.name) / 'fotos' / f'evento-{velho}.jpg').exists()) with self.assertRaises(ValueError): self.h.validar({'dias': 0}) def test_modo_presenca(self): chamadas = [] self.h.ao_presenca = lambda: chamadas.append('chegou') self.h.ao_ausencia = lambda: chamadas.append('saiu') self.h.pir['disponivel'] = True self.h.marcar_presenca(1000) self.assertFalse(self.h.checar_ausencia(1000 + 4 * 60)) # 4 min: ainda presente self.assertTrue(self.h.checar_ausencia(1000 + 5 * 60 + 1)) # 5 min: ausente self.assertFalse(self.h.checar_ausencia(1000 + 10 * 60)) # não repete self.h.marcar_presenca(2000) # voltou alguém self.assertEqual(chamadas, ['saiu', 'chegou']) self.h.config['presenca'] = False self.h.marcar_presenca(3000) self.assertFalse(self.h.checar_ausencia(9999)) def test_pir_segurando_conta_como_presenca(self): avisos = [] self.h.publicar = lambda tipo, dados: avisos.append((tipo, dados)) self.h.pir['disponivel'] = True self.h.config['sensibilidade_pir'] = 'alta' # qualquer disparo conta self.h.nivel_pir(True, agora=1000) # subida: evento + bolinha acesa self.assertTrue(self.h.pir['ativo']) self.assertEqual(self.h.status()['eventos'][0]['tipo'], 'pir') self.assertFalse(self.h.checar_ausencia(1000 + 10 * 60)) # pino ainda alto: não dorme self.h.nivel_pir(False, agora=1000 + 10 * 60) # descida: só apaga a bolinha self.assertFalse(self.h.pir['ativo']) self.assertEqual(self.h.status()['total'], 1) self.assertTrue(self.h.checar_ausencia(1000 + 15 * 60 + 1)) self.assertIn(('pir', True), [(t, d.get('ativo')) for t, d in avisos]) def test_sensibilidade_do_pir_confirma_antes_de_contar(self): detectados = [] self.h.ao_detectar = detectados.append self.h.config['intervalo'] = 0 # média (padrão): um disparo solto só acende a bolinha self.h.nivel_pir(True, agora=1000) self.assertTrue(self.h.pir['ativo']) self.h.nivel_pir(False, agora=1008) # o HC-SR505 segura ~8 s self.assertEqual((self.h.status()['total'], detectados), (0, [])) self.h.nivel_pir(True, agora=1040) # segundo disparo em menos de 1 min self.assertEqual(detectados, ['pir']) self.h.nivel_pir(False, agora=1048) self.h.nivel_pir(True, agora=1200) # 2 min depois: solto de novo self.h.nivel_pir(False, agora=1208) self.assertEqual(len(detectados), 1) # alguém se mexendo na frente: o sensor fica ligado direto e confirma self.h.nivel_pir(True, agora=2000) self.h.checar_pir(2010) self.assertEqual(len(detectados), 1) self.h.checar_pir(2012) self.assertEqual(len(detectados), 2) self.h.nivel_pir(False, agora=2030) # baixa: precisa de 3 disparos em 90 s self.h.config['sensibilidade_pir'] = 'baixa' for t in (3000, 3030): self.h.nivel_pir(True, agora=t); self.h.nivel_pir(False, agora=t + 8) self.assertEqual(len(detectados), 2) self.h.nivel_pir(True, agora=3060) self.assertEqual(len(detectados), 3) with self.assertRaises(ValueError): Historico.validar(dict(sensibilidade_pir='maxima')) 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()