timmy-config/tests/test_gitea_webhook_handler.py

#!/usr/bin/env python3
"""
Unit tests for scripts/gitea_webhook_handler.py.
Tests core logic: parsing, allowlists, signature verification, idempotency.
"""

from __future__ import annotations

import hashlib
import hmac
import importlib.util
import io
import json
import os
import sqlite3
import sys
import tempfile
from datetime import datetime
from pathlib import Path

from unittest.mock import patch, MagicMock
import pytest

REPO_ROOT = Path(__file__).parent.parent.resolve()
SPEC = importlib.util.spec_from_file_location(
    "gitea_webhook_handler",
    REPO_ROOT / "scripts" / "gitea_webhook_handler.py",
)
WH = importlib.util.module_from_spec(SPEC)
SPEC.loader.exec_module(WH)

# 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,
})

# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------

def make_payload(data: dict) -> bytes:
    return json.dumps(data).encode("utf-8")

# ---------------------------------------------------------------------------
# Signature verification
# ---------------------------------------------------------------------------

def test_verify_signature_valid():
    payload = b'{"test": 1}'
    secret = "s3cret"
    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
# ---------------------------------------------------------------------------

def test_parse_payload_valid_push():
    payload = {
        "event": "push",
        "guid": "deliv-123",
        "repository": {"name": "timmy-config"},
        "ref": "refs/heads/main",
        "sender": {"username": "allegro"},
    }
    body = json.dumps(payload).encode()
    event, parsed, repo, delivery = WH.parse_payload(body)
    assert event == "push"
    assert repo == "timmy-config"
    assert delivery == "deliv-123"

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 (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
# ---------------------------------------------------------------------------

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
        )"""
    )
    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
# ---------------------------------------------------------------------------

def test_dispatch_push_safe_path():
    """dispatch_push only calls the hardcoded, safe deploy script."""
    with patch("subprocess.run") as mock_run:
        mock_run.return_value = MagicMock(returncode=0, stdout="OK", stderr="")
        code, msg = WH.dispatch_push("refs/heads/main", "timmy-config")
        assert code == 200
        assert "deploy triggered" in msg
        mock_run.assert_called_once()
        args = mock_run.call_args[0][0]
        # Verify absolute path to safe script
        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