julio/carinha-control-panel
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
ff06f027051f1d30005c9da78fb33d38c4e63383
carinha-control-panel/tests/test_servidor.py
T
julioandClaude Opus 5.5 d7a17b5683 feat(rosto): atividades (momentos dele) e seis gestos novos 
Atividades: a tela se divide, o rosto vai para a esquerda e encolhe, e
à direita um bonequinho do Carinha (silhueta na cor dos olhos) faz algo:
ler (vira a página), cantar (microfone e notas), jogar bola
(embaixadinha), videogame (TV com joguinho), pintar (cavalete que ganha
pinceladas) e malhar (halteres e suor). Os olhos grandes acompanham:
leem as linhas, seguem a bola, olham a TV. Camada própria atrás dos
olhos, vale para todos os rostos; em pé (retrato) a cena vai para baixo.

- Comando {tipo: "atividade", valor: nome|parar}; toque na tela, dormir
  e E-Stop encerram; o celular informa a atividade na telemetria.
- Modo vida escolhe pelo humor (feliz canta e joga bola, tranquilo lê
  e pinta, entediado joga videogame), nunca de madrugada, e não
  interrompe uma atividade em andamento.
- Gestos novos: rir, comemorar, tímido, confuso, bocejar e tonto; o
  humor e o modo vida também os usam.
- Painel: favoritos vão até 20 e o bravo volta junto com os novos;
  botões de atividade no Ao vivo, no Ctrl+K e tópico no guia.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-06 21:12:14 -03:00

257 lines
12 KiB
Python
Raw Blame History

"""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_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()
Reference in New Issue View Git Blame Copy Permalink