feat: implement Gitea integration for repository management
Add full Gitea API client with support for creating, listing, and managing repositories, including authentication via personal access tokens and webhook configuration for automated CI/CD triggers.
This commit is contained in:
+35
-93
@@ -1,5 +1,4 @@
|
||||
from tests.conftest import BASE_URL
|
||||
from devplacepy.database import get_table
|
||||
|
||||
|
||||
def test_bugs_page_loads(alice):
|
||||
@@ -8,6 +7,20 @@ def test_bugs_page_loads(alice):
|
||||
assert page.is_visible("h1:has-text('Bug Reports')")
|
||||
|
||||
|
||||
def test_bugs_filters_visible(alice):
|
||||
page, _ = alice
|
||||
page.goto(f"{BASE_URL}/bugs", wait_until="domcontentloaded")
|
||||
assert page.locator(".bugs-filter:has-text('Open')").is_visible()
|
||||
assert page.locator(".bugs-filter:has-text('Closed')").is_visible()
|
||||
assert page.locator(".bugs-filter:has-text('All')").is_visible()
|
||||
|
||||
|
||||
def test_bugs_not_configured_notice(alice):
|
||||
page, _ = alice
|
||||
page.goto(f"{BASE_URL}/bugs", wait_until="domcontentloaded")
|
||||
assert page.is_visible("text=The bug tracker is not configured yet.")
|
||||
|
||||
|
||||
def test_bugs_create_modal(alice):
|
||||
page, _ = alice
|
||||
page.goto(f"{BASE_URL}/bugs", wait_until="domcontentloaded")
|
||||
@@ -15,34 +28,6 @@ def test_bugs_create_modal(alice):
|
||||
assert page.is_visible("h3:has-text('Report a Bug')")
|
||||
|
||||
|
||||
def test_bugs_empty_state(alice):
|
||||
page, _ = alice
|
||||
get_table("bug_reports").delete()
|
||||
page.goto(f"{BASE_URL}/bugs", wait_until="domcontentloaded")
|
||||
assert page.is_visible("text=No bug reports yet.")
|
||||
|
||||
|
||||
def test_bugs_create_submit(alice):
|
||||
page, _ = alice
|
||||
page.goto(f"{BASE_URL}/bugs", wait_until="domcontentloaded")
|
||||
page.click("button:has-text('Report Bug')")
|
||||
page.fill("#bug-title", "Login button is misaligned")
|
||||
page.fill("#bug-description", "The login button overlaps the logo on mobile.")
|
||||
page.click("button:has-text('Submit Report')")
|
||||
page.wait_for_timeout(500)
|
||||
assert page.is_visible("text=Login button is misaligned")
|
||||
assert not page.is_visible("text=No bug reports yet.")
|
||||
|
||||
|
||||
def test_bugs_footer_link(alice):
|
||||
page, _ = alice
|
||||
page.goto(f"{BASE_URL}/feed", wait_until="domcontentloaded")
|
||||
link = page.locator("a:has-text('Bug Report')")
|
||||
assert link.is_visible()
|
||||
link.click()
|
||||
page.wait_for_url(f"{BASE_URL}/bugs", wait_until="domcontentloaded")
|
||||
|
||||
|
||||
def test_bugs_create_cancel(alice):
|
||||
page, _ = alice
|
||||
page.goto(f"{BASE_URL}/bugs", wait_until="domcontentloaded")
|
||||
@@ -52,77 +37,20 @@ def test_bugs_create_cancel(alice):
|
||||
assert not modal.is_visible()
|
||||
|
||||
|
||||
def test_bugs_unauth_empty(page):
|
||||
def test_bugs_unauth(page):
|
||||
page.goto(f"{BASE_URL}/bugs", wait_until="domcontentloaded")
|
||||
assert page.is_visible("h1:has-text('Bug Reports')")
|
||||
assert page.locator("a.login-required:has-text('Report Bug')").is_visible()
|
||||
assert page.locator("#create-bug-modal").count() == 0
|
||||
|
||||
|
||||
def test_bug_comments_form_visible(alice):
|
||||
def test_bugs_footer_link(alice):
|
||||
page, _ = alice
|
||||
page.goto(f"{BASE_URL}/bugs", wait_until="domcontentloaded")
|
||||
page.click("button:has-text('Report Bug')")
|
||||
page.fill("#bug-title", "Bug with comments")
|
||||
page.fill("#bug-description", "Testing comment form on bug")
|
||||
page.click("button:has-text('Submit Report')")
|
||||
page.wait_for_timeout(500)
|
||||
assert page.is_visible("text=Comments")
|
||||
assert page.is_visible("textarea[name='content']")
|
||||
|
||||
|
||||
def test_bug_comment_create(alice):
|
||||
page, _ = alice
|
||||
page.goto(f"{BASE_URL}/bugs", wait_until="domcontentloaded")
|
||||
page.click("button:has-text('Report Bug')")
|
||||
page.fill("#bug-title", "Bug comment test")
|
||||
page.fill("#bug-description", "Bug description for comment")
|
||||
page.click("button:has-text('Submit Report')")
|
||||
page.wait_for_timeout(500)
|
||||
page.fill("textarea[name='content']", "Bug comment here")
|
||||
page.click("button:has-text('Post')")
|
||||
page.wait_for_timeout(500)
|
||||
assert page.is_visible("text=Bug comment here")
|
||||
|
||||
|
||||
def test_bug_comment_reply(alice):
|
||||
page, _ = alice
|
||||
page.goto(f"{BASE_URL}/bugs", wait_until="domcontentloaded")
|
||||
page.click("button:has-text('Report Bug')")
|
||||
page.fill("#bug-title", "Bug threaded reply")
|
||||
page.fill("#bug-description", "Bug for thread test")
|
||||
page.click("button:has-text('Submit Report')")
|
||||
page.wait_for_timeout(500)
|
||||
page.fill("textarea[name='content']", "Root comment")
|
||||
page.click("button:has-text('Post')")
|
||||
page.wait_for_timeout(500)
|
||||
assert page.is_visible("text=Root comment")
|
||||
page.click("button:has-text('Reply')")
|
||||
reply_form = page.locator(".comment-reply-form").first
|
||||
reply_form.locator("textarea[name='content']").fill("Nested reply")
|
||||
reply_form.locator("button:has-text('Post')").click()
|
||||
page.wait_for_timeout(500)
|
||||
assert page.is_visible("text=Nested reply")
|
||||
|
||||
|
||||
def test_bug_comment_delete(alice):
|
||||
page, _ = alice
|
||||
page.goto(f"{BASE_URL}/bugs", wait_until="domcontentloaded")
|
||||
page.click("button:has-text('Report Bug')")
|
||||
page.fill("#bug-title", "Bug delete comment")
|
||||
page.fill("#bug-description", "Bug for delete test")
|
||||
page.click("button:has-text('Submit Report')")
|
||||
page.wait_for_timeout(500)
|
||||
page.fill("textarea[name='content']", "Delete me")
|
||||
page.click("button:has-text('Post')")
|
||||
page.wait_for_timeout(500)
|
||||
assert page.is_visible("text=Delete me")
|
||||
page.locator(
|
||||
".comment-body:has-text('Delete me') .comment-action-btn:has-text('Delete')"
|
||||
).click()
|
||||
page.locator(".dialog-overlay.visible .dialog-confirm").click()
|
||||
page.wait_for_timeout(500)
|
||||
assert not page.is_visible("text=Delete me")
|
||||
page.goto(f"{BASE_URL}/feed", wait_until="domcontentloaded")
|
||||
link = page.locator("a:has-text('Bug Report')")
|
||||
assert link.is_visible()
|
||||
link.click()
|
||||
page.wait_for_url(f"{BASE_URL}/bugs", wait_until="domcontentloaded")
|
||||
|
||||
|
||||
def test_bug_button_icon_spacing(alice):
|
||||
@@ -136,3 +64,17 @@ def test_bug_button_icon_spacing(alice):
|
||||
page.goto(f"{BASE_URL}/feed", wait_until="domcontentloaded")
|
||||
footer = page.locator("a:has-text('Bug Report')").inner_html()
|
||||
assert "</span> Bug Report" in footer
|
||||
|
||||
|
||||
def test_bug_detail_not_configured(alice):
|
||||
page, _ = alice
|
||||
page.goto(f"{BASE_URL}/bugs/999", wait_until="domcontentloaded")
|
||||
assert page.is_visible("h1.error-code:has-text('404')")
|
||||
assert page.is_visible("text=Page not found")
|
||||
|
||||
|
||||
def test_bug_job_status_not_found(alice):
|
||||
page, _ = alice
|
||||
page.goto(f"{BASE_URL}/bugs/jobs/nonexistent-job", wait_until="domcontentloaded")
|
||||
assert page.is_visible("h1.error-code:has-text('404')")
|
||||
assert page.is_visible("text=Page not found")
|
||||
|
||||
@@ -0,0 +1,225 @@
|
||||
import asyncio
|
||||
|
||||
import pytest
|
||||
|
||||
from devplacepy.database import (
|
||||
get_table,
|
||||
init_db,
|
||||
refresh_snapshot,
|
||||
set_setting,
|
||||
)
|
||||
from devplacepy.services.gitea import runtime, store
|
||||
from devplacepy.services.gitea.config import gitea_config
|
||||
from devplacepy.services.gitea.enhance import enhance_ticket
|
||||
from devplacepy.services.gitea.fake import FakeGiteaClient
|
||||
from devplacepy.services.gitea.service import BugTrackerService
|
||||
from devplacepy.services.jobs import queue
|
||||
from devplacepy.services.jobs.bug_create_service import BugCreateService
|
||||
from tests.conftest import run_async
|
||||
|
||||
_counter = [0]
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _init_db():
|
||||
init_db()
|
||||
yield
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def gitea_env():
|
||||
fake = FakeGiteaClient()
|
||||
runtime.set_client(fake)
|
||||
set_setting("gitea_base_url", "https://gitea.test")
|
||||
set_setting("gitea_owner", "retoor")
|
||||
set_setting("gitea_repo", "pydevplace")
|
||||
set_setting("gitea_token", "test-token")
|
||||
set_setting("bug_ai_enhance", "0")
|
||||
yield fake
|
||||
runtime.set_client(None)
|
||||
for table in ("bug_tickets", "bug_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="bug_create")):
|
||||
get_table("jobs").delete(uid=row["uid"])
|
||||
|
||||
|
||||
def _make_user():
|
||||
_counter[0] += 1
|
||||
uid = f"bugtest-user-{_counter[0]}"
|
||||
username = f"bugtester{_counter[0]}"
|
||||
get_table("users").insert(
|
||||
{"uid": uid, "username": username, "role": "member", "xp": 0, "level": 1}
|
||||
)
|
||||
return uid, username
|
||||
|
||||
|
||||
def _unread(user_uid):
|
||||
return [
|
||||
n
|
||||
for n in get_table("notifications").find(user_uid=user_uid)
|
||||
if n.get("type") == "bug"
|
||||
]
|
||||
|
||||
|
||||
def _drive_jobs():
|
||||
async def run():
|
||||
svc = BugCreateService()
|
||||
for _ in range(200):
|
||||
await svc.run_once()
|
||||
refresh_snapshot()
|
||||
pending = [
|
||||
r
|
||||
for r in get_table("jobs").find(kind="bug_create")
|
||||
if r["status"] in ("pending", "running")
|
||||
]
|
||||
if not pending and not svc._inflight:
|
||||
return
|
||||
await asyncio.sleep(0.02)
|
||||
|
||||
run_async(run())
|
||||
|
||||
|
||||
# ---------------- fake client ----------------
|
||||
|
||||
|
||||
def test_fake_client_create_and_list(gitea_env):
|
||||
async def go():
|
||||
issue = await gitea_env.create_issue("Crash on save", "details")
|
||||
assert issue["number"] == 1
|
||||
assert issue["state"] == "open"
|
||||
issues, total = await gitea_env.list_issues(state="open")
|
||||
assert total == 1
|
||||
assert issues[0]["number"] == 1
|
||||
|
||||
run_async(go())
|
||||
|
||||
|
||||
def test_fake_client_state_and_comments(gitea_env):
|
||||
async def go():
|
||||
issue = await gitea_env.create_issue("t", "b")
|
||||
number = issue["number"]
|
||||
await gitea_env.create_comment(number, "first")
|
||||
fetched = await gitea_env.get_issue(number)
|
||||
assert fetched["comments"] == 1
|
||||
closed = await gitea_env.set_state(number, "closed")
|
||||
assert closed["state"] == "closed"
|
||||
open_issues, _ = await gitea_env.list_issues(state="open")
|
||||
assert open_issues == []
|
||||
all_issues, total = await gitea_env.list_issues(state="all")
|
||||
assert total == 1
|
||||
|
||||
run_async(go())
|
||||
|
||||
|
||||
# ---------------- store ----------------
|
||||
|
||||
|
||||
def test_store_records_and_maps(gitea_env):
|
||||
uid, _ = _make_user()
|
||||
store.record_ticket(7, uid, "orig", "desc", "Better", "https://x/7", "open")
|
||||
assert store.author_uid_for_issue(7) == uid
|
||||
assert store.author_map([7, 8]) == {7: uid}
|
||||
cid = store.record_comment_author(99, 7, uid)
|
||||
assert cid
|
||||
assert store.comment_author_map([99, 100]) == {99: uid}
|
||||
assert store.local_comment_ids(7) == {99}
|
||||
|
||||
|
||||
# ---------------- enhance fallback ----------------
|
||||
|
||||
|
||||
def test_enhance_fallback_when_disabled(gitea_env):
|
||||
result = run_async(enhance_ticket("Title", "Body text", gitea_config()))
|
||||
assert result.enhanced is False
|
||||
assert "## Summary" in result.body
|
||||
assert "## Steps to Reproduce" in result.body
|
||||
assert result.title == "Title"
|
||||
|
||||
|
||||
# ---------------- create job ----------------
|
||||
|
||||
|
||||
def test_create_job_files_issue(gitea_env):
|
||||
uid, username = _make_user()
|
||||
job_uid = queue.enqueue(
|
||||
"bug_create",
|
||||
{"author_uid": uid, "title": "Login broken", "description": "cannot log in"},
|
||||
"user",
|
||||
uid,
|
||||
"Login broken",
|
||||
)
|
||||
_drive_jobs()
|
||||
job = queue.get_job(job_uid)
|
||||
assert job["status"] == "done"
|
||||
number = job["result"]["number"]
|
||||
assert number == 1
|
||||
ticket = store.get_ticket(number)
|
||||
assert ticket["author_uid"] == uid
|
||||
assert _unread(uid)
|
||||
issue = run_async(gitea_env.get_issue(number))
|
||||
assert username in issue["body"]
|
||||
|
||||
|
||||
def test_create_job_fails_without_config(gitea_env):
|
||||
set_setting("gitea_token", "")
|
||||
uid, _ = _make_user()
|
||||
job_uid = queue.enqueue(
|
||||
"bug_create",
|
||||
{"author_uid": uid, "title": "x", "description": "y"},
|
||||
"user",
|
||||
uid,
|
||||
"x",
|
||||
)
|
||||
_drive_jobs()
|
||||
job = queue.get_job(job_uid)
|
||||
assert job["status"] == "failed"
|
||||
assert "not configured" in job["error"].lower()
|
||||
|
||||
|
||||
# ---------------- poller ----------------
|
||||
|
||||
|
||||
def test_poller_notifies_on_developer_reply(gitea_env):
|
||||
uid, _ = _make_user()
|
||||
issue = run_async(gitea_env.create_issue("bug", "body"))
|
||||
number = issue["number"]
|
||||
store.record_ticket(number, uid, "bug", "body", "bug", issue["html_url"], "open")
|
||||
store.update_ticket_cache(number, "open", 0)
|
||||
|
||||
gitea_env.add_external_comment(number, "developer", "looking into it")
|
||||
before = len(_unread(uid))
|
||||
run_async(BugTrackerService().run_once())
|
||||
refresh_snapshot()
|
||||
assert len(_unread(uid)) == before + 1
|
||||
assert int(store.get_ticket(number)["last_comment_count"]) == 1
|
||||
|
||||
|
||||
def test_poller_ignores_local_comment(gitea_env):
|
||||
uid, _ = _make_user()
|
||||
issue = run_async(gitea_env.create_issue("bug", "body"))
|
||||
number = issue["number"]
|
||||
store.record_ticket(number, uid, "bug", "body", "bug", issue["html_url"], "open")
|
||||
store.update_ticket_cache(number, "open", 0)
|
||||
|
||||
comment = run_async(gitea_env.create_comment(number, "me again"))
|
||||
store.record_comment_author(comment["id"], number, uid)
|
||||
before = len(_unread(uid))
|
||||
run_async(BugTrackerService().run_once())
|
||||
refresh_snapshot()
|
||||
assert len(_unread(uid)) == before
|
||||
|
||||
|
||||
def test_poller_notifies_on_status_change(gitea_env):
|
||||
uid, _ = _make_user()
|
||||
issue = run_async(gitea_env.create_issue("bug", "body"))
|
||||
number = issue["number"]
|
||||
store.record_ticket(number, uid, "bug", "body", "bug", issue["html_url"], "open")
|
||||
store.update_ticket_cache(number, "open", 0)
|
||||
|
||||
run_async(gitea_env.set_state(number, "closed"))
|
||||
before = len(_unread(uid))
|
||||
run_async(BugTrackerService().run_once())
|
||||
refresh_snapshot()
|
||||
assert len(_unread(uid)) == before + 1
|
||||
assert store.get_ticket(number)["last_status"] == "closed"
|
||||
Reference in New Issue
Block a user