openmind/backend/tests/test_public_exposure.py

"""Protections d'une instance exposee publiquement.

Couvre la limitation de debit, la coupure du partage, la capacite maximale et
l'identification du client derriere un reverse proxy.
"""

from __future__ import annotations

from fastapi.testclient import TestClient

from app.services.rate_limit import RateLimiter, parse_rate
from tests.conftest import build_map


def _post(client: TestClient) -> int:
    return client.post("/api/shares", json={"title": "x", "map": build_map()}).status_code


class TestRateRuleParsing:
    def test_reads_usual_syntaxes(self) -> None:
        assert parse_rate("10/hour") == parse_rate("10/h")
        assert parse_rate("5/minute").limit == 5
        assert parse_rate("5/minute").window_seconds == 60
        assert parse_rate("2/second").window_seconds == 1

    def test_missing_unit_defaults_to_minute(self) -> None:
        assert parse_rate("3").window_seconds == 60

    def test_disabled_values(self) -> None:
        """Seules ces valeurs desactivent : voir test_configuration.py pour le reste."""
        for spec in ("", "0", "off", "none", "disabled"):
            assert parse_rate(spec) is None


class TestRateLimiter:
    def test_allows_then_blocks_within_window(self) -> None:
        limiter = RateLimiter()
        rule = parse_rate("2/minute")
        assert limiter.check("a", rule, now=0.0) is None
        assert limiter.check("a", rule, now=1.0) is None
        retry = limiter.check("a", rule, now=2.0)
        assert retry is not None and retry > 0

    def test_window_slides(self) -> None:
        limiter = RateLimiter()
        rule = parse_rate("1/minute")
        assert limiter.check("a", rule, now=0.0) is None
        assert limiter.check("a", rule, now=10.0) is not None
        assert limiter.check("a", rule, now=61.0) is None

    def test_clients_are_independent(self) -> None:
        limiter = RateLimiter()
        rule = parse_rate("1/minute")
        assert limiter.check("a", rule, now=0.0) is None
        assert limiter.check("b", rule, now=0.0) is None


def test_share_creation_is_rate_limited(client: TestClient, configure) -> None:
    configure(share_create_rate_limit="3/hour")

    assert [_post(client) for _ in range(3)] == [201, 201, 201]

    response = client.post("/api/shares", json={"title": "x", "map": build_map()})
    assert response.status_code == 429
    assert "Retry-After" in response.headers
    assert int(response.headers["Retry-After"]) > 0
    assert "detail" in response.json()


def test_health_and_ready_are_never_rate_limited(client: TestClient, configure) -> None:
    configure(share_create_rate_limit="1/hour", share_read_rate_limit="1/hour")
    client.post("/api/shares", json={"title": "x", "map": build_map()})
    client.post("/api/shares", json={"title": "x", "map": build_map()})

    for _ in range(30):
        assert client.get("/api/health").status_code == 200
        assert client.get("/api/ready").status_code == 200


def test_reading_a_share_is_rate_limited(client: TestClient, configure) -> None:
    created = client.post("/api/shares", json={"title": "x", "map": build_map()}).json()
    configure(share_read_rate_limit="2/minute")

    token = created["share_token"]
    assert client.get(f"/api/shares/{token}").status_code == 200
    assert client.get(f"/api/shares/{token}").status_code == 200
    assert client.get(f"/api/shares/{token}").status_code == 429


def test_rate_limit_can_be_disabled(client: TestClient, configure) -> None:
    configure(share_create_rate_limit="0")
    assert all(_post(client) == 201 for _ in range(12))


def test_sharing_disabled_blocks_the_whole_api(client: TestClient, configure) -> None:
    created = client.post("/api/shares", json={"title": "x", "map": build_map()}).json()
    configure(sharing_enabled=False)

    expected = "Le partage est desactive sur cette instance."
    assert client.post("/api/shares", json={"title": "x", "map": build_map()}).status_code == 503
    assert client.get(f"/api/shares/{created['share_token']}").status_code == 503
    assert (
        client.put(
            f"/api/shares/{created['share_token']}",
            json={"title": "x", "map": build_map()},
            headers={"X-Management-Token": created["management_token"]},
        ).status_code
        == 503
    )
    assert (
        client.delete(
            f"/api/shares/{created['share_token']}",
            headers={"X-Management-Token": created["management_token"]},
        ).status_code
        == 503
    )
    assert (
        client.post("/api/shares", json={"title": "x", "map": build_map()}).json()["detail"]
        == expected
    )


def test_sharing_disabled_is_announced_in_config(client: TestClient, configure) -> None:
    configure(sharing_enabled=False)
    assert client.get("/api/config").json()["sharingEnabled"] is False


def test_capacity_limit_refuses_new_shares(client: TestClient, configure) -> None:
    configure(max_shared_maps=2)
    first = client.post("/api/shares", json={"title": "1", "map": build_map()}).json()
    assert client.post("/api/shares", json={"title": "2", "map": build_map()}).status_code == 201

    refused = client.post("/api/shares", json={"title": "3", "map": build_map()})
    assert refused.status_code == 503
    assert refused.json()["detail"] == "Cette instance a atteint sa capacite maximale de partages."

    # Les partages existants restent lisibles et modifiables.
    assert client.get(f"/api/shares/{first['share_token']}").status_code == 200
    assert (
        client.put(
            f"/api/shares/{first['share_token']}",
            json={"title": "1 bis", "map": build_map()},
            headers={"X-Management-Token": first["management_token"]},
        ).status_code
        == 200
    )


def test_capacity_zero_means_unlimited(client: TestClient, configure) -> None:
    configure(max_shared_maps=0)
    assert all(_post(client) == 201 for _ in range(6))


def test_share_specific_size_limit(client: TestClient, configure) -> None:
    configure(max_share_size_mb=0.001)
    response = client.post("/api/shares", json={"title": "x", "map": build_map(node_count=50)})
    assert response.status_code == 413
    assert "limite de partage" in response.json()["detail"]


def test_capacity_is_not_crossed_by_simultaneous_creations(client: TestClient, configure) -> None:
    """La derniere place ne doit pas etre prise deux fois.

    Le comptage et l'insertion sont serialises par un verrou de processus :
    avec un seul service applicatif devant SQLite, cela suffit a ce que deux
    requetes concurrentes ne franchissent pas ensemble la limite.
    """
    import threading

    configure(max_shared_maps=5)
    for _ in range(4):
        assert _post(client) == 201

    results: list[int] = []
    lock = threading.Lock()

    def attempt() -> None:
        status = client.post("/api/shares", json={"title": "x", "map": build_map()}).status_code
        with lock:
            results.append(status)

    threads = [threading.Thread(target=attempt) for _ in range(8)]
    for thread in threads:
        thread.start()
    for thread in threads:
        thread.join()

    assert results.count(201) == 1
    assert results.count(503) == 7

    listing = client.get("/api/config")
    assert listing.status_code == 200