Spaces:

AIDAON
/

aida-api

Sleeping

App Files Files Community

aida-api / testes /test_api_rotas.py

AIDAON

Claude Opus 5.5

Valida dimensoes da imagem e versiona a suite de testes

5340e92 about 1 month ago

Raw History Blame Contribute Delete

6.93 kB

"""Rotas HTTP da API e a integração AIDA Image → AIDA Forensics.
O pipeline é um dublê (ver `conftest.py`): o que está sob teste aqui é o
contrato da API — códigos de status, forma da resposta e o caminho que leva do
resultado da análise ao dossiê forense sem recalcular nada.
"""

from __future__ import annotations

import io

import pytest
from PIL import Image


def _imagem_em_memoria(formato: str = "JPEG", nome: str = "foto.jpg"):
    buffer = io.BytesIO()
    Image.new("RGB", (48, 48), (40, 120, 110)).save(buffer, format=formato)
    buffer.seek(0)
    return (buffer, nome)


# --------------------------------------------------------------------------- #
# Rotas de diagnóstico
# --------------------------------------------------------------------------- #


def test_raiz_lista_os_quatro_estados(cliente_api):
    corpo = cliente_api.get("/").get_json()

    assert corpo["status"] == "online"
    assert set(corpo["estados_possiveis"]) == {"REAL", "REAL_EDITADA", "IA/MANIPULADA", "INCONCLUSIVO"}
    assert corpo["forense"] == "/forense/<id_analise>"


def test_saude_reporta_o_cache(cliente_api):
    """Sem cache não há Forensics; a queda precisa aparecer no diagnóstico."""
    corpo = cliente_api.get("/saude").get_json()

    assert "cache_analises" in corpo
    assert corpo["cache_analises"]["entradas"] == 0
    assert corpo["cache_analises"]["ttl_horas"] > 0


def test_contrato_declara_o_endpoint_forense(cliente_api):
    corpo = cliente_api.get("/contrato").get_json()

    assert "forense_url" in corpo["resposta_200"]
    assert "GET /forense/<id>" in corpo["endpoints_relacionados"]


# --------------------------------------------------------------------------- #
# Envio e análise
# --------------------------------------------------------------------------- #


def test_analisar_sem_imagem(cliente_api):
    resposta = cliente_api.post("/analisar", data={})

    assert resposta.status_code == 400
    assert "erro" in resposta.get_json()


def test_analisar_formato_nao_aceito(cliente_api):
    resposta = cliente_api.post(
        "/analisar",
        data={"imagem": (io.BytesIO(b"nao sou imagem"), "documento.txt")},
        content_type="multipart/form-data",
    )

    assert resposta.status_code == 400
    assert "formatos_aceitos" in resposta.get_json()


def test_analisar_arquivo_invalido_com_extensao_valida(cliente_api):
    """Extensão certa não é garantia de conteúdo: o arquivo é verificado."""
    resposta = cliente_api.post(
        "/analisar",
        data={"imagem": (io.BytesIO(b"isto nao e um jpeg"), "mentira.jpg")},
        content_type="multipart/form-data",
    )

    assert resposta.status_code == 400


def test_analisar_devolve_resultado_e_ponte_forense(cliente_api):
    resposta = cliente_api.post(
        "/analisar",
        data={"imagem": _imagem_em_memoria()},
        content_type="multipart/form-data",
    )

    assert resposta.status_code == 200
    corpo = resposta.get_json()

    assert corpo["sucesso"] is True
    assert corpo["resultado"] == "INCONCLUSIVO"
    assert corpo["forense_url"] == f"/forense/{corpo['id_analise']}"
    # O vetor de características não pode viajar na resposta HTTP.
    assert "principais_metricas" not in corpo


def test_analisar_preserva_o_nome_enviado(cliente_api):
    """O arquivo temporário do servidor tem o id por nome; o original se perderia."""
    resposta = cliente_api.post(
        "/analisar",
        data={"imagem": _imagem_em_memoria(nome="paisagem-praia.jpg")},
        content_type="multipart/form-data",
    )

    assert resposta.get_json()["nome_imagem_original"] == "paisagem-praia.jpg"


def test_nome_enviado_e_saneado(cliente_api):
    """O nome vem do cliente e vai para a tela: separadores de caminho saem."""
    resposta = cliente_api.post(
        "/analisar",
        data={"imagem": _imagem_em_memoria(nome="../../etc/passwd.jpg")},
        content_type="multipart/form-data",
    )

    nome = resposta.get_json()["nome_imagem_original"]
    assert "/" not in nome and "\\" not in nome


# --------------------------------------------------------------------------- #
# Integração AIDA Image → AIDA Forensics
# --------------------------------------------------------------------------- #


