Files
jhonny-editor/code/engine/testes/test_consultas_de_persistencia.py
T
João Henrique 19d10e61a1 feat: sincronizado o estado de transcrição, métricas e diarização
- sincronizado o estado de transcrição, métricas e diarização entre SQLite, Scanner e Editar vídeo

Resumo:
- 6 arquivos alterados
- 1 novos
- 5 modificados
- 0 removidos

 5 files changed, 181 insertions(+), 12 deletions(-)

Arquivos:
  - code/cep-plugin/index.html
  - code/cep-plugin/main.js
  - code/cep-plugin/styles.css
  - code/engine/persistencia/consultas.py
  - code/engine/testes/test_consultas_de_persistencia.py
  - code/engine/consultar_status_da_analise.py
2026-09-09 19:45:15 -04:00

218 lines
10 KiB
Python

"""
Testes de ``engine.persistencia.consultas.ConsultasDeAnalises``.
Verifica que cada método devolve apenas os campos esperados para aquele
nível de detalhe, a partir de um banco populado pelos repositórios já
testados em ``test_persistencia_sqlite.py``, e que entradas inválidas são
rejeitadas com ``ErroDeConsultaInvalida``.
"""
import tempfile
import unittest
from pathlib import Path
from engine.dominio import Clipe, Faixa, IntervaloDeTempo, Timeline
from engine.persistencia import (
ConsultasDeAnalises,
ErroDeConsultaInvalida,
RepositorioDeAnalisesSQLite,
RepositorioDeRetakesSQLite,
RepositorioDeTimelineSQLite,
)
from engine.scanner.modelos import Cena, EvidenciaVisual, SegmentoDeTranscricao
from engine.scanner.retakes.modelos_de_retakes import EvidenciaDeRetake, GrupoDeRetake, TomadaDeRetake
from engine.scanner.transcricao_da_timeline import TranscricaoDoClipe
VIDEO_ID = "v1"
class TesteConsultasDeAnalises(unittest.TestCase):
"""Popula um banco temporário com dados de todas as análises e valida as consultas."""
def setUp(self) -> None:
self._pasta_temporaria = tempfile.TemporaryDirectory()
self.banco = Path(self._pasta_temporaria.name) / "analises.db"
self.addCleanup(self._pasta_temporaria.cleanup)
self._popular_banco()
self.consultas = ConsultasDeAnalises(self.banco)
def _popular_banco(self) -> None:
clipe = Clipe(
identificador="clip-1", nome="Take 1",
intervalo_na_timeline=IntervaloDeTempo(0.0, 20.0),
arquivo="/videos/entrada.mp4", identificador_da_faixa="f1",
)
faixa = Faixa(identificador="f1", nome="V1", tipo="video", indice=0, clipes=[clipe])
timeline = Timeline(identificador=VIDEO_ID, nome="Sequência", duracao=20.0, faixas=[faixa])
RepositorioDeTimelineSQLite(self.banco).registrar_timeline(timeline)
segmento = SegmentoDeTranscricao(
inicio=0.0, fim=6.0, texto="Hoje nós vamos mostrar como funciona o sistema.",
confianca=0.9, falante="locutor", emocao="neutro",
)
transcricao = TranscricaoDoClipe("clip-1", "/videos/entrada.mp4", 0.0, 20.0, [segmento])
RepositorioDeAnalisesSQLite(self.banco).registrar_transcricoes(VIDEO_ID, [transcricao])
repositorio_analises = RepositorioDeAnalisesSQLite(self.banco)
repositorio_analises.registrar_evidencias_visuais(VIDEO_ID, "clip-1", [
EvidenciaVisual(tipo="qualidade", inicio=1.0, fim=2.0, valor={"nitidez": 0.9},
confianca=0.95, provider="opencv"),
EvidenciaVisual(tipo="rosto", inicio=1.0, fim=2.0, valor={"quantidade": 1},
confianca=0.4, provider="opencv"),
])
repositorio_analises.registrar_cenas(VIDEO_ID, [Cena(inicio=0.0, fim=20.0, confianca=0.8)],
clipe_id="clip-1")
grupo = GrupoDeRetake(
id="retake_teste", video_id=VIDEO_ID, faixa_id="f1",
tipo="retake_confirmado", confianca=0.92,
tomadas=[
TomadaDeRetake(0, "fala-1", "clip-1", 0.0, 6.0, "Hoje nós vamos mostrar.",
confianca=0.92),
TomadaDeRetake(1, "fala-2", "clip-1", 9.0, 15.0, "Hoje nós vamos mostrar.",
similaridade_com_anterior=1.0, confianca=0.98),
],
evidencias=[EvidenciaDeRetake("similaridade_textual", "Textos quase idênticos.", 0.98)],
)
RepositorioDeRetakesSQLite(self.banco).registrar(VIDEO_ID, [grupo])
# --- consultar_resumo_do_video -----------------------------------------------------
def test_consultar_resumo_do_video_traz_contagens_agregadas(self) -> None:
resumo = self.consultas.consultar_resumo_do_video(VIDEO_ID)
self.assertIsNotNone(resumo)
self.assertEqual(resumo["nome"], "Sequência")
self.assertEqual(resumo["total_clipes"], 1)
self.assertEqual(resumo["total_falas"], 1)
self.assertEqual(resumo["total_retakes"], 1)
self.assertEqual(resumo["total_cenas"], 1)
def test_consultar_resumo_do_video_inexistente_retorna_none(self) -> None:
self.assertIsNone(self.consultas.consultar_resumo_do_video("video_desconhecido"))
def test_consultar_resumo_do_video_rejeita_video_id_vazio(self) -> None:
with self.assertRaises(ErroDeConsultaInvalida):
self.consultas.consultar_resumo_do_video("")
def test_consultar_status_da_analise_expoe_falantes_e_metricas(self) -> None:
status = self.consultas.consultar_status_da_analise(VIDEO_ID)
self.assertEqual(status["audio"]["total_falas"], 1)
self.assertEqual(status["audio"]["falas_com_falante"], 1)
self.assertEqual(status["audio"]["falas_com_metricas"], 0)
self.assertTrue(status["audio"]["diarizacao_disponivel"])
self.assertFalse(status["audio"]["metricas_de_voz_disponiveis"])
self.assertEqual(status["audio"]["falantes"], ["locutor"])
# --- listar_falas_no_intervalo ------------------------------------------------------
def test_listar_falas_no_intervalo_filtra_por_tempo_e_omite_palavras_por_padrao(self) -> None:
falas = self.consultas.listar_falas_no_intervalo(VIDEO_ID, inicio=0.0, fim=10.0)
self.assertEqual(len(falas), 1)
self.assertEqual(falas[0]["clipe_id"], "clip-1")
self.assertEqual(falas[0]["emocao"], "neutro")
self.assertNotIn("palavras", falas[0])
self.assertNotIn("caracteristicas_acusticas", falas[0])
def test_listar_falas_no_intervalo_fora_do_periodo_retorna_lista_vazia(self) -> None:
falas = self.consultas.listar_falas_no_intervalo(VIDEO_ID, inicio=100.0, fim=110.0)
self.assertEqual(falas, [])
def test_listar_falas_no_intervalo_rejeita_inicio_negativo(self) -> None:
with self.assertRaises(ErroDeConsultaInvalida):
self.consultas.listar_falas_no_intervalo(VIDEO_ID, inicio=-1.0)
def test_listar_falas_no_intervalo_rejeita_fim_anterior_ao_inicio(self) -> None:
with self.assertRaises(ErroDeConsultaInvalida):
self.consultas.listar_falas_no_intervalo(VIDEO_ID, inicio=10.0, fim=5.0)
# --- consultar_clipe_no_instante -----------------------------------------------------
def test_consultar_clipe_no_instante_localiza_o_clipe_correto(self) -> None:
clipe = self.consultas.consultar_clipe_no_instante(VIDEO_ID, 5.0)
self.assertIsNotNone(clipe)
self.assertEqual(clipe["clipe_id"], "clip-1")
def test_consultar_clipe_no_instante_fora_da_timeline_retorna_none(self) -> None:
self.assertIsNone(self.consultas.consultar_clipe_no_instante(VIDEO_ID, 999.0))
def test_consultar_clipe_no_instante_rejeita_instante_negativo(self) -> None:
with self.assertRaises(ErroDeConsultaInvalida):
self.consultas.consultar_clipe_no_instante(VIDEO_ID, -1.0)
# --- listar_retakes_do_video / consultar_detalhes_do_retake -------------------------
def test_listar_retakes_do_video_traz_resumo_sem_tomadas_completas(self) -> None:
retakes = self.consultas.listar_retakes_do_video(VIDEO_ID)
self.assertEqual(len(retakes), 1)
resumo = retakes[0]
self.assertEqual(resumo["grupo_id"], "retake_teste")
self.assertEqual(resumo["total_tomadas"], 2)
self.assertEqual(resumo["tomada_principal"]["fala_id"], "fala-2")
self.assertNotIn("tomadas", resumo)
self.assertNotIn("evidencias", resumo)
def test_consultar_detalhes_do_retake_traz_tomadas_e_evidencias_completas(self) -> None:
detalhes = self.consultas.consultar_detalhes_do_retake("retake_teste")
self.assertIsNotNone(detalhes)
self.assertEqual(len(detalhes["tomadas"]), 2)
self.assertEqual(len(detalhes["evidencias"]), 1)
self.assertEqual(detalhes["evidencias"][0]["tipo"], "similaridade_textual")
def test_consultar_detalhes_do_retake_inexistente_retorna_none(self) -> None:
self.assertIsNone(self.consultas.consultar_detalhes_do_retake("retake_inexistente"))
def test_consultar_detalhes_do_retake_rejeita_grupo_id_vazio(self) -> None:
with self.assertRaises(ErroDeConsultaInvalida):
self.consultas.consultar_detalhes_do_retake(" ")
# --- listar_evidencias_visuais_do_clipe ----------------------------------------------
def test_listar_evidencias_visuais_filtra_por_tipo(self) -> None:
evidencias = self.consultas.listar_evidencias_visuais_do_clipe(
VIDEO_ID, "clip-1", tipo="qualidade")
self.assertEqual(len(evidencias), 1)
self.assertEqual(evidencias[0]["valor"], {"nitidez": 0.9})
def test_listar_evidencias_visuais_filtra_por_confianca_minima(self) -> None:
evidencias = self.consultas.listar_evidencias_visuais_do_clipe(
VIDEO_ID, "clip-1", confianca_minima=0.9)
self.assertEqual(len(evidencias), 1)
self.assertEqual(evidencias[0]["tipo"], "qualidade")
def test_listar_evidencias_visuais_rejeita_confianca_fora_da_faixa(self) -> None:
with self.assertRaises(ErroDeConsultaInvalida):
self.consultas.listar_evidencias_visuais_do_clipe(
VIDEO_ID, "clip-1", confianca_minima=1.5)
def test_listar_evidencias_visuais_rejeita_clipe_id_vazio(self) -> None:
with self.assertRaises(ErroDeConsultaInvalida):
self.consultas.listar_evidencias_visuais_do_clipe(VIDEO_ID, "")
# --- listar_cenas_do_video -------------------------------------------------------------
def test_listar_cenas_do_video_filtra_por_clipe(self) -> None:
cenas = self.consultas.listar_cenas_do_video(VIDEO_ID, clipe_id="clip-1")
self.assertEqual(len(cenas), 1)
self.assertEqual(cenas[0]["fim"], 20.0)
def test_listar_cenas_do_video_sem_clipe_traz_todas(self) -> None:
self.assertEqual(len(self.consultas.listar_cenas_do_video(VIDEO_ID)), 1)
def test_listar_cenas_do_video_rejeita_video_id_vazio(self) -> None:
with self.assertRaises(ErroDeConsultaInvalida):
self.consultas.listar_cenas_do_video(None)
if __name__ == "__main__":
unittest.main()