test/test_app.py

from pathlib import Path

import gradio as gr
import pytest
from fastapi import FastAPI
from fastapi.responses import HTMLResponse, JSONResponse
from fastapi.testclient import TestClient
from starlette.middleware import Middleware

import app


def test_classify_verdict_fake():
    text, color = app.classify({"verdict": "deepfake"})
    assert text == "Deepfake"
    assert color == app.DEEPFAKE_COLOR


def test_classify_verdict_genuine():
    text, color = app.classify({"verdict": "genuine"})
    assert text == "Genuine"
    assert color == app.GENUINE_COLOR


def test_classify_score_high_is_deepfake():
    text, color = app.classify({"score": 0.9})
    assert text.startswith("Deepfake")
    assert color == app.DEEPFAKE_COLOR


def test_classify_score_low_is_genuine():
    text, color = app.classify({"score": 0.1})
    assert text.startswith("Genuine")
    assert color == app.GENUINE_COLOR


def test_classify_error_is_failure():
    text, color = app.classify({"error": "NoFaceDetected"})
    assert text.startswith("Failure")
    assert color == app.NEUTRAL_COLOR


def test_classify_garbage_is_unknown():
    assert app.classify({})[0] == "Unknown"
    assert app.classify("nonsense")[0] == "Unknown"


def test_classify_score_at_threshold_is_deepfake():
    text, color = app.classify({"verdict": "authentic", "score": 0.5})
    assert text.startswith("Deepfake")
    assert color == app.DEEPFAKE_COLOR


def test_classify_score_overrides_video_verdict():
    text, color = app.classify(
        {"verdict": "suspicious", "score": 0.009, "status": "completed"}
    )
    assert text.startswith("Genuine")
    assert color == app.GENUINE_COLOR


def test_classify_verdict_suspicious_without_score_is_deepfake():
    text, color = app.classify({"verdict": "suspicious"})
    assert text == "Deepfake"
    assert color == app.DEEPFAKE_COLOR


def test_classify_status_failure_surfaces_message():
    text, color = app.classify(
        {"status": "error", "message": "500: Internal Server Error"}
    )
    assert text == "Failure: 500: Internal Server Error"
    assert color == app.NEUTRAL_COLOR


def test_classify_timeout_surfaces_message():
    text, _ = app.classify(
        {"status": "timeout", "message": "Job x did not complete in time"}
    )
    assert text == "Failure: Job x did not complete in time"


class _Resp:
    def __init__(self, payload):
        self._payload = payload

    def json(self):
        return self._payload

    def raise_for_status(self):
        pass


def test_post_file_sends_token_and_hits_endpoint(tmp_path, monkeypatch):
    f = tmp_path / "x.jpg"
    f.write_bytes(b"data")
    monkeypatch.setenv("API_KEY", "secret")
    captured = {}

    def fake_post(url, headers=None, files=None, timeout=None):
        captured["url"] = url
        captured["headers"] = headers
        return _Resp({"score": 0.7})

    monkeypatch.setattr(app.requests, "post", fake_post)
    out = app.post_file(app.IMAGE_ENDPOINT, str(f))
    assert out == {"score": 0.7}
    assert captured["url"] == app.BASE_URL + app.IMAGE_ENDPOINT
    assert captured["headers"]["ds-api-token"] == "secret"


def test_poll_job_stops_on_terminal(monkeypatch):
    seq = [
        _Resp({"status": "processing"}),
        _Resp({"status": "completed", "score": 0.2}),
    ]
    monkeypatch.setattr(
        app.requests, "get", lambda url, headers=None, timeout=None: seq.pop(0)
    )
    monkeypatch.setattr(app.time, "sleep", lambda s: None)
    out = app.poll_job("job-123", interval=0, max_retries=5)
    assert out["status"] == "completed"
    assert out["score"] == 0.2


def test_poll_job_times_out(monkeypatch):
    monkeypatch.setattr(
        app.requests,
        "get",
        lambda url, headers=None, timeout=None: _Resp({"status": "processing"}),
    )
    monkeypatch.setattr(app.time, "sleep", lambda s: None)
    out = app.poll_job("job-123", interval=0, max_retries=3)
    assert out["status"] == "timeout"
    assert "message" in out


