Azimkin/multi-bot-tg
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor: remove GitHub webhook support

Browse Source
This commit is contained in:
Azimkin
2026-09-14 13:39:41 +02:00
parent 3f0cce6732
commit 3a8c198caa
7 changed files with 1 additions and 682 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
AGENTS.md
View File

@@ -23,8 +23,6 @@ Copy `config.example.toml` to `config.toml` and fill in credentials, or use envi
Key env vars: `TG_API_ID`, `TG_API_HASH`, `SESSION_DIR`, `ACCOUNTS_JSON` (JSON list) or `ACCOUNT_SESSIONS` (comma-separated session names), `BOT_CONFIG` (path to config file), and `BOT_DOTENV_PATH` (path to `.env`).
The optional GitHub webhook uses `GITHUB_WEBHOOK_ENABLED`, `GITHUB_WEBHOOK_HOST`, `GITHUB_WEBHOOK_PORT`, and `GITHUB_WEBHOOK_SECRET`.
The `.txt` transcription command uses `[elevenlabs]` settings or `ELEVENLABS_API_KEY`, `ELEVENLABS_MODEL_ID`, `ELEVENLABS_LANGUAGE_CODE`, and `ELEVENLABS_REQUEST_TIMEOUT_SECONDS`.
The `.ai` agent command uses `[ai.glm]` by default and `[ai.deepseek]` with the `.ds` flag. Provider environment variables are `GLM_API_URL`, `GLM_API_KEY`, `GLM_MODEL`, `DEEPSEEK_API_URL`, `DEEPSEEK_API_KEY`, and `DEEPSEEK_MODEL`. `.th` enables reasoning. Brave Search is available in every AI session and requires `BRAVE_SEARCH_API_KEY` or `ai.brave_api_key`.
@@ -46,10 +44,6 @@ On supported platforms, `main.py` runs the application with `uvloop`; Windows au
- **Check system** (`core.py`): `CommandCheck` functions (e.g. `owner_only_check()`, `shared_access_check()`, `require_reply_check()`, `require_args_check()`) gate handler execution. Chain them with `.check(...)` or `.owner_only()` / `.require_reply()` / `.require_args()` on the builder.
- **ACL** (`acl.py`): `RuntimeAclStore` persists per-session `<session>.acl.json` allow/deny rules. Only shared (`.is_shared`) commands can be granted to guests.
### GitHub webhook (`app/github_webhook.py`)
When enabled, the `aiohttp` server accepts signed GitHub events at `/github/{account}/{chat_id}` and forwards formatted messages through the selected running account.
### Transcription (`app/transcription.py`)
`.txt` / `..txt` must reply to a voice message, video note, or video. The command downloads the Telegram media temporarily and sends it to ElevenLabs Speech to Text. Invalid usage and transcription failures react to the command message with 👎 and send no error reply. Guests need an ACL grant for `txt`.

71
app/config.py
View File

