262 lines
9.8 KiB
Python
262 lines
9.8 KiB
Python
#!/usr/bin/env python3
|
|
import json
|
|
import hmac
|
|
import hashlib
|
|
from typing import Any, Dict, Optional
|
|
from dataclasses import dataclass
|
|
import time
|
|
|
|
# FastAPI TestClient for local calls without a real web server
|
|
from fastapi.testclient import TestClient
|
|
|
|
# Import your classes from the main code (ensure this file is next to gitea_to_github_sync.py)
|
|
from gitea_to_github_sync import PRAutocloserServer
|
|
|
|
# ---------------------------
|
|
# In-memory fakes and helpers
|
|
# ---------------------------
|
|
|
|
class InMemoryState:
|
|
def __init__(self):
|
|
self.map: Dict[str, int] = {}
|
|
|
|
def set_pr_bug(self, pr_key: str, bug_id: int):
|
|
self.map[pr_key] = bug_id
|
|
|
|
def get_pr_bug(self, pr_key: str) -> Optional[int]:
|
|
return self.map.get(pr_key)
|
|
|
|
@dataclass
|
|
class RecordedComment:
|
|
pr_number: int
|
|
text: str
|
|
|
|
@dataclass
|
|
class RecordedLabel:
|
|
pr_number: int
|
|
label: str
|
|
|
|
class FakeIssue:
|
|
def __init__(self, pr_number: int, tracker: "FakeRepo"):
|
|
self.pr_number = pr_number
|
|
self.tracker = tracker
|
|
|
|
def add_to_labels(self, label: str):
|
|
self.tracker.labels.append(RecordedLabel(self.pr_number, label))
|
|
|
|
class FakePR:
|
|
def __init__(self, number: int, tracker: "FakeRepo"):
|
|
self.number = number
|
|
self.state = "open"
|
|
self.tracker = tracker
|
|
|
|
def create_issue_comment(self, text: str):
|
|
self.tracker.comments.append(RecordedComment(self.number, text))
|
|
|
|
def edit(self, state: str):
|
|
if state in ("open", "closed"):
|
|
self.state = state
|
|
|
|
def as_issue(self):
|
|
return FakeIssue(self.number, self.tracker)
|
|
|
|
class FakeRepo:
|
|
def __init__(self, full_name: str):
|
|
self.full_name = full_name
|
|
self._prs: Dict[int, FakePR] = {}
|
|
self.comments: list[RecordedComment] = []
|
|
self.labels: list[RecordedLabel] = []
|
|
self.created_labels: set[str] = set()
|
|
|
|
def get_pull(self, number: int) -> FakePR:
|
|
if number not in self._prs:
|
|
self._prs[number] = FakePR(number, self)
|
|
return self._prs[number]
|
|
|
|
def get_issue(self, number: int) -> FakeIssue:
|
|
return FakeIssue(number, self)
|
|
|
|
def create_label(self, name: str, color: str):
|
|
self.created_labels.add(name)
|
|
|
|
class FakeGhApi:
|
|
def __init__(self, repo_full_name: str):
|
|
self.repo_full_name = repo_full_name
|
|
self.repo = FakeRepo(repo_full_name)
|
|
|
|
def get_repo(self, full_name: str) -> FakeRepo:
|
|
assert full_name == self.repo_full_name, f"Unexpected repo requested: {full_name}"
|
|
return self.repo
|
|
|
|
class FakeGitHubClient:
|
|
def __init__(self, repo_full_name: str, token: str = "gh_test_token"):
|
|
self.token = token
|
|
self.gh = FakeGhApi(repo_full_name)
|
|
|
|
# Match PRAutocloserServer usage
|
|
def close_pr_with_comment_and_label(self, repo, pr_number: int, comment: str, label: Optional[str] = None):
|
|
pr = repo.get_pull(pr_number)
|
|
pr.create_issue_comment(comment)
|
|
if label:
|
|
repo.create_label(label, "ededed")
|
|
repo.get_issue(pr_number).add_to_labels(label)
|
|
if pr.state != "closed":
|
|
pr.edit(state="closed")
|
|
|
|
def comment_on_pr(self, repo, pr_number: int, comment: str, label: Optional[str] = None):
|
|
pr = repo.get_pull(pr_number)
|
|
pr.create_issue_comment(comment)
|
|
if label:
|
|
repo.create_label(label, "ededed")
|
|
repo.get_issue(pr_number).add_to_labels(label)
|
|
|
|
class FakeBugzillaClient:
|
|
def __init__(self, fail_create: bool = False):
|
|
self.fail_create = fail_create
|
|
self.created_bugs: list[int] = []
|
|
self.attachments: list[tuple[int, str]] = [] # (bug_id, filename)
|
|
self.comments: list[tuple[int, str]] = [] # (bug_id, text)
|
|
self._next_id = 1000
|
|
|
|
def create_bug(self, summary: str, description: str, component: str, visibility_groups=None) -> int:
|
|
if self.fail_create:
|
|
raise RuntimeError("Simulated Bugzilla create failure")
|
|
bug_id = self._next_id
|
|
self._next_id += 1
|
|
self.created_bugs.append(bug_id)
|
|
return bug_id
|
|
|
|
def add_attachment(self, bug_id: int, file_name: str, content_type: str, summary: str, data_bytes: bytes):
|
|
# No-op, just record
|
|
self.attachments.append((bug_id, file_name))
|
|
|
|
def add_comment(self, bug_id: int, comment: str):
|
|
self.comments.append((bug_id, comment))
|
|
|
|
# HMAC signature helper
|
|
def sign(secret: str, body: bytes) -> str:
|
|
return "sha256=" + hmac.new(secret.encode("utf-8"), body, hashlib.sha256).hexdigest()
|
|
|
|
# Sample GitHub PR payload (minimized to fields used by the endpoint)
|
|
def sample_pr_payload(action="opened") -> Dict[str, Any]:
|
|
return {
|
|
"action": action,
|
|
"repository": {
|
|
"full_name": "Koozali-SME-Server/smeserver-manager",
|
|
"owner": {"login": "Koozali-SME-Server"},
|
|
"name": "smeserver-manager",
|
|
},
|
|
"pull_request": {
|
|
"number": 42,
|
|
"html_url": "https://github.com/Koozali-SME-Server/smeserver-manager/pull/42",
|
|
"title": "Test PR: improve logging",
|
|
"body": "Please review these changes.",
|
|
"user": {"login": "octocat", "html_url": "https://github.com/octocat"},
|
|
"base": {"ref": "master", "sha": "9f8e7d6cafebabe0000deadbeef0000000000000"},
|
|
"head": {"ref": "feature/logs", "sha": "a1b2c3ddeeddbb0000deadbeef0000000000000", "repo": {"owner": {"login": "octocat"}}},
|
|
"created_at": "2025-09-16T12:34:56Z",
|
|
"labels": [{"name": "enhancement"}],
|
|
},
|
|
}
|
|
|
|
# ---------------------------
|
|
# Test runner
|
|
# ---------------------------
|
|
|
|
def run_tests():
|
|
# Config for server (only the parts the endpoint uses)
|
|
cfg = {
|
|
"sync": {
|
|
"gitea": {"base_url": "https://src.koozali.org", "org": "smeserver"},
|
|
"github": {"webhook": {"secret": "test_secret"}},
|
|
},
|
|
"bugzilla": {
|
|
"base_url": "https://bugs.koozali.org",
|
|
"templates": {
|
|
"pr_comment_success": "Bug {bug_id}: {bugzilla_base_url}/show_bug.cgi?id={bug_id}\nCanonical: {gitea_repo_url}\n",
|
|
"pr_comment_failure": "Unable to create bug. Please file at {bugzilla_base_url} (Component: {repo})\nPR: {pr_url}\n",
|
|
"bug_summary": "[GH PR #{pr_number}] {org}/{repo}: {pr_title}",
|
|
"bug_body": "PR: {pr_url}\nNotes:\n{pr_body}\n",
|
|
"bug_update_comment": "Update for PR #{pr_number}",
|
|
"pr_sync_short_comment": "Updated bug {bug_id}: {bug_url}",
|
|
},
|
|
"attach_diff": False, # disable network to GitHub patch API during tests
|
|
"failure_policy": {"close_pr_on_bugzilla_failure": False, "label_on_bugzilla_failure": "bugzilla-needed"},
|
|
"component_template": "{repo}",
|
|
},
|
|
}
|
|
|
|
repo_full = "Koozali-SME-Server/smeserver-manager"
|
|
|
|
# Test 1: Success path (bug created, PR closed)
|
|
print("\n=== Test 1: Success path ===")
|
|
ghc = FakeGitHubClient(repo_full)
|
|
bzc = FakeBugzillaClient(fail_create=False)
|
|
state = InMemoryState()
|
|
server = PRAutocloserServer(cfg, state, ghc, bzc)
|
|
client = TestClient(server.app)
|
|
|
|
body = json.dumps(sample_pr_payload("opened")).encode("utf-8")
|
|
headers = {
|
|
"X-GitHub-Event": "pull_request",
|
|
"X-Hub-Signature-256": sign(cfg["sync"]["github"]["webhook"]["secret"], body),
|
|
"Content-Type": "application/json",
|
|
}
|
|
resp = client.post("/webhook/github", data=body, headers=headers)
|
|
print("Response:", resp.status_code, resp.json())
|
|
# Inspect effects
|
|
repo = ghc.gh.repo
|
|
print("Created bugs:", bzc.created_bugs)
|
|
print("PR comments:", [c.text.strip() for c in repo.comments])
|
|
print("PR state:", repo.get_pull(42).state)
|
|
assert repo.get_pull(42).state == "closed", "PR should be closed on success"
|
|
|
|
# Test 2: Failure path (bugzilla create fails, PR left open, guidance comment posted)
|
|
print("\n=== Test 2: Failure path ===")
|
|
ghc2 = FakeGitHubClient(repo_full)
|
|
bzc2 = FakeBugzillaClient(fail_create=True)
|
|
state2 = InMemoryState()
|
|
server2 = PRAutocloserServer(cfg, state2, ghc2, bzc2)
|
|
client2 = TestClient(server2.app)
|
|
|
|
body2 = json.dumps(sample_pr_payload("opened")).encode("utf-8")
|
|
headers2 = {
|
|
"X-GitHub-Event": "pull_request",
|
|
"X-Hub-Signature-256": sign(cfg["sync"]["github"]["webhook"]["secret"], body2),
|
|
"Content-Type": "application/json",
|
|
}
|
|
resp2 = client2.post("/webhook/github", data=body2, headers=headers2)
|
|
print("Response:", resp2.status_code, resp2.json())
|
|
repo2 = ghc2.gh.repo
|
|
print("Created bugs:", bzc2.created_bugs)
|
|
print("PR comments:", [c.text.strip() for c in repo2.comments])
|
|
print("PR labels:", [l.label for l in repo2.labels])
|
|
print("PR state:", repo2.get_pull(42).state)
|
|
assert repo2.get_pull(42).state == "open", "PR should remain open on bugzilla failure"
|
|
|
|
# Test 3: Synchronize event with existing bug (attachments disabled here)
|
|
print("\n=== Test 3: Synchronize with existing bug ===")
|
|
ghc3 = FakeGitHubClient(repo_full)
|
|
bzc3 = FakeBugzillaClient(fail_create=False)
|
|
state3 = InMemoryState()
|
|
server3 = PRAutocloserServer(cfg, state3, ghc3, bzc3)
|
|
client3 = TestClient(server3.app)
|
|
|
|
# Seed a bug mapping as if created earlier
|
|
state3.set_pr_bug(f"{repo_full}#42", 2001)
|
|
|
|
body3 = json.dumps(sample_pr_payload("synchronize")).encode("utf-8")
|
|
headers3 = {
|
|
"X-GitHub-Event": "pull_request",
|
|
"X-Hub-Signature-256": sign(cfg["sync"]["github"]["webhook"]["secret"], body3),
|
|
"Content-Type": "application/json",
|
|
}
|
|
resp3 = client3.post("/webhook/github", data=body3, headers=headers3)
|
|
print("Response:", resp3.status_code, resp3.json())
|
|
repo3 = ghc3.gh.repo
|
|
print("PR comments:", [c.text.strip() for c in repo3.comments])
|
|
print("Bug attachments recorded:", bzc3.attachments)
|
|
print("PR state:", repo3.get_pull(42).state)
|
|
|
|
if __name__ == "__main__":
|
|
run_tests() |