def test_post_file_returns_failure_on_request_error(tmp_path, monkeypatch):
    f = tmp_path / "x.jpg"
    f.write_bytes(b"d")

    def boom(*a, **k):
        raise app.requests.RequestException("network down")

    monkeypatch.setattr(app.requests, "post", boom)
    out = app.post_file(app.IMAGE_ENDPOINT, str(f))
    assert out["status"] == "error"
    assert "network down" in out["message"]


def test_poll_job_returns_error_on_request_error(monkeypatch):
    def boom(*a, **k):
        raise app.requests.RequestException("boom")

    monkeypatch.setattr(app.requests, "get", boom)
    out = app.poll_job("job-x", interval=0, max_retries=3)
    assert out["status"] == "error"


def test_demo_is_blocks():
    assert isinstance(app.Demo, gr.Blocks)


def test_header_links_to_deepfake_landing():
    assert "https://dataspike.io/deepfake-detection" in app.HTML_HEADER


def test_header_explains_supported_kyc_media():
    assert app.SUBTITLE in app.HTML_HEADER
    assert app.SUBTITLE == (
        "Deepfakes, AI-generated and manipulated human faces. "
        "Selfie-style photo, video or voice. Built for KYC liveness."
    )


@pytest.mark.parametrize(
    "preset",
    app.IMAGE_EXAMPLES + app.VIDEO_EXAMPLES + app.AUDIO_EXAMPLES,
)
def test_preset_file_exists(preset):
    assert app.PRESETS_DIR in Path(preset).parents
    assert Path(preset).is_file()


def test_audio_presets_cover_generated_and_genuine_voice():
    assert [Path(preset).name for preset in app.AUDIO_EXAMPLES] == [
        "ai-generated-voice.wav",
        "genuine-voice.wav",
    ]


def test_face_media_hint_uses_scan_face_icon_and_clear_copy():
    assert 'class="face-media-icon"' in app.FACE_MEDIA_HINT_HTML
    assert 'viewBox="0 0 24 24"' in app.FACE_MEDIA_HINT_HTML
    assert "Human faces only" in app.FACE_MEDIA_HINT_HTML
    assert "selfie-style media" in app.FACE_MEDIA_HINT_HTML


def test_face_media_hint_is_shown_for_image_and_video_only():
    hints = [
        block
        for block in app.Demo.blocks.values()
        if isinstance(block, gr.HTML) and block.value == app.FACE_MEDIA_HINT_HTML
    ]
    assert len(hints) == 2


def test_verdict_text_never_carries_the_score():
    """The score meant the opposite of the verdict it sat next to."""

    assert app.classify({"score": 0.1})[0] == "Genuine"
    assert app.classify({"score": 0.9})[0] == "Deepfake"


def test_verdict_html_places_marker_at_the_score():
    panel = app.verdict_html({"score": 0.1})
    assert "left: 10.0%" in panel
    assert "Score 0.10" in panel


def test_verdict_html_marks_the_threshold():
    panel = app.verdict_html({"score": 0.42})
    assert f"left: {app.DEEPFAKE_SCORE_THRESHOLD:.1%}" in panel
    assert "Threshold 0.50" in panel


def test_verdict_html_labels_both_ends_of_the_scale():
    panel = app.verdict_html({"score": 0.42})
    assert app.AUTHENTIC_END_LABEL in panel
    assert app.DEEPFAKE_END_LABEL in panel


def test_verdict_html_omits_the_scale_without_a_score():
    panel = app.verdict_html({"verdict": "genuine"})
    assert ">Genuine</span>" in panel
    assert "verdict-track" not in panel
    assert "Threshold" not in panel


def test_verdict_html_escapes_api_text():
    panel = app.verdict_html({"status": "error", "message": "<img src=x onerror=1>"})
    assert "<img" not in panel
    assert "&lt;img" in panel


@pytest.mark.parametrize("score,expected", [(-0.4, 0.0), (1.7, 1.0), (0.25, 0.25)])
def test_score_of_clamps_into_range(score, expected):
    assert app.score_of({"score": score}) == expected