def test_forense_reabre_a_analise_sem_recalcular(cliente_api):
    analise = cliente_api.post(
        "/analisar",
        data={"imagem": _imagem_em_memoria()},
        content_type="multipart/form-data",
    ).get_json()

    chamadas_antes = len(cliente_api.analises_feitas)

    dossie = cliente_api.get(analise["forense_url"]).get_json()

    # A garantia central desta funcionalidade: abrir o Forensics não dispara
    # uma segunda execução do pipeline.
    assert len(cliente_api.analises_feitas) == chamadas_antes

    assert dossie["id_analise"] == analise["id_analise"]
    assert dossie["origem"] == "cache"
    assert dossie["resultado"]["resultado"] == analise["resultado"]
    assert dossie["resultado"]["probabilidade_ia_exibicao"] == pytest.approx(
        analise["probabilidade_ia_exibicao"]
    )
    assert dossie["metodos"]
    assert dossie["evidencias"]["mapas"]


def test_mapas_do_dossie_sao_servidos(cliente_api):
    analise = cliente_api.post(
        "/analisar",
        data={"imagem": _imagem_em_memoria()},
        content_type="multipart/form-data",
    ).get_json()

    dossie = cliente_api.get(analise["forense_url"]).get_json()

    for mapa in dossie["evidencias"]["mapas"]:
        assert cliente_api.get(mapa["url"]).status_code == 200


def test_forense_de_analise_inexistente(cliente_api):
    resposta = cliente_api.get("/forense/nao-existe")

    assert resposta.status_code == 404
    corpo = resposta.get_json()
    # A mensagem precisa dizer o que fazer, não só que falhou.
    assert corpo["acao"]
    assert corpo["motivo"]


def test_forense_recusa_identificador_com_travessia(cliente_api):
    """`..%2f..` não pode alcançar arquivo fora do diretório de cache."""
    resposta = cliente_api.get("/forense/..%2f..%2fsegredo")

    assert resposta.status_code in (404, 400)


def test_delete_forense_remove_do_cache(cliente_api):
    analise = cliente_api.post(
        "/analisar",
        data={"imagem": _imagem_em_memoria()},
        content_type="multipart/form-data",
    ).get_json()

    apagar = cliente_api.delete(analise["forense_url"]).get_json()
    assert apagar["removida"] is True

    assert cliente_api.get(analise["forense_url"]).status_code == 404
    # Segunda remoção não é erro; é apenas nada a remover.
    assert cliente_api.delete(analise["forense_url"]).get_json()["removida"] is False


def test_evidencia_com_identificador_invalido(cliente_api):
    assert cliente_api.get("/evidencia/..%2f..%2fetc/passwd").status_code in (400, 404)


