devplacepy/tests/api/content/permissions.py

# retoor <retoor@molodetz.nl>

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

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


@pytest.fixture(scope="module", autouse=True)
def _perm_settings(app_server):
    for key, value in {
        "rate_limit_per_minute": "1000000",
        "rate_limit_window_seconds": "60",
        "registration_open": "1",
        "maintenance_mode": "0",
        "max_attachments_per_resource": "10",
        "allowed_file_types": "",
        "happy_404_enabled": "0",
    }.items():
        set_setting(key, value)
    time.sleep(CACHE_VERSION_PROPAGATION_SECONDS)
    yield
    set_setting("happy_404_enabled", "1")
    time.sleep(CACHE_VERSION_PROPAGATION_SECONDS)


def _member():
    _counter_perm[0] += 1
    name = f"perm{int(time.time() * 1000)}{_counter_perm[0]}"
    s = requests.Session()
    s.post(
        f"{BASE_URL}/auth/signup",
        data={
            "username": name,
            "email": f"{name}@t.dev",
            "password": "secret123",
            "confirm_password": "secret123",
            "birth_date": "1990-01-01",
            "accept_terms": "1",
        },
        allow_redirects=True,
    )
    refresh_snapshot()
    return s, name, get_table("users").find_one(username=name)["uid"]


def _admin_session(seeded_db):
    refresh_snapshot()
    key = get_table("users").find_one(username="alice_test")["api_key"]
    s = requests.Session()
    s.headers.update({"X-API-KEY": key, **JSON})
    return s, get_table("users").find_one(username="alice_test")["uid"]


def _create_post(session):
    session.post(
        f"{BASE_URL}/posts/create",
        headers=JSON,
        data={
            "title": f"perm post {_counter_perm[0]}",
            "content": "a post body for permission tests",
            "topic": "devlog",
        },
    )
    refresh_snapshot()


def _latest_post(uid):
    refresh_snapshot()
    rows = list(get_table("posts").find(user_uid=uid, deleted_at=None))
    return max(rows, key=lambda r: r["id"])


def test_admin_deletes_member_post(seeded_db):
    member, _, muid = _member()
    _create_post(member)
    post = _latest_post(muid)
    admin, admin_uid = _admin_session(seeded_db)
    r = admin.post(f"{BASE_URL}/posts/delete/{post['slug']}")
    assert r.status_code in (200, 302), r.text[:200]
    refresh_snapshot()
    row = get_table("posts").find_one(uid=post["uid"])
    assert row["deleted_at"] is not None
    assert row["deleted_by"] == admin_uid


def test_member_cannot_delete_other_members_post(seeded_db):
    owner, _, ouid = _member()
    _create_post(owner)
    post = _latest_post(ouid)
    other, _, _ = _member()
    r = other.post(f"{BASE_URL}/posts/delete/{post['slug']}", headers=JSON, allow_redirects=False)
    assert r.status_code in (302, 403)
    refresh_snapshot()
    assert get_table("posts").find_one(uid=post["uid"])["deleted_at"] is None


def test_admin_deletes_member_comment(seeded_db):
    member, _, muid = _member()
    _create_post(member)
    post = _latest_post(muid)
    member.post(
        f"{BASE_URL}/comments/create",
        headers=JSON,
        data={
            "content": "member comment to be removed by admin",
            "target_type": "post",
            "target_uid": post["uid"],
            "post_uid": post["uid"],
        },
    )
    refresh_snapshot()
    comment = max(
        get_table("comments").find(target_uid=post["uid"], deleted_at=None),
        key=lambda c: c["id"],
    )
    admin, admin_uid = _admin_session(seeded_db)
    r = admin.post(f"{BASE_URL}/comments/delete/{comment['uid']}")
    assert r.status_code in (200, 302), r.text[:200]
    refresh_snapshot()
    assert get_table("comments").find_one(uid=comment["uid"])["deleted_at"] is not None


def test_admin_deletes_member_attachment(seeded_db):
    member, _, muid = _member()
    refresh_snapshot()
    key = get_table("users").find_one(uid=muid)["api_key"]
    up = requests.post(
        f"{BASE_URL}/uploads/upload",
        headers={"X-API-KEY": key},
        files={"file": ("note.txt", b"attachment bytes", "text/plain")},
    )
    assert up.status_code == 201, up.text[:200]
    att_uid = up.json()["uid"]
    admin, admin_uid = _admin_session(seeded_db)
    r = admin.delete(f"{BASE_URL}/uploads/delete/{att_uid}")
    assert r.status_code == 200, r.text[:200]
    refresh_snapshot()
    assert get_table("attachments").find_one(uid=att_uid)["deleted_at"] is not None


def test_soft_deleted_post_returns_404(app_server):
    member, _, muid = _member()
    _create_post(member)
    post = _latest_post(muid)
    member.post(f"{BASE_URL}/posts/delete/{post['slug']}", headers=JSON)
    refresh_snapshot()
    r = requests.get(f"{BASE_URL}/posts/{post['slug']}")
    assert r.status_code == 404


def test_soft_deleted_gist_returns_404(app_server):
    member, _, muid = _member()
    member.post(
        f"{BASE_URL}/gists/create",
        headers=JSON,
        data={
            "title": "perm gist",
            "description": "gist for soft delete check",
            "language": "python",
            "source_code": "print('hi')",
        },
    )
    refresh_snapshot()
    gist = max(get_table("gists").find(user_uid=muid, deleted_at=None), key=lambda g: g["id"])
    member.post(f"{BASE_URL}/gists/delete/{gist['slug']}", headers=JSON)
    refresh_snapshot()
    r = requests.get(f"{BASE_URL}/gists/{gist['slug']}")
    assert r.status_code == 404