""" 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, EdicaoDeVideo, RepositorioDeAnalisesSQLite, RepositorioDeEdicoes, 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_contexto_da_edicao_carrega_edicao_mais_recente_e_falas(self) -> None: RepositorioDeEdicoes(self.consultas.conexao).registrar(EdicaoDeVideo( video_id=VIDEO_ID, sequencia="Sequência", origem="/videos/entrada.mp4", tipo_de_video="depoimento", configuracao={"objetivo": "Gerar confiança"}, )) contexto = self.consultas.consultar_contexto_da_edicao() self.assertIsNotNone(contexto) self.assertEqual(contexto["edicao"]["tipo_de_video"], "depoimento") self.assertEqual(contexto["edicao"]["configuracao"]["objetivo"], "Gerar confiança") self.assertEqual(contexto["video"]["video_id"], VIDEO_ID) self.assertEqual(len(contexto["falas"]), 1) self.assertEqual(contexto["falas"][0]["texto"], "Hoje nós vamos mostrar como funciona o sistema.") 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"]) def test_listar_intervalos_de_preview_converte_timeline_para_origem(self) -> None: self.consultas.conexao.executescript( """INSERT INTO faixas (id, video_id, nome, tipo, indice) VALUES ('audio-2', 'v1', 'A2', 'audio', 1); INSERT INTO faixas (id, video_id, nome, tipo, indice) VALUES ('video-2', 'v1', 'V2', 'video', 1); INSERT INTO clipes (id, video_id, faixa_id, nome, inicio_na_timeline, fim_na_timeline, inicio_na_origem, fim_na_origem, arquivo) VALUES ('audio-clip-2', 'v1', 'audio-2', 'A2', 30.0, 40.0, 100.0, 110.0, '/videos/entrevista.mp4'); INSERT INTO clipes (id, video_id, faixa_id, nome, inicio_na_timeline, fim_na_timeline, inicio_na_origem, fim_na_origem, arquivo) VALUES ('video-clip-2', 'v1', 'video-2', 'V2', 30.0, 40.0, 100.0, 110.0, '/videos/entrevista.mp4'); INSERT INTO segmentos_de_transcricao (video_id, clipe_id, inicio, fim, texto, falante) VALUES ('v1', 'audio-clip-2', 32.5, 35.0, 'trecho', 'locutor-2');""" ) intervalos = self.consultas.listar_intervalos_de_preview_dos_falantes(VIDEO_ID) intervalo = next(item for item in intervalos if item["falante"] == "locutor-2") self.assertEqual(intervalo["clipe_de_video"], "video-clip-2") self.assertEqual(intervalo["inicio"], 102.5) self.assertEqual(intervalo["fim"], 105.0) self.assertEqual(intervalo["arquivo"], "/videos/entrevista.mp4") # --- 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()