#!/usr/bin/env python3 import json import hmac import hashlib from typing import Optional, Dict, Any from dataclasses import dataclass from fastapi.testclient import TestClient import webhook_endpoint as ws # your standalone endpoint module # --------- Fakes --------- @dataclass class RecordedComment: pr_number: int text: str @dataclass class RecordedLabel: pr_number: int label: str class FakeIssue: def __init__(self, pr_number: int, repo: "FakeRepo"): self.pr_number = pr_number self.repo = repo def add_to_labels(self, label: str): self.repo.labels.append(RecordedLabel(self.pr_number, label)) class FakePR: def __init__(self, number: int, repo: "FakeRepo"): self.number = number self.repo = repo self.state = "open" def create_issue_comment(self, text: str): self.repo.comments.append(RecordedComment(self.number, text)) def edit(self, state: str): if state in ("open", "closed"): self.state = state 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: self._prs.setdefault(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 return self.repo class FakeGhSingleton: def __init__(self, repo_full_name: str): self._api = FakeGhApi(repo_full_name) def get_repo(self, full_name: str): return self._api.get_repo(full_name) class FakeBugzilla: def __init__(self, fail_create: bool = False): self.fail_create = fail_create self.created_bugs: list[int] = [] self.attachments: list[tuple[int, str]] = [] self._next = 1000 def create_bug(self, summary: str, description: str, component: str) -> int: if self.fail_create: raise RuntimeError("Simulated Bugzilla create failure") bug_id = self._next; self._next += 1 self.created_bugs.append(bug_id) return bug_id def add_attachment(self, bug_id: int, filename: str, summary: str, data_bytes: bytes): self.attachments.append((bug_id, filename)) def add_comment(self, bug_id: int, comment: str): pass class InMemoryState: def __init__(self): self.map: Dict[str, int] = {} def get_bug(self, pr_key: str) -> Optional[int]: return self.map.get(pr_key) def set_bug(self, pr_key: str, bug_id: int): self.map[pr_key] = bug_id # --------- Helpers --------- def sign(secret: str, body: bytes) -> str: return "sha256=" + hmac.new(secret.encode("utf-8"), body, hashlib.sha256).hexdigest() def sample_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": "Improve logging", "body": "Please review.", "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"}], }, } def run_tests(): # Inject fakes into the running module singletons ws.gh = FakeGhSingleton("Koozali-SME-Server/smeserver-manager") ws.bz = FakeBugzilla(fail_create=False) ws.state = InMemoryState() client = TestClient(ws.app) secret = os.environ.get("WEBHOOK_SECRET", "test_secret") # must match your server config payload = sample_payload("opened") body = json.dumps(payload).encode("utf-8") headers = { "X-GitHub-Event": "pull_request", "X-Hub-Signature-256": sign(secret, body), "Content-Type": "application/json", } resp = client.post("/webhook/github", data=body, headers=headers) print("Opened response:", resp.status_code, resp.json()) repo = ws.gh.get_repo("Koozali-SME-Server/smeserver-manager") print("Comments:", [c.text.strip() for c in repo.comments]) print("PR state:", repo.get_pull(42).state) print("Created bugs:", ws.bz.created_bugs) # Test synchronize on same PR (existing bug id used) ws.state.set_bug("Koozali-SME-Server/smeserver-manager#42", ws.bz.created_bugs[0]) payload_sync = sample_payload("synchronize") body_sync = json.dumps(payload_sync).encode("utf-8") headers_sync = { "X-GitHub-Event": "pull_request", "X-Hub-Signature-256": sign(secret, body_sync), "Content-Type": "application/json", } resp2 = client.post("/webhook/github", data=body_sync, headers=headers_sync) print("Sync response:", resp2.status_code, resp2.json()) print("Attachments recorded:", ws.bz.attachments) if __name__ == "__main__": import os run_tests()