@@ -33,14 +33,6 @@ class AccountConfig:
hide_password: bool = False
@dataclass(slots=True)
class GitHubWebhookConfig:
enabled: bool = False
host: str = "0.0.0.0"
port: int = 8080
secret: str | None = None
@dataclass(slots=True)
class ElevenLabsConfig:
api_key: str | None = None
@@ -85,7 +77,6 @@ class AppConfig:
api_hash: str | None
session_dir: Path
accounts: list[AccountConfig]
github_webhook: GitHubWebhookConfig = field(default_factory=GitHubWebhookConfig)
elevenlabs: ElevenLabsConfig = field(default_factory=ElevenLabsConfig)
ai: AiConfig = field(default_factory=AiConfig)
@@ -102,20 +93,6 @@ def _parse_int(value: Any, *, field_name: str) -> int:
raise ValueError(f"{field_name} must be an integer") from exc
def _parse_bool(value: Any, *, field_name: str) -> bool:
if isinstance(value, bool):
return value
if isinstance(value, str):
normalized = value.strip().lower()
if normalized in {"1", "true", "yes", "on"}:
return True
if normalized in {"0", "false", "no", "off"}:
return False
if isinstance(value, int):
return bool(value)
raise ValueError(f"{field_name} must be a boolean")
def _parse_float(value: Any, *, field_name: str) -> float:
try:
return float(value)
@@ -131,52 +108,6 @@ def _parse_mapping(value: Any, *, field_name: str) -> dict[str, Any]:
return value
def _parse_github_webhook_config(raw: Any) -> GitHubWebhookConfig:
raw_config = _parse_mapping(raw, field_name="github_webhook")
enabled = _parse_bool(
raw_config.get("enabled", False),
field_name="github_webhook.enabled",
)
host = str(raw_config.get("host", "0.0.0.0") or "0.0.0.0").strip()
if not host:
raise ValueError("github_webhook.host is required")
port = _parse_int(raw_config.get("port", 8080), field_name="github_webhook.port")
if port < 1 or port > 65535:
raise ValueError("github_webhook.port must be between 1 and 65535")
raw_secret = raw_config.get("secret")
secret = str(raw_secret).strip() if raw_secret is not None else None
secret = secret or None
if os.environ.get("GITHUB_WEBHOOK_ENABLED") is not None:
enabled = _parse_bool(
os.environ["GITHUB_WEBHOOK_ENABLED"],
field_name="GITHUB_WEBHOOK_ENABLED",
)
if os.environ.get("GITHUB_WEBHOOK_HOST") is not None:
host = os.environ["GITHUB_WEBHOOK_HOST"].strip()
if not host:
raise ValueError("GITHUB_WEBHOOK_HOST is required")
if os.environ.get("GITHUB_WEBHOOK_PORT") is not None:
port = _parse_int(os.environ["GITHUB_WEBHOOK_PORT"], field_name="GITHUB_WEBHOOK_PORT")
if port < 1 or port > 65535:
raise ValueError("GITHUB_WEBHOOK_PORT must be between 1 and 65535")
if os.environ.get("GITHUB_WEBHOOK_SECRET") is not None:
secret = os.environ["GITHUB_WEBHOOK_SECRET"].strip() or None
if enabled and not secret:
raise ValueError("github_webhook.secret is required when github_webhook.enabled is true")
return GitHubWebhookConfig(
enabled=enabled,
host=host,
port=port,
secret=secret,
)
def _parse_elevenlabs_config(raw: Any) -> ElevenLabsConfig:
raw_config = _parse_mapping(raw, field_name="elevenlabs")
@@ -461,7 +392,6 @@ def load_app_config(root: Path) -> AppConfig:
default_session_dir=session_dir,
)
github_webhook = _parse_github_webhook_config(raw.get("github_webhook"))
elevenlabs = _parse_elevenlabs_config(raw.get("elevenlabs"))
ai = _parse_ai_config(raw.get("ai"))
ai.prompt_dir = root / "prompts"
@@ -485,7 +415,6 @@ def load_app_config(root: Path) -> AppConfig:
api_hash=api_hash,
session_dir=session_dir,
accounts=accounts,
github_webhook=github_webhook,
elevenlabs=elevenlabs,
ai=ai,
)

388
app/github_webhook.py
View File

