devplacepy/tests/unit/database/skills.py

# retoor <retoor@molodetz.nl>

import json

import pytest

from devplacepy.database import SOFT_DELETE_TABLES, UNREPORTABLE_TABLES, get_table
from devplacepy.database.skills import (
    SkillLimitError,
    count_skills,
    delete_skill,
    enabled_skill_summaries,
    get_skill_row,
    list_skills,
    set_skill_enabled,
    skill_view,
    upsert_skill,
)
from devplacepy.services.moderation.deletion import OWNED_TABLES
from devplacepy.services.skills import limits
from devplacepy.services.skills.prompt import skill_index_lines
from devplacepy.utils import generate_uid


def fields(name, description="Does a thing.", **extra):
    return {
        "name": name,
        "description": description,
        "body": "# Body",
        "files": json.dumps({"scripts/run.py": "print(1)"}),
        "source_url": f"https://example.org/{name}/SKILL.md",
        "license": "",
        "compatibility": "",
        "metadata": json.dumps({"author": "x"}),
        "allowed_tools": "",
        "sha256": "0" * 64,
        "skipped": json.dumps([]),
        "warnings": json.dumps([]),
        **extra,
    }


@pytest.fixture
def owner(local_db):
    return generate_uid()


def test_skills_is_soft_deletable_private_and_cascaded_on_account_deletion():
    assert "skills" in SOFT_DELETE_TABLES
    assert "skills" in UNREPORTABLE_TABLES
    assert ("skills", "user_uid") in OWNED_TABLES


def test_install_then_update_in_place(owner):
    row, action = upsert_skill(owner, fields("alpha"))
    assert action == "install"
    assert row["enabled"] == 1
    again, action = upsert_skill(owner, fields("alpha", description="Changed."))
    assert action == "update"
    assert again["uid"] == row["uid"]
    assert again["description"] == "Changed."
    assert count_skills(owner) == 1


def test_skills_are_private_to_their_owner(owner, local_db):
    other = generate_uid()
    upsert_skill(owner, fields("shared-name"))
    assert get_skill_row(other, "shared-name") is None
    assert list_skills(other) == []
    row, action = upsert_skill(other, fields("shared-name"))
    assert action == "install"
    assert get_skill_row(owner, "shared-name")["uid"] != row["uid"]


def test_toggle_controls_the_prompt_index(owner):
    upsert_skill(owner, fields("beta", description="Beta description."))
    assert "beta: Beta description." in skill_index_lines(owner)
    assert set_skill_enabled(owner, "beta", False)["enabled"] == 0
    assert skill_index_lines(owner) == ""
    assert set_skill_enabled(owner, "beta", True)["enabled"] == 1
    assert set_skill_enabled(owner, "missing", True) is None


def test_delete_is_soft_and_reinstall_revives_the_same_row(owner):
    row, _ = upsert_skill(owner, fields("gamma"))
    assert delete_skill(owner, "gamma", owner) is True
    assert get_skill_row(owner, "gamma") is None
    assert delete_skill(owner, "gamma", owner) is False
    physical = get_table("skills").find_one(uid=row["uid"])
    assert physical["deleted_at"] is not None
    assert physical["deleted_by"] == owner
    revived, action = upsert_skill(owner, fields("gamma"))
    assert action == "update"
    assert revived["uid"] == row["uid"]
    assert revived["deleted_at"] is None
    assert revived["enabled"] == 1


def test_per_user_cap_is_enforced_including_on_revive(owner):
    for index in range(limits.SKILLS_PER_USER_MAX):
        upsert_skill(owner, fields(f"cap-{index}"))
    with pytest.raises(SkillLimitError):
        upsert_skill(owner, fields("one-too-many"))
    delete_skill(owner, "cap-0", owner)
    upsert_skill(owner, fields("fits-now"))
    with pytest.raises(SkillLimitError):
        upsert_skill(owner, fields("cap-0"))


def test_listing_is_alphabetical_and_excludes_deleted(owner):
    for name in ("zeta", "alpha", "mid"):
        upsert_skill(owner, fields(name))
    delete_skill(owner, "mid", owner)
    assert [row["name"] for row in list_skills(owner)] == ["alpha", "zeta"]


def test_skill_view_summary_and_detail(owner):
    upsert_skill(owner, fields("delta"))
    row = get_skill_row(owner, "delta")
    summary = skill_view(row)
    assert summary["file_count"] == 1
    assert "body" not in summary and "files" not in summary
    detail = skill_view(row, detail=True)
    assert detail["body"] == "# Body"
    assert detail["files"] == {"scripts/run.py": "print(1)"}
    assert detail["metadata"] == {"author": "x"}
    assert detail["skipped"] == [] and detail["warnings"] == []


def test_prompt_index_is_bounded(owner):
    long_description = "word " * 500
    for index in range(limits.PROMPT_MAX_SKILLS + 10):
        upsert_skill(owner, fields(f"p{index:02d}", description=long_description))
    lines = skill_index_lines(owner).split("\n")
    assert len(enabled_skill_summaries(owner)) == limits.PROMPT_MAX_SKILLS
    assert len(lines) <= limits.PROMPT_MAX_SKILLS
    assert len("\n".join(lines)) <= limits.PROMPT_MAX_CHARS
    for line in lines:
        assert len(line) <= len("- p00: ") + limits.PROMPT_DESCRIPTION_MAX_CHARS


def test_prompt_index_collapses_whitespace_and_markdown_prefix(owner):
    upsert_skill(owner, fields("epsilon", description="## Heading\n\n  spaced   out"))
    assert skill_index_lines(owner) == "- epsilon: Heading spaced out"