| """Cache de análises: gravação, leitura, invalidação e o que NÃO é persistido. |
| |
| O cache é a peça de que o AIDA Forensics depende inteiramente. Se ele guardar |
| demais, o disco cresce sem controle; se guardar de menos ou entregar dados de |
| outra versão do modelo, o dossiê passa a contradizer o resultado que o originou. |
| """ |
|
|
| from __future__ import annotations |
|
|
| import json |
| import time |
|
|
| import pytest |
|
|
|
|
| def test_guarda_e_recupera(cache, payload_analise): |
| cache.guardar(payload_analise("ABC-1")) |
| lido = cache.recuperar_analise("ABC-1") |
|
|
| assert lido is not None |
| assert lido["resultado"] == "INCONCLUSIVO" |
| assert lido["scores_modulos"] == {"forense": 0.31, "clip": 0.55} |
|
|
|
|
| def test_nao_persiste_o_vetor_de_caracteristicas(cache, payload_analise): |
| """659 colunas que nenhuma tela usa não podem ir para o disco.""" |
| cache.guardar(payload_analise("ABC-2")) |
| lido = cache.recuperar_analise("ABC-2") |
|
|
| assert "principais_metricas" not in lido |
| |
| assert lido["qualidade"]["qualidade_score"] == pytest.approx(0.62) |
|
|
|
|
| def test_nao_persiste_caminho_do_servidor(cache, payload_analise): |
| """Caminho absoluto expõe a estrutura de diretórios da implantação.""" |
| cache.guardar(payload_analise("ABC-3")) |
| lido = cache.recuperar_analise("ABC-3") |
|
|
| assert lido["caminho_imagem"] == "" |
| |
| for nome in lido["evidencias"]["arquivos"].values(): |
| assert "/" not in nome and "\\" not in nome |
|
|
|
|
| def test_tipos_do_numpy_sao_serializados(cache, payload_analise): |
| """`np.float32` no payload não pode derrubar a gravação inteira.""" |
| numpy = pytest.importorskip("numpy") |
|
|
| payload = payload_analise("ABC-4") |
| payload["scores_modulos"] = {"forense": numpy.float32(0.42)} |
| payload["duracao_s"] = numpy.float64(2.5) |
| payload["evidencias"]["perfil_radial"] = numpy.array([1.0, 2.0]) |
|
|
| assert cache.guardar(payload) is not None |
| lido = cache.recuperar_analise("ABC-4") |
| assert lido["scores_modulos"]["forense"] == pytest.approx(0.42, abs=1e-6) |
|
|
|
|
| def test_identificador_invalido_nao_grava(cache, payload_analise): |
| """`../` na chave escaparia do diretório de cache.""" |
| assert cache.guardar(payload_analise("ABC-5"), id_analise="../fuga") is None |
| assert cache.recuperar_analise("../fuga") is None |
|
|
|
|
| def test_expira_por_tempo(cache, payload_analise): |
| cache.guardar(payload_analise("ABC-6")) |
|
|
| caminho = cache.caminho_da_analise("ABC-6") |
| registro = json.loads(caminho.read_text(encoding="utf-8")) |
| registro["gravado_em"] = time.time() - 3600 |
| caminho.write_text(json.dumps(registro), encoding="utf-8") |
|
|
| assert cache.recuperar_analise("ABC-6", ttl_s=60) is None |
| assert cache.recuperar_analise("ABC-6", ttl_s=7200) is not None |
|
|
|
|
| def test_ttl_zero_desliga_expiracao(cache, payload_analise, monkeypatch): |
| monkeypatch.setenv("AIDA_CACHE_TTL_H", "0") |
| cache.guardar(payload_analise("ABC-7")) |
|
|
| caminho = cache.caminho_da_analise("ABC-7") |
| registro = json.loads(caminho.read_text(encoding="utf-8")) |
| registro["gravado_em"] = 0.0 |
| caminho.write_text(json.dumps(registro), encoding="utf-8") |
|
|
| assert cache.recuperar_analise("ABC-7") is not None |
|
|
|
|
| def test_versao_de_modelo_diferente_invalida(cache, payload_analise): |
| """Dossiê precisa corresponder ao modelo que produziu a decisão.""" |
| cache.guardar(payload_analise("ABC-8")) |
|
|
| assert cache.recuperar_analise("ABC-8", versao_modelo="v-outra") is None |
| assert cache.recuperar_analise("ABC-8") is not None |
|
|
|
|
| def test_invalidacao_manual_remove_tambem_o_indice(cache, payload_analise): |
| cache.guardar(payload_analise("ABC-9")) |
| assert cache.recuperar_por_conteudo("a" * 64) is not None |
|
|
| assert cache.invalidar("ABC-9") is True |
| assert cache.recuperar_analise("ABC-9") is None |
| |
| assert cache.recuperar_por_conteudo("a" * 64) is None |
| assert cache.invalidar("ABC-9") is False |
|
|
|
|
| def test_indice_por_conteudo_encontra_analise_anterior(cache, payload_analise): |
| cache.guardar(payload_analise("ABC-10", hash_sha256="b" * 64)) |
| registro = cache.recuperar_por_conteudo("b" * 64) |
|
|
| assert registro is not None |
| assert registro["id_analise"] == "ABC-10" |
|
|
|
|
| def test_arquivo_corrompido_e_descartado(cache, payload_analise): |
| cache.guardar(payload_analise("ABC-11")) |
| cache.caminho_da_analise("ABC-11").write_text("{ isto não é json", encoding="utf-8") |
|
|
| assert cache.recuperar_analise("ABC-11") is None |
| |
| |
| assert not cache.caminho_da_analise("ABC-11").exists() |
|
|
|
|
| def test_limpar_remove_apenas_o_que_expirou(cache, payload_analise): |
| cache.guardar(payload_analise("NOVA")) |
| cache.guardar(payload_analise("VELHA")) |
|
|
| velha = cache.caminho_da_analise("VELHA") |
| antigo = time.time() - 10 * 24 * 3600 |
| import os |
|
|
| os.utime(velha, (antigo, antigo)) |
|
|
| assert cache.limpar() == 1 |
| assert cache.recuperar_analise("NOVA") is not None |
| assert cache.recuperar_analise("VELHA") is None |
|
|
|
|
| def test_listar_devolve_apenas_validas(cache, payload_analise): |
| cache.guardar(payload_analise("L-1")) |
| cache.guardar(payload_analise("L-2")) |
|
|
| assert sorted(cache.listar()) == ["L-1", "L-2"] |
|
|