def test_analise_inexistente_no_historico(cliente_api):
    assert cliente_api.get("/analise/nao-existe-123").status_code == 404

	"""Rotas HTTP da API e a integração AIDA Image → AIDA Forensics.

	O pipeline é um dublê (ver `conftest.py`): o que está sob teste aqui é o
	contrato da API — códigos de status, forma da resposta e o caminho que leva do
	resultado da análise ao dossiê forense sem recalcular nada.
	"""

	from __future__ import annotations

	import io

	import pytest
	from PIL import Image


	def _imagem_em_memoria(formato: str = "JPEG", nome: str = "foto.jpg"):
	buffer = io.BytesIO()
	Image.new("RGB", (48, 48), (40, 120, 110)).save(buffer, format=formato)
	buffer.seek(0)
	return (buffer, nome)


	# --------------------------------------------------------------------------- #
	# Rotas de diagnóstico
	# --------------------------------------------------------------------------- #


	def test_raiz_lista_os_quatro_estados(cliente_api):
	corpo = cliente_api.get("/").get_json()

	assert corpo["status"] == "online"
	assert set(corpo["estados_possiveis"]) == {"REAL", "REAL_EDITADA", "IA/MANIPULADA", "INCONCLUSIVO"}
	assert corpo["forense"] == "/forense/<id_analise>"


	def test_saude_reporta_o_cache(cliente_api):
	"""Sem cache não há Forensics; a queda precisa aparecer no diagnóstico."""
	corpo = cliente_api.get("/saude").get_json()

	assert "cache_analises" in corpo
	assert corpo["cache_analises"]["entradas"] == 0
	assert corpo["cache_analises"]["ttl_horas"] > 0


	def test_contrato_declara_o_endpoint_forense(cliente_api):
	corpo = cliente_api.get("/contrato").get_json()

	assert "forense_url" in corpo["resposta_200"]
	assert "GET /forense/<id>" in corpo["endpoints_relacionados"]


	# --------------------------------------------------------------------------- #
	# Envio e análise
	# --------------------------------------------------------------------------- #


	def test_analisar_sem_imagem(cliente_api):
	resposta = cliente_api.post("/analisar", data={})

	assert resposta.status_code == 400
	assert "erro" in resposta.get_json()


	def test_analisar_formato_nao_aceito(cliente_api):
	resposta = cliente_api.post(
	"/analisar",
	data={"imagem": (io.BytesIO(b"nao sou imagem"), "documento.txt")},
	content_type="multipart/form-data",
	)

	assert resposta.status_code == 400
	assert "formatos_aceitos" in resposta.get_json()


	def test_analisar_arquivo_invalido_com_extensao_valida(cliente_api):
	"""Extensão certa não é garantia de conteúdo: o arquivo é verificado."""
	resposta = cliente_api.post(
	"/analisar",
	data={"imagem": (io.BytesIO(b"isto nao e um jpeg"), "mentira.jpg")},
	content_type="multipart/form-data",
	)

	assert resposta.status_code == 400


	def test_analisar_devolve_resultado_e_ponte_forense(cliente_api):
	resposta = cliente_api.post(
	"/analisar",
	data={"imagem": _imagem_em_memoria()},
	content_type="multipart/form-data",
	)

	assert resposta.status_code == 200
	corpo = resposta.get_json()

	assert corpo["sucesso"] is True
	assert corpo["resultado"] == "INCONCLUSIVO"
	assert corpo["forense_url"] == f"/forense/{corpo['id_analise']}"
	# O vetor de características não pode viajar na resposta HTTP.
	assert "principais_metricas" not in corpo


	def test_analisar_preserva_o_nome_enviado(cliente_api):
	"""O arquivo temporário do servidor tem o id por nome; o original se perderia."""
	resposta = cliente_api.post(
	"/analisar",
	data={"imagem": _imagem_em_memoria(nome="paisagem-praia.jpg")},
	content_type="multipart/form-data",
	)

	assert resposta.get_json()["nome_imagem_original"] == "paisagem-praia.jpg"


	def test_nome_enviado_e_saneado(cliente_api):
	"""O nome vem do cliente e vai para a tela: separadores de caminho saem."""
	resposta = cliente_api.post(
	"/analisar",
	data={"imagem": _imagem_em_memoria(nome="../../etc/passwd.jpg")},
	content_type="multipart/form-data",
	)

	nome = resposta.get_json()["nome_imagem_original"]
	assert "/" not in nome and "\\" not in nome


	# --------------------------------------------------------------------------- #
	# Integração AIDA Image → AIDA Forensics
	# --------------------------------------------------------------------------- #


	def test_forense_reabre_a_analise_sem_recalcular(cliente_api):
	analise = cliente_api.post(
	"/analisar",
	data={"imagem": _imagem_em_memoria()},
	content_type="multipart/form-data",
	).get_json()

	chamadas_antes = len(cliente_api.analises_feitas)

	dossie = cliente_api.get(analise["forense_url"]).get_json()

	# A garantia central desta funcionalidade: abrir o Forensics não dispara
	# uma segunda execução do pipeline.
	assert len(cliente_api.analises_feitas) == chamadas_antes

	assert dossie["id_analise"] == analise["id_analise"]
	assert dossie["origem"] == "cache"
	assert dossie["resultado"]["resultado"] == analise["resultado"]
	assert dossie["resultado"]["probabilidade_ia_exibicao"] == pytest.approx(
	analise["probabilidade_ia_exibicao"]
	)
	assert dossie["metodos"]
	assert dossie["evidencias"]["mapas"]


	def test_mapas_do_dossie_sao_servidos(cliente_api):
	analise = cliente_api.post(
	"/analisar",
	data={"imagem": _imagem_em_memoria()},
	content_type="multipart/form-data",
	).get_json()

	dossie = cliente_api.get(analise["forense_url"]).get_json()

	for mapa in dossie["evidencias"]["mapas"]:
	assert cliente_api.get(mapa["url"]).status_code == 200


	def test_forense_de_analise_inexistente(cliente_api):
	resposta = cliente_api.get("/forense/nao-existe")

	assert resposta.status_code == 404
	corpo = resposta.get_json()
	# A mensagem precisa dizer o que fazer, não só que falhou.
	assert corpo["acao"]
	assert corpo["motivo"]


	def test_forense_recusa_identificador_com_travessia(cliente_api):
	"""`..%2f..` não pode alcançar arquivo fora do diretório de cache."""
	resposta = cliente_api.get("/forense/..%2f..%2fsegredo")

	assert resposta.status_code in (404, 400)


	def test_delete_forense_remove_do_cache(cliente_api):
	analise = cliente_api.post(
	"/analisar",
	data={"imagem": _imagem_em_memoria()},
	content_type="multipart/form-data",
	).get_json()

	apagar = cliente_api.delete(analise["forense_url"]).get_json()
	assert apagar["removida"] is True

	assert cliente_api.get(analise["forense_url"]).status_code == 404
	# Segunda remoção não é erro; é apenas nada a remover.
	assert cliente_api.delete(analise["forense_url"]).get_json()["removida"] is False


	def test_evidencia_com_identificador_invalido(cliente_api):
	assert cliente_api.get("/evidencia/..%2f..%2fetc/passwd").status_code in (400, 404)


	def test_analise_inexistente_no_historico(cliente_api):
	assert cliente_api.get("/analise/nao-existe-123").status_code == 404