github_release_monitor/tests/test_integration.py

"""End-to-end integration tests."""

import sqlite3
import threading
import tempfile
import urllib.request
from pathlib import Path

import httpx
import pytest
import respx

from src.db import _create_tables, add_repo, upsert_entry, get_connection
from src.server import create_handler
from http.server import HTTPServer


@pytest.fixture
def mock_github():
    """Mock only GitHub API calls."""
    with respx.mock:
        yield respx


def fetch_feed(port: int) -> str:
    """Fetch feed from localhost using urllib (bypasses respx)."""
    url = f"http://127.0.0.1:{port}/feed.xml"
    with urllib.request.urlopen(url) as resp:
        return resp.read().decode("utf-8")


class TestFullFlowWithMockedAPI:
    """Test the full flow: add repo, check (with mocked API), serve, fetch feed."""

    def test_release_mode_full_flow(self, mock_github):
        mock_github.get("https://api.github.com/repos/signoz/signoz/releases").mock(
            return_value=httpx.Response(200, json=[
                {
                    "tag_name": "v1.0.0",
                    "name": "Signoz v1.0.0",
                    "body": "# Release v1.0.0\n\n- New dashboard features",
                    "published_at": "2025-01-15T10:30:00Z",
                    "html_url": "https://github.com/signoz/signoz/releases/tag/v1.0.0",
                }
            ])
        )

        with tempfile.TemporaryDirectory() as tmp:
            db_path = Path(tmp) / "test.db"
            conn = get_connection(db_path)
            add_repo(conn, "signoz/signoz", "release")
            conn.close()

            from src.github import GitHubClient
            from src.db import get_repos_with_mode, upsert_entry as db_upsert

            conn = get_connection(db_path)
            repos = get_repos_with_mode(conn)
            assert len(repos) == 1

            with GitHubClient(token="fake_token") as gh:
                for repo_id, owner_repo, mode in repos:
                    items = gh.fetch_releases(owner_repo)
                    for item in items:
                        db_upsert(conn, repo_id, mode,
                                  item["tag_name"], item["title"], item["body"],
                                  item["published_at"], item["html_url"])

            conn.close()

            handler = create_handler(str(db_path))
            server = HTTPServer(("127.0.0.1", 0), handler)
            port = server.server_address[1]
            thread = threading.Thread(target=server.serve_forever, daemon=True)
            thread.start()

            try:
                text = fetch_feed(port)
                assert "signoz/signoz" in text
                assert "v1.0.0" in text
                assert "Signoz v1.0.0" in text
            finally:
                server.shutdown()

    def test_tag_mode_full_flow(self, mock_github):
        mock_github.get("https://api.github.com/repos/zammad/zammad/tags").mock(
            return_value=httpx.Response(200, json=[
                {"name": "6.3.0", "zipball_url": "https://api.github.com/..."},
            ])
        )
        mock_github.get("https://api.github.com/repos/zammad/zammad/git/ref/tags/6.3.0").mock(
            return_value=httpx.Response(200, json={
                "object": {
                    "type": "commit",
                    "author": {"date": "2025-02-01T12:00:00Z"},
                }
            })
        )

        with tempfile.TemporaryDirectory() as tmp:
            db_path = Path(tmp) / "test.db"
            conn = get_connection(db_path)
            add_repo(conn, "zammad/zammad", "tag")
            conn.close()

            from src.github import GitHubClient
            from src.db import get_repos_with_mode, upsert_entry as db_upsert

            conn = get_connection(db_path)
            repos = get_repos_with_mode(conn)

            with GitHubClient(token="fake_token") as gh:
                for repo_id, owner_repo, mode in repos:
                    items = gh.fetch_tags(owner_repo)
                    for item in items:
                        db_upsert(conn, repo_id, "tag",
                                  item["tag_name"], item["title"], item["body"],
                                  item["published_at"], item["html_url"])

            conn.close()

            handler = create_handler(str(db_path))
            server = HTTPServer(("127.0.0.1", 0), handler)
            port = server.server_address[1]
            thread = threading.Thread(target=server.serve_forever, daemon=True)
            thread.start()

            try:
                text = fetch_feed(port)
                assert "zammad/zammad" in text
                assert "6.3.0" in text
                assert "Tag 6.3.0" in text
            finally:
                server.shutdown()


class TestMultipleRepos:
    def test_mixed_repos_in_feed(self, mock_github):
        """Feed should contain entries from both release and tag repos, sorted by date."""
        mock_github.get("https://api.github.com/repos/a/b/releases").mock(
            return_value=httpx.Response(200, json=[
                {
                    "tag_name": "v1.0.0",
                    "name": "v1.0.0",
                    "body": "Release body",
                    "published_at": "2025-01-10T00:00:00Z",
                    "html_url": "https://github.com/a/b/releases/tag/v1.0.0",
                }
            ])
        )
        mock_github.get("https://api.github.com/repos/c/d/tags").mock(
            return_value=httpx.Response(200, json=[
                {"name": "v2.0.0", "zipball_url": "https://api.github.com/..."},
            ])
        )
        mock_github.get("https://api.github.com/repos/c/d/git/ref/tags/v2.0.0").mock(
            return_value=httpx.Response(200, json={
                "object": {
                    "type": "commit",
                    "author": {"date": "2025-01-20T00:00:00Z"},
                }
            })
        )

        with tempfile.TemporaryDirectory() as tmp:
            db_path = Path(tmp) / "test.db"
            conn = get_connection(db_path)
            add_repo(conn, "a/b", "release")
            add_repo(conn, "c/d", "tag")
            conn.close()

            from src.github import GitHubClient
            from src.db import get_repos_with_mode, upsert_entry as db_upsert

            conn = get_connection(db_path)
            repos = get_repos_with_mode(conn)

            with GitHubClient(token="fake_token") as gh:
                for repo_id, owner_repo, mode in repos:
                    if mode == "tag":
                        items = gh.fetch_tags(owner_repo)
                        kind = "tag"
                    else:
                        items = gh.fetch_releases(owner_repo)
                        kind = "release"
                    for item in items:
                        db_upsert(conn, repo_id, kind,
                                  item["tag_name"], item["title"], item["body"],
                                  item["published_at"], item["html_url"])

            conn.close()

            handler = create_handler(str(db_path))
            server = HTTPServer(("127.0.0.1", 0), handler)
            port = server.server_address[1]
            thread = threading.Thread(target=server.serve_forever, daemon=True)
            thread.start()

            try:
                text = fetch_feed(port)
                c_d_pos = text.find("c/d")
                a_b_pos = text.find("a/b")
                assert c_d_pos < a_b_pos
            finally:
                server.shutdown()