@@ -1,388 +0,0 @@
from __future__ import annotations
import hashlib
import hmac
import html
import json
import logging
from dataclasses import dataclass
from textwrap import wrap
from typing import TYPE_CHECKING, Any
from aiohttp import web
from pyrogram import enums
from app.config import GitHubWebhookConfig
if TYPE_CHECKING:
from app.manager import AccountManager, AccountState
log = logging.getLogger(__name__)
MAX_TELEGRAM_HTML = 3900
MAX_FALLBACK_JSON = 2000
class GitHubWebhookError(Exception):
def __init__(self, status: int, message: str) -> None:
super().__init__(message)
self.status = status
self.message = message
@dataclass(slots=True)
class GitHubWebhookServer:
manager: "AccountManager"
config: GitHubWebhookConfig
runner: web.AppRunner | None = None
site: web.TCPSite | None = None
async def start(self) -> None:
if not self.config.enabled:
return
app = web.Application()
app.add_routes([web.post("/github/{account}/{chat_id}", self.handle)])
self.runner = web.AppRunner(app)
await self.runner.setup()
self.site = web.TCPSite(self.runner, self.config.host, self.config.port)
await self.site.start()
log.info(
"GitHub webhook server listening on %s:%s",
self.config.host,
self.config.port,
)
async def stop(self) -> None:
if self.runner is None:
return
await self.runner.cleanup()
self.runner = None
self.site = None
async def handle(self, request: web.Request) -> web.Response:
try:
body = await request.read()
verify_signature(body, request.headers.get("X-Hub-Signature-256"), self.config.secret)
event = require_header(request, "X-GitHub-Event")
delivery = require_header(request, "X-GitHub-Delivery")
payload = parse_json_body(body)
state = self.resolve_account(request.match_info["account"])
chat_id = parse_chat_id(request.match_info["chat_id"])
if state.client is None or state.status != "running":
raise GitHubWebhookError(503, "account is not running")
text = format_github_event(event, payload, delivery=delivery)
for chunk in chunk_text(text):
await state.client.send_message(
chat_id,
chunk,
parse_mode=enums.ParseMode.HTML,
disable_web_page_preview=True,
)
except GitHubWebhookError as exc:
return web.json_response({"ok": False, "error": exc.message}, status=exc.status)
except Exception:
log.exception("GitHub webhook delivery failed")
return web.json_response({"ok": False, "error": "internal server error"}, status=500)
return web.json_response({"ok": True})
def resolve_account(self, account_name: str) -> "AccountState":
for state in self.manager.states:
if account_name in {state.config.name, state.config.session}:
return state
raise GitHubWebhookError(404, "account not found")
def require_header(request: web.Request, name: str) -> str:
value = request.headers.get(name)
if not value:
raise GitHubWebhookError(400, f"missing {name} header")
return value
def verify_signature(body: bytes, signature: str | None, secret: str | None) -> None:
if not secret:
raise GitHubWebhookError(500, "github webhook secret is not configured")
if not signature:
raise GitHubWebhookError(401, "missing X-Hub-Signature-256 header")
if not signature.startswith("sha256="):
raise GitHubWebhookError(401, "invalid X-Hub-Signature-256 header")
expected = "sha256=" + hmac.new(secret.encode("utf-8"), body, hashlib.sha256).hexdigest()
if not hmac.compare_digest(expected, signature):
raise GitHubWebhookError(401, "invalid signature")
def parse_json_body(body: bytes) -> dict[str, Any]:
try:
payload = json.loads(body.decode("utf-8"))
except (UnicodeDecodeError, json.JSONDecodeError) as exc:
raise GitHubWebhookError(400, "request body must be valid JSON") from exc
if not isinstance(payload, dict):
raise GitHubWebhookError(400, "request JSON must be an object")
return payload
def parse_chat_id(value: str) -> int | str:
chat_id = value.strip()
if not chat_id:
raise GitHubWebhookError(400, "chat_id is required")
if chat_id.startswith("@"):
return chat_id
try:
return int(chat_id)
except ValueError as exc:
raise GitHubWebhookError(400, "chat_id must be an integer or @username") from exc
def chunk_text(text: str, limit: int = MAX_TELEGRAM_HTML) -> list[str]:
if len(text) <= limit:
return [text]
chunks: list[str] = []
current: list[str] = []
current_len = 0
for paragraph in text.split("\n"):
piece = paragraph if not current else "\n" + paragraph
if current_len + len(piece) <= limit:
current.append(piece.lstrip("\n") if not current else piece)
current_len += len(piece)
continue
if current:
chunks.append("".join(current))
current = []
current_len = 0
if len(paragraph) <= limit:
current = [paragraph]
current_len = len(paragraph)
continue
chunks.extend(wrap(paragraph, width=limit) or [paragraph[:limit]])
if current:
chunks.append("".join(current))
return [chunk for chunk in chunks if chunk]
def format_github_event(event: str, payload: dict[str, Any], *, delivery: str | None = None) -> str:
formatter = {
"push": _format_push,
"pull_request": _format_pull_request,
"issues": _format_issue,
"issue_comment": _format_issue_comment,
"release": _format_release,
"workflow_run": _format_workflow_run,
}.get(event, _format_fallback)
lines = formatter(event, payload)
if delivery:
lines.append("")
lines.append(f"<code>delivery: {html.escape(delivery)}</code>")
return "\n".join(line for line in lines if line is not None)
def _format_push(event: str, payload: dict[str, Any]) -> list[str]:
repo = _repo_name(payload)
sender = _sender_name(payload)
ref = _short_ref(_text(payload.get("ref"), "unknown"))
commits = _list(payload.get("commits"))
compare_url = _text(payload.get("compare"))
head_commit = _dict(payload.get("head_commit"))
lines = [
f"<b>GitHub push</b> in {_link(repo, _repo_url(payload))}",
f"Branch: <code>{_e(ref)}</code>",
f"Sender: {_e(sender)}",
f"Commits: <b>{len(commits)}</b>",
]
if compare_url:
lines.append(f"Compare: {_link('open diff', compare_url)}")
for commit in commits[:5]:
commit_data = _dict(commit)
sha = _text(commit_data.get("id"))[:7]
message = _first_line(_text(commit_data.get("message"), "No commit message"))
author = _text(_dict(commit_data.get("author")).get("name"), "unknown")
url = _text(commit_data.get("url"))
label = f"{sha} - {message}" if sha else message
lines.append(f"- {_link(label, url)} by {_e(author)}")
if len(commits) > 5:
lines.append(f"- ...and {len(commits) - 5} more")
if not commits and head_commit:
message = _first_line(_text(head_commit.get("message"), "No commit message"))
lines.append(f"Head: {_link(message, _text(head_commit.get('url')))}")
return lines
def _format_pull_request(event: str, payload: dict[str, Any]) -> list[str]:
action = _text(payload.get("action"), "updated")
repo = _repo_name(payload)
pr = _dict(payload.get("pull_request"))
number = _text(pr.get("number") or payload.get("number"), "?")
title = _text(pr.get("title"), "Untitled pull request")
state = _text(pr.get("state"), "unknown")
merged = pr.get("merged")
user = _login(pr.get("user")) or _sender_name(payload)
url = _text(pr.get("html_url"))
additions = pr.get("additions")
deletions = pr.get("deletions")
changed_files = pr.get("changed_files")
lines = [
f"<b>GitHub pull request {html.escape(action)}</b> in {_link(repo, _repo_url(payload))}",
f"PR: {_link('#' + number + ' ' + title, url)}",
f"State: <code>{_e(state)}</code>{' merged' if merged else ''}",
f"Author: {_e(user)}",
]
stats = []
if changed_files is not None:
stats.append(f"{changed_files} files")
if additions is not None:
stats.append(f"+{additions}")
if deletions is not None:
stats.append(f"-{deletions}")
if stats:
lines.append("Changes: " + ", ".join(stats))
return lines
def _format_issue(event: str, payload: dict[str, Any]) -> list[str]:
action = _text(payload.get("action"), "updated")
repo = _repo_name(payload)
issue = _dict(payload.get("issue"))
number = _text(issue.get("number") or payload.get("number"), "?")
title = _text(issue.get("title"), "Untitled issue")
state = _text(issue.get("state"), "unknown")
user = _login(issue.get("user")) or _sender_name(payload)
return [
f"<b>GitHub issue {html.escape(action)}</b> in {_link(repo, _repo_url(payload))}",
f"Issue: {_link('#' + number + ' ' + title, _text(issue.get('html_url')))}",
f"State: <code>{_e(state)}</code>",
f"Author: {_e(user)}",
]
def _format_issue_comment(event: str, payload: dict[str, Any]) -> list[str]:
action = _text(payload.get("action"), "updated")
repo = _repo_name(payload)
issue = _dict(payload.get("issue"))
comment = _dict(payload.get("comment"))
number = _text(issue.get("number"), "?")
title = _text(issue.get("title"), "Untitled issue")
user = _login(comment.get("user")) or _sender_name(payload)
body = _first_line(_text(comment.get("body"), ""), limit=220)
lines = [
f"<b>GitHub issue comment {html.escape(action)}</b> in {_link(repo, _repo_url(payload))}",
f"Thread: {_link('#' + number + ' ' + title, _text(issue.get('html_url')))}",
f"Comment: {_link('open comment', _text(comment.get('html_url')))} by {_e(user)}",
]
if body:
lines.append(f"<blockquote>{_e(body)}</blockquote>")
return lines
def _format_release(event: str, payload: dict[str, Any]) -> list[str]:
action = _text(payload.get("action"), "updated")
repo = _repo_name(payload)
release = _dict(payload.get("release"))
name = _text(release.get("name") or release.get("tag_name"), "Untitled release")
tag = _text(release.get("tag_name"), "unknown")
author = _login(release.get("author")) or _sender_name(payload)
return [
f"<b>GitHub release {html.escape(action)}</b> in {_link(repo, _repo_url(payload))}",
f"Release: {_link(name, _text(release.get('html_url')))}",
f"Tag: <code>{_e(tag)}</code>",
f"Author: {_e(author)}",
]
def _format_workflow_run(event: str, payload: dict[str, Any]) -> list[str]:
action = _text(payload.get("action"), "updated")
repo = _repo_name(payload)
run = _dict(payload.get("workflow_run"))
name = _text(run.get("name"), "Workflow")
status = _text(run.get("status"), "unknown")
conclusion = _text(run.get("conclusion"), "")
branch = _text(run.get("head_branch"), "unknown")
actor = _login(run.get("actor")) or _sender_name(payload)
state = conclusion if conclusion else status
return [
f"<b>GitHub workflow {html.escape(action)}</b> in {_link(repo, _repo_url(payload))}",
f"Run: {_link(name, _text(run.get('html_url')))}",
f"State: <code>{_e(state)}</code>",
f"Branch: <code>{_e(branch)}</code>",
f"Actor: {_e(actor)}",
]
def _format_fallback(event: str, payload: dict[str, Any]) -> list[str]:
action = _text(payload.get("action"), "")
repo = _repo_name(payload)
sender = _sender_name(payload)
pretty = json.dumps(payload, ensure_ascii=False, indent=2, default=str)
if len(pretty) > MAX_FALLBACK_JSON:
pretty = pretty[:MAX_FALLBACK_JSON].rstrip() + "\n..."
lines = [
f"<b>GitHub {html.escape(event)}</b> in {_link(repo, _repo_url(payload))}",
f"Sender: {_e(sender)}",
]
if action:
lines.append(f"Action: <code>{_e(action)}</code>")
lines.append(f"<pre><code>{_e(pretty)}</code></pre>")
return lines
def _repo_name(payload: dict[str, Any]) -> str:
repo = _dict(payload.get("repository"))
return _text(repo.get("full_name") or repo.get("name"), "unknown/repository")
def _repo_url(payload: dict[str, Any]) -> str:
return _text(_dict(payload.get("repository")).get("html_url"))
def _sender_name(payload: dict[str, Any]) -> str:
return _login(payload.get("sender")) or "unknown"
def _login(value: Any) -> str | None:
data = _dict(value)
login = _text(data.get("login"))
return login or None
def _dict(value: Any) -> dict[str, Any]:
return value if isinstance(value, dict) else {}
def _list(value: Any) -> list[Any]:
return value if isinstance(value, list) else []
def _text(value: Any, default: str = "") -> str:
if value is None:
return default
return str(value)
def _first_line(value: str, *, limit: int = 140) -> str:
line = value.strip().splitlines()[0] if value.strip() else ""
if len(line) <= limit:
return line
return line[: limit - 1].rstrip() + "..."
def _short_ref(ref: str) -> str:
for prefix in ("refs/heads/", "refs/tags/"):
if ref.startswith(prefix):
return ref[len(prefix) :]
return ref
def _link(label: str, url: str) -> str:
safe_label = _e(label)
if not url:
return safe_label
return f'<a href="{html.escape(url, quote=True)}">{safe_label}</a>'
def _e(value: Any) -> str:
return html.escape(str(value), quote=False)

