wehub-resource-sync
603 lines
21 KiB
Python
603 lines
21 KiB
Python
"""Coverage for `GitHubProxyClient` + `ProxyGitTransport` against an
|
|
ASGI-wrapped proxy app and a hand-rolled `httpx.MockTransport`."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import json
|
|
import os
|
|
import subprocess
|
|
from collections.abc import Callable
|
|
from pathlib import Path
|
|
|
|
import httpx
|
|
import pytest
|
|
from pydantic import SecretStr
|
|
|
|
from robomp.config import Settings
|
|
from robomp.git_ops import HeadDriftError
|
|
from robomp.github_client import (
|
|
CommentInfo,
|
|
GitHubClient,
|
|
GitHubError,
|
|
IssueInfo,
|
|
IssueSummary,
|
|
PullRequestFileInfo,
|
|
PullRequestInfo,
|
|
PullRequestReviewInfo,
|
|
ReactionInfo,
|
|
RepoInfo,
|
|
ReviewCommentInfo,
|
|
)
|
|
from robomp.proxy.server import create_proxy_app
|
|
from robomp.proxy_client import GitHubProxyClient, ProxyGitTransport
|
|
from robomp.proxy_hmac import HEADER_SIGNATURE, HEADER_TIMESTAMP, verify
|
|
from robomp.sandbox import workspace_key
|
|
|
|
_HMAC = "test-hmac-key-aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
|
|
_HMAC_BYTES = _HMAC.encode("utf-8")
|
|
_TOKEN = "ghp_test_token_value"
|
|
|
|
|
|
# ---------- shared helpers ----------
|
|
|
|
|
|
def _build_settings(tmp_path: Path) -> Settings:
|
|
cfg = Settings.model_construct(
|
|
github_token=SecretStr(_TOKEN),
|
|
github_webhook_secret=SecretStr("webhook-secret"),
|
|
bot_login="robomp-bot",
|
|
git_author_email="robomp-bot@example.invalid",
|
|
repo_allowlist_raw="octo/widget",
|
|
gh_proxy_url=None,
|
|
gh_proxy_hmac_key=SecretStr(_HMAC),
|
|
gh_proxy_bind_host="0.0.0.0",
|
|
gh_proxy_bind_port=8081,
|
|
workspace_root=tmp_path / "workspaces",
|
|
sqlite_path=tmp_path / "robomp.sqlite",
|
|
log_dir=tmp_path / "logs",
|
|
)
|
|
cfg.ensure_paths()
|
|
return cfg
|
|
|
|
|
|
@pytest.fixture
|
|
def proxy_settings(tmp_path: Path) -> Settings:
|
|
return _build_settings(tmp_path)
|
|
|
|
|
|
def _git(args: list[str], cwd: Path) -> subprocess.CompletedProcess[str]:
|
|
return subprocess.run(
|
|
["git", *args],
|
|
cwd=str(cwd),
|
|
check=True,
|
|
capture_output=True,
|
|
text=True,
|
|
env=os.environ
|
|
| {
|
|
"GIT_AUTHOR_NAME": "t",
|
|
"GIT_AUTHOR_EMAIL": "t@t",
|
|
"GIT_COMMITTER_NAME": "t",
|
|
"GIT_COMMITTER_EMAIL": "t@t",
|
|
},
|
|
)
|
|
|
|
|
|
@pytest.fixture
|
|
def upstream_repo(tmp_path: Path) -> Path:
|
|
repo = tmp_path / "upstream.git"
|
|
repo.mkdir()
|
|
_git(["init", "--initial-branch=main", "--bare", str(repo)], tmp_path)
|
|
seed = tmp_path / "seed"
|
|
seed.mkdir()
|
|
_git(["init", "--initial-branch=main", str(seed)], tmp_path)
|
|
(seed / "README.md").write_text("hello\n", encoding="utf-8")
|
|
_git(["-C", str(seed), "add", "."], tmp_path)
|
|
_git(["-C", str(seed), "commit", "-m", "init"], tmp_path)
|
|
_git(["-C", str(seed), "remote", "add", "origin", str(repo)], tmp_path)
|
|
_git(["-C", str(seed), "push", "origin", "main"], tmp_path)
|
|
return repo
|
|
|
|
|
|
def _stage_workspace(cfg: Settings, upstream: Path, repo: str, number: int, branch: str) -> tuple[Path, str]:
|
|
ws_dir = Path(cfg.workspace_root) / workspace_key(repo, number)
|
|
ws_dir.mkdir(parents=True, exist_ok=True)
|
|
repo_dir = ws_dir / "repo"
|
|
_git(["clone", str(upstream), str(repo_dir)], ws_dir)
|
|
_git(["-C", str(repo_dir), "config", "user.email", "t@t"], ws_dir)
|
|
_git(["-C", str(repo_dir), "config", "user.name", "t"], ws_dir)
|
|
_git(["-C", str(repo_dir), "checkout", "-b", branch], ws_dir)
|
|
(repo_dir / "x.txt").write_text("x", encoding="utf-8")
|
|
_git(["-C", str(repo_dir), "add", "."], ws_dir)
|
|
_git(["-C", str(repo_dir), "commit", "-m", "x"], ws_dir)
|
|
proc = _git(["-C", str(repo_dir), "rev-parse", "HEAD"], ws_dir)
|
|
return repo_dir, proc.stdout.strip()
|
|
|
|
|
|
def _bare_has_branch(bare: Path, branch: str) -> bool:
|
|
proc = subprocess.run(
|
|
["git", "-C", str(bare), "branch", "--list", branch],
|
|
capture_output=True,
|
|
text=True,
|
|
check=False,
|
|
)
|
|
return bool(proc.stdout.strip())
|
|
|
|
|
|
def _attach_gh(app, handler: Callable[[httpx.Request], httpx.Response]) -> None:
|
|
app.state.github = GitHubClient(_TOKEN, transport=httpx.MockTransport(handler))
|
|
|
|
|
|
# Sync httpx.Client cannot accept httpx.ASGITransport (which is async-only).
|
|
# Bridge by running the async transport inside a one-shot event loop per call.
|
|
class _SyncASGIBridge(httpx.BaseTransport):
|
|
def __init__(self, app) -> None:
|
|
self._async = httpx.ASGITransport(app=app)
|
|
|
|
def handle_request(self, request: httpx.Request) -> httpx.Response: # type: ignore[override]
|
|
async def _drain() -> tuple[int, httpx.Headers, bytes]:
|
|
async_resp = await self._async.handle_async_request(request)
|
|
body = await async_resp.aread()
|
|
await async_resp.aclose()
|
|
return async_resp.status_code, async_resp.headers, body
|
|
|
|
status, headers, body = asyncio.run(_drain())
|
|
# Wrap the bytes in a fresh sync Response so httpx.Client's
|
|
# `isinstance(response.stream, SyncByteStream)` assertion holds.
|
|
return httpx.Response(
|
|
status_code=status,
|
|
headers=headers,
|
|
content=body,
|
|
request=request,
|
|
)
|
|
|
|
|
|
# ============================================================================
|
|
# 1. HMAC headers + signature verify
|
|
# ============================================================================
|
|
|
|
|
|
async def test_signed_headers_present_and_verify() -> None:
|
|
captured: list[httpx.Request] = []
|
|
|
|
def handler(request: httpx.Request) -> httpx.Response:
|
|
captured.append(request)
|
|
# Echo a minimal valid payload for whichever endpoint was hit.
|
|
return httpx.Response(
|
|
200,
|
|
json={
|
|
"full_name": "octo/widget",
|
|
"default_branch": "main",
|
|
"clone_url": "https://example/octo/widget.git",
|
|
"private": False,
|
|
},
|
|
)
|
|
|
|
client = GitHubProxyClient(
|
|
base_url="http://proxy.test",
|
|
hmac_key=_HMAC,
|
|
transport=httpx.MockTransport(handler),
|
|
)
|
|
info = await client.get_repo("octo/widget")
|
|
assert isinstance(info, RepoInfo)
|
|
assert len(captured) == 1
|
|
req = captured[0]
|
|
ts = req.headers.get(HEADER_TIMESTAMP)
|
|
sig = req.headers.get(HEADER_SIGNATURE)
|
|
assert ts is not None and sig is not None
|
|
raw_query = req.url.query.decode("ascii")
|
|
target = f"{req.url.path}?{raw_query}" if raw_query else req.url.path
|
|
result = verify(
|
|
method=req.method,
|
|
path=target,
|
|
body=req.content or b"",
|
|
timestamp=ts,
|
|
signature=sig,
|
|
key=_HMAC_BYTES,
|
|
)
|
|
assert result.ok, result.reason
|
|
|
|
|
|
# ============================================================================
|
|
# 2. Round-trip via ASGI against a real proxy app
|
|
# ============================================================================
|
|
|
|
|
|
@pytest.fixture
|
|
def round_trip_app(proxy_settings: Settings):
|
|
"""A proxy app whose GitHub-side `app.state.github` answers every GH
|
|
endpoint the GitHubProxyClient exercises in the round-trip test."""
|
|
app = create_proxy_app(proxy_settings)
|
|
app.state.settings = proxy_settings
|
|
|
|
def gh(req: httpx.Request) -> httpx.Response:
|
|
path = req.url.path
|
|
if path == "/repos/octo/widget":
|
|
return httpx.Response(
|
|
200,
|
|
json={
|
|
"full_name": "octo/widget",
|
|
"default_branch": "main",
|
|
"clone_url": "https://example/octo/widget.git",
|
|
"private": False,
|
|
},
|
|
)
|
|
if path == "/repos/octo/widget/issues/1" and req.method == "GET":
|
|
return httpx.Response(
|
|
200,
|
|
json={
|
|
"number": 1,
|
|
"title": "T",
|
|
"body": "B",
|
|
"state": "open",
|
|
"user": {"login": "alice"},
|
|
"labels": [{"name": "bug"}],
|
|
},
|
|
)
|
|
if path == "/repos/octo/widget/issues" and req.method == "GET":
|
|
return httpx.Response(
|
|
200,
|
|
json=[
|
|
{
|
|
"number": 1,
|
|
"title": "first",
|
|
"state": "open",
|
|
"user": {"login": "alice"},
|
|
"labels": [],
|
|
"comments": 0,
|
|
"updated_at": "2026-01-01T00:00:00Z",
|
|
"created_at": "2026-01-01T00:00:00Z",
|
|
"html_url": "https://example/1",
|
|
}
|
|
],
|
|
)
|
|
if path == "/repos/octo/widget/issues/1/comments" and req.method == "GET":
|
|
return httpx.Response(
|
|
200,
|
|
json=[
|
|
{"id": 7, "user": {"login": "u"}, "body": "hi", "created_at": "2026-01-01T00:00:00Z"},
|
|
],
|
|
)
|
|
if path == "/repos/octo/widget/issues/1/comments" and req.method == "POST":
|
|
return httpx.Response(
|
|
201,
|
|
json={"id": 11, "user": {"login": "bot"}, "body": "posted", "created_at": "2026-01-01T00:00:00Z"},
|
|
)
|
|
if path == "/repos/octo/widget/pulls/2/comments":
|
|
return httpx.Response(
|
|
200,
|
|
json=[
|
|
{
|
|
"id": 9,
|
|
"user": {"login": "rev"},
|
|
"body": "nit",
|
|
"path": "a.py",
|
|
"line": 5,
|
|
"created_at": "2026-01-01T00:00:00Z",
|
|
}
|
|
],
|
|
)
|
|
if path == "/repos/octo/widget/pulls/2/reviews" and req.method == "GET":
|
|
return httpx.Response(
|
|
200,
|
|
json=[
|
|
{
|
|
"id": 12,
|
|
"user": {"login": "rev"},
|
|
"body": "approved",
|
|
"state": "APPROVED",
|
|
"submitted_at": "2026-01-01T00:00:00Z",
|
|
}
|
|
],
|
|
)
|
|
if path == "/repos/octo/widget/pulls/2/files":
|
|
return httpx.Response(
|
|
200,
|
|
json=[{"filename": "src/app.py", "status": "modified", "additions": 2, "deletions": 1}],
|
|
)
|
|
if path == "/repos/octo/widget/pulls/2/reviews" and req.method == "POST":
|
|
body = json.loads(req.content)
|
|
assert body["event"] == "COMMENT"
|
|
assert body["comments"] == [{"path": "src/app.py", "line": 12, "side": "RIGHT", "body": "finding"}]
|
|
return httpx.Response(
|
|
200,
|
|
json={
|
|
"id": 55,
|
|
"user": {"login": "robomp-bot"},
|
|
"body": body["body"],
|
|
"state": "COMMENTED",
|
|
"submitted_at": "2026-01-01T00:00:00Z",
|
|
},
|
|
)
|
|
if path == "/user":
|
|
return httpx.Response(200, json={"login": "robomp-bot"})
|
|
if path == "/repos/octo/widget/pulls/4" and req.method == "GET":
|
|
return httpx.Response(
|
|
200,
|
|
json={
|
|
"number": 4,
|
|
"html_url": "https://example/4",
|
|
"head": {"ref": "feat", "repo": {"full_name": "octo/widget"}},
|
|
"base": {"ref": "main"},
|
|
"state": "open",
|
|
"user": {"login": "robomp-bot"},
|
|
},
|
|
)
|
|
if path == "/repos/octo/widget/pulls" and req.method == "POST":
|
|
return httpx.Response(
|
|
201,
|
|
json={
|
|
"number": 4,
|
|
"html_url": "https://example/4",
|
|
"head": {"ref": "feat"},
|
|
"base": {"ref": "main"},
|
|
"state": "open",
|
|
},
|
|
)
|
|
if path == "/repos/octo/widget/pulls/4/requested_reviewers":
|
|
return httpx.Response(201, json={})
|
|
if path == "/repos/octo/widget/issues/1/labels":
|
|
return httpx.Response(200, json=[{"name": "triage"}])
|
|
if path == "/repos/octo/widget/issues/1/labels/needs-info" and req.method == "DELETE":
|
|
return httpx.Response(200, json={})
|
|
if path == "/repos/octo/widget/issues/1/assignees":
|
|
return httpx.Response(201, json={})
|
|
return httpx.Response(404, json={"message": f"unrouted {req.method} {path}"})
|
|
|
|
_attach_gh(app, gh)
|
|
return app
|
|
|
|
|
|
async def test_round_trip_all_endpoints(round_trip_app) -> None:
|
|
client = GitHubProxyClient(
|
|
base_url="http://proxy.test",
|
|
hmac_key=_HMAC,
|
|
transport=httpx.ASGITransport(app=round_trip_app),
|
|
)
|
|
repo = await client.get_repo("octo/widget")
|
|
assert isinstance(repo, RepoInfo)
|
|
assert repo.full_name == "octo/widget"
|
|
|
|
issue = await client.get_issue("octo/widget", 1)
|
|
assert isinstance(issue, IssueInfo)
|
|
assert issue.labels == ("bug",)
|
|
|
|
issues = await client.list_issues("octo/widget")
|
|
assert len(issues) == 1 and isinstance(issues[0], IssueSummary)
|
|
|
|
comments = await client.list_comments("octo/widget", 1)
|
|
assert len(comments) == 1 and isinstance(comments[0], CommentInfo)
|
|
|
|
rcs = await client.list_review_comments("octo/widget", 2)
|
|
assert len(rcs) == 1 and isinstance(rcs[0], ReviewCommentInfo)
|
|
assert rcs[0].line == 5
|
|
|
|
prs = await client.list_pr_reviews("octo/widget", 2)
|
|
assert len(prs) == 1 and isinstance(prs[0], PullRequestReviewInfo)
|
|
|
|
files = await client.list_pr_files("octo/widget", 2)
|
|
assert len(files) == 1 and isinstance(files[0], PullRequestFileInfo)
|
|
assert files[0].path == "src/app.py"
|
|
|
|
submitted = await client.submit_pr_review(
|
|
repo="octo/widget",
|
|
pr_number=2,
|
|
body="summary",
|
|
event="COMMENT",
|
|
comments=[{"path": "src/app.py", "line": 12, "side": "RIGHT", "body": "finding"}],
|
|
)
|
|
assert submitted.id == 55
|
|
|
|
assert await client.get_authenticated_login() == "robomp-bot"
|
|
|
|
existing_pr = await client.get_pull_request("octo/widget", 4)
|
|
assert isinstance(existing_pr, PullRequestInfo)
|
|
assert existing_pr.head_ref == "feat"
|
|
assert existing_pr.author == "robomp-bot"
|
|
|
|
posted = await client.post_comment("octo/widget", 1, "hi")
|
|
assert isinstance(posted, CommentInfo)
|
|
assert posted.id == 11
|
|
|
|
pr = await client.open_pull_request(repo="octo/widget", head="feat", base="main", title="t", body="b")
|
|
assert isinstance(pr, PullRequestInfo)
|
|
assert pr.number == 4
|
|
|
|
# request_reviewers returns None on success.
|
|
assert await client.request_reviewers(repo="octo/widget", pr_number=4, reviewers=["alice"]) is None
|
|
|
|
labels = await client.add_issue_labels("octo/widget", 1, ["triage"])
|
|
assert await client.remove_issue_label("octo/widget", 1, "needs-info") is None
|
|
assert labels == ("triage",)
|
|
|
|
assert await client.add_assignees("octo/widget", 1, ["alice"]) is None
|
|
|
|
|
|
async def test_list_comment_reactions_round_trip(proxy_settings: Settings) -> None:
|
|
app = create_proxy_app(proxy_settings)
|
|
app.state.settings = proxy_settings
|
|
|
|
def gh(req: httpx.Request) -> httpx.Response:
|
|
if req.url.path == "/repos/octo/widget/issues/comments/999/reactions":
|
|
assert req.url.params.get("content") == "-1"
|
|
return httpx.Response(
|
|
200,
|
|
json=[
|
|
{"content": "-1", "user": {"login": "alice", "type": "User"}},
|
|
],
|
|
)
|
|
return httpx.Response(404, json={"message": "unrouted"})
|
|
|
|
_attach_gh(app, gh)
|
|
client = GitHubProxyClient(
|
|
base_url="http://proxy.test",
|
|
hmac_key=_HMAC,
|
|
transport=httpx.ASGITransport(app=app),
|
|
)
|
|
reactions = await client.list_comment_reactions("octo/widget", 999)
|
|
assert reactions == (ReactionInfo(content="-1", user_login="alice", user_type="User"),)
|
|
|
|
|
|
async def test_close_issue_round_trip(proxy_settings: Settings) -> None:
|
|
captured: dict[str, object] = {}
|
|
app = create_proxy_app(proxy_settings)
|
|
app.state.settings = proxy_settings
|
|
|
|
def gh(req: httpx.Request) -> httpx.Response:
|
|
if req.url.path == "/repos/octo/widget/issues/7" and req.method == "PATCH":
|
|
captured["body"] = json.loads(req.content)
|
|
return httpx.Response(200, json={})
|
|
return httpx.Response(404, json={"message": "unrouted"})
|
|
|
|
_attach_gh(app, gh)
|
|
client = GitHubProxyClient(
|
|
base_url="http://proxy.test",
|
|
hmac_key=_HMAC,
|
|
transport=httpx.ASGITransport(app=app),
|
|
)
|
|
assert await client.close_issue("octo/widget", 7) is None
|
|
assert captured["body"] == {"state": "closed", "state_reason": "completed"}
|
|
|
|
|
|
# ============================================================================
|
|
# 3. Error decode
|
|
# ============================================================================
|
|
|
|
|
|
async def test_error_decode_github_422() -> None:
|
|
def handler(_: httpx.Request) -> httpx.Response:
|
|
return httpx.Response(
|
|
422,
|
|
json={"error": {"kind": "github", "status": 422, "message": "x"}},
|
|
)
|
|
|
|
client = GitHubProxyClient(
|
|
base_url="http://proxy.test",
|
|
hmac_key=_HMAC,
|
|
transport=httpx.MockTransport(handler),
|
|
)
|
|
with pytest.raises(GitHubError) as exc:
|
|
await client.post_comment("octo/widget", 1, "hi")
|
|
assert exc.value.status == 422
|
|
assert exc.value.message == "x"
|
|
|
|
|
|
# ============================================================================
|
|
# 4 + 5. ProxyGitTransport push (happy + HEAD drift)
|
|
# ============================================================================
|
|
|
|
|
|
def test_proxy_git_transport_push_happy(proxy_settings: Settings, upstream_repo: Path) -> None:
|
|
branch = "farm/abc/feat"
|
|
_, head = _stage_workspace(proxy_settings, upstream_repo, "octo/widget", 1, branch)
|
|
app = create_proxy_app(proxy_settings)
|
|
app.state.settings = proxy_settings
|
|
_attach_gh(app, lambda _: httpx.Response(500, json={"message": "should not be hit"}))
|
|
|
|
transport = ProxyGitTransport(
|
|
base_url="http://proxy.test",
|
|
hmac_key=_HMAC,
|
|
transport=_SyncASGIBridge(app),
|
|
)
|
|
result = transport.push_branch(
|
|
repo="octo/widget",
|
|
workspace_key=workspace_key("octo/widget", 1),
|
|
repo_dir=Path(proxy_settings.workspace_root) / workspace_key("octo/widget", 1) / "repo",
|
|
branch=branch,
|
|
expected_head=head,
|
|
)
|
|
assert result.head == head
|
|
assert result.branch == branch
|
|
assert _bare_has_branch(upstream_repo, branch)
|
|
|
|
|
|
def test_proxy_git_transport_push_head_drift(proxy_settings: Settings, upstream_repo: Path) -> None:
|
|
branch = "farm/abc/drift"
|
|
_, _ = _stage_workspace(proxy_settings, upstream_repo, "octo/widget", 1, branch)
|
|
app = create_proxy_app(proxy_settings)
|
|
app.state.settings = proxy_settings
|
|
_attach_gh(app, lambda _: httpx.Response(500, json={"message": "should not be hit"}))
|
|
|
|
transport = ProxyGitTransport(
|
|
base_url="http://proxy.test",
|
|
hmac_key=_HMAC,
|
|
transport=_SyncASGIBridge(app),
|
|
)
|
|
with pytest.raises(HeadDriftError):
|
|
transport.push_branch(
|
|
repo="octo/widget",
|
|
workspace_key=workspace_key("octo/widget", 1),
|
|
repo_dir=Path(proxy_settings.workspace_root) / workspace_key("octo/widget", 1) / "repo",
|
|
branch=branch,
|
|
expected_head="0" * 40,
|
|
)
|
|
assert not _bare_has_branch(upstream_repo, branch)
|
|
|
|
|
|
def test_proxy_git_transport_push_slot_uid_body() -> None:
|
|
captured: list[dict[str, object]] = []
|
|
|
|
def handler(request: httpx.Request) -> httpx.Response:
|
|
captured.append(json.loads(request.content))
|
|
return httpx.Response(200, json={"head": "abc123", "branch": "farm/abc/feat"})
|
|
|
|
transport = ProxyGitTransport(
|
|
base_url="http://proxy.test",
|
|
hmac_key=_HMAC,
|
|
transport=httpx.MockTransport(handler),
|
|
)
|
|
transport.push_branch(
|
|
repo="octo/widget",
|
|
workspace_key="octo__widget__1",
|
|
repo_dir=Path("/unused"),
|
|
branch="farm/abc/feat",
|
|
expected_head="abc123",
|
|
slot_uid=2001,
|
|
)
|
|
transport.push_branch(
|
|
repo="octo/widget",
|
|
workspace_key="octo__widget__1",
|
|
repo_dir=Path("/unused"),
|
|
branch="farm/abc/feat",
|
|
expected_head="abc123",
|
|
)
|
|
|
|
assert captured[0]["slot_uid"] == 2001
|
|
assert "slot_uid" not in captured[1]
|
|
|
|
|
|
# Sanity: signed POST headers from ProxyGitTransport._post verify cleanly.
|
|
def test_proxy_git_transport_post_headers_verify() -> None:
|
|
captured: list[httpx.Request] = []
|
|
|
|
def handler(request: httpx.Request) -> httpx.Response:
|
|
captured.append(request)
|
|
return httpx.Response(200, json={"pool_dir": "/tmp/x"})
|
|
|
|
transport = ProxyGitTransport(
|
|
base_url="http://proxy.test",
|
|
hmac_key=_HMAC,
|
|
transport=httpx.MockTransport(handler),
|
|
)
|
|
transport.clone_pool(
|
|
repo="octo/widget",
|
|
clone_url="https://example/widget.git",
|
|
default_branch="main",
|
|
target=Path("/tmp/unused"),
|
|
)
|
|
assert len(captured) == 1
|
|
req = captured[0]
|
|
ts = req.headers.get(HEADER_TIMESTAMP)
|
|
sig = req.headers.get(HEADER_SIGNATURE)
|
|
assert ts and sig
|
|
result = verify(
|
|
method="POST",
|
|
path="/gh/v1/git/clone",
|
|
body=req.content or b"",
|
|
timestamp=ts,
|
|
signature=sig,
|
|
key=_HMAC_BYTES,
|
|
)
|
|
assert result.ok, result.reason
|
|
assert json.loads(req.content)["repo"] == "octo/widget"
|