devplacepy/tests/api/devrant/auth.py

# retoor <retoor@molodetz.nl>

import time
import pytest
import requests
from tests.conftest import BASE_URL
from devplacepy.database import get_table, refresh_snapshot, set_setting

_counter_dr = [0]


@pytest.fixture(scope="module", autouse=True)
def _devrant_settings(app_server):
    for key, value in {
        "rate_limit_per_minute": "1000000",
        "rate_limit_window_seconds": "60",
        "registration_open": "1",
        "maintenance_mode": "0",
        "devrant_api_enabled": "1",
        "session_max_age_days": "7",
    }.items():
        set_setting(key, value)
    yield


def _name(prefix="dra"):
    _counter_dr[0] += 1
    return f"{prefix}{int(time.time() * 1000)}{_counter_dr[0]}"


def _register(password="secret123"):
    name = _name()
    email = f"{name}@t.dev"
    body = None
    for _ in range(20):
        body = requests.post(
            f"{BASE_URL}/api/users",
            data={"username": name, "email": email, "password": password},
        ).json()
        if body.get("success"):
            break
        time.sleep(0.3)
    return name, email, password, body


def _auth(name, password):
    r = requests.post(
        f"{BASE_URL}/api/users/auth-token",
        data={"username": name, "password": password},
    )
    return r


def _token_params(name, password):
    at = _auth(name, password).json()["auth_token"]
    return {"token_id": at["id"], "token_key": at["key"], "user_id": at["user_id"]}


def test_auth_token_success(app_server):
    name, email, password, _ = _register()
    r = _auth(name, password)
    assert r.status_code == 200, r.text[:300]
    body = r.json()
    assert body["success"] is True
    at = body["auth_token"]
    assert at["id"] and at["key"] and at["user_id"]


def test_auth_token_accepts_email(app_server):
    name, email, password, _ = _register()
    r = requests.post(
        f"{BASE_URL}/api/users/auth-token",
        data={"username": email, "password": password},
    )
    assert r.status_code == 200
    assert r.json()["success"] is True


def test_auth_token_wrong_password(app_server):
    name, email, password, _ = _register()
    r = _auth(name, "wrongpass")
    assert r.status_code == 400
    assert r.json()["success"] is False


def test_auth_token_missing_fields(app_server):
    r = requests.post(f"{BASE_URL}/api/users/auth-token", data={"username": "x"})
    assert r.status_code == 400
    assert r.json()["success"] is False


def test_register_creates_user(app_server):
    name, email, password, body = _register()
    assert body["success"] is True
    assert body["auth_token"]["key"]
    refresh_snapshot()
    assert get_table("users").find_one(username=name) is not None


def test_register_duplicate_username(app_server):
    name, email, password, _ = _register()
    r = requests.post(
        f"{BASE_URL}/api/users",
        data={"username": name, "email": f"{_name()}@t.dev", "password": password},
    )
    assert r.json()["success"] is False
    assert r.json().get("error_field") == "username"


def test_register_short_password(app_server):
    r = requests.post(
        f"{BASE_URL}/api/users",
        data={"username": _name(), "email": f"{_name()}@t.dev", "password": "x"},
    )
    assert r.json()["success"] is False
    assert r.json().get("error_field") == "password"


def test_register_closed(app_server):
    set_setting("registration_open", "0")
    try:
        closed = False
        for _ in range(20):
            r = requests.post(
                f"{BASE_URL}/api/users",
                data={
                    "username": _name(),
                    "email": f"{_name()}@t.dev",
                    "password": "secret123",
                },
            )
            if r.json()["success"] is False:
                closed = True
                break
            time.sleep(0.3)
        assert closed, "registration should be reported closed"
    finally:
        set_setting("registration_open", "1")
        for _ in range(20):
            probe = requests.post(
                f"{BASE_URL}/api/users",
                data={
                    "username": _name(),
                    "email": f"{_name()}@t.dev",
                    "password": "secret123",
                },
            ).json()
            if probe.get("success"):
                break
            time.sleep(0.3)


def test_get_user_id(app_server):
    name, email, password, _ = _register()
    refresh_snapshot()
    r = requests.get(f"{BASE_URL}/api/get-user-id", params={"username": name})
    assert r.json()["success"] is True
    assert isinstance(r.json()["user_id"], int)


def test_get_user_id_unknown(app_server):
    r = requests.get(
        f"{BASE_URL}/api/get-user-id", params={"username": "no_such_zzz"}
    )
    assert r.json()["success"] is False


def test_profile_by_id(app_server):
    name, email, password, _ = _register()
    refresh_snapshot()
    uid = requests.get(
        f"{BASE_URL}/api/get-user-id", params={"username": name}
    ).json()["user_id"]
    r = requests.get(f"{BASE_URL}/api/users/{uid}")
    assert r.json()["success"] is True
    assert r.json()["profile"]


def test_edit_profile_requires_auth(app_server):
    r = requests.post(
        f"{BASE_URL}/api/users/me/edit-profile", data={"profile_about": "hi"}
    )
    assert r.status_code == 401
    assert r.json()["success"] is False


def test_edit_profile_updates_bio(app_server):
    name, email, password, _ = _register()
    params = _token_params(name, password)
    r = requests.post(
        f"{BASE_URL}/api/users/me/edit-profile",
        data={**params, "profile_about": "devrant bio text"},
    )
    assert r.json()["success"] is True
    refresh_snapshot()
    assert get_table("users").find_one(username=name)["bio"] == "devrant bio text"


def test_delete_account_deletes_and_revokes(app_server):
    name, email, password, _ = _register()
    refresh_snapshot()
    uid = get_table("users").find_one(username=name)["uid"]
    params = _token_params(name, password)
    r = requests.delete(f"{BASE_URL}/api/users/me", data=params)
    assert r.json()["success"] is True
    refresh_snapshot()
    assert get_table("users").find_one(username=name) is None
    user = get_table("users").find_one(uid=uid)
    assert user["is_active"] in (False, 0)
    assert user["username"].startswith("deleted_")
    assert not user["email"]
    assert not user["password_hash"]
    assert (
        get_table("devrant_tokens").count(user_uid=uid, deleted_at=None) == 0
    )


def test_avatar_png(app_server):
    r = requests.get(f"{BASE_URL}/api/avatars/u/seedvalue.png")
    assert r.status_code == 200
    assert "image" in r.headers.get("content-type", "")


def test_disabled_api_returns_404(app_server):
    set_setting("devrant_api_enabled", "0")
    try:
        disabled = False
        for _ in range(20):
            if requests.get(f"{BASE_URL}/api/devrant/rants").status_code == 404:
                disabled = True
                break
            time.sleep(0.3)
        assert disabled, "devrant api should report disabled"
    finally:
        set_setting("devrant_api_enabled", "1")
        for _ in range(20):
            if requests.get(f"{BASE_URL}/api/devrant/rants").status_code == 200:
                break
            time.sleep(0.3)