devplacepy/tests/unit/content.py

# retoor <retoor@molodetz.nl>

from datetime import datetime, timedelta, timezone
from devplacepy.content import (
    is_owner,
    can_view_project,
    can_manage_instance,
    can_view_instance,
    can_view_project_containers,
    canonical_redirect,
    first_image_url,
    enrich_items,
    owns_instance,
)
from devplacepy.database import get_table, get_users_by_uids, invalidate_admins_cache
from devplacepy.utils import generate_uid
import time
import pytest
import requests
from tests.conftest import BASE_URL
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 test_is_owner_matches_user_uid():
    assert is_owner({"user_uid": "u1"}, {"uid": "u1"}) is True
    assert is_owner({"user_uid": "u1"}, {"uid": "u2"}) is False
    assert is_owner(None, {"uid": "u1"}) is False
    assert is_owner({"user_uid": "u1"}, None) is False


def _make_user(role="Member"):
    uid = generate_uid()
    username = f"cv_{uid[:8]}"
    get_table("users").insert(
        {
            "uid": uid,
            "username": username,
            "terms_version": "1",
            "email": f"{username}@t.dev",
            "role": role,
            "created_at": datetime.now(timezone.utc).isoformat(),
        }
    )
    return get_table("users").find_one(uid=uid)


def test_can_view_project_public_visible_to_everyone(local_db):
    owner = _make_user()
    other = _make_user()
    admin = _make_user("Admin")
    project = {"user_uid": owner["uid"], "is_private": 0}
    assert can_view_project(project, owner) is True
    assert can_view_project(project, other) is True
    assert can_view_project(project, admin) is True
    assert can_view_project(project, None) is True


def test_can_view_member_private_visible_to_owner_and_admin(local_db):
    owner = _make_user()
    other = _make_user()
    admin = _make_user("Admin")
    project = {"user_uid": owner["uid"], "is_private": 1}
    assert can_view_project(project, owner) is True
    assert can_view_project(project, admin) is True
    assert can_view_project(project, other) is False
    assert can_view_project(project, None) is False


def test_can_view_admin_private_visible_to_owner_admin_only(local_db):
    owner_admin = _make_user("Admin")
    other_admin = _make_user("Admin")
    member = _make_user()
    project = {"user_uid": owner_admin["uid"], "is_private": 1}
    assert can_view_project(project, owner_admin) is True
    assert can_view_project(project, other_admin) is False
    assert can_view_project(project, member) is False
    assert can_view_project(project, None) is False


def _make_user_at(role, created_at):
    uid = generate_uid()
    username = f"cv_{uid[:8]}"
    get_table("users").insert(
        {
            "uid": uid,
            "username": username,
            "terms_version": "1",
            "email": f"{username}@t.dev",
            "role": role,
            "created_at": created_at.isoformat(),
        }
    )
    invalidate_admins_cache()
    return get_table("users").find_one(uid=uid)


def _demote_existing_admins():
    existing = [r["uid"] for r in get_table("users").find(role="Admin")]
    for uid in existing:
        get_table("users").update({"uid": uid, "role": "Member"}, ["uid"])
    invalidate_admins_cache()
    return existing


def _restore_admins(uids):
    for uid in uids:
        get_table("users").update({"uid": uid, "role": "Admin"}, ["uid"])
    invalidate_admins_cache()


def _purge_users(*rows):
    for row in rows:
        if row:
            get_table("users").delete(uid=row["uid"])


def test_owns_instance_true_for_creator():
    owner = {"uid": "cnt-u1"}
    other = {"uid": "cnt-u2"}
    instance = {"created_by": owner["uid"]}
    project = {"user_uid": "cnt-u3"}
    assert owns_instance(instance, project, owner) is True
    assert owns_instance(instance, project, other) is False


def test_owns_instance_true_for_project_owner():
    owner = {"uid": "cnt-u1"}
    creator = {"uid": "cnt-u2"}
    instance = {"created_by": creator["uid"]}
    project = {"user_uid": owner["uid"]}
    assert owns_instance(instance, project, owner) is True


def test_owns_instance_false_when_missing_data():
    assert owns_instance(None, {}, {"uid": "cnt-u1"}) is False
    assert owns_instance({"created_by": "cnt-u1"}, {}, None) is False
    assert owns_instance({"created_by": "cnt-u1"}, {}, {}) is False
    assert owns_instance({"created_by": "cnt-u1"}, None, {"uid": "cnt-u2"}) is False


def test_can_view_project_containers_requires_admin(local_db):
    owner = _make_user()
    project = {"user_uid": owner["uid"], "is_private": 0}
    try:
        assert can_view_project_containers(project, owner) is False
        assert can_view_project_containers(project, None) is False
        assert can_view_project_containers(None, owner) is False
    finally:
        _purge_users(owner)


