devplacepy/tests/api/issues/create.py

# retoor <retoor@molodetz.nl>

import time
from datetime import datetime, timezone
import pytest
import requests
from tests.conftest import BASE_URL
from devplacepy.database import get_table, refresh_snapshot, set_setting
from devplacepy.utils import generate_uid, make_combined_slug
JSON_audit_log = {"Accept": "application/json"}
_counter_audit_log = [0]
@pytest.fixture(scope="module", autouse=True)
def _audit_test_settings(app_server):
    # On a fresh DB init_db skips seeding the operational/upload settings (its
    # `tables` snapshot predates site_settings creation), so those rows are
    # absent and the admin settings form would INSERT them as "" - which both
    # closes registration and makes consumers that do int("") crash. Seed sane
    # values here so the form's empty submissions are skipped (existing key), and
    # lift the per-IP rate limit since this file fires many mutating requests.
    for key, value in {
        "rate_limit_per_minute": "1000000",
        "rate_limit_window_seconds": "60",
        "registration_open": "1",
        "maintenance_mode": "0",
        "max_upload_size_mb": "10",
        "allowed_file_types": "",
        "max_attachments_per_resource": "10",
        "session_max_age_days": "7",
        "session_remember_days": "30",
        "news_service_interval": "3600",
        "news_grade_threshold": "7",
    }.items():
        set_setting(key, value)
    yield
def _db_user(name):
    # the user is created by the server subprocess; refresh the test-process
    # SQLite snapshot before reading it back across the process boundary.
    refresh_snapshot()
    return get_table("users").find_one(username=name)
def _unique(prefix="au"):
    _counter_audit_log[0] += 1
    return f"{prefix}{int(time.time() * 1000)}{_counter_audit_log[0]}"
def _member():
    name = _unique("aumem")
    s = requests.Session()
    s.post(
        f"{BASE_URL}/auth/signup",
        data={
            "username": name,
            "email": f"{name}@t.dev",
            "password": "secret123",
            "confirm_password": "secret123",
        },
        allow_redirects=True,
    )
    return s, name
def _member_key():
    _, name = _member()
    return _db_user(name)["api_key"]
def _admin(seeded_db):
    # authenticate via the seeded admin's API key (header auth) rather than a
    # login POST - GET reads are exempt from the rate limiter, so reusing this
    # across the file's many tests never counts against the per-IP write budget.
    key = _db_user("alice_test")["api_key"]
    s = requests.Session()
    s.headers.update({"X-API-KEY": key})
    return s
def _audit(admin, **params):
    r = admin.get(f"{BASE_URL}/admin/audit-log", headers=JSON_audit_log, params=params)
    assert r.status_code == 200, r.text[:300]
    return r.json()
def _find(admin, event_key, predicate):
    data = _audit(admin, event_key=event_key)
    for entry in data["entries"]:
        if predicate(entry):
            return entry
    return None
def _new_post(session, body="audited post body here"):
    return session.post(
        f"{BASE_URL}/posts/create",
        headers=JSON_audit_log,
        data={"title": _unique("aup"), "content": body, "topic": "devlog"},
    ).json()["data"]
def _new_project(session):
    return session.post(
        f"{BASE_URL}/projects/create",
        headers=JSON_audit_log,
        data={
            "title": _unique("aupr"),
            "description": "audited project description text",
            "project_type": "software",
            "status": "In Development",
            "platforms": "",
        },
    ).json()["data"]
def _seed_news_audit_log():
    uid = generate_uid()
    title = _unique("aunews")
    get_table("news").insert(
        {
            "deleted_at": None,
            "deleted_by": None,
            "uid": uid,
            "slug": make_combined_slug(title, uid),
            "title": title,
            "external_id": uid,
            "status": "draft",
            "featured": 0,
            "show_on_landing": 0,
            "grade": 5,
            "source_name": "AuditTest",
            "synced_at": datetime.now(timezone.utc).isoformat(),
            "description": "audited news article",
        }
    )
    refresh_snapshot()
    return uid
import asyncio
from devplacepy.database import (
    get_table,
    init_db,
    refresh_snapshot,
    set_setting,
)
from devplacepy.services.gitea import runtime, store
from devplacepy.services.gitea.config import gitea_config
from devplacepy.services.gitea.enhance import enhance_ticket
from devplacepy.services.gitea.fake import FakeGiteaClient
from devplacepy.services.gitea.service import IssueTrackerService
from devplacepy.services.jobs import queue
from devplacepy.services.jobs.issue_create_service import IssueCreateService
from tests.conftest import run_async
_counter_issues_gitea = [0]
@pytest.fixture(autouse=True)
def _init_db_issues_gitea():
    init_db()
    yield
