devplacepy/tests/api/skills/index.py

# retoor <retoor@molodetz.nl>

import json
import uuid

import requests

from devplacepy.database import get_table
from devplacepy.database.skills import get_skill_row, upsert_skill
from tests.conftest import BASE_URL

JSON = {"Accept": "application/json"}


def _user_skills(name):
    return get_table("users").find_one(username=name)


def _member_skills():
    name = f"skl{uuid.uuid4().hex[:10]}"
    requests.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 = _user_skills(name)
    session = requests.Session()
    session.headers.update({"X-API-KEY": row["api_key"], **JSON})
    return session, row


def _admin_skills(seeded_db):
    session = requests.Session()
    session.headers.update({"X-API-KEY": _user_skills("alice_test")["api_key"], **JSON})
    return session


def _seed_skills(user_uid, name, description="Does a thing."):
    row, _ = upsert_skill(
        user_uid,
        {
            "name": name,
            "description": description,
            "body": "# Body\n\nSteps.",
            "files": json.dumps({"scripts/run.py": "print(1)"}),
            "source_url": f"https://example.org/{name}/SKILL.md",
            "license": "",
            "compatibility": "",
            "metadata": json.dumps({}),
            "allowed_tools": "",
            "sha256": "0" * 64,
            "skipped": json.dumps([]),
            "warnings": json.dumps([]),
        },
    )
    return row


def test_skills_require_a_signed_in_user(app_server):
    assert requests.get(f"{BASE_URL}/skills", headers=JSON, allow_redirects=False).status_code == 401
    assert requests.post(
        f"{BASE_URL}/skills/install",
        data={"url": "https://example.org"},
        headers=JSON,
        allow_redirects=False,
    ).status_code == 401
    assert requests.get(f"{BASE_URL}/skills/anything", headers=JSON, allow_redirects=False).status_code == 401


def test_list_and_detail_json(app_server):
    session, user = _member_skills()
    _seed_skills(user["uid"], "listed", "Listed skill.")
    listing = session.get(f"{BASE_URL}/skills").json()
    assert listing["skill_total"] == 1
    assert listing["skill_limit"] == 50
    assert [item["name"] for item in listing["items"]] == ["listed"]
    assert "body" not in listing["items"][0]
    detail = session.get(f"{BASE_URL}/skills/listed").json()["skill"]
    assert detail["body"] == "# Body\n\nSteps."
    assert detail["files"] == {"scripts/run.py": "print(1)"}


def test_html_pages_render_and_are_noindex(app_server):
    session, user = _member_skills()
    _seed_skills(user["uid"], "paged", "Paged skill.")
    html_headers = {"Accept": "text/html"}
    listing = session.get(f"{BASE_URL}/skills", headers=html_headers)
    assert listing.status_code == 200
    assert "paged" in listing.text
    assert "noindex" in listing.text
    detail = session.get(f"{BASE_URL}/skills/paged", headers=html_headers)
    assert detail.status_code == 200
    assert "noindex" in detail.text


def test_skills_are_invisible_to_other_users(app_server):
    owner_session, owner = _member_skills()
    other_session, _ = _member_skills()
    _seed_skills(owner["uid"], "private-one")
    assert other_session.get(f"{BASE_URL}/skills/private-one").status_code == 404
    assert other_session.get(f"{BASE_URL}/skills").json()["items"] == []
    assert other_session.post(f"{BASE_URL}/skills/private-one/enabled", data={"enabled": "false"}).status_code == 404
    assert other_session.post(f"{BASE_URL}/skills/private-one/delete").status_code == 404
    assert get_skill_row(owner["uid"], "private-one") is not None


def test_toggle_flips_and_sets_explicitly(app_server):
    session, user = _member_skills()
    _seed_skills(user["uid"], "toggled")
    assert session.post(f"{BASE_URL}/skills/toggled/enabled").json() == {"name": "toggled", "enabled": False}
    assert session.post(f"{BASE_URL}/skills/toggled/enabled").json() == {"name": "toggled", "enabled": True}
    assert session.post(f"{BASE_URL}/skills/toggled/enabled", data={"enabled": "false"}).json()["enabled"] is False
    assert get_skill_row(user["uid"], "toggled")["enabled"] == 0
    assert session.post(f"{BASE_URL}/skills/nope/enabled").status_code == 404


def test_delete_is_soft_and_audited(app_server):
    session, user = _member_skills()
    row = _seed_skills(user["uid"], "doomed")
    assert session.post(f"{BASE_URL}/skills/doomed/delete").json() == {"deleted": True, "name": "doomed"}
    assert get_skill_row(user["uid"], "doomed") is None
    physical = get_table("skills").find_one(uid=row["uid"])
    assert physical["deleted_at"] is not None and physical["deleted_by"] == user["uid"]
    assert session.get(f"{BASE_URL}/skills/doomed").status_code == 404
    keys = {event["event_key"] for event in get_table("audit_log").find(target_uid=row["uid"])}
    assert "skill.delete" in keys


def test_a_member_cannot_delete_another_users_skill(app_server):
    member, _ = _member_skills()
    _, owner = _member_skills()
    _seed_skills(owner["uid"], "protected")
    response = member.post(f"{BASE_URL}/skills/protected/delete", data={"owner": owner["username"]})
    assert response.status_code == 403
    assert get_skill_row(owner["uid"], "protected") is not None


def test_an_administrator_can_delete_another_users_skill(seeded_db):
    _, owner = _member_skills()
    _seed_skills(owner["uid"], "admin-removed")
    admin = _admin_skills(seeded_db)
    assert admin.post(
        f"{BASE_URL}/skills/admin-removed/delete", data={"owner": owner["username"]}
    ).status_code == 200
    assert get_skill_row(owner["uid"], "admin-removed") is None
    missing = admin.post(f"{BASE_URL}/skills/x/delete", data={"owner": "no_such_user_xyz"})
    assert missing.status_code == 404


def test_install_failures_are_400_with_a_reason(app_server):
    session, _ = _member_skills()
    for url in (
        "ftp://example.org/SKILL.md",
        "https://user:pw@example.org/SKILL.md",
        "http://127.0.0.1:1/SKILL.md",
        "http://169.254.169.254/SKILL.md",
    ):
        response = session.post(f"{BASE_URL}/skills/install", data={"url": url})
        assert response.status_code == 400, url
        assert response.json()["error"]["message"], url
    assert session.get(f"{BASE_URL}/skills").json()["skill_total"] == 0


def test_install_failure_from_a_browser_redirects_with_the_error(app_server):
    session, _ = _member_skills()
    response = session.post(
        f"{BASE_URL}/skills/install",
        data={"url": "ftp://example.org/SKILL.md"},
        headers={"Accept": "text/html"},
        allow_redirects=False,
    )
    assert response.status_code == 302
    assert response.headers["location"].startswith("/skills?error=")


def test_install_validates_the_form(app_server):
    session, _ = _member_skills()
    assert session.post(f"{BASE_URL}/skills/install", data={"url": ""}).status_code in (400, 422)