def test_can_view_project_containers_public_visible_to_any_admin(local_db):
    restore = _demote_existing_admins()
    owner_admin = other_admin = None
    try:
        base = datetime.now(timezone.utc)
        owner_admin = _make_user_at("Admin", base)
        other_admin = _make_user_at("Admin", base + timedelta(seconds=1))
        project = {"user_uid": owner_admin["uid"], "is_private": 0}
        assert can_view_project_containers(project, owner_admin) is True
        assert can_view_project_containers(project, other_admin) is True
    finally:
        _purge_users(owner_admin, other_admin)
        _restore_admins(restore)


def test_can_view_project_containers_private_hidden_from_other_admin(local_db):
    restore = _demote_existing_admins()
    owner_admin = other_admin = None
    try:
        base = datetime.now(timezone.utc)
        owner_admin = _make_user_at("Admin", base)
        other_admin = _make_user_at("Admin", base + timedelta(seconds=1))
        project = {"user_uid": owner_admin["uid"], "is_private": 1}
        assert can_view_project_containers(project, owner_admin) is True
        assert can_view_project_containers(project, other_admin) is False
    finally:
        _purge_users(owner_admin, other_admin)
        _restore_admins(restore)


def test_can_view_project_containers_primary_admin_sees_all(local_db):
    restore = _demote_existing_admins()
    primary = other_admin = None
    try:
        base = datetime.now(timezone.utc)
        primary = _make_user_at("Admin", base)
        other_admin = _make_user_at("Admin", base + timedelta(seconds=1))
        project = {"user_uid": other_admin["uid"], "is_private": 1}
        assert can_view_project_containers(project, primary) is True
    finally:
        _purge_users(primary, other_admin)
        _restore_admins(restore)


def test_can_view_instance_requires_admin(local_db):
    owner = _make_user()
    instance = {"created_by": owner["uid"], "project_uid": "p1"}
    project = {"user_uid": owner["uid"], "is_private": 0}
    try:
        assert can_view_instance(instance, project, owner) is False
        assert can_manage_instance(instance, project, owner) is False
    finally:
        _purge_users(owner)


def test_can_view_instance_owner_sees_own_private_hidden_from_other_admin(local_db):
    restore = _demote_existing_admins()
    owner_admin = other_admin = None
    try:
        base = datetime.now(timezone.utc)
        owner_admin = _make_user_at("Admin", base)
        other_admin = _make_user_at("Admin", base + timedelta(seconds=1))
        project = {"user_uid": owner_admin["uid"], "is_private": 1}
        instance = {"created_by": owner_admin["uid"], "project_uid": "p1"}
        assert can_view_instance(instance, project, owner_admin) is True
        assert can_view_instance(instance, project, other_admin) is False
    finally:
        _purge_users(owner_admin, other_admin)
        _restore_admins(restore)


def test_can_view_instance_public_project_visible_to_any_admin(local_db):
    restore = _demote_existing_admins()
    creator = other_admin = None
    try:
        base = datetime.now(timezone.utc)
        creator = _make_user_at("Admin", base)
        other_admin = _make_user_at("Admin", base + timedelta(seconds=1))
        project = {"user_uid": creator["uid"], "is_private": 0}
        instance = {"created_by": creator["uid"], "project_uid": "p1"}
        assert can_view_instance(instance, project, other_admin) is True
    finally:
        _purge_users(creator, other_admin)
        _restore_admins(restore)


def test_can_view_instance_primary_admin_sees_others_private_instance(local_db):
    restore = _demote_existing_admins()
    primary = owner_admin = None
    try:
        base = datetime.now(timezone.utc)
        primary = _make_user_at("Admin", base)
        owner_admin = _make_user_at("Admin", base + timedelta(seconds=1))
        project = {"user_uid": owner_admin["uid"], "is_private": 1}
        instance = {"created_by": owner_admin["uid"], "project_uid": "p1"}
        assert can_view_instance(instance, project, primary) is True
    finally:
        _purge_users(primary, owner_admin)
        _restore_admins(restore)


def test_can_manage_instance_only_owner_or_primary_admin(local_db):
    restore = _demote_existing_admins()
    primary = owner_admin = other_admin = None
    try:
        base = datetime.now(timezone.utc)
        primary = _make_user_at("Admin", base)
        owner_admin = _make_user_at("Admin", base + timedelta(seconds=1))
        other_admin = _make_user_at("Admin", base + timedelta(seconds=2))
        project = {"user_uid": owner_admin["uid"], "is_private": 1}
        instance = {"created_by": owner_admin["uid"], "project_uid": "p1"}
        assert can_manage_instance(instance, project, owner_admin) is True
        assert can_manage_instance(instance, project, primary) is True
        assert can_manage_instance(instance, project, other_admin) is False
    finally:
        _purge_users(primary, owner_admin, other_admin)
        _restore_admins(restore)


