host
8fc4e27241
接上轮仓库最大权限：用户明确要求无论什么身份，取该身份 授权范围内的所有权限。manifest 现按四类全拉满（74 项权限、 39 个事件）：仓库补 codespaces_lifecycle_admin、repository_ creation、metadata 只读；组织 members 升 write 并补 organization_administration 等 15 项；企业补 8 项；账号补 blocking、email_addresses、gists、密钥类等 22 项。 身份边界写进注释：GitHub 只授予安装身份能给的那部分，个人 装不上组织/企业、企业装不上仓库/组织；换 token 不传子集， token 本就是安装全量；企业安装另需 App 公开/内部 （GITHUB_APP_PUBLIC=1）。诚实边界：企业类键名未用真实 GitHub 建 App 实测，新建时表单会校验，某键被拒删键再建， 线上旧 App 不受影响。
536 lines
22 KiB
Python
536 lines
22 KiB
Python
"""GitHub App 一键安装单测：manifest、setup state、换 code、
|
||
配置、安装记录、内容去重、双 secret 验签、HTTP 端点。
|
||
|
||
不联网：换 code 挂 server.github_app_exchange_client 的 fake；
|
||
测试里出现的 secret、pem、key 一律是 fake 占位，绝无真实
|
||
secret、真实 GitHub 账号。
|
||
"""
|
||
|
||
import hashlib
|
||
import hmac
|
||
import http.client
|
||
import json
|
||
import os
|
||
import sqlite3
|
||
import sys
|
||
import tempfile
|
||
import threading
|
||
import unittest
|
||
from pathlib import Path
|
||
|
||
try:
|
||
from cloud.auth import AuthStore
|
||
from cloud.db import connect as _db_connect
|
||
from cloud.github_app import (
|
||
AppError,
|
||
SetupStateStore,
|
||
build_app_manifest,
|
||
claim_fix_review,
|
||
exchange_manifest_code,
|
||
get_app_config,
|
||
list_installations,
|
||
parse_installation_event,
|
||
public_app_status,
|
||
record_installation_repos,
|
||
remove_installation,
|
||
remove_installation_repos,
|
||
save_app_config,
|
||
set_fix_review_task_id,
|
||
)
|
||
from cloud.server import create_server
|
||
from cloud.webhooks import verify_github_signature_any
|
||
except ImportError:
|
||
sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
|
||
from cloud.auth import AuthStore
|
||
from cloud.db import connect as _db_connect
|
||
from cloud.github_app import (
|
||
AppError,
|
||
SetupStateStore,
|
||
build_app_manifest,
|
||
claim_fix_review,
|
||
exchange_manifest_code,
|
||
get_app_config,
|
||
list_installations,
|
||
parse_installation_event,
|
||
public_app_status,
|
||
record_installation_repos,
|
||
remove_installation,
|
||
remove_installation_repos,
|
||
save_app_config,
|
||
set_fix_review_task_id,
|
||
)
|
||
from cloud.server import create_server
|
||
from cloud.webhooks import verify_github_signature_any
|
||
|
||
BASE = "https://example.invalid"
|
||
|
||
|
||
def _fake_exchange(result=None):
|
||
def fake(url, headers):
|
||
assert url.startswith(
|
||
"https://api.github.com/app-manifests/"), url
|
||
return dict(result or {
|
||
"id": 424242, "slug": "fake-codex-app",
|
||
"client_id": "Iv1.fake", "webhook_secret": "fake-app-secret",
|
||
"pem": "-----BEGIN RSA PRIVATE KEY-----\nfake-pem\n"
|
||
"-----END RSA PRIVATE KEY-----\n",
|
||
"html_url": "https://github.com/apps/fake-codex-app"})
|
||
return fake
|
||
|
||
|
||
class ManifestTestCase(unittest.TestCase):
|
||
def test_manifest_urls_and_permissions(self):
|
||
m = build_app_manifest(BASE)
|
||
self.assertEqual(m["hook_attributes"]["url"],
|
||
BASE + "/webhooks/github")
|
||
self.assertEqual(m["redirect_url"], BASE + "/github-app/callback")
|
||
self.assertEqual(m["default_permissions"]["contents"], "write")
|
||
self.assertEqual(m["default_permissions"]["pull_requests"], "write")
|
||
self.assertEqual(m["default_permissions"]["issues"], "write")
|
||
self.assertEqual(m["default_permissions"]["workflows"], "write")
|
||
self.assertEqual(m["default_permissions"]["administration"], "write")
|
||
self.assertEqual(m["default_permissions"]["actions"], "write")
|
||
self.assertEqual(m["default_permissions"]["checks"], "write")
|
||
# 全量：组织 / 企业 / 账号三类也必须在 manifest 里，不只仓库。
|
||
self.assertEqual(m["default_permissions"]["members"], "write")
|
||
self.assertEqual(
|
||
m["default_permissions"]["organization_administration"],
|
||
"write")
|
||
self.assertEqual(
|
||
m["default_permissions"]["enterprise_organization_installations"],
|
||
"write")
|
||
self.assertEqual(m["default_permissions"]["email_addresses"], "write")
|
||
self.assertEqual(m["default_permissions"]["gists"], "write")
|
||
self.assertIn("issue_comment", m["default_events"])
|
||
self.assertIn("pull_request", m["default_events"])
|
||
self.assertIn("push", m["default_events"])
|
||
self.assertIn("check_run", m["default_events"])
|
||
self.assertIn("workflow_run", m["default_events"])
|
||
self.assertIn("organization", m["default_events"])
|
||
self.assertIn("member", m["default_events"])
|
||
self.assertFalse(m["public"])
|
||
|
||
def test_manifest_requires_base(self):
|
||
with self.assertRaises(ValueError):
|
||
build_app_manifest("")
|
||
|
||
|
||
class SetupStateTestCase(unittest.TestCase):
|
||
def setUp(self):
|
||
self.tmp = tempfile.TemporaryDirectory()
|
||
self.db_path = os.path.join(self.tmp.name, "control.db")
|
||
|
||
def tearDown(self):
|
||
self.tmp.cleanup()
|
||
|
||
def test_consume_once_and_expired(self):
|
||
store = SetupStateStore(self.db_path)
|
||
state = store.create_state()
|
||
self.assertTrue(store.consume_state(state))
|
||
self.assertFalse(store.consume_state(state))
|
||
store.conn.close()
|
||
|
||
def test_expired_state_rejected(self):
|
||
store = SetupStateStore(self.db_path)
|
||
state = store.create_state()
|
||
# 注入“1 小时之后”的 now：state 应过期且被删。
|
||
self.assertFalse(store.consume_state(
|
||
state, now=9999999999.0))
|
||
self.assertFalse(store.consume_state(state))
|
||
store.conn.close()
|
||
|
||
|
||
class ExchangeTestCase(unittest.TestCase):
|
||
def test_exchange_ok(self):
|
||
config = exchange_manifest_code(
|
||
"fake-code", http_client=_fake_exchange())
|
||
self.assertEqual(config["app_id"], 424242)
|
||
self.assertEqual(config["slug"], "fake-codex-app")
|
||
self.assertEqual(config["webhook_secret"], "fake-app-secret")
|
||
|
||
def test_exchange_missing_fields(self):
|
||
with self.assertRaises(AppError) as ctx:
|
||
exchange_manifest_code(
|
||
"fake-code",
|
||
http_client=_fake_exchange({"id": 1, "pem": "x"}))
|
||
self.assertEqual(ctx.exception.reason, "config_incomplete")
|
||
|
||
def test_exchange_http_error(self):
|
||
def boom(url, headers):
|
||
raise RuntimeError("boom")
|
||
with self.assertRaises(AppError) as ctx:
|
||
exchange_manifest_code("fake-code", http_client=boom)
|
||
self.assertEqual(ctx.exception.reason, "exchange_failed")
|
||
|
||
|
||
class ConfigAndInstallationTestCase(unittest.TestCase):
|
||
def setUp(self):
|
||
self.tmp = tempfile.TemporaryDirectory()
|
||
self.db_path = os.path.join(self.tmp.name, "control.db")
|
||
self.conn = _db_connect(self.db_path)
|
||
|
||
def tearDown(self):
|
||
self.conn.close()
|
||
self.tmp.cleanup()
|
||
|
||
def test_config_roundtrip_and_public_status(self):
|
||
self.assertIsNone(get_app_config(self.conn))
|
||
save_app_config(self.conn, {
|
||
"app_id": 7, "slug": "fake-app", "client_id": "Iv1.x",
|
||
"webhook_secret": "fake-secret", "pem": "fake-pem",
|
||
"html_url": "https://github.com/apps/fake-app"})
|
||
config = get_app_config(self.conn)
|
||
self.assertEqual(config["slug"], "fake-app")
|
||
status = public_app_status(self.conn)
|
||
self.assertTrue(status["configured"])
|
||
self.assertNotIn("webhook_secret", status)
|
||
self.assertNotIn("pem", status)
|
||
|
||
def test_installations_created_added_removed_deleted(self):
|
||
record_installation_repos(self.conn, 11, "octo",
|
||
["octo/a", "octo/b"])
|
||
record_installation_repos(self.conn, 11, "octo", ["octo/a"])
|
||
rows = list_installations(self.conn)
|
||
self.assertEqual(len(rows), 2) # 重复同步不翻倍
|
||
record_installation_repos(self.conn, 11, "octo", ["octo/c"])
|
||
remove_installation_repos(self.conn, 11, ["octo/b"])
|
||
repos = [r["repo_full_name"] for r in list_installations(self.conn)]
|
||
self.assertEqual(sorted(repos), ["octo/a", "octo/c"])
|
||
removed = remove_installation(self.conn, 11)
|
||
self.assertEqual(removed, 2)
|
||
self.assertEqual(list_installations(self.conn), [])
|
||
|
||
def test_parse_installation_event(self):
|
||
payload = {"action": "created",
|
||
"installation": {"id": 11, "account": {"login": "octo"}},
|
||
"repositories": [{"full_name": "octo/a"}]}
|
||
inst = parse_installation_event(payload)
|
||
self.assertEqual(inst["installation_id"], 11)
|
||
self.assertEqual(inst["repos"], ["octo/a"])
|
||
added = dict(payload, action="added", repositories_added=[
|
||
{"full_name": "octo/b"}])
|
||
self.assertEqual(parse_installation_event(added)["repos"],
|
||
["octo/b"])
|
||
self.assertIsNone(parse_installation_event(
|
||
{"action": "created"}))
|
||
self.assertIsNone(parse_installation_event(
|
||
{"action": "opened", "installation": {"id": 1}}))
|
||
|
||
|
||
class DedupTestCase(unittest.TestCase):
|
||
def setUp(self):
|
||
self.tmp = tempfile.TemporaryDirectory()
|
||
self.db_path = os.path.join(self.tmp.name, "control.db")
|
||
self.conn = _db_connect(self.db_path)
|
||
|
||
def tearDown(self):
|
||
self.conn.close()
|
||
self.tmp.cleanup()
|
||
|
||
def test_content_dedup(self):
|
||
is_new, task_id = claim_fix_review(
|
||
self.conn, "octo/a", 3, 99)
|
||
self.assertTrue(is_new)
|
||
self.assertIsNone(task_id)
|
||
set_fix_review_task_id(self.conn, "octo/a", 3, 99, "task-1")
|
||
is_new, task_id = claim_fix_review(
|
||
self.conn, "octo/a", 3, 99)
|
||
self.assertFalse(is_new)
|
||
self.assertEqual(task_id, "task-1")
|
||
# 另一个评论 id 不受影响。
|
||
self.assertTrue(claim_fix_review(
|
||
self.conn, "octo/a", 3, 100)[0])
|
||
# 无评论 id 时无法去重，直接放行。
|
||
self.assertTrue(claim_fix_review(
|
||
self.conn, "octo/a", 3, None)[0])
|
||
|
||
|
||
class SignatureAnyTestCase(unittest.TestCase):
|
||
def _sig(self, secret, raw):
|
||
return "sha256=" + hmac.new(
|
||
secret.encode(), raw, hashlib.sha256).hexdigest()
|
||
|
||
def test_any_secret_matches(self):
|
||
raw = b'{"zen":"hi"}'
|
||
self.assertTrue(verify_github_signature_any(
|
||
["app-secret", "legacy-secret"], raw,
|
||
self._sig("legacy-secret", raw)))
|
||
self.assertFalse(verify_github_signature_any(
|
||
["app-secret", "legacy-secret"], raw,
|
||
self._sig("wrong-secret", raw)))
|
||
self.assertFalse(verify_github_signature_any([], raw, "sha256=x"))
|
||
|
||
|
||
class DialectTranslationTestCase(unittest.TestCase):
|
||
"""Postgres 适配层翻译：2026-10-26 实弹回归——安装事件在生产
|
||
Postgres 上因 INSERT OR IGNORE 未翻译直接报语法错。
|
||
|
||
直接测 _translate_sql（纯函数）：函数发出的 SQL 在这里翻
|
||
译成合法 Postgres，SQLite 路径不受影响（翻译只走 Pg 游标）。
|
||
"""
|
||
|
||
def test_or_ignore_translates(self):
|
||
from cloud.db import _translate_sql
|
||
sql, kind = _translate_sql(
|
||
"INSERT OR IGNORE INTO github_app_installations("
|
||
"installation_id, account_login, repo_full_name, created_at)"
|
||
" VALUES(?, ?, ?, ?)")
|
||
self.assertEqual(kind, "normal")
|
||
self.assertNotIn("OR IGNORE", sql)
|
||
self.assertIn("ON CONFLICT DO NOTHING", sql)
|
||
self.assertIn("%s", sql)
|
||
self.assertNotIn("?", sql)
|
||
|
||
def test_or_ignore_with_null_value_translates(self):
|
||
from cloud.db import _translate_sql
|
||
sql, _ = _translate_sql(
|
||
"INSERT OR IGNORE INTO github_fix_review_dedup("
|
||
"repo, pr_number, comment_id, task_id, created_at)"
|
||
" VALUES(?, ?, ?, NULL, ?)")
|
||
self.assertNotIn("OR IGNORE", sql)
|
||
self.assertIn("ON CONFLICT DO NOTHING", sql)
|
||
self.assertIn("NULL", sql)
|
||
|
||
def test_or_replace_app_config_updates_on_conflict(self):
|
||
from cloud.db import _translate_sql
|
||
sql, _ = _translate_sql(
|
||
"INSERT OR REPLACE INTO github_app_config(id, app_id, slug,"
|
||
" client_id, webhook_secret, pem, html_url, created_at)"
|
||
" VALUES(1, ?, ?, ?, ?, ?, ?, ?)")
|
||
self.assertIn("ON CONFLICT (id) DO UPDATE SET", sql)
|
||
self.assertIn("slug = EXCLUDED.slug", sql)
|
||
self.assertNotIn("OR REPLACE", sql)
|
||
|
||
|
||
class AppHttpTestCase(unittest.TestCase):
|
||
def setUp(self):
|
||
self.tmp = tempfile.TemporaryDirectory()
|
||
self.db_path = os.path.join(self.tmp.name, "control.db")
|
||
self.auth = AuthStore(self.db_path)
|
||
self.user_a, self.key_a = self.auth.create_user("user-a")
|
||
# P0 安全批后：setup state 只限全局管理员，本用例里
|
||
# user-a 即管理员。
|
||
self._old_admins = os.environ.get("CODEX_GLOBAL_ADMINS")
|
||
os.environ["CODEX_GLOBAL_ADMINS"] = self.user_a
|
||
self.server = create_server(self.db_path, "127.0.0.1", 0)
|
||
self.port = self.server.server_address[1]
|
||
self.thread = threading.Thread(
|
||
target=self.server.serve_forever, daemon=True)
|
||
self.thread.start()
|
||
self.server.github_app_exchange_client = _fake_exchange()
|
||
|
||
def tearDown(self):
|
||
self.server.shutdown()
|
||
self.server.server_close()
|
||
self.thread.join(timeout=5)
|
||
try:
|
||
self.auth.conn.close()
|
||
except Exception:
|
||
pass
|
||
self.tmp.cleanup()
|
||
os.environ.pop("GITHUB_APP_PUBLIC_BASE", None)
|
||
os.environ.pop("GITHUB_WEBHOOK_SECRET", None)
|
||
if self._old_admins is None:
|
||
os.environ.pop("CODEX_GLOBAL_ADMINS", None)
|
||
else:
|
||
os.environ["CODEX_GLOBAL_ADMINS"] = self._old_admins
|
||
|
||
def _req(self, method, path, body=None, key=None, headers=None):
|
||
conn = http.client.HTTPConnection("127.0.0.1", self.port, timeout=10)
|
||
try:
|
||
hdrs = dict(headers or {})
|
||
payload = None
|
||
if body is not None:
|
||
payload = body if isinstance(body, str) else json.dumps(body)
|
||
hdrs.setdefault("Content-Type", "application/json")
|
||
if key is not None:
|
||
hdrs["X-API-Key"] = key
|
||
conn.request(method, path, body=payload, headers=hdrs)
|
||
resp = conn.getresponse()
|
||
raw = resp.read().decode()
|
||
try:
|
||
parsed = json.loads(raw)
|
||
except ValueError:
|
||
parsed = raw
|
||
return resp.status, parsed
|
||
finally:
|
||
conn.close()
|
||
|
||
def _sig(self, secret, raw):
|
||
return "sha256=" + hmac.new(
|
||
secret.encode(), raw, hashlib.sha256).hexdigest()
|
||
|
||
def _webhook(self, event, payload, secret, delivery):
|
||
raw = json.dumps(payload)
|
||
return self._req(
|
||
"POST", "/webhooks/github", body=raw,
|
||
headers={"X-GitHub-Event": event,
|
||
"X-GitHub-Delivery": delivery,
|
||
"X-Hub-Signature-256": self._sig(secret, raw.encode())})
|
||
|
||
def _issue_setup_state(self):
|
||
os.environ["GITHUB_APP_PUBLIC_BASE"] = BASE
|
||
status, body = self._req(
|
||
"POST", "/github-app/setup/state", {}, key=self.key_a)
|
||
self.assertEqual(status, 200, body)
|
||
return body["state"], body["setup_url"]
|
||
|
||
def test_setup_state_requires_auth_and_base(self):
|
||
status, _ = self._req("POST", "/github-app/setup/state", {})
|
||
self.assertEqual(status, 401)
|
||
os.environ.pop("GITHUB_APP_PUBLIC_BASE", None)
|
||
status, body = self._req(
|
||
"POST", "/github-app/setup/state", {}, key=self.key_a)
|
||
self.assertEqual(status, 400, body)
|
||
|
||
def test_setup_page_and_callback_flow(self):
|
||
state, setup_url = self._issue_setup_state()
|
||
self.assertTrue(setup_url.startswith(BASE + "/github-app/setup"))
|
||
# 无 state 不渲染表单。
|
||
status, page = self._req("GET", "/github-app/setup")
|
||
self.assertEqual(status, 400, page)
|
||
# 有 state 渲染表单：manifest 与 GitHub 建 App 地址都在。
|
||
status, page = self._req(
|
||
"GET", "/github-app/setup?state=%s" % state)
|
||
self.assertEqual(status, 200, page)
|
||
self.assertIn("github.com/settings/apps/new", page)
|
||
self.assertIn("codex-app-server", page)
|
||
self.assertIn("/webhooks/github", page)
|
||
# 回调：坏 state 失败；好 state 换 code 存配置。
|
||
status, page = self._req(
|
||
"GET", "/github-app/callback?code=x&state=nope")
|
||
self.assertEqual(status, 400, page)
|
||
status, page = self._req(
|
||
"GET", "/github-app/callback?code=fake-code&state=%s" % state)
|
||
self.assertEqual(status, 200, page)
|
||
self.assertIn("fake-codex-app", page)
|
||
self.assertIn("/installations/new", page)
|
||
self.assertNotIn("fake-app-secret", page)
|
||
self.assertNotIn("fake-pem", page)
|
||
# 同一 state 第二次回调必然失败。
|
||
status, page = self._req(
|
||
"GET", "/github-app/callback?code=fake-code&state=%s" % state)
|
||
self.assertEqual(status, 400, page)
|
||
# 状态接口：已配置、无 secret。
|
||
status, body = self._req(
|
||
"GET", "/github-app/status", key=self.key_a)
|
||
self.assertEqual(status, 200, body)
|
||
self.assertTrue(body["configured"])
|
||
self.assertEqual(body["slug"], "fake-codex-app")
|
||
self.assertNotIn("webhook_secret", body)
|
||
self.assertNotIn("pem", body)
|
||
|
||
def test_webhook_dual_secret_and_installation(self):
|
||
# 先存一个 App 配置（secret 为 fake-app-secret）。
|
||
conn = _db_connect(self.db_path)
|
||
try:
|
||
save_app_config(conn, {
|
||
"app_id": 424242, "slug": "fake-codex-app",
|
||
"client_id": "Iv1.fake", "webhook_secret": "fake-app-secret",
|
||
"pem": "fake-pem",
|
||
"html_url": "https://github.com/apps/fake-codex-app"})
|
||
finally:
|
||
conn.close()
|
||
os.environ["GITHUB_WEBHOOK_SECRET"] = "legacy-secret"
|
||
ping = {"zen": "hi"}
|
||
# App secret 与 legacy secret 都能验过；错 secret 401。
|
||
status, _ = self._webhook(
|
||
"ping", ping, "fake-app-secret", "d-app-1")
|
||
self.assertEqual(status, 200)
|
||
status, _ = self._webhook(
|
||
"ping", ping, "legacy-secret", "d-legacy-1")
|
||
self.assertEqual(status, 200)
|
||
status, _ = self._webhook("ping", ping, "wrong", "d-wrong-1")
|
||
self.assertEqual(status, 401)
|
||
# installation 事件记安装记录。
|
||
inst_payload = {
|
||
"action": "created",
|
||
"installation": {"id": 11, "account": {"login": "octo"}},
|
||
"repositories": [{"full_name": "octo/a"},
|
||
{"full_name": "octo/b"}]}
|
||
status, body = self._webhook(
|
||
"installation", inst_payload, "fake-app-secret", "d-inst-1")
|
||
self.assertEqual(status, 200, body)
|
||
self.assertEqual(body["result"], "installation_recorded")
|
||
status, body = self._req(
|
||
"GET", "/github-app/status", key=self.key_a)
|
||
self.assertEqual(status, 200, body)
|
||
self.assertEqual(len(body["installations"]), 2)
|
||
self.assertEqual(
|
||
body["installations"][0]["repo_full_name"], "octo/a")
|
||
|
||
def test_installation_repositories_sync(self):
|
||
# 安装后增减仓库的同步事件：added 增行。
|
||
conn = _db_connect(self.db_path)
|
||
try:
|
||
save_app_config(conn, {
|
||
"app_id": 424242, "slug": "fake-codex-app",
|
||
"client_id": "Iv1.fake", "webhook_secret": "fake-app-secret",
|
||
"pem": "fake-pem",
|
||
"html_url": "https://github.com/apps/fake-codex-app"})
|
||
finally:
|
||
conn.close()
|
||
payload = {
|
||
"action": "added",
|
||
"installation": {"id": 11, "account": {"login": "octo"}},
|
||
"repositories_added": [{"full_name": "octo/c"}]}
|
||
status, body = self._webhook(
|
||
"installation_repositories", payload, "fake-app-secret",
|
||
"d-inst-repos-1")
|
||
self.assertEqual(status, 200, body)
|
||
self.assertEqual(body["result"], "installation_recorded")
|
||
status, body = self._req(
|
||
"GET", "/github-app/status", key=self.key_a)
|
||
self.assertEqual(status, 200, body)
|
||
self.assertEqual(
|
||
[r["repo_full_name"] for r in body["installations"]],
|
||
["octo/c"])
|
||
|
||
def test_webhook_no_secret_is_503(self):
|
||
os.environ.pop("GITHUB_WEBHOOK_SECRET", None)
|
||
status, _ = self._webhook(
|
||
"ping", {"zen": "hi"}, "anything", "d-none-1")
|
||
self.assertEqual(status, 503)
|
||
|
||
def test_fix_review_content_dedup_over_http(self):
|
||
# 项目名 == 仓库 full_name，/fix-review 才能建任务。
|
||
conn = sqlite3.connect(self.db_path)
|
||
try:
|
||
conn.execute(
|
||
"INSERT INTO projects(id, name, owner_id, created_at,"
|
||
" repo_url, auto_pr) VALUES(?, ?, ?, 0, NULL, 0)",
|
||
("proj-1", "octo/a", self.user_a))
|
||
conn.commit()
|
||
finally:
|
||
conn.close()
|
||
conn = _db_connect(self.db_path)
|
||
try:
|
||
save_app_config(conn, {
|
||
"app_id": 424242, "slug": "fake-codex-app",
|
||
"client_id": "Iv1.fake", "webhook_secret": "fake-app-secret",
|
||
"pem": "fake-pem",
|
||
"html_url": "https://github.com/apps/fake-codex-app"})
|
||
finally:
|
||
conn.close()
|
||
payload = {
|
||
"action": "created",
|
||
"comment": {"id": 777, "body": "please /fix-review this",
|
||
"user": {"login": "octo",
|
||
"author_association": "COLLABORATOR"}},
|
||
"issue": {"number": 5},
|
||
"repository": {"full_name": "octo/a"}}
|
||
# 同一评论送两遍，delivery_id 不同（模拟手动 + App 双通道）。
|
||
status, first = self._webhook(
|
||
"issue_comment", payload, "fake-app-secret", "d-fix-1")
|
||
self.assertEqual(status, 200, first)
|
||
self.assertIsNotNone(first.get("task_id"), first)
|
||
status, second = self._webhook(
|
||
"issue_comment", payload, "fake-app-secret", "d-fix-2")
|
||
self.assertEqual(status, 200, second)
|
||
self.assertTrue(second.get("duplicate"), second)
|
||
self.assertEqual(second.get("task_id"), first.get("task_id"))
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|