devplacepy/tests/api/projects/index.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",
            "birth_date": "1990-01-01",
            "accept_terms": "1",
        },
        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
from devplacepy.database import get_table
from devplacepy.utils import clear_user_cache
from devplacepy import project_files
from devplacepy.project_files import ProjectFileError
from devplacepy.services.devii.actions.dispatcher import (
    confirmation_error,
    _is_confirmed,
)
_counter_project_visibility = [0]
def _signup_project_visibility():
    _counter_project_visibility[0] += 1
    name = f"pv{int(time.time() * 1000)}{_counter_project_visibility[0]}"
    session = requests.Session()
    session.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,
    )
    row = get_table("users").find_one(username=name)
    return name, row["uid"], row["api_key"]
def _make_admin_project_visibility():
    name, uid, key = _signup_project_visibility()
    get_table("users").update({"uid": uid, "role": "Admin"}, ["uid"])
    clear_user_cache(uid)
    return name, uid, key
def _h_project_visibility(key=None):
    headers = {"Accept": "application/json"}
    if key:
        headers["X-API-KEY"] = key
    return headers
def _create_project_project_visibility(key, title, is_private=False):
    data = {
        "title": title,
        "description": "visibility test",
        "project_type": "software",
        "status": "In Development",
    }
    if is_private:
        data["is_private"] = "on"
    r = requests.post(f"{BASE_URL}/projects/create", headers=_h_project_visibility(key), data=data)
    assert r.status_code == 200, r.text
    return r.json()["data"]
def _project_uid(slug):
    return get_table("projects").find_one(slug=slug)["uid"]
def _write_project_visibility(key, slug, path, content):
    return requests.post(
        f"{BASE_URL}/projects/{slug}/files/write",
        headers=_h_project_visibility(key),
        data={"path": path, "content": content},
        allow_redirects=False,
    )
def _set_private(key, slug, value):
    return requests.post(
        f"{BASE_URL}/projects/{slug}/private",
        headers=_h_project_visibility(key),
        data={"value": 1 if value else 0},
        allow_redirects=False,
    )
def _set_readonly(key, slug, value):
    return requests.post(
        f"{BASE_URL}/projects/{slug}/readonly",
        headers=_h_project_visibility(key),
        data={"value": 1 if value else 0},
        allow_redirects=False,
    )
def _list_slugs(key=None, user_uid=None):
    params = {"user_uid": user_uid} if user_uid else None
    r = requests.get(f"{BASE_URL}/projects", headers=_h_project_visibility(key), params=params)
    return [p["slug"] for p in r.json()["projects"]]
def _seed_news_seo():
    from datetime import datetime, timezone
    from devplacepy.database import get_table
    from devplacepy.utils import generate_uid, make_combined_slug

    uid = generate_uid()
    title = f"SEO Test News Article {uid.split('-')[-1]}"
    slug = make_combined_slug(title, uid)
    get_table("news").insert(
        {
            "deleted_at": None,
            "deleted_by": None,
            "uid": uid,
            "slug": slug,
            "title": title,
            "description": "A seeded news article for SEO tests.",
            "content": "Body content for the seeded article.",
            "url": "https://example.com/article",
            "source_name": "ExampleSource",
            "status": "published",
            "synced_at": datetime.now(timezone.utc).isoformat(),
            "show_on_landing": 0,
            "grade": 8,
        }
    )
    return slug, uid
def _seed_owner():
    from datetime import datetime, timezone
    from uuid import uuid4
    from devplacepy.database import get_table

    uid = str(uuid4())
    get_table("users").insert(
        {
            "uid": uid,
            "username": f"seo_{uid[:8]}",
            "terms_version": "1",
            "email": f"{uid[:8]}@seo.test",
            "password_hash": "x",
            "role": "Member",
            "is_active": True,
            "created_at": datetime.now(timezone.utc).isoformat(),
        }
    )
    return uid
def _seed_post_seo(image=None):
    from datetime import datetime, timezone
    from uuid import uuid4
    from devplacepy.database import get_table
    from devplacepy.utils import make_combined_slug

    owner = _seed_owner()
    uid = str(uuid4())
    slug = make_combined_slug("SEO Detail Post", uid)
    get_table("posts").insert(
        {
            "deleted_at": None,
            "deleted_by": None,
            "uid": uid,
            "user_uid": owner,
            "slug": slug,
            "title": "SEO Detail Post",
            "content": "Body text for the SEO detail post.",
            "topic": "general",
            "project_uid": None,
            "image": image,
            "stars": 0,
            "created_at": datetime.now(timezone.utc).isoformat(),
        }
    )
    return slug, uid
