devplacepy/tests/api/auth/terms.py

# retoor <retoor@molodetz.nl>

import time

import requests

from devplacepy.database import get_setting, get_table, refresh_snapshot, set_setting
from devplacepy.routers.auth.terms import TERMS_ACCEPTANCE_CODE
from tests.conftest import BASE_URL

JSON = {"Accept": "application/json"}
_counter = [0]


def _unique(prefix="terms"):
    _counter[0] += 1
    return f"{prefix}{int(time.time() * 1000)}{_counter[0]}"


def _signup(**overrides):
    name = _unique()
    payload = {
        "username": name,
        "email": f"{name}@t.dev",
        "password": "secret123",
        "confirm_password": "secret123",
        "birth_date": "1990-01-01",
        "accept_terms": "1",
    }
    payload.update(overrides)
    session = requests.Session()
    response = session.post(
        f"{BASE_URL}/auth/signup", data=payload, headers=JSON, allow_redirects=False
    )
    return session, payload["username"], response


def test_signup_without_accepting_the_terms_is_refused(app_server):
    _, name, response = _signup(accept_terms="")
    assert response.status_code == 422
    assert "accept" in response.text.lower()
    refresh_snapshot()
    assert get_table("users").find_one(username=name) is None


def test_signup_below_the_minimum_age_is_refused(app_server):
    from datetime import date

    young = date.today().replace(year=date.today().year - 12).isoformat()
    _, name, response = _signup(birth_date=young)
    assert response.status_code == 422
    assert "years old" in response.text
    refresh_snapshot()
    assert get_table("users").find_one(username=name) is None


def test_signup_stores_the_band_and_the_acceptance_but_not_the_date(app_server):
    _, name, response = _signup(birth_date="1990-06-15")
    assert response.status_code == 200, response.text
    refresh_snapshot()
    row = get_table("users").find_one(username=name)
    assert row["age_band"] == "adult"
    assert row["terms_version"] == (get_setting("terms_version", "1") or "1")
    assert row["terms_accepted_at"]
    assert "1990-06-15" not in str(dict(row))
    consents = {
        entry["kind"]
        for entry in get_table("user_consents").find(
            owner_kind="user", owner_id=row["uid"], state="granted", deleted_at=None
        )
    }
    assert {"terms", "privacy", "activity_recording"} <= consents
    assert "ai_third_party" not in consents


def test_the_acceptance_page_reports_both_versions(app_server):
    session, name, response = _signup()
    assert response.status_code == 200
    page = session.get(f"{BASE_URL}/auth/accept-terms", headers=JSON)
    assert page.status_code == 200, page.text
    body = page.json()
    in_force = get_setting("terms_version", "1") or "1"
    assert body["terms_version"] == in_force
    assert body["accepted_version"] == in_force
    assert "consents" not in body
    refresh_snapshot()
    assert get_table("users").find_one(username=name)["terms_version"] == in_force


def test_the_acceptance_page_is_signed_in_only(app_server):
    assert (
        requests.get(f"{BASE_URL}/auth/accept-terms", headers=JSON).status_code == 401
    )


def test_the_acceptance_page_shows_the_gap_after_a_bump(app_server):
    session, _, response = _signup()
    assert response.status_code == 200
    original = get_setting("terms_version", "1") or "1"
    set_setting("terms_version", f"{original}-gap")
    try:
        body = {}
        for _ in range(40):
            body = session.get(f"{BASE_URL}/auth/accept-terms", headers=JSON).json()
            if body.get("terms_version") == f"{original}-gap":
                break
            time.sleep(0.5)
        assert body["terms_version"] == f"{original}-gap"
        assert body["accepted_version"] == original
    finally:
        set_setting("terms_version", original)


def test_bumping_the_version_gates_writes_but_never_reads(app_server):
    session, name, response = _signup()
    assert response.status_code == 200
    original = get_setting("terms_version", "1") or "1"
    set_setting("terms_version", f"{original}-bumped")
    try:
        blocked = None
        for _ in range(40):
            blocked = session.post(
                f"{BASE_URL}/posts/create",
                data={
                    "title": "gated",
                    "content": "This write must be gated by the terms version.",
                },
                headers=JSON,
            )
            if blocked.status_code == 403:
                break
            time.sleep(0.5)
        assert blocked.status_code == 403
        assert session.get(f"{BASE_URL}/feed", headers=JSON).status_code == 200
        assert session.get(f"{BASE_URL}/docs/terms.html").status_code == 200
        assert (
            session.get(f"{BASE_URL}/profile/{name}/delete", headers=JSON).status_code
            == 200
        )

        accepted = session.post(f"{BASE_URL}/auth/accept-terms", headers=JSON)
        assert accepted.status_code == 200
        refresh_snapshot()
        assert (
            get_table("users").find_one(username=name)["terms_version"]
            == f"{original}-bumped"
        )
        allowed = session.post(
            f"{BASE_URL}/posts/create",
            data={"title": "ok", "content": "Writing works again after acceptance."},
            headers=JSON,
        )
        assert allowed.status_code == 200
    finally:
        set_setting("terms_version", original)


def test_the_refusal_tells_a_fetch_client_how_to_resolve_it(app_server):
    session, _, response = _signup()
    assert response.status_code == 200
    original = get_setting("terms_version", "1") or "1"
    bumped = f"{original}-contract"
    set_setting("terms_version", bumped)
    try:
        blocked = None
        for _ in range(40):
            blocked = session.post(
                f"{BASE_URL}/posts/create",
                data={"title": "gated", "content": "This write must be gated."},
                headers=JSON,
            )
            if blocked.status_code == 403:
                break
            time.sleep(0.5)
        assert blocked.status_code == 403
        error = blocked.json()["error"]
        assert error["code"] == TERMS_ACCEPTANCE_CODE
        assert error["redirect"] == "/auth/accept-terms"
        assert error["terms_version"] == bumped
        assert error["message"]
    finally:
        set_setting("terms_version", original)


def test_a_browser_form_post_is_redirected_to_the_acceptance_page(app_server):
    session, _, response = _signup()
    assert response.status_code == 200
    original = get_setting("terms_version", "1") or "1"
    set_setting("terms_version", f"{original}-redirect")
    try:
        blocked = None
        for _ in range(40):
            blocked = session.post(
                f"{BASE_URL}/posts/create",
                data={"title": "gated", "content": "This write must be gated."},
                headers={"Accept": "text/html"},
                allow_redirects=False,
            )
            if blocked.status_code == 303:
                break
            time.sleep(0.5)
        assert blocked.status_code == 303
        assert blocked.headers["location"] == "/auth/accept-terms"
    finally:
        set_setting("terms_version", original)