devplacepy/tests/api/topics.py

# retoor <retoor@molodetz.nl>

import time
from datetime import datetime, timedelta, timezone
import pytest
import requests
from tests.conftest import BASE_URL, CACHE_VERSION_PROPAGATION_SECONDS
from devplacepy.database import get_table, refresh_snapshot, set_setting
from devplacepy.database.pagination import PAGE_SIZE
from devplacepy.utils import generate_uid

JSON_topics = {"Accept": "application/json"}
_counter_topics = [0]


@pytest.fixture
def _no_happy_404(app_server):
    set_setting("happy_404_enabled", "0")
    time.sleep(CACHE_VERSION_PROPAGATION_SECONDS)
    yield
    set_setting("happy_404_enabled", "1")
    time.sleep(CACHE_VERSION_PROPAGATION_SECONDS)


def _session_topics():
    _counter_topics[0] += 1
    name = f"tpc{int(time.time() * 1000)}{_counter_topics[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,
    )
    return s, name


def _create_post_topics(session, title, topic):
    r = session.post(
        f"{BASE_URL}/posts/create",
        data={"content": f"Post body for {title}", "title": title, "topic": topic},
        allow_redirects=False,
    )
    return r.headers["location"].split("/posts/")[-1]


def _create_post_direct_topics(user_uid, topic, order, marker=None):
    uid = generate_uid()
    marker = marker or f"tpc-{uid[:8]}"
    get_table("posts").insert(
        {
            "deleted_at": None,
            "deleted_by": None,
            "uid": uid,
            "user_uid": user_uid,
            "slug": f"{uid[:8]}-{topic}-post",
            "title": marker,
            "content": f"Direct topic post content {order}",
            "topic": topic,
            "project_uid": None,
            "image": None,
            "stars": 0,
            "created_at": (
                datetime.now(timezone.utc) + timedelta(seconds=order)
            ).isoformat(),
        }
    )
    return uid, marker


def _topic_hub_entry(topic):
    r = requests.get(f"{BASE_URL}/topics", headers=JSON_topics)
    data = r.json()
    return next(t for t in data["topics"] if t["key"] == topic)


def test_topics_hub_lists_every_topic(app_server):
    r = requests.get(f"{BASE_URL}/topics", headers=JSON_topics)
    assert r.status_code == 200
    data = r.json()
    keys = {t["key"] for t in data["topics"]}
    assert keys == {"devlog", "showcase", "question", "rant", "fun", "random", "politics"}


def test_topics_hub_post_count_reflects_new_posts(app_server):
    s, name = _session_topics()
    refresh_snapshot()
    user_uid = get_table("users").find_one(username=name)["uid"]
    before = _topic_hub_entry("rant")["post_count"]
    for i in range(3):
        _create_post_direct_topics(user_uid, "rant", i)
    refresh_snapshot()
    after = _topic_hub_entry("rant")["post_count"]
    assert after - before == 3


def test_topic_page_lists_only_that_topics_posts(app_server):
    s, _ = _session_topics()
    unique = int(time.time() * 1000)
    devlog_title = f"devlog-only-{unique}"
    showcase_title = f"showcase-only-{unique}"
    _create_post_topics(s, devlog_title, "devlog")
    _create_post_topics(s, showcase_title, "showcase")

    r = requests.get(f"{BASE_URL}/topics/devlog", headers=JSON_topics)
    assert r.status_code == 200
    data = r.json()
    assert data["topic"] == "devlog"
    titles = [item["post"]["title"] for item in data["posts"]]
    assert devlog_title in titles
    assert showcase_title not in titles


def test_topic_page_rejects_an_unknown_topic(app_server, _no_happy_404):
    r = requests.get(f"{BASE_URL}/topics/not-a-real-topic", allow_redirects=False)
    assert r.status_code == 404


def test_topic_page_canonical_and_breadcrumbs_are_topic_specific(app_server):
    r = requests.get(f"{BASE_URL}/topics/showcase", allow_redirects=False)
    assert r.status_code == 200
    assert 'href="' in r.text
    assert "/topics/showcase" in r.text
    assert "Topics" in r.text


def test_topic_page_pagination_crosses_a_page_boundary(app_server):
    s, name = _session_topics()
    refresh_snapshot()
    user_uid = get_table("users").find_one(username=name)["uid"]
    count = PAGE_SIZE + 1
    markers = []
    for i in range(count):
        _, marker = _create_post_direct_topics(user_uid, "fun", i, marker=f"tpcpag-{i}")
        markers.append(marker)
    refresh_snapshot()

    r = requests.get(f"{BASE_URL}/topics/fun", headers=JSON_topics)
    assert r.status_code == 200
    body = r.json()
    assert len(body["posts"]) == PAGE_SIZE
    assert body["next_cursor"] is not None

    r2 = requests.get(
        f"{BASE_URL}/topics/fun",
        headers=JSON_topics,
        params={"before": body["next_cursor"]},
    )
    assert r2.status_code == 200
    body2 = r2.json()
    assert len(body2["posts"]) >= 1

    seen_titles = {item["post"]["title"] for item in body["posts"]} | {
        item["post"]["title"] for item in body2["posts"]
    }
    assert set(markers).issubset(seen_titles)