def _seed_gist():
    from datetime import datetime, timezone
    from uuid import uuid4
    from devplacepy.database import get_table
    from devplacepy.utils import make_combined_slug

    owner = _seed_owner()
    uid = str(uuid4())
    slug = make_combined_slug("SEO Detail Gist", uid)
    get_table("gists").insert(
        {
            "deleted_at": None,
            "deleted_by": None,
            "uid": uid,
            "user_uid": owner,
            "slug": slug,
            "title": "SEO Detail Gist",
            "description": "Gist description for SEO tests.",
            "source_code": "print('seo')",
            "language": "python",
            "stars": 0,
            "created_at": datetime.now(timezone.utc).isoformat(),
        }
    )
    return slug, uid
def _seed_project():
    from datetime import datetime, timezone
    from uuid import uuid4
    from devplacepy.database import get_table
    from devplacepy.utils import make_combined_slug

    owner = _seed_owner()
    uid = str(uuid4())
    slug = make_combined_slug("SEO Detail Project", uid)
    get_table("projects").insert(
        {
            "deleted_at": None,
            "deleted_by": None,
            "uid": uid,
            "user_uid": owner,
            "slug": slug,
            "title": "SEO Detail Project",
            "description": "Project description for SEO tests.",
            "project_type": "software",
            "platforms": "Linux",
            "status": "Released",
            "stars": 0,
            "created_at": datetime.now(timezone.utc).isoformat(),
        }
    )
    return slug, uid
def _seed_news_image(news_uid):
    from devplacepy.database import get_table

    get_table("news_images").insert(
        {
            "deleted_at": None,
            "deleted_by": None,
            "news_uid": news_uid,
            "url": "https://example.com/seo-news-image.jpg",
        }
    )
def _seed_feed_posts(count):
    from datetime import datetime, timezone, timedelta
    from uuid import uuid4
    from devplacepy.database import get_table
    from devplacepy.utils import make_combined_slug

    owner = _seed_owner()
    topic = f"seopag{owner[:8]}"
    base = datetime(2026, 1, 1, tzinfo=timezone.utc)
    posts = get_table("posts")
    for i in range(count):
        uid = str(uuid4())
        posts.insert(
            {
                "deleted_at": None,
                "deleted_by": None,
                "uid": uid,
                "user_uid": owner,
                "slug": make_combined_slug(f"seo pag {i}", uid),
                "title": None,
                "content": f"seo pag post {i}",
                "topic": topic,
                "project_uid": None,
                "image": None,
                "stars": 0,
                "created_at": (base - timedelta(seconds=i)).isoformat(),
            }
        )
    return topic


def test_project_create_visibility_readonly_recorded(seeded_db):
    s, _ = _member()
    proj = _new_project(s)
    admin = _admin(seeded_db)
    assert _find(admin, "project.create", lambda e: e.get("target_uid") == proj["uid"]) is not None
    s.post(f"{BASE_URL}/projects/{proj['slug']}/private", headers=JSON_audit_log, data={"value": "true"}, allow_redirects=False)
    assert _find(admin, "project.visibility.private", lambda e: e.get("target_uid") == proj["uid"]) is not None
    s.post(f"{BASE_URL}/projects/{proj['slug']}/readonly", headers=JSON_audit_log, data={"value": "true"}, allow_redirects=False)
    assert _find(admin, "project.readonly.enable", lambda e: e.get("target_uid") == proj["uid"]) is not None


def test_project_fork_and_zip_request_recorded(seeded_db):
    s, _ = _member()
    proj = _new_project(s)
    s.post(f"{BASE_URL}/projects/{proj['slug']}/fork", headers=JSON_audit_log, data={"title": _unique("aufork")}, allow_redirects=False)
    s.post(f"{BASE_URL}/projects/{proj['slug']}/zip", headers=JSON_audit_log, allow_redirects=False)
    admin = _admin(seeded_db)
    assert _find(admin, "project.fork.request", lambda e: e.get("target_uid") == proj["uid"]) is not None
    assert _find(admin, "project.zip.request", lambda e: e.get("target_uid") == proj["uid"]) is not None


def test_project_file_write_and_readonly_denied_recorded(seeded_db):
    s, _ = _member()
    proj = _new_project(s)
    s.post(
        f"{BASE_URL}/projects/{proj['slug']}/files/write",
        headers=JSON_audit_log,
        data={"path": "audit_a.txt", "content": "hello world"},
        allow_redirects=False,
    )
    admin = _admin(seeded_db)
    assert _find(
        admin, "file.write.create", lambda e: e.get("target_uid") == f"{proj['uid']}:audit_a.txt"
    ) is not None
    # turn the project read-only, then a write must be recorded as denied
    s.post(f"{BASE_URL}/projects/{proj['slug']}/readonly", headers=JSON_audit_log, data={"value": "true"}, allow_redirects=False)
    s.post(
        f"{BASE_URL}/projects/{proj['slug']}/files/write",
        headers=JSON_audit_log,
        data={"path": "audit_b.txt", "content": "blocked"},
        allow_redirects=False,
    )
    denied = _find(
        admin,
        "file.write.create",
        lambda e: e.get("target_uid") == f"{proj['uid']}:audit_b.txt" and e.get("result") == "denied",
    )
    assert denied is not None