5
app/manager.py
View File

@@ -9,7 +9,6 @@ from pyrogram import Client, idle
from app.agentic import AgentSessionStore
from app.commands import DEFAULT_REGISTRY, CommandRegistry, RuntimeAclStore
from app.config import AccountConfig, AppConfig
from app.github_webhook import GitHubWebhookServer
from app.handlers import register_common_handlers
log = logging.getLogger(__name__)
@@ -31,12 +30,10 @@ class AccountManager:
config: AppConfig
command_registry: CommandRegistry = field(init=False)
states: list[AccountState] = field(init=False)
github_webhook_server: GitHubWebhookServer = field(init=False)
def __post_init__(self) -> None:
self.command_registry = DEFAULT_REGISTRY
self.states = [AccountState(config=account) for account in self.config.accounts]
self.github_webhook_server = GitHubWebhookServer(self, self.config.github_webhook)
def _make_client(self, account: AccountConfig) -> Client:
api_id = account.api_id if account.api_id is not None else self.config.api_id
@@ -146,8 +143,6 @@ class AccountManager:
raise RuntimeError("No accounts were started successfully")
try:
await self.github_webhook_server.start()
await idle()
finally:
await self.github_webhook_server.stop()
await self.stop_all()

6
config.example.toml
View File

@@ -10,12 +10,6 @@ session = "personal"
name = "work"
session = "work"
[github_webhook]
enabled = false
host = "0.0.0.0"
port = 8080
secret = "change-me"
[elevenlabs]
api_key = "your_elevenlabs_api_key_here"
model_id = "scribe_v2"

