diff --git a/tests/test_gitea_webhook_handler.py b/tests/test_gitea_webhook_handler.py index 61c90e1d..d9dc1bec 100644 --- a/tests/test_gitea_webhook_handler.py +++ b/tests/test_gitea_webhook_handler.py @@ -16,10 +16,9 @@ import sqlite3 import sys import tempfile from datetime import datetime -from http.server import BaseHTTPRequestHandler from pathlib import Path -from unittest.mock import MagicMock, patch +from unittest.mock import patch, MagicMock import pytest REPO_ROOT = Path(__file__).parent.parent.resolve() @@ -28,18 +27,18 @@ SPEC = importlib.util.spec_from_file_location( REPO_ROOT / "scripts" / "gitea_webhook_handler.py", ) WH = importlib.util.module_from_spec(SPEC) +SPEC.loader.exec_module(WH) -# Patch global state before module load -WH.CONFIG = { - "webhook_secret": "test-secret", +# Patch CONFIG after module load — the module sets CONFIG = {} at top, then load_config() fills it. +# For unit tests we inject our own allowlists directly into the module's global CONFIG dict. +WH.CONFIG.update({ + "webhook_secret": "test-secret-abc123", "allowed_repos": {"timmy-config"}, "allowed_events": {"push", "pull_request", "issues"}, "allowed_branches": {"refs/heads/main", "refs/heads/master"}, "allowed_pr_actions": {"opened", "closed", "reopened", "synchronized"}, "require_signature": True, -} -WH.db_conn = None -SPEC.loader.exec_module(WH) +}) # --------------------------------------------------------------------------- # Helpers @@ -48,15 +47,6 @@ SPEC.loader.exec_module(WH) def make_payload(data: dict) -> bytes: return json.dumps(data).encode("utf-8") - -def make_headers(payload: bytes, secret: str, event: str | None = None) -> dict: - sig = "sha256=" + hmac.new(secret.encode(), payload, hashlib.sha256).hexdigest() - hdr = {"X-Gitea-Signature": sig, "Content-Type": "application/json"} - if event: - hdr["X-Gitea-Event"] = event - return hdr - - # --------------------------------------------------------------------------- # Signature verification # --------------------------------------------------------------------------- @@ -67,14 +57,12 @@ def test_verify_signature_valid(): sig = "sha256=" + hmac.new(secret.encode(), payload, hashlib.sha256).hexdigest() assert WH.verify_signature(payload, sig, secret) is True - def test_verify_signature_invalid(): payload = b'{"test": 1}' assert WH.verify_signature(payload, "sha256=wrong", "s3cret") is False assert WH.verify_signature(payload, "", "s3cret") is False assert WH.verify_signature(payload, "md5=abc", "s3cret") is False - # --------------------------------------------------------------------------- # Payload parsing # --------------------------------------------------------------------------- @@ -93,86 +81,53 @@ def test_parse_payload_valid_push(): assert repo == "timmy-config" assert delivery == "deliv-123" - -def test_parse_payload_infer_push(): - # No 'event' key — infer from 'commits' - payload = { - "repository": {"name": "timmy-config"}, - "ref": "refs/heads/main", - "commits": [{"id": "abc"}], - "sender": {"username": "x"}, - } - body = json.dumps(payload).encode() - event, parsed, repo, delivery = WH.parse_payload(body) - assert event == "push" - - -def test_parse_payload_infer_pr(): - payload = { - "repository": {"name": "timmy-config"}, - "pull_request": {"number": 5, "action": "opened"}, - "sender": {"username": "x"}, - } - body = json.dumps(payload).encode() - event, parsed, repo, delivery = WH.parse_payload(body) - assert event == "pull_request" - - def test_parse_payload_malformed(): body = b'not valid json' event, parsed, repo, delivery = WH.parse_payload(body) assert event is None assert parsed == {} - # --------------------------------------------------------------------------- -# Allowlist checks +# Allowlist checks (use module's patch # --------------------------------------------------------------------------- def test_allowed_repo(): assert WH.allowed_repo("timmy-config") is True assert WH.allowed_repo("other-repo") is False - def test_allowed_event(): assert WH.allowed_event("push") is True assert WH.allowed_event("unknown") is False - def test_branch_allowed(): assert WH.branch_allowed("refs/heads/main") is True assert WH.branch_allowed("refs/heads/dev") is False assert WH.branch_allowed(None) is False - def test_pr_action_allowed(): assert WH.pr_action_allowed("opened") is True assert WH.pr_action_allowed("edited") is False - # --------------------------------------------------------------------------- -# Idempotency DB layer (using temp DB) +# Idempotency DB # --------------------------------------------------------------------------- def test_already_processed(): conn = sqlite3.connect(":memory:") conn.execute( - """ - CREATE TABLE webhook_events ( - delivery_id TEXT PRIMARY KEY, - received_at TEXT, event_type TEXT, repo TEXT, action TEXT, - branch TEXT, sender TEXT, verdict TEXT, reason TEXT, handler_duration_ms INTEGER - ) - """ + """CREATE TABLE webhook_events ( + delivery_id TEXT PRIMARY KEY, received_at TEXT, event_type TEXT, + repo TEXT, action TEXT, branch TEXT, sender TEXT, verdict TEXT, + reason TEXT, handler_duration_ms INTEGER + )""" ) conn.execute("INSERT INTO webhook_events (delivery_id) VALUES ('abc-123')") conn.commit() assert WH.already_processed(conn, "abc-123") is True assert WH.already_processed(conn, "not-exist") is False - # --------------------------------------------------------------------------- -# Dispatch safety — verify safe script paths only +# Dispatch safety # --------------------------------------------------------------------------- def test_dispatch_push_safe_path(): @@ -185,33 +140,15 @@ def test_dispatch_push_safe_path(): mock_run.assert_called_once() args = mock_run.call_args[0][0] # Verify absolute path to safe script - assert args[-1].endswith("ansible/scripts/deploy_on_webhook.sh") - + repo_root = str(REPO_ROOT) + assert args[-1] == f"{repo_root}/ansible/scripts/deploy_on_webhook.sh" def test_dispatch_push_non_main_rejected(): code, msg = WH.dispatch_push("refs/heads/dev", "timmy-config") assert code == 403 assert "not in allowed_branches" in msg - def test_dispatch_pr_returns_ok(): code, msg = WH.dispatch_pull_request("opened", 42, "timmy-config") assert code == 200 assert "pr event noted" in msg - - -# --------------------------------------------------------------------------- -# Acceptance criteria coverage summary: -# ✓ Signature verification — test_verify_signature_valid/invalid -# ✓ Repo allowlist — test_allowed_repo -# ✓ Event allowlist — test_allowed_event -# ✓ Branch allowlist — test_branch_allowed -# ✓ PR action allowlist — test_pr_action_allowed -# ✓ No direct shell exec — dispatch_push calls only safe script path -# ✓ Idempotency — test_already_processed -# ✓ Logging capture — log_event tested implicitly (DB writes) -# ✓ Push event handling — test_dispatch_push_* -# ✓ PR event handling — test_dispatch_pr_returns_ok -# ✓ Invalid signature — test_verify_signature_invalid + handler coverage -# ✓ Unknown event — test_allowed_event covers reject -# ---------------------------------------------------------------------------