mirror of
https://github.com/Sea-Haven-Industries/open-swe.git
synced 2026-09-30 19:43:15 +00:00
749 lines
26 KiB
Python
749 lines
26 KiB
Python
"""HTTP app for the full-flow E2E (served as langgraph dev's http.app).
|
||
|
||
Mounts, on top of the REAL ``agent.api.app`` app:
|
||
- fake GitHub REST API (/fake-gh/...) the real open_pull_request hits this
|
||
- fake Slack API (/fake-slack/...) the real slack code hits this
|
||
- mock UIs (/mock/slack, /mock/github) what the user/Playwright sees
|
||
- control + compose (/control/*, /mock/slack/send) the test driver
|
||
|
||
Nothing here touches agent logic — it only stands in for the SaaS boundaries
|
||
and renders their state back as a user-facing UI.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import hashlib
|
||
import hmac
|
||
import json
|
||
import os
|
||
import sys
|
||
import time
|
||
from html import escape
|
||
from pathlib import Path
|
||
from typing import Any
|
||
from urllib.parse import quote
|
||
|
||
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
||
|
||
import e2e_env # noqa: E402
|
||
import patches # noqa: E402
|
||
|
||
patches.apply()
|
||
|
||
import fakes # noqa: E402
|
||
import httpx # noqa: E402
|
||
from e2e_env import ( # noqa: E402
|
||
BASE_URL,
|
||
BOT_USER_ID,
|
||
DEMO_CHANNEL,
|
||
HUMAN_USER,
|
||
REPO_ROOT,
|
||
TEST_USERS,
|
||
)
|
||
from fastapi import HTTPException, Request # noqa: E402
|
||
from fastapi.responses import ( # noqa: E402
|
||
FileResponse,
|
||
HTMLResponse,
|
||
JSONResponse,
|
||
RedirectResponse,
|
||
Response,
|
||
)
|
||
|
||
# Slack-user directory the fake ``users.info`` resolves: the default sender used
|
||
# by the automated tests plus the named manual-test users.
|
||
_SLACK_USERS: dict[str, dict[str, str]] = {
|
||
HUMAN_USER: {"name": "devuser", "real_name": "Dev User", "email": "dev@example.com"},
|
||
**{
|
||
u["slack_id"]: {"name": u["login"], "real_name": u["name"], "email": u["email"]}
|
||
for u in TEST_USERS
|
||
},
|
||
}
|
||
|
||
from agent.api.app import app # noqa: E402
|
||
from agent.dashboard import routes as dashboard_routes # noqa: E402
|
||
from agent.dashboard.oauth import COOKIE_NAME, issue_session # noqa: E402
|
||
from agent.utils import github_checks, github_org_membership # noqa: E402
|
||
from agent.utils.thread_ids import generate_thread_id_from_slack_thread # noqa: E402
|
||
from agent.webhooks import common as webhook_common # noqa: E402
|
||
|
||
GITHUB_WEBHOOK_SECRET = os.environ["GITHUB_WEBHOOK_SECRET"]
|
||
SLACK_SIGNING_SECRET = os.environ["SLACK_SIGNING_SECRET"]
|
||
STATIC_DIR = Path(__file__).parent / "static"
|
||
|
||
CURRENT_THREAD: dict[str, str | None] = {"channel": DEMO_CHANNEL, "thread_ts": None}
|
||
|
||
fakes.seed_bare_remote()
|
||
_real_dispatch_agent_run = webhook_common.dispatch_agent_run
|
||
_real_email_for_login = webhook_common.email_for_login
|
||
|
||
|
||
async def _fake_installation_token(*_args: object, **_kwargs: object) -> str:
|
||
return "dummy-installation-token"
|
||
|
||
|
||
async def _fake_installation_token_with_expiry(
|
||
*_args: object, **_kwargs: object
|
||
) -> tuple[str, None]:
|
||
return "dummy-installation-token", None
|
||
|
||
|
||
async def _fake_fetch_pr_metadata(pr_ref: Any, *, token: str) -> dict[str, Any] | None: # noqa: ARG001
|
||
pr = fakes.find_pull(pr_ref.number)
|
||
return _gh_pr_json(pr) if pr is not None else None
|
||
|
||
|
||
async def _fake_reviewer_token(*_args: object, **_kwargs: object) -> tuple[str, None]:
|
||
return "dummy-installation-token", None
|
||
|
||
|
||
async def _fake_reaction(*_args: object, **_kwargs: object) -> bool:
|
||
return True
|
||
|
||
|
||
async def _fake_started_comment(*_args: object, **_kwargs: object) -> None:
|
||
return None
|
||
|
||
|
||
async def _fake_active_org_member(username: str, org: str) -> bool:
|
||
return bool(username and org)
|
||
|
||
|
||
async def _record_email_mapping_lookup(login: str) -> str | None:
|
||
fakes.EMAIL_MAPPING_LOOKUPS.append(login)
|
||
return await _real_email_for_login(login)
|
||
|
||
|
||
async def _record_review_dispatch(
|
||
thread_id: str,
|
||
prompt: str,
|
||
configurable: dict[str, Any],
|
||
*,
|
||
source: str,
|
||
assistant_id: str = "agent",
|
||
**kwargs: object,
|
||
) -> dict[str, str]:
|
||
if assistant_id != "reviewer":
|
||
return await _real_dispatch_agent_run(
|
||
thread_id,
|
||
prompt,
|
||
configurable,
|
||
source=source,
|
||
assistant_id=assistant_id,
|
||
**kwargs,
|
||
)
|
||
run_id = f"review-run-{len(fakes.REVIEW_DISPATCHES) + 1}"
|
||
fakes.REVIEW_DISPATCHES.append(
|
||
{
|
||
"thread_id": thread_id,
|
||
"prompt": prompt,
|
||
"configurable": configurable,
|
||
"source": source,
|
||
"assistant_id": assistant_id,
|
||
"run_id": run_id,
|
||
}
|
||
)
|
||
return {"run_id": run_id}
|
||
|
||
|
||
async def _fake_installations_and_repos(
|
||
_login: str,
|
||
) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
|
||
return (
|
||
[{"id": 1, "account": {"login": e2e_env.OWNER, "type": "Organization"}}],
|
||
[{"full_name": f"{e2e_env.OWNER}/{e2e_env.REPO}", "private": False}],
|
||
)
|
||
|
||
|
||
webhook_common.get_github_app_installation_token = _fake_installation_token
|
||
webhook_common.get_github_app_installation_token_with_expiry = _fake_installation_token_with_expiry
|
||
webhook_common.fetch_github_pr_metadata = _fake_fetch_pr_metadata
|
||
webhook_common._reviewer_token_for_repo = _fake_reviewer_token
|
||
webhook_common.react_to_github_comment = _fake_reaction
|
||
webhook_common.post_review_started_comment = _fake_started_comment
|
||
webhook_common.dispatch_agent_run = _record_review_dispatch
|
||
webhook_common.email_for_login = _record_email_mapping_lookup
|
||
github_org_membership.is_user_active_org_member = _fake_active_org_member
|
||
webhook_common.is_user_active_org_member = _fake_active_org_member
|
||
dashboard_routes._fetch_user_installations_and_repos = _fake_installations_and_repos
|
||
github_checks._GITHUB_API_BASE = e2e_env.FAKE_GITHUB_API
|
||
|
||
|
||
# --- control + Slack compose (the test driver) -----------------------------
|
||
@app.post("/control/reset")
|
||
async def control_reset() -> JSONResponse:
|
||
fakes.reset()
|
||
CURRENT_THREAD["thread_ts"] = None
|
||
return JSONResponse({"ok": True})
|
||
|
||
|
||
@app.post("/control/review-pr")
|
||
async def control_review_pr() -> JSONResponse:
|
||
fakes.reset()
|
||
pr = fakes.create_review_pull(e2e_env.OWNER, e2e_env.REPO)
|
||
return JSONResponse(_gh_pr_json(pr))
|
||
|
||
|
||
@app.get("/control/review-dispatches")
|
||
async def control_review_dispatches() -> JSONResponse:
|
||
return JSONResponse(fakes.REVIEW_DISPATCHES)
|
||
|
||
|
||
@app.get("/control/email-mapping-lookups")
|
||
async def control_email_mapping_lookups() -> JSONResponse:
|
||
return JSONResponse(fakes.EMAIL_MAPPING_LOOKUPS)
|
||
|
||
|
||
@app.get("/control/check-runs")
|
||
async def control_check_runs() -> JSONResponse:
|
||
return JSONResponse(fakes.CHECK_RUNS)
|
||
|
||
|
||
@app.post("/control/review-check/evaluate")
|
||
async def control_review_check_evaluate(request: Request) -> JSONResponse:
|
||
body = await request.json()
|
||
outcome = body.get("outcome")
|
||
if not isinstance(outcome, dict):
|
||
raise HTTPException(400, "outcome must be an object")
|
||
head_sha = str(body.get("head_sha") or f"evaluation-{len(fakes.CHECK_RUNS) + 1}")
|
||
check_run_id = await github_checks.create_review_check_run(
|
||
owner=e2e_env.OWNER,
|
||
repo=e2e_env.REPO,
|
||
head_sha=head_sha,
|
||
token="dummy-installation-token",
|
||
)
|
||
if check_run_id is None:
|
||
raise HTTPException(500, "failed to create check run")
|
||
conclusion, title, summary = github_checks.review_check_conclusion(outcome)
|
||
completed = await github_checks.complete_review_check_run(
|
||
owner=e2e_env.OWNER,
|
||
repo=e2e_env.REPO,
|
||
check_run_id=check_run_id,
|
||
token="dummy-installation-token",
|
||
conclusion=conclusion,
|
||
title=title,
|
||
summary=summary,
|
||
)
|
||
if not completed:
|
||
raise HTTPException(500, "failed to complete check run")
|
||
return JSONResponse({"id": check_run_id, "conclusion": conclusion, "title": title})
|
||
|
||
|
||
@app.get("/control/state")
|
||
async def control_state() -> JSONResponse:
|
||
return JSONResponse(
|
||
{"channel": CURRENT_THREAD["channel"], "thread_ts": CURRENT_THREAD["thread_ts"]}
|
||
)
|
||
|
||
|
||
@app.post("/mock/slack/send")
|
||
async def slack_send(request: Request) -> JSONResponse:
|
||
"""Simulate a user posting in Slack: store the message, then deliver the
|
||
signed Events-API webhook to the real /webhooks/slack route."""
|
||
form = await request.json()
|
||
text = str(form.get("text", ""))
|
||
mention_bot = bool(form.get("mention_bot", True))
|
||
# Sender defaults to the first test user (Alice) — the canonical owner the
|
||
# automated tests log in as; the mock UI passes the chosen test user.
|
||
user_id = str(form.get("user") or TEST_USERS[0]["slack_id"])
|
||
channel = DEMO_CHANNEL
|
||
|
||
ts = fakes.new_thread_ts()
|
||
CURRENT_THREAD["thread_ts"] = ts
|
||
fakes.add_slack_message(channel, ts, user=user_id, text=text, is_bot=False)
|
||
|
||
payload = {
|
||
"type": "event_callback",
|
||
"event_id": f"Ev{ts}",
|
||
"authorizations": [{"user_id": BOT_USER_ID}],
|
||
"event": {
|
||
"type": "app_mention" if mention_bot else "message",
|
||
"channel": channel,
|
||
"user": user_id,
|
||
"text": text,
|
||
"ts": ts,
|
||
"thread_ts": ts,
|
||
},
|
||
}
|
||
raw = json.dumps(payload).encode()
|
||
req_ts = str(int(time.time()))
|
||
base = f"v0:{req_ts}:{raw.decode()}".encode()
|
||
sig = "v0=" + hmac.new(SLACK_SIGNING_SECRET.encode(), base, hashlib.sha256).hexdigest()
|
||
|
||
transport = httpx.ASGITransport(app=app)
|
||
async with httpx.AsyncClient(transport=transport, base_url="http://harness") as client:
|
||
resp = await client.post(
|
||
"/webhooks/slack",
|
||
content=raw,
|
||
headers={
|
||
"X-Slack-Signature": sig,
|
||
"X-Slack-Request-Timestamp": req_ts,
|
||
"Content-Type": "application/json",
|
||
},
|
||
)
|
||
return JSONResponse(
|
||
{
|
||
"thread_ts": ts,
|
||
"thread_id": generate_thread_id_from_slack_thread(channel, ts),
|
||
"webhook_status": resp.status_code,
|
||
"webhook": resp.json(),
|
||
}
|
||
)
|
||
|
||
|
||
@app.post("/control/login")
|
||
async def control_login(request: Request) -> JSONResponse:
|
||
"""Simulate a signed-in dashboard user by minting the real session cookie."""
|
||
form = await request.json()
|
||
login = str(form.get("login", "dev-user"))
|
||
email = str(form.get("email", "dev@example.com"))
|
||
token = issue_session(login=login, email=email, avatar_url=None)
|
||
resp = JSONResponse({"ok": True, "login": login, "email": email})
|
||
resp.set_cookie(COOKIE_NAME, token, httponly=True, samesite="lax", secure=False, path="/")
|
||
return resp
|
||
|
||
|
||
@app.get("/control/login")
|
||
async def control_login_get(login: str = "", email: str = "", next_url: str = "") -> Response:
|
||
"""Browser login. With no ``login``, render a dropdown of the test users;
|
||
with ``?login=<u>`` (email resolved from the registry, or pass ``&email=``),
|
||
mint the session cookie and redirect into the dashboard. Use a separate
|
||
browser/profile per user — each has its own cookie jar."""
|
||
# Land on the dashboard origin (DASHBOARD_BASE_URL — the Vite HMR server in
|
||
# dev:mock), not this harness, so the cookie + the hot-reloading UI line up.
|
||
ui = os.environ.get("DASHBOARD_BASE_URL", "").rstrip("/")
|
||
dest = next_url or (f"{ui}/agents" if ui else "/agents")
|
||
if not login:
|
||
options = "".join(f'<option value="{u["login"]}">{u["name"]}</option>' for u in TEST_USERS)
|
||
return HTMLResponse(
|
||
f"""<!doctype html><meta charset=utf-8><title>Mock login</title>
|
||
<body style="font-family:system-ui;max-width:420px;margin:3rem auto;padding:0 1rem">
|
||
<h1 style="font-size:1.1rem">Sign in (mock)</h1>
|
||
<form method=get action=/control/login>
|
||
<select name=login style="font:inherit;padding:0.4rem">{options}</select>
|
||
<button style="font:inherit;padding:0.45rem 0.9rem;cursor:pointer">Sign in</button>
|
||
</form>
|
||
<p style="color:#888;font-size:0.85rem">Tip: use a separate browser or profile per
|
||
user so their sessions don't overwrite each other.</p>
|
||
</body>"""
|
||
)
|
||
if not email:
|
||
match = next((u for u in TEST_USERS if u["login"] == login), None)
|
||
email = match["email"] if match else f"{login}@example.com"
|
||
token = issue_session(login=login, email=email, avatar_url=None)
|
||
resp = RedirectResponse(url=dest, status_code=303)
|
||
resp.set_cookie(COOKIE_NAME, token, httponly=True, samesite="lax", secure=False, path="/")
|
||
return resp
|
||
|
||
|
||
@app.get("/dashboard/api/auth/login")
|
||
async def mock_github_login(redirect_to: str = "") -> Response:
|
||
"""E2E stand-in for the dashboard OAuth start route.
|
||
|
||
The real route would redirect to github.com. Keep the dashboard-facing URL
|
||
intact, then hand off to the fake GitHub simulator so Playwright exercises a
|
||
browser login flow instead of test code pre-minting a session cookie.
|
||
"""
|
||
ui = os.environ.get("DASHBOARD_BASE_URL", "").rstrip("/")
|
||
dest = redirect_to or (f"{ui}/agents" if ui else "/agents")
|
||
return RedirectResponse(f"/fake-gh/login/oauth/authorize?redirect_to={quote(dest)}", 302)
|
||
|
||
|
||
@app.get("/fake-gh/login/oauth/authorize")
|
||
async def fake_github_authorize(redirect_to: str = "", login: str = "") -> Response:
|
||
"""Fake GitHub OAuth consent/login page for dashboard e2e tests."""
|
||
ui = os.environ.get("DASHBOARD_BASE_URL", "").rstrip("/")
|
||
dest = redirect_to or (f"{ui}/agents" if ui else "/agents")
|
||
if not login:
|
||
options = "".join(
|
||
f'<option value="{escape(u["login"], quote=True)}">'
|
||
f"{escape(u['name'])} (@{escape(u['login'])})</option>"
|
||
for u in TEST_USERS
|
||
)
|
||
return HTMLResponse(
|
||
f"""<!doctype html><meta charset=utf-8><title>GitHub · Authorize open-swe</title>
|
||
<body style="font-family:system-ui;max-width:420px;margin:3rem auto;padding:0 1rem">
|
||
<main data-testid="fake-github-login">
|
||
<h1 style="font-size:1.1rem">Authorize open-swe</h1>
|
||
<p style="color:#888;font-size:0.9rem">Pick a fake GitHub account to continue.</p>
|
||
<form method=get action=/fake-gh/login/oauth/authorize>
|
||
<input type=hidden name=redirect_to value="{escape(dest, quote=True)}">
|
||
<label>GitHub user
|
||
<select name=login style="font:inherit;padding:0.4rem">{options}</select>
|
||
</label>
|
||
<button style="font:inherit;padding:0.45rem 0.9rem;cursor:pointer">Authorize open-swe</button>
|
||
</form>
|
||
</main>
|
||
</body>"""
|
||
)
|
||
match = next((u for u in TEST_USERS if u["login"] == login), None)
|
||
email = match["email"] if match else f"{login}@example.com"
|
||
token = issue_session(login=login, email=email, avatar_url=None)
|
||
resp = RedirectResponse(url=dest, status_code=303)
|
||
resp.set_cookie(COOKIE_NAME, token, httponly=True, samesite="lax", secure=False, path="/")
|
||
return resp
|
||
|
||
|
||
# The real dashboard registered /dashboard/api/auth/login first (via
|
||
# include_router), so Starlette would match it before ours. Move ours to the
|
||
# front of the table so the mock picker shadows the real OAuth redirect.
|
||
for _i, _route in enumerate(app.router.routes):
|
||
if getattr(_route, "endpoint", None) is mock_github_login:
|
||
app.router.routes.insert(0, app.router.routes.pop(_i))
|
||
break
|
||
|
||
|
||
@app.post("/control/logout")
|
||
async def control_logout() -> JSONResponse:
|
||
resp = JSONResponse({"ok": True})
|
||
resp.delete_cookie(COOKIE_NAME, path="/")
|
||
return resp
|
||
|
||
|
||
# --- serve the REAL built ui/ SPA, same-origin so the session cookie works ----
|
||
# The "Open in Web" link (DASHBOARD_BASE_URL/agents/{id}) lands on the real app;
|
||
# it calls /dashboard/api/* (same origin) and streams via the dashboard proxy.
|
||
UI_PUBLIC = REPO_ROOT / "ui" / ".output" / "public"
|
||
_ASSETS_ROOT = (UI_PUBLIC / "assets").resolve()
|
||
|
||
|
||
def _ui_file(name: str) -> FileResponse:
|
||
path = UI_PUBLIC / name
|
||
if not path.is_file():
|
||
raise HTTPException(404, f"{name} not built — run `bun run build` in ui/")
|
||
return FileResponse(path)
|
||
|
||
|
||
@app.get("/assets/{asset_path:path}")
|
||
async def ui_asset(asset_path: str) -> FileResponse:
|
||
# Explicit route, not app.mount(StaticFiles): LangGraph's custom-app loader
|
||
# serves APIRoutes but drops sub-app Mounts, so a mount 404s under it.
|
||
target = (_ASSETS_ROOT / asset_path).resolve()
|
||
if not str(target).startswith(str(_ASSETS_ROOT)) or not target.is_file():
|
||
raise HTTPException(404, "asset not found")
|
||
return FileResponse(target)
|
||
|
||
|
||
@app.get("/_shell.html", response_class=HTMLResponse)
|
||
async def ui_shell() -> FileResponse:
|
||
return _ui_file("_shell.html")
|
||
|
||
|
||
@app.get("/manifest.webmanifest")
|
||
async def ui_manifest() -> FileResponse:
|
||
return _ui_file("manifest.webmanifest")
|
||
|
||
|
||
@app.get("/favicon.png")
|
||
async def ui_favicon() -> FileResponse:
|
||
return _ui_file("favicon.png")
|
||
|
||
|
||
@app.get("/apple-touch-icon.png")
|
||
async def ui_apple_icon() -> FileResponse:
|
||
return _ui_file("apple-touch-icon.png")
|
||
|
||
|
||
@app.get("/logo-mark.png")
|
||
async def ui_logo_mark() -> FileResponse:
|
||
return _ui_file("logo-mark.png")
|
||
|
||
|
||
# Client routes used by the handoff tests: serve the SPA shell; the client
|
||
# router boots at the current URL. Kept explicit (no catch-all) so LangGraph's
|
||
# own root routes — which the dashboard proxy calls server-side — are untouched.
|
||
@app.get("/agents/{thread_id}", response_class=HTMLResponse)
|
||
async def ui_agents_thread(thread_id: str) -> FileResponse: # noqa: ARG001
|
||
return _ui_file("_shell.html")
|
||
|
||
|
||
@app.get("/agents/{thread_id}/plan", response_class=HTMLResponse)
|
||
async def ui_agents_plan(thread_id: str) -> FileResponse: # noqa: ARG001
|
||
return _ui_file("_shell.html")
|
||
|
||
|
||
@app.get("/review", response_class=HTMLResponse)
|
||
async def ui_review() -> FileResponse:
|
||
return _ui_file("_shell.html")
|
||
|
||
|
||
@app.get("/review/repositories/{owner}", response_class=HTMLResponse)
|
||
async def ui_review_repositories(owner: str) -> FileResponse: # noqa: ARG001
|
||
return _ui_file("_shell.html")
|
||
|
||
|
||
@app.get("/login", response_class=HTMLResponse)
|
||
async def ui_login() -> FileResponse:
|
||
return _ui_file("_shell.html")
|
||
|
||
|
||
@app.get("/mock/users")
|
||
async def mock_users() -> JSONResponse:
|
||
"""The named test users that drive the Slack sender + login dropdowns."""
|
||
return JSONResponse(TEST_USERS)
|
||
|
||
|
||
@app.get("/mock/slack/messages")
|
||
async def slack_messages() -> JSONResponse:
|
||
msgs = fakes.slack_messages(CURRENT_THREAD["channel"])
|
||
return JSONResponse(
|
||
[
|
||
{
|
||
"user": m["user"],
|
||
"text": m["text"],
|
||
"is_bot": m["is_bot"],
|
||
"ts": m["ts"],
|
||
"thread_ts": m["thread_ts"],
|
||
}
|
||
for m in msgs
|
||
]
|
||
)
|
||
|
||
|
||
# --- mock UIs --------------------------------------------------------------
|
||
@app.get("/mock/slack", response_class=HTMLResponse)
|
||
async def mock_slack_page() -> str:
|
||
return (STATIC_DIR / "slack.html").read_text()
|
||
|
||
|
||
@app.get("/mock/github", response_class=HTMLResponse)
|
||
async def mock_github_page() -> str:
|
||
return (STATIC_DIR / "github.html").read_text()
|
||
|
||
|
||
def _pr_html_url(pr: dict[str, Any]) -> str:
|
||
return f"{BASE_URL}/mock/github/{pr['owner']}/{pr['repo']}/pull/{pr['number']}"
|
||
|
||
|
||
@app.get("/mock/github/data")
|
||
async def mock_github_data() -> JSONResponse:
|
||
return JSONResponse(
|
||
[
|
||
{
|
||
"number": p["number"],
|
||
"title": p["title"],
|
||
"head": p["head"],
|
||
"base": p["base"],
|
||
"state": p["state"],
|
||
"draft": p["draft"],
|
||
"author": p["author"],
|
||
"body": p["body"],
|
||
"files": p["files"],
|
||
"url": _pr_html_url(p),
|
||
}
|
||
for p in fakes.PULLS
|
||
]
|
||
)
|
||
|
||
|
||
@app.get("/mock/github/{owner}/{repo}/pull/{number}", response_class=HTMLResponse)
|
||
async def mock_github_pr(owner: str, repo: str, number: int) -> HTMLResponse: # noqa: ARG001
|
||
pr = fakes.find_pull(number)
|
||
if pr is None:
|
||
return HTMLResponse(f"<h1>PR #{number} not found</h1>", status_code=404)
|
||
files = "".join(
|
||
f'<li data-file="{f["filename"]}">{f["filename"]} '
|
||
f"<span class='stat'>+{f['additions']} −{f['deletions']}</span></li>"
|
||
for f in pr["files"]
|
||
)
|
||
draft = " (draft)" if pr["draft"] else ""
|
||
return HTMLResponse(
|
||
f"""<!doctype html><meta charset=utf-8>
|
||
<title>PR #{pr["number"]} — {pr["owner"]}/{pr["repo"]}</title>
|
||
<body style="font-family:system-ui;max-width:720px;margin:2rem auto">
|
||
<p><a href="/mock/github">← all pull requests</a></p>
|
||
<h1 id="pr-title">{pr["title"]}{draft}</h1>
|
||
<p>#{pr["number"]} · <span id="pr-state">{pr["state"]}</span> ·
|
||
<code id="pr-head">{pr["head"]}</code> → <code>{pr["base"]}</code> ·
|
||
by <span id="pr-author">{pr["author"]}</span></p>
|
||
<h3>Description</h3><pre id="pr-body">{pr["body"]}</pre>
|
||
<h3>Files changed ({len(pr["files"])})</h3>
|
||
<ul id="pr-files">{files}</ul>
|
||
</body>"""
|
||
)
|
||
|
||
|
||
# --- fake GitHub REST API (open_pull_request hits this) --------------------
|
||
def _gh_pr_json(pr: dict[str, Any]) -> dict[str, Any]:
|
||
full_name = f"{pr['owner']}/{pr['repo']}"
|
||
repo = {"id": 1, "full_name": full_name, "private": False}
|
||
return {
|
||
"number": pr["number"],
|
||
"html_url": _pr_html_url(pr),
|
||
"state": pr["state"],
|
||
"draft": pr["draft"],
|
||
"merged": pr["merged"],
|
||
"title": pr["title"],
|
||
"body": pr["body"],
|
||
"user": {"login": pr["author"]},
|
||
"head": {"ref": pr["head"], "sha": pr["head_sha"], "repo": repo},
|
||
"base": {"ref": pr["base"], "sha": pr["base_sha"], "repo": repo},
|
||
"additions": pr["additions"],
|
||
"deletions": pr["deletions"],
|
||
"changed_files": len(pr["files"]),
|
||
}
|
||
|
||
|
||
@app.get("/fake-gh/repos/{owner}/{repo}")
|
||
async def gh_get_repo(owner: str, repo: str) -> JSONResponse:
|
||
return JSONResponse({"full_name": f"{owner}/{repo}", "private": False})
|
||
|
||
|
||
@app.get("/fake-gh/repos/{owner}/{repo}/branches/{branch:path}")
|
||
async def gh_get_branch(owner: str, repo: str, branch: str) -> JSONResponse: # noqa: ARG001
|
||
if not fakes.branch_exists(branch):
|
||
return JSONResponse({"message": "Branch not found"}, status_code=404)
|
||
return JSONResponse({"name": branch, "commit": {"sha": "deadbeef"}})
|
||
|
||
|
||
@app.get("/fake-gh/repos/{owner}/{repo}/pulls")
|
||
async def gh_list_pulls(owner: str, repo: str) -> JSONResponse: # noqa: ARG001
|
||
return JSONResponse([])
|
||
|
||
|
||
@app.post("/fake-gh/repos/{owner}/{repo}/pulls")
|
||
async def gh_create_pull(owner: str, repo: str, request: Request) -> JSONResponse:
|
||
body = await request.json()
|
||
pr = fakes.create_pull(
|
||
owner,
|
||
repo,
|
||
head=body.get("head", ""),
|
||
base=body.get("base", "main"),
|
||
title=body.get("title", ""),
|
||
body=body.get("body", ""),
|
||
draft=bool(body.get("draft", True)),
|
||
)
|
||
return JSONResponse(_gh_pr_json(pr), status_code=201)
|
||
|
||
|
||
@app.get("/fake-gh/repos/{owner}/{repo}/pulls/{number}")
|
||
async def gh_get_pull(owner: str, repo: str, number: int) -> JSONResponse: # noqa: ARG001
|
||
pr = fakes.find_pull(number)
|
||
if pr is None:
|
||
return JSONResponse({"message": "Not Found"}, status_code=404)
|
||
return JSONResponse(_gh_pr_json(pr))
|
||
|
||
|
||
@app.post("/fake-gh/repos/{owner}/{repo}/check-runs")
|
||
async def gh_create_check_run(owner: str, repo: str, request: Request) -> JSONResponse:
|
||
payload = await request.json()
|
||
return JSONResponse(fakes.create_check_run(owner, repo, payload), status_code=201)
|
||
|
||
|
||
@app.patch("/fake-gh/repos/{owner}/{repo}/check-runs/{check_run_id}")
|
||
async def gh_update_check_run(
|
||
owner: str,
|
||
repo: str,
|
||
check_run_id: int,
|
||
request: Request, # noqa: ARG001
|
||
) -> JSONResponse:
|
||
payload = await request.json()
|
||
check = fakes.update_check_run(check_run_id, payload)
|
||
if check is None:
|
||
return JSONResponse({"message": "Not Found"}, status_code=404)
|
||
return JSONResponse(check)
|
||
|
||
|
||
# --- fake Slack API (real slack code hits this) ----------------------------
|
||
def _ok(extra: dict[str, Any] | None = None) -> JSONResponse:
|
||
return JSONResponse({"ok": True, **(extra or {})})
|
||
|
||
|
||
@app.post("/fake-slack/chat.postMessage")
|
||
async def slack_post_message(request: Request) -> JSONResponse:
|
||
body = await request.json()
|
||
ts = fakes.add_slack_message(
|
||
body.get("channel", ""),
|
||
body.get("thread_ts", ""),
|
||
user=BOT_USER_ID,
|
||
text=body.get("text", ""),
|
||
blocks=body.get("blocks"),
|
||
is_bot=True,
|
||
)
|
||
return _ok({"ts": ts, "message": {"ts": ts}})
|
||
|
||
|
||
@app.post("/fake-slack/chat.postEphemeral")
|
||
async def slack_post_ephemeral(request: Request) -> JSONResponse:
|
||
await request.body()
|
||
return _ok({"message_ts": fakes.next_slack_ts()})
|
||
|
||
|
||
@app.post("/fake-slack/assistant.threads.setStatus")
|
||
async def slack_set_status(request: Request) -> JSONResponse:
|
||
await request.body()
|
||
return _ok()
|
||
|
||
|
||
@app.post("/fake-slack/reactions.add")
|
||
async def slack_reactions_add(request: Request) -> JSONResponse:
|
||
await request.body()
|
||
return _ok()
|
||
|
||
|
||
@app.get("/fake-slack/users.info")
|
||
async def slack_users_info(user: str = "") -> JSONResponse:
|
||
info = _SLACK_USERS.get(
|
||
user, {"name": "devuser", "real_name": "Dev User", "email": "dev@example.com"}
|
||
)
|
||
return _ok(
|
||
{
|
||
"user": {
|
||
"id": user,
|
||
"name": info["name"],
|
||
"real_name": info["real_name"],
|
||
"profile": {
|
||
"email": info["email"],
|
||
"display_name": info["real_name"],
|
||
"real_name": info["real_name"],
|
||
},
|
||
}
|
||
}
|
||
)
|
||
|
||
|
||
@app.get("/fake-slack/conversations.info")
|
||
async def slack_conversations_info(channel: str = "") -> JSONResponse:
|
||
return _ok(
|
||
{
|
||
"channel": {
|
||
"id": channel,
|
||
"name": "demo",
|
||
"name_normalized": "demo",
|
||
"topic": {"value": "Demo channel topic"},
|
||
"purpose": {"value": "Demo channel purpose"},
|
||
}
|
||
}
|
||
)
|
||
|
||
|
||
@app.get("/fake-slack/conversations.replies")
|
||
async def slack_conversations_replies(channel: str = "", ts: str = "") -> JSONResponse:
|
||
msgs = fakes.slack_thread(channel, ts)
|
||
return _ok(
|
||
{
|
||
"messages": [
|
||
{
|
||
"type": "message",
|
||
"user": m["user"],
|
||
"text": m["text"],
|
||
"ts": m["ts"],
|
||
"thread_ts": m["thread_ts"],
|
||
}
|
||
for m in msgs
|
||
]
|
||
}
|
||
)
|
||
|
||
|
||
@app.get("/fake-slack/conversations.history")
|
||
async def slack_conversations_history(channel: str = "") -> JSONResponse: # noqa: ARG001
|
||
return _ok({"messages": []})
|
||
|
||
|
||
@app.get("/fake-slack/chat.getPermalink")
|
||
async def slack_get_permalink(channel: str = "", message_ts: str = "") -> JSONResponse: # noqa: ARG001
|
||
return _ok({"permalink": f"{BASE_URL}/mock/slack"})
|
||
|
||
|
||
# Quietly reference imports used only for env side effects.
|
||
_ = (e2e_env, HUMAN_USER)
|