julioandClaude Opus 5.5
82f9673692
- comando "telao" (validado: texto curto, tom, símbolos permitidos, até 2 lados) e letreiro por cima do rosto: título enorme que estoura e pisca, as duas jogadas (VOCÊ ✊ × ✋ CARINHA), cor por resultado; - pedra, papel e tesoura: 3, 2, 1 piscando, "JÁ!", o que ele leu da sua mão, "VENCI!!!" / "VOCÊ GANHOU!" / "EMPATE!" com placar e o campeão; quantos dedos e achou! também usam o telão; - o cartão do painel mostra o mesmo destaque. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
293 lines
15 KiB
Python
293 lines
15 KiB
Python
"""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': '<b>'}), ('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_telao_dos_jogos(self):
|
||
ordem = kiosk.ordem_de_comando('telao', {'titulo': 'VENCI!!!', 'tom': 'eu', 'sub': 'Você 0 × 1 Carinha', 'segundos': 3,
|
||
'lados': [{'rotulo': 'Você', 'simbolo': '✊', 'nome': 'pedra'},
|
||
{'rotulo': 'Carinha', 'simbolo': '✋', 'nome': 'papel'}]})
|
||
self.assertTrue(ordem.startswith('telao {'))
|
||
t = json.loads(ordem[6:])
|
||
self.assertEqual((t['titulo'], t['tom'], len(t['lados'])), ('VENCI!!!', 'eu', 2))
|
||
self.assertEqual(json.loads(kiosk.ordem_de_comando('telao', {'titulo': 'x' * 99, 'tom': 'roxo'})[6:])['tom'], 'neutro')
|
||
for ruim in ({}, {'titulo': 'a', 'lados': [{'simbolo': '<img>'}]}, {'titulo': 'a', 'lados': [{}, {}, {}]}, 'texto'):
|
||
with self.assertRaises(ValueError):
|
||
kiosk.ordem_de_comando('telao', ruim)
|
||
|
||
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()
|