julioandClaude Opus 5.5
fd6a14054d
O HC-SR505 segura a saída em alto ~8 s e rearma a cada movimento; com alguém na frente o pino ficava alto direto e passar a mão não gerava borda nova, então o painel parecia não detectar nada. Agora o gpiomon escuta as duas bordas, o nível inicial vem do gpioget e o status/SSE levam `ativo`. A bolinha do Ao vivo fica acesa enquanto o pino está alto, e o modo presença não manda dormir com o sensor ainda segurando. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
161 lines
7.4 KiB
Python
161 lines
7.4 KiB
Python
"""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.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_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()
|