def test_can_manage_instance_other_admin_denied_even_on_public_project(local_db):
    restore = _demote_existing_admins()
    owner_admin = other_admin = None
    try:
        base = datetime.now(timezone.utc)
        owner_admin = _make_user_at("Admin", base)
        other_admin = _make_user_at("Admin", base + timedelta(seconds=1))
        project = {"user_uid": owner_admin["uid"], "is_private": 0}
        instance = {"created_by": owner_admin["uid"], "project_uid": "p1"}
        assert can_view_instance(instance, project, other_admin) is True
        assert can_manage_instance(instance, project, other_admin) is False
    finally:
        _purge_users(owner_admin, other_admin)
        _restore_admins(restore)


def test_canonical_redirect_when_slug_differs():
    item = {"slug": "abcd1234-title", "uid": "abcd1234"}
    assert canonical_redirect("posts", item, "abcd1234-title") is None
    response = canonical_redirect("posts", item, "abcd1234")
    assert response is not None
    assert response.status_code == 301
    assert response.headers["location"] == "/posts/abcd1234-title"


def test_first_image_url_prefers_inline():
    assert first_image_url({"image": "pic.png"}, None) == "/static/uploads/pic.png"


def test_first_image_url_uses_first_image_attachment():
    item = {"image": None}
    attachments = [{"is_image": False, "url": "/a"}, {"is_image": True, "url": "/b"}]
    assert first_image_url(item, attachments) == "/b"


def test_first_image_url_none_when_absent():
    assert first_image_url({}, []) is None


def test_enrich_items_attaches_author_and_extras(local_db):
    uid = generate_uid()
    username = f"enrich_{uid[:8]}"
    get_table("users").insert(
        {
            "uid": uid,
            "username": username,
            "terms_version": "1",
            "email": f"{username}@t.dev",
            "created_at": datetime.now(timezone.utc).isoformat(),
        }
    )
    item_uid = generate_uid()
    items = [
        {
            "uid": item_uid,
            "user_uid": uid,
            "created_at": datetime.now(timezone.utc).isoformat(),
        }
    ]
    authors = get_users_by_uids([uid])
    enriched = enrich_items(
        items,
        "post",
        authors,
        extra_maps={"comment_count": {item_uid: 4}, "flag": lambda item: "ok"},
    )
    entry = enriched[0]
    assert entry["post"] is items[0]
    assert entry["author"]["username"] == username
    assert entry["time_ago"] == "just now"
    assert entry["my_vote"] == 0
    assert entry["comment_count"] == 4
    assert entry["flag"] == "ok"


def test_readonly_action_requires_confirmation():
    assert confirmation_error("project_set_readonly", {}) is not None
    assert confirmation_error("project_set_readonly", {"confirm": "false"}) is not None
    assert confirmation_error("project_set_readonly", {"confirm": "true"}) is None


def test_private_action_requires_confirmation():
    assert confirmation_error("project_set_private", {"value": "true"}) is not None
    assert confirmation_error("project_set_private", {"value": "true", "confirm": "true"}) is None
    assert confirmation_error("vote", {}) is None


def test_is_confirmed_accepts_common_truthy():
    assert _is_confirmed({"confirm": True})
    assert _is_confirmed({"confirm": "yes"})
    assert not _is_confirmed({"confirm": ""})
    assert not _is_confirmed({})


def test_delete_actions_require_confirmation():
    assert confirmation_error("project_delete_file", {"path": "a.py"}) is not None
    assert confirmation_error("project_delete_file", {"path": "a.py", "confirm": "true"}) is None
    assert confirmation_error("delete_project", {"project_slug": "p"}) is not None
    assert confirmation_error("delete_project", {"project_slug": "p", "confirm": "true"}) is None


def test_container_delete_requires_confirmation_but_not_other_actions():
    assert confirmation_error("container_instance_action", {"action": "delete"}) is not None
    assert confirmation_error("container_instance_action", {"action": "restart"}) is None
    assert confirmation_error("container_instance_action", {"action": "start"}) is None


def test_destructive_container_exec_requires_confirmation():
    assert confirmation_error("container_exec", {"command": "rm -f /app/cms/cms.db"}) is not None
    assert confirmation_error("container_exec", {"command": "sudo rm -rf /app"}) is not None
    assert confirmation_error("container_exec", {"command": "find . -delete"}) is not None
    assert confirmation_error("container_exec", {"command": "echo x > /etc/hosts"}) is not None
    assert (
        confirmation_error("container_exec", {"command": "rm -rf /app", "confirm": "true"})
        is None
    )


def test_benign_container_exec_runs_without_confirmation():
    assert confirmation_error("container_exec", {"command": "pip install requests"}) is None
    assert confirmation_error("container_exec", {"command": "ls -la /app"}) is None
    assert confirmation_error("container_exec", {"command": "grep -rm 5 foo ."}) is None