"""Servidor base: token, /eventos (SSE), /estado, /comando, E-Stop, trava de hardware e fila.""" 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)) spec = importlib.util.spec_from_file_location('kiosk', ROOT / 'kiosk-server.py') kiosk = importlib.util.module_from_spec(spec) spec.loader.exec_module(kiosk) from braco import Braco from comum import carregar_token from servos import ControleServos TOKEN = 'segredo-de-teste' class HW: def __init__(self): self.calls = [] def pulse(self, ch, us): self.calls.append((ch, us)) def off(self, ch): self.calls.append((ch, None)) class ServidorTest(unittest.TestCase): def setUp(self): self.tmp = tempfile.TemporaryDirectory() pasta = Path(self.tmp.name) self.hw = HW() self.server = kiosk.Servidor(('127.0.0.1', 0), kiosk.Handler) self.server.token = TOKEN self.server.controle = kiosk.ControlePC(pasta / 'app.json') self.server.servos = ControleServos(pasta / 'servos.json', self.hw) self.server.braco = Braco(pasta / 'braco.json', self.server.servos) kiosk.ligar_eventos(self.server) kiosk.FILA.limpar() kiosk.ESTOP.update(ativo=False, desde=None) self.thread = threading.Thread(target=self.server.serve_forever, daemon=True) self.thread.start() self.porta = self.server.server_port def tearDown(self): self.server.shutdown() self.server.server_close() self.server.servos.close() self.thread.join() self.tmp.cleanup() def pedir(self, metodo, rota, dados=None, token=TOKEN, cabecalhos=None): conn = http.client.HTTPConnection('127.0.0.1', self.porta, timeout=5) h = {'Content-Type': 'application/json'} if token: h['Authorization'] = 'Bearer ' + token h.update(cabecalhos or {}) corpo = None if dados is None else json.dumps(dados) conn.request(metodo, rota, body=corpo, headers=h) r = conn.getresponse() texto = r.read() conn.close() return r.status, json.loads(texto) if texto else None # ------------------------------------------------------------- token def test_rotas_de_controle_exigem_token(self): for metodo, rota, dados in [('POST', '/braco/mover', {'angulos': [90] * 7}), ('POST', '/comando', {'tipo': 'gesto', 'valor': 'feliz'}), ('POST', '/parar-tudo', {}), ('GET', '/estado', None), ('GET', '/fila?aparelho=1', None), ('GET', '/app/status', None), ('GET', '/gesto/feliz', None)]: for token in (None, 'errado'): codigo, corpo = self.pedir(metodo, rota, dados, token=token) self.assertEqual(codigo, 401, (metodo, rota, token)) self.assertEqual(self.hw.calls, []) self.assertEqual(kiosk.FILA.ordens(), []) def test_rotas_publicas_e_token_na_query(self): self.assertEqual(self.pedir('GET', '/gestos', token=None)[0], 200) self.assertEqual(self.pedir('GET', '/app/health', token=None)[0], 200) self.assertEqual(self.pedir('GET', '/estado?t=' + TOKEN, token=None)[0], 200) def test_comandos_dos_jogos_falar_e_olhos(self): self.assertEqual(kiosk.ordem_de_comando('dizer', ' Pedra... '), 'dizer Pedra...') self.assertEqual(kiosk.ordem_de_comando('olhos', {'simbolo': '✌️', 'segundos': 3}), 'olhos ✌️ 3.0') self.assertEqual(kiosk.ordem_de_comando('olhos', {'simbolo': '3', 'segundos': 99}), 'olhos 3 8.0') self.assertEqual(kiosk.ordem_de_comando('olhos', {}), 'olhos - 2.5') for tipo, valor in (('olhos', {'simbolo': ''}), ('olhos', {'simbolo': '1', 'segundos': 'x'}), ('dizer', ' ')): with self.assertRaises(ValueError): kiosk.ordem_de_comando(tipo, valor) codigo, _ = self.pedir('POST', '/comando', {'tipo': 'dizer', 'valor': 'Já!'}) self.assertEqual(codigo, 200) self.assertIn('dizer Já!', kiosk.FILA.ordens()) def test_pagina_do_rosto_sempre_conferida_e_api_sem_cache(self): # o WebView do celular não pode seguir com uma carinha.html velha depois de um deploy for rota, esperado in (('/carinha.html', 'no-cache'), ('/gestos', 'no-store')): conn = http.client.HTTPConnection('127.0.0.1', self.porta, timeout=5) conn.request('GET', rota) r = conn.getresponse() r.read() conn.close() self.assertEqual(r.status, 200, rota) self.assertEqual(r.getheader('Cache-Control'), esperado, rota) self.assertEqual(len(r.headers.get_all('Cache-Control')), 1, rota) def test_sem_token_configurado_fica_aberto(self): self.server.token = None self.assertEqual(self.pedir('GET', '/estado', token=None)[0], 200) def test_carregar_token(self): pasta = Path(self.tmp.name) self.assertIsNone(carregar_token(pasta)) (pasta / 'seguranca.json').write_text(json.dumps({'token': 'abcdefgh123'})) self.assertEqual(carregar_token(pasta), 'abcdefgh123') (pasta / 'seguranca.json').write_text(json.dumps({'token': 'curto'})) with self.assertRaises(ValueError): carregar_token(pasta) def test_origem_estrangeira_recusada(self): codigo, _ = self.pedir('POST', '/comando', {'tipo': 'gesto', 'valor': 'feliz'}, cabecalhos={'Origin': 'http://malicioso.example'}) self.assertEqual(codigo, 400) # ------------------------------------------------------------- comando e fila def test_comando_valida_e_enfileira(self): self.assertEqual(self.pedir('POST', '/comando', {'tipo': 'gesto', 'valor': 'feliz'})[0], 200) self.assertEqual(self.pedir('POST', '/comando', {'tipo': 'rosto', 'valor': 'miudo'})[0], 200) self.assertEqual(self.pedir('POST', '/comando', {'tipo': 'gesto', 'valor': 'rm -rf'})[0], 400) self.assertEqual(self.pedir('POST', '/comando', {'tipo': 'rosto', 'valor': 'nada'})[0], 400) self.assertEqual(kiosk.FILA.ordens(), ['feliz', 'rosto miudo']) for novo in ('acenar', 'esconder', 'espiar', 'pensar', 'musica', # gestos com mãos e faixa 'rir', 'bocejar', 'confuso', 'tonto', 'timido', 'comemorar', 'bravo'): self.assertEqual(self.pedir('POST', '/comando', {'tipo': 'gesto', 'valor': novo})[0], 200) kiosk.FILA.limpar() # atividades (momentos dele) self.assertEqual(self.pedir('POST', '/comando', {'tipo': 'atividade', 'valor': 'videogame'})[0], 200) self.assertEqual(self.pedir('POST', '/comando', {'tipo': 'atividade', 'valor': 'parar'})[0], 200) self.assertEqual(self.pedir('POST', '/comando', {'tipo': 'atividade', 'valor': 'voar'})[0], 400) self.assertEqual(kiosk.FILA.ordens(), ['atividade videogame', 'atividade parar']) kiosk.FILA.limpar() kiosk.FILA.adicionar('feliz'); kiosk.FILA.adicionar('rosto miudo') self.assertEqual(self.pedir('GET', '/fila?aparelho=1')[1], ['feliz', 'rosto miudo']) self.assertEqual(self.pedir('GET', '/fila?aparelho=1')[1], []) def test_rostos_novos_e_formato_do_olho(self): kiosk.ARQUIVO_APARENCIA = str(Path(self.tmp.name) / 'aparencia.json') self.assertEqual(self.pedir('POST', '/comando', {'tipo': 'rosto', 'valor': 'bloco'})[0], 200) self.assertEqual(self.pedir('POST', '/comando', {'tipo': 'formato', 'valor': 'quadrado'})[0], 200) self.assertEqual(self.pedir('POST', '/comando', {'tipo': 'formato', 'valor': 'triangulo'})[0], 400) self.assertEqual(kiosk.FILA.ordens(), ['rosto bloco', 'formato quadrado']) estado = self.pedir('GET', '/estado')[1] self.assertEqual(estado['aparencia']['formato'], 'quadrado') self.assertEqual(estado['aparencia']['rosto'], 'bloco') # a escolha fica salva self.assertIn('pilula', estado['formatos']) def test_post_face_usa_prefixo_rosto(self): conn = http.client.HTTPConnection('127.0.0.1', self.porta, timeout=5) conn.request('POST', '/face', body='miudo', headers={'Authorization': 'Bearer ' + TOKEN}) self.assertEqual(conn.getresponse().status, 200) conn.close() self.assertEqual(kiosk.FILA.ordens(), ['rosto miudo']) def test_fila_limitada_e_com_validade(self): fila = kiosk.Fila(maximo=3, validade=10) for i in range(5): fila.adicionar(f'g{i}', agora=0) self.assertEqual(fila.ordens(), ['g2', 'g3', 'g4']) self.assertEqual(fila.consumir(agora=11), []) fila.adicionar('recarregar'); fila.adicionar('recarregar'); fila.adicionar('feliz') self.assertEqual(fila.ordens(), ['recarregar', 'feliz']) # Rosto no navegador não rouba o 'recarregar' do celular. self.assertEqual(fila.consumir(aparelho=False), ['feliz']) self.assertEqual(fila.consumir(aparelho=True), ['recarregar']) # ------------------------------------------------------------- SSE def test_eventos_entrega_comando(self): conn = http.client.HTTPConnection('127.0.0.1', self.porta, timeout=5) conn.request('GET', '/eventos?t=' + TOKEN) r = conn.getresponse() self.assertEqual(r.status, 200) self.assertIn('text/event-stream', r.getheader('Content-Type')) linhas = [] while not linhas or linhas[-1] != '': linhas.append(r.fp.readline().decode().rstrip('\n')) self.assertIn('event: pronto', linhas) self.pedir('POST', '/comando', {'tipo': 'gesto', 'valor': 'coracao'}) bloco = [] while not bloco or bloco[-1] != '': bloco.append(r.fp.readline().decode().rstrip('\n')) conn.close() self.assertIn('event: expressao', bloco) self.assertIn('data: {"ordem": "coracao"}', bloco) def test_estado_agrega(self): codigo, estado = self.pedir('GET', '/estado') self.assertEqual(codigo, 200) for chave in ('versao', 'seq', 'estop', 'aparencia', 'gestos', 'app', 'motores'): self.assertIn(chave, estado) self.assertTrue(estado['autenticacao']) # ------------------------------------------------------------- E-Stop e trava def test_parar_tudo_desliga_e_trava_ate_rearmar(self): self.assertEqual(self.pedir('POST', '/braco/mover', {'angulos': [90] * 7})[0], 200) self.hw.calls.clear() codigo, corpo = self.pedir('POST', '/parar-tudo', {}) self.assertEqual(codigo, 200) self.assertTrue(corpo['estop']['ativo']) self.assertEqual({c for c, us in self.hw.calls if us is None}, set(range(16))) self.assertEqual(kiosk.FILA.ordens(), ['calma']) self.assertEqual(self.pedir('POST', '/braco/mover', {'angulos': [90] * 7})[0], 409) self.assertEqual(self.pedir('POST', '/servos/mover', {'indice': 0, 'pulso': 1500})[0], 409) self.assertEqual(self.pedir('POST', '/rearmar', {})[0], 200) self.assertEqual(self.pedir('POST', '/braco/mover', {'angulos': [90] * 7})[0], 200) def test_braco_e_diagnostico_nao_disputam_motores(self): self.assertEqual(self.pedir('POST', '/braco/mover', {'angulos': [90] * 7})[0], 200) codigo, corpo = self.pedir('POST', '/servos/mover', {'indice': 0, 'pulso': 1500}) self.assertEqual(codigo, 409) self.assertIn('braco', corpo['erro']) self.pedir('POST', '/braco/parar', {}) self.assertEqual(self.pedir('POST', '/servos/mover', {'indice': 0, 'pulso': 1500})[0], 200) class BracoMovimentoTest(unittest.TestCase): def setUp(self): self.tmp = tempfile.TemporaryDirectory() self.hw = HW() self.servos = ControleServos(Path(self.tmp.name) / 's.json', self.hw) self.servos.closed.set() # sem threads: os passos são chamados à mão self.arm = Braco(Path(self.tmp.name) / 'b.json', self.servos) cal = self.arm.status() for row in cal['calibracao']: row.update(minimo=500, maximo=2500, centro=1500) self.arm.save(dict(calibracao=cal['calibracao'], posturas=[])) def tearDown(self): self.tmp.cleanup() def test_interpolacao_no_servidor_respeita_velocidade(self): self.arm.move([90] * 7) resposta = self.arm.move([90, 90, 90, 90, 90, 90, 130], velocidade=100, ttl_ms=2000) self.assertTrue(resposta['movendo']) agora = self.arm.lease - 1.5 self.arm.passo(agora, .1) # 100 °/s * 0,1 s = 10° self.assertEqual(self.arm.atual[6], 100) for i in range(10): self.arm.passo(agora + .1 * (i + 1), .1) self.assertEqual(self.arm.atual[6], 130) self.assertIsNone(self.arm.alvo) def test_lease_vencido_interrompe_movimento(self): self.arm.move([90] * 7) self.arm.move([90] * 6 + [170], velocidade=50, ttl_ms=300) self.arm.passo(self.arm.lease + .01, .1) self.assertIsNone(self.arm.alvo) self.assertEqual(self.arm.atual[6], 90) # O watchdog desliga os canais no mesmo prazo. self.servos.expire(self.arm.lease + .01) self.assertEqual(self.servos.active, {}) self.assertIsNone(self.servos.dono) def test_renovar_no_alvo_informa_que_chegou(self): self.arm.move([90] * 7) self.hw.calls.clear() r = self.arm.move([90] * 7, velocidade=60, ttl_ms=400) self.assertFalse(r['movendo']) self.assertIsNone(self.arm.alvo) self.assertEqual(len(self.hw.calls), 8) # reenergiza os 8 canais com o novo prazo def test_parametros_invalidos(self): for kwargs in (dict(velocidade=1000), dict(velocidade=True), dict(ttl_ms=50), dict(ttl_ms=5000)): with self.assertRaises(ValueError): self.arm.move([90] * 7, **kwargs) self.assertEqual(self.hw.calls, []) if __name__ == '__main__': unittest.main()