@pytest.mark.parametrize("payload", [{}, {"score": None}, {"score": "0.4"}, "nonsense"])
def test_score_of_returns_none_without_a_number(payload):
    assert app.score_of(payload) is None


def test_score_of_rejects_booleans():
    assert app.score_of({"score": True}) is None


def test_verdict_uses_html_panel_not_label():
    assert not [b for b in app.Demo.blocks.values() if isinstance(b, gr.Label)]


def test_every_tab_starts_with_the_empty_verdict_panel():
    empty = [
        b
        for b in app.Demo.blocks.values()
        if isinstance(b, gr.HTML) and b.value == app.EMPTY_VERDICT_HTML
    ]
    assert len(empty) == 3


def test_verdict_panel_always_captions_itself():
    for payload in ({"score": 0.1}, {"verdict": "genuine"}, {"status": "error"}):
        assert app.VERDICT_CAPTION in app.verdict_html(payload)


def test_video_input_has_bounded_height():
    video = next(b for b in app.Demo.blocks.values() if isinstance(b, gr.Video))
    assert video.height == 300


def _locale_client() -> TestClient:
    api = FastAPI(middleware=[Middleware(app.EnglishLocaleMiddleware)])

    @api.get("/")
    def index() -> HTMLResponse:
        return HTMLResponse("<html><head><title>t</title></head><body></body></html>")

    @api.get("/data")
    def data() -> JSONResponse:
        return JSONResponse({"ok": True})

    return TestClient(api)


def test_locale_middleware_pins_english_before_page_scripts():
    body = _locale_client().get("/").text
    assert 'Object.defineProperty(navigator, "language"' in body
    assert body.index("navigator") < body.index("<title>")


def test_locale_middleware_leaves_non_html_untouched():
    response = _locale_client().get("/data")
    assert response.json() == {"ok": True}


def test_analyze_image_no_file_returns_failure():
    result, panel = app.analyze_image(None)
    assert result["overallStatus"] == "Failure"
    assert "Please submit an image first." in panel


def test_analyze_image_maps_verdict(tmp_path, monkeypatch):
    f = tmp_path / "x.jpg"
    f.write_bytes(b"d")
    monkeypatch.setattr(app, "post_file", lambda e, p: {"verdict": "deepfake"})
    result, panel = app.analyze_image(str(f))
    assert result == {"verdict": "deepfake"}
    assert ">Deepfake</span>" in panel


def test_analyze_video_no_file_returns_failure():
    result, panel = app.analyze_video(None)
    assert result["overallStatus"] == "Failure"
    assert "Please submit a video first." in panel


def test_analyze_video_missing_id_returns_submit(monkeypatch):
    monkeypatch.setattr(app, "post_file", lambda e, p: {"error": "RuntimeError"})
    result, panel = app.analyze_video("x.mp4", progress=lambda *a, **k: None)
    assert result == {"error": "RuntimeError"}
    assert "Failure: RuntimeError" in panel


def test_analyze_video_polls_when_id_present(monkeypatch):
    monkeypatch.setattr(app, "post_file", lambda e, p: {"id": "job-1"})
    monkeypatch.setattr(
        app, "poll_job", lambda jid: {"verdict": "genuine", "status": "completed"}
    )
    result, panel = app.analyze_video("x.mp4", progress=lambda *a, **k: None)
    assert result["verdict"] == "genuine"
    assert ">Genuine</span>" in panel


def test_analyze_audio_no_file_returns_failure():
    result, panel = app.analyze_audio(None)
    assert result["overallStatus"] == "Failure"
    assert "Please submit an audio file first." in panel


def test_analyze_audio_polls_when_id_present(monkeypatch):
    monkeypatch.setattr(app, "post_file", lambda e, p: {"id": "job-2"})
    monkeypatch.setattr(app, "poll_job", lambda jid: {"score": 0.9, "status": "done"})
    result, panel = app.analyze_audio("x.wav", progress=lambda *a, **k: None)
    assert result["score"] == 0.9
    assert "Score 0.90" in panel