AstrBot/tests/test_github_webhook.py

"""Test GitHub webhook platform adapter"""

import asyncio
import hashlib
import hmac
from unittest.mock import MagicMock

import pytest

from astrbot.core.platform.sources.github_webhook.github_webhook_adapter import (
    GitHubWebhookPlatformAdapter,
)


@pytest.fixture
def event_queue():
    """Create a test event queue"""
    return asyncio.Queue()


@pytest.fixture
def platform_config():
    """Create test platform configuration"""
    return {
        "type": "github_webhook",
        "enable": True,
        "id": "test_github_webhook",
        "unified_webhook_mode": True,
        "webhook_uuid": "test-uuid-123",
        "webhook_secret": "",  # No secret by default for easier testing
    }


@pytest.fixture
def platform_settings():
    """Create test platform settings"""
    return {"unique_session": False}


@pytest.fixture
def adapter(platform_config, platform_settings, event_queue):
    """Create test adapter instance"""
    return GitHubWebhookPlatformAdapter(platform_config, platform_settings, event_queue)


class TestGitHubWebhookAdapter:
    """Test cases for GitHub webhook adapter"""

    def test_adapter_initialization(self, adapter):
        """Test adapter is initialized correctly"""
        assert adapter.unified_webhook_mode is True
        assert adapter.webhook_secret == ""
        assert adapter.meta().name == "github_webhook"
        assert adapter.meta().description == "GitHub Webhook 适配器"

    @pytest.mark.asyncio
    async def test_ping_event(self, adapter):
        """Test GitHub ping event"""
        # Mock request
        request = MagicMock()
        request.headers.get.return_value = "ping"

        async def mock_json():
            return {}

        request.json = mock_json()

        response = await adapter.webhook_callback(request)
        assert response == {"message": "pong"}

    @pytest.mark.asyncio
    async def test_issue_created_event(self, adapter, event_queue):
        """Test GitHub issue created event"""
        # Mock request with issue created payload
        request = MagicMock()
        request.headers.get.return_value = "issues"
        payload = {
            "action": "opened",
            "issue": {
                "title": "Test Issue",
                "body": "This is a test issue",
                "html_url": "https://github.com/test/repo/issues/1",
            },
            "repository": {"full_name": "test/repo"},
            "sender": {"login": "testuser"},
        }

        async def mock_json():
            return payload

        request.json = mock_json()

        response = await adapter.webhook_callback(request)
        assert response == {"status": "ok"}

        # Verify event was queued
        assert not event_queue.empty()
        event = event_queue.get_nowait()
        assert event.event_type == "issues"
        assert "新 Issue 创建" in event.message_str
        assert "Test Issue" in event.message_str

    @pytest.mark.asyncio
    async def test_issue_comment_event(self, adapter, event_queue):
        """Test GitHub issue comment event"""
        request = MagicMock()
        request.headers.get.return_value = "issue_comment"
        payload = {
            "action": "created",
            "issue": {"title": "Test Issue"},
            "comment": {
                "body": "Test comment",
                "html_url": "https://github.com/test/repo/issues/1#comment",
            },
            "repository": {"full_name": "test/repo"},
            "sender": {"login": "commenter"},
        }

        async def mock_json():
            return payload

        request.json = mock_json()

        response = await adapter.webhook_callback(request)
        assert response == {"status": "ok"}

        # Verify event was queued
        assert not event_queue.empty()
        event = event_queue.get_nowait()
        assert event.event_type == "issue_comment"
        assert "新 Issue 评论" in event.message_str
        assert "Test comment" in event.message_str

    @pytest.mark.asyncio
    async def test_pull_request_event(self, adapter, event_queue):
        """Test GitHub pull request opened event"""
        request = MagicMock()
        request.headers.get.return_value = "pull_request"
        payload = {
            "action": "opened",
            "pull_request": {
                "title": "Test PR",
                "body": "This is a test PR",
                "html_url": "https://github.com/test/repo/pull/1",
            },
            "repository": {"full_name": "test/repo"},
            "sender": {"login": "prauthor"},
        }

        async def mock_json():
            return payload

        request.json = mock_json()

        response = await adapter.webhook_callback(request)
        assert response == {"status": "ok"}

        # Verify event was queued
        assert not event_queue.empty()
        event = event_queue.get_nowait()
        assert event.event_type == "pull_request"
        assert "新 Pull Request" in event.message_str
        assert "Test PR" in event.message_str

    @pytest.mark.asyncio
    async def test_unsupported_event(self, adapter, event_queue):
        """Test unsupported GitHub event type"""
        request = MagicMock()
        request.headers.get.return_value = "push"

        async def mock_json():
            return {"action": "created"}

        request.json = mock_json()

        response = await adapter.webhook_callback(request)
        assert response == {"status": "ok"}

        # Verify no event was queued for unsupported events
        assert event_queue.empty()

    @pytest.mark.asyncio
    async def test_issue_closed_ignored(self, adapter, event_queue):
        """Test that issue closed action is ignored"""
        request = MagicMock()
        request.headers.get.return_value = "issues"
        payload = {
            "action": "closed",  # Should be ignored
            "issue": {"title": "Test Issue"},
            "repository": {"full_name": "test/repo"},
            "sender": {"login": "testuser"},
        }

        async def mock_json():
            return payload

        request.json = mock_json()

        response = await adapter.webhook_callback(request)
        assert response == {"status": "ok"}

        # Verify no event was queued
        assert event_queue.empty()

    @pytest.mark.asyncio
    async def test_signature_verification(self, platform_settings, event_queue):
        """Test webhook signature verification"""
        # Create adapter with webhook secret
        config_with_secret = {
            "type": "github_webhook",
            "enable": True,
            "id": "test_github_webhook",
            "unified_webhook_mode": True,
            "webhook_uuid": "test-uuid-123",
            "webhook_secret": "test-secret",
        }
        adapter = GitHubWebhookPlatformAdapter(
            config_with_secret, platform_settings, event_queue
        )

        # Create a valid signature
        body = b'{"action": "opened"}'
        signature = hmac.new(b"test-secret", body, hashlib.sha256).hexdigest()

        # Mock request with valid signature
        request = MagicMock()
        request.headers.get = lambda key, default="": {
            "X-GitHub-Event": "ping",
            "X-Hub-Signature-256": f"sha256={signature}",
        }.get(key, default)

        async def mock_get_data():
            return body

        request.get_data = mock_get_data

        async def mock_json():
            return {"action": "opened"}

        request.json = mock_json()

        response = await adapter.webhook_callback(request)
        assert response == {"message": "pong"}

    @pytest.mark.asyncio
    async def test_invalid_signature(self, platform_settings, event_queue):
        """Test webhook with invalid signature is rejected"""
        # Create adapter with webhook secret
        config_with_secret = {
            "type": "github_webhook",
            "enable": True,
            "id": "test_github_webhook",
            "unified_webhook_mode": True,
            "webhook_uuid": "test-uuid-123",
            "webhook_secret": "test-secret",
        }
        adapter = GitHubWebhookPlatformAdapter(
            config_with_secret, platform_settings, event_queue
        )

        # Mock request with invalid signature
        request = MagicMock()
        request.headers.get = lambda key, default="": {
            "X-GitHub-Event": "ping",
            "X-Hub-Signature-256": "sha256=invalidsignature",
        }.get(key, default)

        async def mock_get_data():
            return b'{"action": "opened"}'

        request.get_data = mock_get_data

        async def mock_json():
            return {"action": "opened"}

        request.json = mock_json()

        response = await adapter.webhook_callback(request)
        assert response == ({"error": "Invalid signature"}, 401)