54
tests/test_config_github_webhook.py → tests/test_config_services.py
View File

@@ -3,59 +3,7 @@ from __future__ import annotations
import unittest
from unittest.mock import patch
from app.config import _parse_ai_config, _parse_elevenlabs_config, _parse_github_webhook_config
class ConfigGitHubWebhookTests(unittest.TestCase):
def test_github_webhook_config_parses_values(self) -> None:
with patch.dict("os.environ", {}, clear=True):
config = _parse_github_webhook_config(
{
"enabled": True,
"host": "127.0.0.1",
"port": 9000,
"secret": "config-secret",
}
)
self.assertTrue(config.enabled)
self.assertEqual(config.host, "127.0.0.1")
self.assertEqual(config.port, 9000)
self.assertEqual(config.secret, "config-secret")
def test_github_webhook_env_overrides(self) -> None:
env = {
"GITHUB_WEBHOOK_ENABLED": "true",
"GITHUB_WEBHOOK_HOST": "localhost",
"GITHUB_WEBHOOK_PORT": "7777",
"GITHUB_WEBHOOK_SECRET": "env-secret",
}
with patch.dict("os.environ", env, clear=True):
config = _parse_github_webhook_config(
{
"enabled": False,
"host": "0.0.0.0",
"port": 8080,
"secret": "config-secret",
}
)
self.assertTrue(config.enabled)
self.assertEqual(config.host, "localhost")
self.assertEqual(config.port, 7777)
self.assertEqual(config.secret, "env-secret")
def test_github_webhook_requires_secret_when_enabled(self) -> None:
with patch.dict("os.environ", {}, clear=True):
with self.assertRaisesRegex(ValueError, "github_webhook.secret"):
_parse_github_webhook_config({"enabled": True})
def test_github_webhook_allows_missing_secret_when_disabled(self) -> None:
with patch.dict("os.environ", {}, clear=True):
config = _parse_github_webhook_config({"enabled": False})
self.assertFalse(config.enabled)
self.assertIsNone(config.secret)
from app.config import _parse_ai_config, _parse_elevenlabs_config
class ConfigElevenLabsTests(unittest.TestCase):