@pytest.fixture
def gitea_env():
    fake = FakeGiteaClient()
    runtime.set_client(fake)
    set_setting("gitea_base_url", "https://gitea.test")
    set_setting("gitea_owner", "retoor")
    set_setting("gitea_repo", "pydevplace")
    set_setting("gitea_token", "test-token")
    set_setting("issue_ai_enhance", "0")
    yield fake
    runtime.set_client(None)
    for key in ("gitea_base_url", "gitea_owner", "gitea_repo", "gitea_token"):
        set_setting(key, "")
    for table in ("issue_tickets", "issue_comment_authors"):
        for row in list(get_table(table).find()):
            get_table(table).delete(uid=row["uid"])
    for row in list(get_table("jobs").find(kind="issue_create")):
        get_table("jobs").delete(uid=row["uid"])
def _make_user_issues_gitea():
    _counter_issues_gitea[0] += 1
    uid = f"issuetest-user-{_counter_issues_gitea[0]}"
    username = f"issuetester{_counter_issues_gitea[0]}"
    # upsert, not insert: tests/unit/services/gitea.py is a twin fixture sharing this
    # uid namespace with its own counter, so a combined unit+api run collides on the uid
    get_table("users").upsert(
        {
            "uid": uid,
            "username": username,
            "role": "member",
            "xp": 0,
            "level": 1,
            "is_active": True,
        },
        ["uid"],
    )
    return uid, username
def _unread(user_uid):
    return [
        n
        for n in get_table("notifications").find(user_uid=user_uid)
        if n.get("type") == "issue"
    ]
def _drive_jobs():
    async def run():
        svc = IssueCreateService()
        for _ in range(200):
            await svc.run_once()
            refresh_snapshot()
            pending = [
                r
                for r in get_table("jobs").find(kind="issue_create")
                if r["status"] in ("pending", "running")
            ]
            if not pending and not svc._inflight:
                return
            await asyncio.sleep(0.02)

    run_async(run())
def _role_user(role: str):
    uid, username = _make_user_issues_gitea()
    get_table("users").update({"uid": uid, "role": role}, ["uid"])
    return uid, username
def _open_ticket(fake, role: str):
    import devplacepy.main as m
    from starlette.testclient import TestClient
    from devplacepy.utils import create_session

    uid, _ = _role_user(role)
    issue = run_async(fake.create_issue("Submit does nothing", "details"))
    number = int(issue["number"])
    store.record_ticket(
        number=number,
        author_uid=uid,
        original_title="Submit does nothing",
        original_description="details",
        enhanced_title="Submit does nothing",
        html_url=issue.get("html_url", ""),
        status="open",
    )
    client = TestClient(m.app)
    client.cookies.set("session", create_session(uid))
    return number, client


def test_issue_create_recorded(seeded_db):
    # Filing is async + Gitea-gated: POST /issues/create only enqueues a job (returns
    # {uid, status_url}) and records the synchronous "issue.create.request" event; the
    # final "issue.create" fires from IssueCreateService, which the tests never run
    # (DEVPLACE_DISABLE_SERVICES=1). Configure Gitea so the enqueue path is taken
    # (no network call happens here) and assert the request event. Restore after.
    s, _ = _member()
    title = _unique("auissue")
    for key, value in {
        "gitea_base_url": "https://gitea.test",
        "gitea_owner": "o",
        "gitea_repo": "devplacepy",
        "gitea_token": "audit-test-token",
    }.items():
        set_setting(key, value)
    try:
        resp = s.post(
            f"{BASE_URL}/issues/create",
            headers=JSON_audit_log,
            data={"title": title, "description": "audited issue description text"},
        )
        assert resp.status_code == 200, resp.text[:300]
        assert resp.json().get("uid")
        admin = _admin(seeded_db)
        assert (
            _find(
                admin,
                "issue.create.request",
                lambda e: title in (e.get("summary") or ""),
            )
            is not None
        )
    finally:
        for key in ("gitea_base_url", "gitea_owner", "gitea_repo", "gitea_token"):
            set_setting(key, "")


def test_issue_detail_renders_for_member(gitea_env):
    number, client = _open_ticket(gitea_env, "Member")
    html = client.get(f"/issues/{number}")
    assert html.status_code == 200, html.text[:300]
    assert "Submit does nothing" in html.text
    assert "Close ticket" not in html.text
    data = client.get(
        f"/issues/{number}", headers={"Accept": "application/json"}
    ).json()
    assert data["viewer_is_admin"] is False


def test_issue_detail_renders_for_admin(gitea_env):
    number, client = _open_ticket(gitea_env, "Admin")
    html = client.get(f"/issues/{number}")
    assert html.status_code == 200, html.text[:300]
    assert "Close ticket" in html.text
    data = client.get(
        f"/issues/{number}", headers={"Accept": "application/json"}
    ).json()
    assert data["viewer_is_admin"] is True