def test_private_detail_404_for_guest_200_for_owner(app_server):
    _, _, key = _signup_project_visibility()
    slug = _create_project_project_visibility(key, "Private Detail", is_private=True)["slug"]
    assert requests.get(f"{BASE_URL}/projects/{slug}", headers=_h_project_visibility()).status_code == 404
    assert (
        requests.get(f"{BASE_URL}/projects/{slug}", headers=_h_project_visibility(key)).status_code == 200
    )


def test_private_detail_visible_to_admin(app_server):
    _, _, owner_key = _signup_project_visibility()
    _, _, admin_key = _make_admin_project_visibility()
    slug = _create_project_project_visibility(owner_key, "Private AdminDetail", is_private=True)["slug"]
    assert (
        requests.get(f"{BASE_URL}/projects/{slug}", headers=_h_project_visibility(admin_key)).status_code
        == 200
    )


def _await_workspace_flag(url, key, expected, timeout=10.0):
    deadline = time.time() + timeout
    while time.time() < deadline:
        body = requests.get(url, headers=_h_project_visibility(key)).json()
        if body["viewer_can_workspace"] is expected:
            return True
        time.sleep(0.2)
    return False


def test_detail_json_exposes_viewer_can_workspace(app_server):
    _, _, owner_key = _signup_project_visibility()
    _, _, other_key = _signup_project_visibility()
    slug = _create_project_project_visibility(owner_key, "Workspace Flag")["slug"]
    url = f"{BASE_URL}/projects/{slug}"
    previous = get_table("site_settings").find_one(key="workspace_enabled")
    set_setting("workspace_enabled", "1")
    try:
        assert _await_workspace_flag(url, owner_key, True)
        other = requests.get(url, headers=_h_project_visibility(other_key)).json()
        assert other["viewer_can_workspace"] is False

        set_setting("workspace_enabled", "0")
        assert _await_workspace_flag(url, owner_key, False)
    finally:
        set_setting(
            "workspace_enabled", previous.get("value") if previous else "0"
        )


def test_project_uid_redirects_to_canonical_slug(app_server):
    slug, uid = _seed_project()
    r = requests.get(f"{BASE_URL}/projects/{uid}", allow_redirects=False)
    assert r.status_code == 301
    assert r.headers["location"].endswith(f"/projects/{slug}"), r.headers.get(
        "location"
    )


def test_private_detail_is_noindex_for_owner(app_server):
    _, _, key = _signup_project_visibility()
    slug = _create_project_project_visibility(key, "Private NoIndex", is_private=True)["slug"]
    r = requests.get(f"{BASE_URL}/projects/{slug}", headers={"X-API-KEY": key})
    assert r.status_code == 200, r.text[:300]
    assert '<meta name="robots" content="noindex,nofollow">' in r.text


def test_public_detail_is_indexable(app_server):
    _, _, key = _signup_project_visibility()
    slug = _create_project_project_visibility(key, "Public Indexable", is_private=False)["slug"]
    r = requests.get(f"{BASE_URL}/projects/{slug}", headers={"X-API-KEY": key})
    assert r.status_code == 200, r.text[:300]
    assert '<meta name="robots" content="index,follow">' in r.text


def test_viewing_project_marks_notification_read(app_server):
    from devplacepy.database import resolve_object_url

    name, uid, key = _signup_project_visibility()
    project = _create_project_project_visibility(key, "Notif Mark Read")
    refresh_snapshot()
    target_url = resolve_object_url("project", project["uid"])

    notif_uid = generate_uid()
    get_table("notifications").insert(
        {
            "uid": notif_uid,
            "user_uid": uid,
            "type": "comment",
            "message": "someone commented on your project",
            "related_uid": project["uid"],
            "target_url": target_url,
            "read": False,
            "created_at": datetime.now(timezone.utc).isoformat(),
        }
    )
    refresh_snapshot()

    r = requests.get(f"{BASE_URL}{target_url}", headers={"X-API-KEY": key})
    assert r.status_code == 200, r.text[:200]

    refresh_snapshot()
    assert bool(get_table("notifications").find_one(uid=notif_uid)["read"]) is True