forked from retoor/devplacepy
feat: add file attachment support to issue tracker with Gitea mirroring
Implement full attachment CRUD for issues and comments, restricted to open issues only. Attachments are stored locally and mirrored to the Gitea tracker via new `mirror_attachment_to_gitea` and `set_gitea_asset_id` functions. Add `gitea_asset_id` column to the attachments table, extend `IssueForm` and `IssueCommentForm` with `attachment_uids` field, and expose new API endpoints for listing, adding, and deleting issue attachments. Update agent configuration to include web research tools, and document the new capability in README, AGENTS.md, and CLAUDE.md.
This commit is contained in:
@@ -0,0 +1,311 @@
|
||||
# retoor <retoor@molodetz.nl>
|
||||
|
||||
import asyncio
|
||||
import io
|
||||
|
||||
import pytest
|
||||
from PIL import Image
|
||||
from starlette.testclient import TestClient
|
||||
|
||||
from devplacepy.database import get_table, init_db, refresh_snapshot, set_setting
|
||||
from devplacepy.services.gitea import runtime, store
|
||||
from devplacepy.services.gitea.fake import FakeGiteaClient
|
||||
from devplacepy.services.jobs import queue
|
||||
from devplacepy.services.jobs.issue_create_service import IssueCreateService
|
||||
from devplacepy.utils import create_session
|
||||
from tests.conftest import run_async
|
||||
|
||||
JSON = {"Accept": "application/json"}
|
||||
_seq = [0]
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _init_db_issue_attachments():
|
||||
init_db()
|
||||
yield
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def gitea_env():
|
||||
fake = FakeGiteaClient()
|
||||
runtime.set_client(fake)
|
||||
for key, value in {
|
||||
"gitea_base_url": "https://gitea.test",
|
||||
"gitea_owner": "retoor",
|
||||
"gitea_repo": "pydevplace",
|
||||
"gitea_token": "test-token",
|
||||
"issue_ai_enhance": "0",
|
||||
"max_upload_size_mb": "10",
|
||||
"allowed_file_types": "",
|
||||
"max_attachments_per_resource": "10",
|
||||
}.items():
|
||||
set_setting(key, value)
|
||||
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 _png():
|
||||
buf = io.BytesIO()
|
||||
Image.new("RGB", (8, 8), (255, 0, 0)).save(buf, "PNG")
|
||||
return buf.getvalue()
|
||||
|
||||
|
||||
def _user(role="member"):
|
||||
_seq[0] += 1
|
||||
uid = f"iatt-user-{_seq[0]}"
|
||||
get_table("users").insert(
|
||||
{
|
||||
"uid": uid,
|
||||
"username": f"iatt{_seq[0]}",
|
||||
"role": role,
|
||||
"xp": 0,
|
||||
"level": 1,
|
||||
"is_active": True,
|
||||
}
|
||||
)
|
||||
return uid
|
||||
|
||||
|
||||
def _client(uid=None):
|
||||
import devplacepy.main as m
|
||||
|
||||
client = TestClient(m.app)
|
||||
if uid:
|
||||
client.cookies.set("session", create_session(uid))
|
||||
return client
|
||||
|
||||
|
||||
def _issue(fake, state="open"):
|
||||
issue = run_async(fake.create_issue("Attach issue", "body"))
|
||||
number = int(issue["number"])
|
||||
store.record_ticket(
|
||||
number=number,
|
||||
author_uid="iatt-owner",
|
||||
original_title="Attach issue",
|
||||
original_description="body",
|
||||
enhanced_title="Attach issue",
|
||||
html_url=issue.get("html_url", ""),
|
||||
status=state,
|
||||
)
|
||||
if state == "closed":
|
||||
run_async(fake.set_state(number, "closed"))
|
||||
return number
|
||||
|
||||
|
||||
def _upload(client):
|
||||
r = client.post(
|
||||
"/uploads/upload", files={"file": ("a.png", _png(), "image/png")}
|
||||
)
|
||||
assert r.status_code == 201, r.text
|
||||
return r.json()["uid"]
|
||||
|
||||
|
||||
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 test_list_empty(gitea_env):
|
||||
number = _issue(gitea_env)
|
||||
client = _client(_user())
|
||||
r = client.get(f"/issues/{number}/attachments", headers=JSON)
|
||||
assert r.status_code == 200, r.text
|
||||
body = r.json()
|
||||
assert body["number"] == number
|
||||
assert body["attachments"] == []
|
||||
|
||||
|
||||
def test_add_links_mirrors_and_lists(gitea_env):
|
||||
number = _issue(gitea_env)
|
||||
client = _client(_user())
|
||||
file_uid = _upload(client)
|
||||
r = client.post(
|
||||
f"/issues/{number}/attachments",
|
||||
headers=JSON,
|
||||
data={"attachment_uids": file_uid},
|
||||
)
|
||||
assert r.status_code == 200, r.text
|
||||
assert r.json()["data"]["count"] == 1
|
||||
row = get_table("attachments").find_one(uid=file_uid)
|
||||
assert row["target_type"] == "issue"
|
||||
assert row["target_uid"] == str(number)
|
||||
assert len(gitea_env._issue_assets.get(number, [])) == 1
|
||||
listed = client.get(f"/issues/{number}/attachments", headers=JSON).json()
|
||||
assert len(listed["attachments"]) == 1
|
||||
assert listed["attachments"][0]["can_modify"] is True
|
||||
|
||||
|
||||
def test_delete_owner_allowed_removes_mirror(gitea_env):
|
||||
number = _issue(gitea_env)
|
||||
client = _client(_user())
|
||||
file_uid = _upload(client)
|
||||
client.post(
|
||||
f"/issues/{number}/attachments",
|
||||
headers=JSON,
|
||||
data={"attachment_uids": file_uid},
|
||||
)
|
||||
r = client.delete(f"/issues/{number}/attachments/{file_uid}", headers=JSON)
|
||||
assert r.status_code == 200, r.text
|
||||
assert get_table("attachments").find_one(uid=file_uid)["deleted_at"]
|
||||
assert gitea_env._issue_assets.get(number, []) == []
|
||||
|
||||
|
||||
def test_delete_other_member_forbidden_admin_allowed(gitea_env):
|
||||
number = _issue(gitea_env)
|
||||
owner = _client(_user())
|
||||
file_uid = _upload(owner)
|
||||
owner.post(
|
||||
f"/issues/{number}/attachments",
|
||||
headers=JSON,
|
||||
data={"attachment_uids": file_uid},
|
||||
)
|
||||
other = _client(_user())
|
||||
assert (
|
||||
other.delete(
|
||||
f"/issues/{number}/attachments/{file_uid}", headers=JSON
|
||||
).status_code
|
||||
== 403
|
||||
)
|
||||
admin = _client(_user("Admin"))
|
||||
assert (
|
||||
admin.delete(
|
||||
f"/issues/{number}/attachments/{file_uid}", headers=JSON
|
||||
).status_code
|
||||
== 200
|
||||
)
|
||||
|
||||
|
||||
def test_closed_issue_rejects_add(gitea_env):
|
||||
number = _issue(gitea_env, state="closed")
|
||||
client = _client(_user())
|
||||
file_uid = _upload(client)
|
||||
r = client.post(
|
||||
f"/issues/{number}/attachments",
|
||||
headers=JSON,
|
||||
data={"attachment_uids": file_uid},
|
||||
)
|
||||
assert r.status_code == 409, r.text
|
||||
assert not get_table("attachments").find_one(uid=file_uid)["target_uid"]
|
||||
|
||||
|
||||
def test_delete_rejected_after_close(gitea_env):
|
||||
number = _issue(gitea_env)
|
||||
client = _client(_user())
|
||||
file_uid = _upload(client)
|
||||
client.post(
|
||||
f"/issues/{number}/attachments",
|
||||
headers=JSON,
|
||||
data={"attachment_uids": file_uid},
|
||||
)
|
||||
run_async(gitea_env.set_state(number, "closed"))
|
||||
r = client.delete(f"/issues/{number}/attachments/{file_uid}", headers=JSON)
|
||||
assert r.status_code == 409, r.text
|
||||
|
||||
|
||||
def test_guest_cannot_add(gitea_env):
|
||||
number = _issue(gitea_env)
|
||||
guest = _client()
|
||||
r = guest.post(
|
||||
f"/issues/{number}/attachments",
|
||||
headers=JSON,
|
||||
data={"attachment_uids": "anything"},
|
||||
)
|
||||
assert r.status_code == 401, r.text
|
||||
|
||||
|
||||
def test_cannot_attach_another_users_orphan(gitea_env):
|
||||
number = _issue(gitea_env)
|
||||
owner = _client(_user())
|
||||
file_uid = _upload(owner)
|
||||
attacker = _client(_user())
|
||||
r = attacker.post(
|
||||
f"/issues/{number}/attachments",
|
||||
headers=JSON,
|
||||
data={"attachment_uids": file_uid},
|
||||
)
|
||||
assert r.status_code == 400, r.text
|
||||
assert not get_table("attachments").find_one(uid=file_uid)["target_uid"]
|
||||
|
||||
|
||||
def test_comment_post_links_and_mirrors_attachment(gitea_env):
|
||||
number = _issue(gitea_env)
|
||||
client = _client(_user())
|
||||
file_uid = _upload(client)
|
||||
r = client.post(
|
||||
f"/issues/{number}/comment",
|
||||
headers=JSON,
|
||||
data={"body": "see attached", "attachment_uids": file_uid},
|
||||
)
|
||||
assert r.status_code == 200, r.text
|
||||
cid = r.json()["data"]["comment_id"]
|
||||
row = get_table("attachments").find_one(uid=file_uid)
|
||||
assert row["target_type"] == "issue_comment"
|
||||
assert row["target_uid"] == str(cid)
|
||||
assert len(gitea_env._comment_assets.get(cid, [])) == 1
|
||||
listed = client.get(
|
||||
f"/issues/{number}/comments/{cid}/attachments", headers=JSON
|
||||
).json()
|
||||
assert len(listed["attachments"]) == 1
|
||||
|
||||
|
||||
def test_comment_attachment_add_endpoint_and_delete(gitea_env):
|
||||
number = _issue(gitea_env)
|
||||
client = _client(_user())
|
||||
comment = run_async(gitea_env.create_comment(number, "c"))
|
||||
cid = int(comment["id"])
|
||||
file_uid = _upload(client)
|
||||
r = client.post(
|
||||
f"/issues/{number}/comments/{cid}/attachments",
|
||||
headers=JSON,
|
||||
data={"attachment_uids": file_uid},
|
||||
)
|
||||
assert r.status_code == 200, r.text
|
||||
r = client.delete(
|
||||
f"/issues/{number}/comments/{cid}/attachments/{file_uid}", headers=JSON
|
||||
)
|
||||
assert r.status_code == 200, r.text
|
||||
assert get_table("attachments").find_one(uid=file_uid)["deleted_at"]
|
||||
|
||||
|
||||
def test_attach_at_issue_creation(gitea_env):
|
||||
uid = _user()
|
||||
client = _client(uid)
|
||||
file_uid = _upload(client)
|
||||
queue.enqueue(
|
||||
"issue_create",
|
||||
{
|
||||
"author_uid": uid,
|
||||
"title": "Created with file",
|
||||
"description": "desc",
|
||||
"attachment_uids": [file_uid],
|
||||
},
|
||||
"user",
|
||||
uid,
|
||||
"Created with file",
|
||||
)
|
||||
_drive_jobs()
|
||||
refresh_snapshot()
|
||||
row = get_table("attachments").find_one(uid=file_uid)
|
||||
assert row["target_type"] == "issue"
|
||||
number = int(row["target_uid"])
|
||||
assert len(gitea_env._issue_assets.get(number, [])) == 1
|
||||
Reference in New Issue
Block a user