#!/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()