smeserver-gitutils/TestPrograms/test_webhook_endpoint_standalone.py

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