153
tests/test_github_webhook.py
View File

@@ -1,153 +0,0 @@
from __future__ import annotations
import hashlib
import hmac
import json
import unittest
from types import SimpleNamespace
from unittest.mock import AsyncMock
from aiohttp import web
from aiohttp.test_utils import TestClient, TestServer
from app.config import GitHubWebhookConfig
from app.github_webhook import GitHubWebhookServer, chunk_text, format_github_event
def _signature(body: bytes, secret: str = "secret") -> str:
digest = hmac.new(secret.encode("utf-8"), body, hashlib.sha256).hexdigest()
return f"sha256={digest}"
def _push_payload() -> dict:
return {
"ref": "refs/heads/main",
"compare": "https://github.com/acme/project/compare/a...b",
"repository": {
"full_name": "acme/project",
"html_url": "https://github.com/acme/project",
},
"sender": {"login": "octocat"},
"commits": [
{
"id": "abcdef123456",
"message": "Add webhook receiver\n\nDetails",
"url": "https://github.com/acme/project/commit/abcdef1",
"author": {"name": "Mona"},
}
],
}
class GitHubWebhookFormattingTests(unittest.TestCase):
def test_format_push_event(self) -> None:
text = format_github_event("push", _push_payload(), delivery="delivery-1")
self.assertIn("GitHub push", text)
self.assertIn("acme/project", text)
self.assertIn("main", text)
self.assertIn("abcdef1", text)
self.assertIn("delivery-1", text)
def test_format_pull_request_event(self) -> None:
text = format_github_event(
"pull_request",
{
"action": "opened",
"repository": {"full_name": "acme/project"},
"sender": {"login": "octocat"},
"pull_request": {
"number": 42,
"title": "Improve webhooks",
"html_url": "https://github.com/acme/project/pull/42",
"state": "open",
"user": {"login": "octocat"},
"changed_files": 3,
"additions": 12,
"deletions": 4,
},
},
)
self.assertIn("pull request opened", text)
self.assertIn("#42 Improve webhooks", text)
self.assertIn("3 files", text)
def test_format_unknown_event_uses_compact_json(self) -> None:
text = format_github_event(
"star",
{
"action": "created",
"repository": {"full_name": "acme/project"},
"sender": {"login": "octocat"},
"extra": "value",
},
)
self.assertIn("GitHub star", text)
self.assertIn("Action:", text)
self.assertIn('"extra": "value"', text)
def test_chunk_text_splits_long_messages(self) -> None:
chunks = chunk_text("a" * 5000, limit=1000)
self.assertGreater(len(chunks), 1)
self.assertTrue(all(len(chunk) <= 1000 for chunk in chunks))
class GitHubWebhookHandlerTests(unittest.IsolatedAsyncioTestCase):
async def asyncSetUp(self) -> None:
self.telegram_client = SimpleNamespace(send_message=AsyncMock())
state = SimpleNamespace(
config=SimpleNamespace(name="personal", session="personal_session"),
client=self.telegram_client,
status="running",
)
manager = SimpleNamespace(states=[state])
self.server = GitHubWebhookServer(
manager=manager,
config=GitHubWebhookConfig(enabled=True, secret="secret"),
)
app = web.Application()
app.add_routes([web.post("/github/{account}/{chat_id}", self.server.handle)])
self.client = TestClient(TestServer(app))
await self.client.start_server()
async def asyncTearDown(self) -> None:
await self.client.close()
async def _post(self, payload: dict, *, signature: str | None = None, account: str = "personal"):
body = json.dumps(payload).encode("utf-8")
headers = {
"Content-Type": "application/json",
"X-GitHub-Event": "push",
"X-GitHub-Delivery": "delivery-1",
"X-Hub-Signature-256": signature if signature is not None else _signature(body),
}
return await self.client.post(f"/github/{account}/-100123", data=body, headers=headers)
async def test_handler_accepts_valid_signature_and_sends_message(self) -> None:
response = await self._post(_push_payload())
self.assertEqual(response.status, 200)
self.telegram_client.send_message.assert_awaited_once()
args, kwargs = self.telegram_client.send_message.await_args
self.assertEqual(args[0], -100123)
self.assertIn("GitHub push", args[1])
self.assertIn("parse_mode", kwargs)
async def test_handler_rejects_invalid_signature(self) -> None:
response = await self._post(_push_payload(), signature="sha256=bad")
self.assertEqual(response.status, 401)
self.telegram_client.send_message.assert_not_awaited()
async def test_handler_rejects_unknown_account(self) -> None:
response = await self._post(_push_payload(), account="missing")
self.assertEqual(response.status, 404)
self.telegram_client.send_message.assert_not_awaited()
if __name__ == "__main__":
unittest.main()