""" 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("") # --- 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()