Newer
Older
navi-1 / tests / unit / api / test_export.py
"""Export endpoints: token gate, manifest payload, conditional GET."""

from __future__ import annotations

import json

import pytest
from fastapi.testclient import TestClient

from navi.api.routes import export as export_mod
from navi.config import Settings
from navi.export import builder

TOKEN = "e" * 48


@pytest.fixture
def export_env(tmp_path, monkeypatch):
    out = tmp_path / "export_files"
    out.mkdir()
    settings = Settings(
        _env_file=None,
        navi_persona_file="",
        navi_export_dir=str(out),
        navi_export_token=TOKEN,
        navi_export_age_recipient="age1qqqq",
    )
    monkeypatch.setattr(export_mod, "settings", settings)
    monkeypatch.setattr(builder, "settings", settings)

    from navi.main import app

    return TestClient(app), out


def _publish(out, content_id="a" * 64, payload=b"ciphertext", built_at="2026-11-04T12:00:00+00:00"):
    (out / builder.FILE_NAME).write_bytes(payload)
    manifest = {
        "export_version": 1,
        "built_at": built_at,
        "content_id": content_id,
        "file": {"name": builder.FILE_NAME, "bytes": len(payload), "sha256": "b" * 64},
        "rows": 3,
        "sessions": 1,
        "sessions_total": 1,
        "fields": builder.FIELDS,
        "roles": ["user", "assistant"],
        "window": {"from": None, "to": None, "truncated": False, "max_bytes": 1},
        "age_recipient": "age1qqqq",
    }
    (out / builder.MANIFEST_NAME).write_text(json.dumps(manifest))
    return manifest


def test_missing_or_wrong_token_is_rejected(export_env):
    client, out = export_env
    _publish(out)
    assert client.get("/export/v1/manifest").status_code == 401
    bad = client.get("/export/v1/manifest", headers={"Authorization": "Bearer nope"})
    assert bad.status_code == 401
    assert bad.headers["www-authenticate"] == "Bearer"
    # Same answer for a wrong token and no token: the endpoint confirms nothing.
    assert client.get("/export/v1/messages").status_code == 401


def test_manifest_is_served_with_a_valid_token(export_env):
    client, out = export_env
    _publish(out)
    resp = client.get("/export/v1/manifest", headers={"Authorization": f"Bearer {TOKEN}"})
    assert resp.status_code == 200
    assert resp.json()["content_id"] == "a" * 64
    assert resp.headers["cache-control"] == "no-store"


def test_nothing_built_answers_503(export_env):
    client, _ = export_env
    resp = client.get("/export/v1/manifest", headers={"Authorization": f"Bearer {TOKEN}"})
    assert resp.status_code == 503


def test_file_is_served_with_an_etag_and_304_when_unchanged(export_env):
    client, out = export_env
    _publish(out)
    headers = {"Authorization": f"Bearer {TOKEN}"}

    first = client.get("/export/v1/messages", headers=headers)
    assert first.status_code == 200
    assert first.content == b"ciphertext"
    etag = first.headers["etag"]
    assert etag.startswith('"v1-')
    assert first.headers["x-export-content-id"] == "a" * 64

    again = client.get("/export/v1/messages", headers={**headers, "If-None-Match": etag})
    assert again.status_code == 304
    assert again.content == b""


def test_new_content_id_changes_the_etag(export_env):
    client, out = export_env
    headers = {"Authorization": f"Bearer {TOKEN}"}
    _publish(out, content_id="a" * 64)
    first = client.get("/export/v1/messages", headers=headers).headers["etag"]

    out.joinpath(builder.MANIFEST_NAME).unlink()
    _publish(out, content_id="c" * 64)
    second = client.get("/export/v1/messages", headers=headers).headers["etag"]
    assert first != second


def test_unconfigured_token_disables_the_endpoints(export_env, monkeypatch):
    client, out = export_env
    _publish(out)
    monkeypatch.setattr(
        export_mod, "settings", Settings(_env_file=None, navi_persona_file="", navi_export_token="")
    )
    resp = client.get("/export/v1/manifest", headers={"Authorization": "Bearer anything"})
    assert resp.status_code == 503


def test_health_reports_export_freshness_without_touching_status(export_env, monkeypatch):
    _, out = export_env
    from datetime import datetime, timedelta, timezone

    from navi.api.routes import health as health_mod
    from navi.export import loop as loop_mod

    monkeypatch.setattr(health_mod, "settings", builder.settings)
    monkeypatch.setattr(loop_mod, "settings", builder.settings)

    assert health_mod._export_status()["built"] is False

    now = datetime.now(timezone.utc)
    _publish(out, built_at=now.isoformat())
    fresh = health_mod._export_status()
    assert fresh["built"] is True and fresh["stale"] is False and fresh["rows"] == 3

    _publish(out, built_at=(now - timedelta(hours=24)).isoformat())
    assert health_mod._export_status()["stale"] is True


def test_health_omits_export_when_not_configured(export_env, monkeypatch):
    from navi.api.routes import health as health_mod

    monkeypatch.setattr(
        health_mod, "settings", Settings(_env_file=None, navi_persona_file="")
    )
    assert health_mod._export_status() is None