from __future__ import annotations import importlib.util import json import os from pathlib import Path import sys from types import SimpleNamespace import types import pytest from fastapi import FastAPI from httpx import ASGITransport, AsyncClient os.environ.setdefault("DATABASE_URL", "postgresql+asyncpg://tester:tester@localhost:5432/pmflow_test") from app.core.database import get_db from app.dependencies.auth import get_current_user def _load_dev_sync_module(): package_name = "app.api" module_name = "app.api.dev_sync" module_path = Path(__file__).resolve().parents[1] / "app" / "api" / "dev_sync.py" if package_name not in sys.modules: package = types.ModuleType(package_name) package.__path__ = [str(module_path.parent)] # type: ignore[attr-defined] sys.modules[package_name] = package if module_name in sys.modules: return sys.modules[module_name] spec = importlib.util.spec_from_file_location(module_name, module_path) if spec is None or spec.loader is None: raise RuntimeError("无法加载 dev_sync 模块") module = importlib.util.module_from_spec(spec) sys.modules[module_name] = module spec.loader.exec_module(module) return module dev_sync = _load_dev_sync_module() def _build_app(*, role: str = "OWNER") -> FastAPI: app = FastAPI() app.include_router(dev_sync.router) async def _fake_get_db(): yield object() app.dependency_overrides[get_db] = _fake_get_db app.dependency_overrides[get_current_user] = lambda: SimpleNamespace( id=7, tenant_id=1, username="owner", role=role, ) return app @pytest.mark.asyncio async def test_github_install_url_route_returns_install_center_payload(monkeypatch) -> None: class FakeService: def __init__(self, db, tenant_id: int): _ = db self.tenant_id = tenant_id async def get_github_app_install_center(self): return { "configured": True, "install_url": "https://github.com/apps/pmflow/installations/new", "callback_url": "https://pmflow.local/github/callback", "webhook_signature_configured": True, "app_slug": "pmflow", } monkeypatch.setattr(dev_sync, "AIProjectControlService", FakeService) app = _build_app() async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client: response = await client.get("/integrations/github/app/install-url") assert response.status_code == 200 payload = response.json() assert payload["code"] == 0 assert payload["data"]["configured"] is True assert payload["data"]["install_url"] == "https://github.com/apps/pmflow/installations/new" @pytest.mark.asyncio async def test_github_webhook_route_passes_raw_body_to_service(monkeypatch) -> None: captured: dict[str, object] = {} class FakeService: def __init__(self, db, tenant_id: int): _ = db captured["tenant_id"] = tenant_id async def ingest_github_webhook(self, **kwargs): captured.update(kwargs) return { "delivery_id": kwargs["delivery_id"], "matched_links": 1, } monkeypatch.setattr(dev_sync, "AIProjectControlService", FakeService) raw_body = json.dumps( {"tenant_id": 9, "event_type": "pull_request", "repository": {"full_name": "pmflow/control-plane"}}, separators=(",", ":"), ).encode("utf-8") app = _build_app() async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client: response = await client.post( "/integrations/github/webhook", content=raw_body, headers={ "Content-Type": "application/json", "X-GitHub-Event": "pull_request", "X-GitHub-Delivery": "delivery-1", "X-Hub-Signature-256": "sha256=test", }, ) assert response.status_code == 200 assert captured["tenant_id"] == 9 assert captured["event_type"] == "pull_request" assert captured["delivery_id"] == "delivery-1" assert captured["signature_256"] == "sha256=test" assert captured["raw_body"] == raw_body @pytest.mark.asyncio async def test_github_installations_sync_route_returns_installations(monkeypatch) -> None: class FakeService: def __init__(self, db, tenant_id: int): _ = db, tenant_id async def sync_github_installations(self): return { "installations": [ { "id": "row-1", "installation_id": "101", "account_login": "pmflow", } ], "repository_count": 3, } monkeypatch.setattr(dev_sync, "AIProjectControlService", FakeService) app = _build_app() async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client: response = await client.post("/integrations/github/app/installations/sync") assert response.status_code == 200 payload = response.json()["data"] assert payload["repository_count"] == 3 assert payload["installations"][0]["account_login"] == "pmflow" @pytest.mark.asyncio async def test_github_repo_bind_route_surfaces_installation_repo_errors(monkeypatch) -> None: class FakeService: def __init__(self, db, tenant_id: int): _ = db, tenant_id async def bind_github_repository(self, *, project_id: int, github_repository_id: int): if github_repository_id == 99: raise ValueError("当前仓库不在 GitHub App 已授权仓库列表中") return { "id": "binding-1", "project_id": str(project_id), "provider": "github", "full_name": "pmflow/control-plane", "status": "active", } async def _allow_project(*args, **kwargs): return None monkeypatch.setattr(dev_sync, "AIProjectControlService", FakeService) monkeypatch.setattr(dev_sync, "_ensure_project_member_or_manager", _allow_project) app = _build_app() async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client: ok_response = await client.post( "/integrations/github/repos/bind", json={"project_id": 17, "github_repository_id": 1}, ) bad_response = await client.post( "/integrations/github/repos/bind", json={"project_id": 17, "github_repository_id": 99}, ) assert ok_response.status_code == 200 assert ok_response.json()["data"]["full_name"] == "pmflow/control-plane" assert bad_response.status_code == 400 assert "已授权仓库列表" in bad_response.json()["detail"]