kuhyx/signal-bot
1
0
Fork 0
mirror of https://github.com/kuhyx/signal-bot.git synced 2026-11-11 13:17:31 +01:00
Code Issues Actions Packages Projects Releases Wiki Activity

feat: name every release after a pope, and tell the group when one lands
Some checks are pending
CI / check (push) Waiting to run
Details

Browse Source 
Release n is pope n, so the version is a fact about the ordinal rather than a
judgement about the size of a change. The constraint that fixes the scheme is
Jan Pawel II at 2.13.7: eight patches per minor with majors opening at ordinals
33 and 153 gives (264 - 153) = 13 * 8 + 7 exactly, and both earlier majors
happen to divide evenly by eight too, so no bracket needs a fudge factor. The
anchors are not arbitrary -- 33 is Sylwester I, pope through the Edict of Milan
and Nicaea, and 153 is Wiktor II, the first pope elected after the 1054 schism.

popes.json is scraped by tool/gen_popes.py rather than recalled, because
antipopes, Benedykt IX's three separate reigns and the Stefan II question all
shift the count and being wrong by one moves Jan Pawel II off 2.13.7 silently.
The generator refuses to write a list that is not 267 contiguous entries with
Jan Pawel II at 264. A second, independent parse of the same page agreed on
every ordinal, which is the only reason to believe either.

On startup each watched group is told about the releases it has not seen. A
group with no record has seen none of them, so the first announcement starts at
0.0.0 "Swiety Piotr Apostol" rather than skipping to whatever is current --
watching the ladder start at Peter is the point of naming releases after popes.
A group added years from now would otherwise be handed the whole backlog, so
the message is capped at ten releases and says how many it left out. The
version is recorded before the message is sent: with Restart=always and
RestartSec=5, a crash between the two would re-announce on every restart, and a
missed announcement is far cheaper than a looping one.

DOCS-changelog.md is the single source -- the bot parses it, so there is no
second machine-readable copy to drift. It is DOCS-, not CHANGELOG.md, because
the md-naming gate allows four namespaces and that is not one of them.

Deliberately NOT bumping SCHEMA_VERSION for the new group_meta table:
_apply_schema drops every table on a bump, so bumping to add a table would have
thrown away the live tallies to gain nothing. Verified against the real
database -- group_meta appeared and all four counts survived the restart.

The schema moves to its own module so store.py stays under the 250-line cap,
and /healthz now names the build, because there was previously no way to ask a
running bot which version it is.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_015d48t4qsQLBJnDwDeo2Nom
This commit is contained in:
Krzysztof Rudnicki 2026-08-29 22:51:20 +02:00
parent 180e53210b
commit 4bbb1e1519
17 changed files with 2488 additions and 55 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

27
DOCS-changelog.md Normal file
View File

@ -0,0 +1,27 @@
# Changelog
Releases are named after the popes, in order: release *n* is pope *n*. The
version number is a pure function of that ordinal — see `signal_bot/versioning.py`
for the two constants and why they are what they are. The short version is that
Jan Paweł II has to land on 2.13.7, and everything else follows.
The bot reads this file. The heading format is load-bearing:
```
## <version> — <pope name>
- <one change, in Polish, as the group will read it>
```
Add the new section at the top when you bump `RELEASE_ORDINAL`. Anything that
is not a `##` heading or a `-` bullet under one is ignored, so prose between
releases is free.
## 0.0.1 — Linus
- wersje bota nazywają się teraz imionami kolejnych papieży
- po aktualizacji bot ogłasza w grupie nową wersję i listę zmian
- `/healthz` pokazuje wersję i papieża, więc widać co naprawdę działa
## 0.0.0 — Święty Piotr Apostoł
- bot taki, jaki był przed drabinką papieską: `!cat`, `!dog` i dzienne podsumowanie

6
README.md
View File

@ -237,13 +237,13 @@ Counts live in SQLite and survive a restart.
## Develop
```sh
pytest # 189 tests, 100% branch coverage, no network
pytest # 359 tests, 100% branch coverage, no network
pre-commit run --all-files # ruff ALL + mypy strict + 250-line cap + gates
./scripts/e2e.sh # the real bot against the fake server
```
No `# noqa`, no `# type: ignore`, no coverage omits. Docs on the wire format
are in [DOCS-signal-envelopes.md](DOCS-signal-envelopes.md).
No `# noqa`, no `# type: ignore`, no coverage omits. Wire format:
[DOCS-signal-envelopes.md](DOCS-signal-envelopes.md). Releases are named after the popes, in order: [DOCS-changelog.md](DOCS-changelog.md).
## Licence

2
pyproject.toml
View File

@ -1,6 +1,6 @@
[project]
name = "signal-bot"
version = "1.0.0"
version = "0.0.1"
description = "A Signal group bot: image commands and daily message tallies."
readme = "README.md"
license = "MIT"

100
signal_bot/announce.py Normal file
View File

@ -0,0 +1,100 @@
# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence.
"""Telling a group that the bot has been upgraded.
A group is told about every release it has not seen, and a group that has
never been told anything has not seen any of them -- so the first
announcement starts at 0.0.0 "Swiety Piotr Apostol" rather than skipping
straight to whatever is current. Watching the ladder start at Peter is the
point of naming releases after popes at all.
That does mean a group added years from now would be handed the whole
backlog, so the message is capped at MAX_RELEASES and says how many older
versions it left out.
The version is recorded *before* the message is sent. `Restart=always` with
`RestartSec=5` means a crash between the two would otherwise re-announce on
every restart, and a missed announcement is far cheaper than a looping one.
"""
from __future__ import annotations
import logging
from typing import TYPE_CHECKING, Final
from signal_bot.changelog import since
if TYPE_CHECKING:
from collections.abc import Iterable
from signal_bot.changelog import Release
from signal_bot.store import Store
from signal_bot.transport_base import Transport
logger = logging.getLogger(__name__)
ANNOUNCED_KEY: Final = "announced_version"
# How many releases one announcement may list before it becomes a wall.
MAX_RELEASES: Final = 10
def render(version: str, pope: str, releases: Iterable[Release]) -> str:
"""Return the message announcing a new version, in Polish.
One release is rendered flat, because the header already names it. Several
get a block each, so the group can see the ladder it is climbing.
"""
shown = list(releases)
header = f"Wersja {version} \u00ab{pope}\u00bb wydana"
omitted = max(0, len(shown) - MAX_RELEASES)
shown = shown[-MAX_RELEASES:]
if len(shown) == 1:
body = "\n".join(f"- {line}" for line in shown[0].lines)
return f"{header}\nChangelog:\n{body}"
blocks = [
"\n".join(
[f"{release.version} \u00ab{release.pope}\u00bb", *(f"- {x}" for x in release.lines)]
)
for release in shown
]
tail = f"\n\n(pominieto {omitted} starszych wersji)" if omitted else ""
return f"{header}\nChangelog:\n\n" + "\n\n".join(blocks) + tail
async def announce_to(
transport: Transport,
store: Store,
group_id: str,
releases: tuple[Release, ...],
version: str,
pope: str,
) -> None:
"""Announce the current version to one group, if it has not seen it."""
previous = store.meta(group_id, ANNOUNCED_KEY)
if previous == version:
return
# No record at all means the group has seen nothing, not that it should be
# skipped: that is how Swiety Piotr Apostol gets announced rather than
# silently assumed.
pending = releases if previous is None else since(releases, previous)
store.set_meta(group_id, ANNOUNCED_KEY, version)
if not pending:
logger.info("group %s recorded at %s with nothing to announce", group_id, version)
return
await transport.send_text(render(version, pope, pending), group_id)
async def announce_all(
transport: Transport,
store: Store,
group_ids: Iterable[str],
releases: tuple[Release, ...],
version: str,
pope: str,
) -> None:
"""Announce to every watched group; one failure must not silence the rest."""
for group_id in group_ids:
try:
await announce_to(transport, store, group_id, releases, version, pope)
except Exception:
logger.exception("release announcement failed for group %s", group_id)

22
signal_bot/app.py
View File

@ -19,6 +19,8 @@ import httpx
from fastapi import FastAPI, Response, status
from signal_bot.animals import register_animal_commands
from signal_bot.announce import announce_all
from signal_bot.changelog import load as load_changelog
from signal_bot.config import Config, load_config
from signal_bot.counting import format_tally
from signal_bot.decorators import cooldown
@ -31,6 +33,7 @@ from signal_bot.scheduler import DailyScheduler
from signal_bot.store import Store
from signal_bot.transport_rest import RestTransport
from signal_bot.trust import InviteTrust
from signal_bot.versioning import POPE, __version__
logger = logging.getLogger(__name__)
@ -78,6 +81,17 @@ class Bot:
continue
self.store.reset_tallies(group_id)
async def announce_release(self) -> None:
"""Tell every watched group about a version it has not seen yet."""
await announce_all(
self.transport,
self.store,
list(self.directory.groups),
load_changelog(),
__version__,
POPE.name,
)
async def listen(self) -> None:
"""Consume frames forever, handing each to the dispatcher."""
async for frame in self.transport.receive():
@ -149,6 +163,9 @@ class Runtime:
# Fourth loop: without it, a dead connection to Signal is
# indistinguishable from a quiet evening, for hours.
asyncio.create_task(self.bot.liveness.run()),
# Fifth, and one-shot: says hello with a changelog when the
# running version is newer than what a group was last told.
asyncio.create_task(self.bot.announce_release()),
]
async def stop(self) -> None:
@ -218,7 +235,8 @@ async def healthz(response: Response) -> dict[str, str]:
thinks to look at the logs.
"""
liveness: Liveness = app.state.runtime.bot.liveness
build = {"version": __version__, "pope": POPE.name}
if liveness.healthy:
return {"status": "ok"}
return {"status": "ok", **build}
response.status_code = status.HTTP_503_SERVICE_UNAVAILABLE
return {"status": "deaf", "reason": liveness.reason}
return {"status": "deaf", "reason": liveness.reason, **build}

80
signal_bot/changelog.py Normal file
View File

@ -0,0 +1,80 @@
# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence.
"""The changelog the bot announces, read from DOCS-changelog.md.
That file is the single source: there is no second machine-readable copy to
drift out of sync with it. Only two shapes are recognised -- a `## <version> —
<pope>` heading and a `- ` bullet beneath one -- so prose around them is free.
"""
from __future__ import annotations
import re
from dataclasses import dataclass
from pathlib import Path
from typing import Final
from signal_bot.versioning import ordinal_for
# Repo root, since the changelog is documentation rather than package data.
DEFAULT_PATH: Final = Path(__file__).resolve().parent.parent / "DOCS-changelog.md"
_HEADING: Final = re.compile(r"^##\s+(\d+\.\d+\.\d+)\s*[—-]\s*(.+?)\s*$")
_BULLET: Final = re.compile(r"^-\s+(.+?)\s*$")
@dataclass(frozen=True, slots=True)
class Release:
"""One released version and the lines describing it."""
version: str
pope: str
lines: tuple[str, ...]
@property
def ordinal(self) -> int:
"""The papal ordinal this release carries."""
return ordinal_for(self.version)
def parse(text: str) -> tuple[Release, ...]:
"""Return every release in the document, oldest first."""
releases: list[Release] = []
version = pope = ""
lines: list[str] = []
for raw in text.splitlines():
heading = _HEADING.match(raw)
if heading is not None:
if version:
releases.append(Release(version, pope, tuple(lines)))
version, pope, lines = heading.group(1), heading.group(2), []
continue
bullet = _BULLET.match(raw)
if bullet is not None and version:
lines.append(bullet.group(1))
if version:
releases.append(Release(version, pope, tuple(lines)))
return tuple(sorted(releases, key=lambda release: release.ordinal))
def load(path: Path | None = None) -> tuple[Release, ...]:
"""Return every release, or nothing if the changelog is not on disk.
A missing changelog must not stop the bot: it is documentation that lives
outside the package, so an odd install layout should cost an announcement
rather than the whole process.
"""
source = DEFAULT_PATH if path is None else path
try:
text = source.read_text(encoding="utf-8")
except OSError:
return ()
return parse(text)
def since(releases: tuple[Release, ...], previous: str) -> tuple[Release, ...]:
"""Return the releases newer than ``previous``, oldest first."""
try:
floor = ordinal_for(previous)
except ValueError:
return ()
return tuple(release for release in releases if release.ordinal > floor)

1607
signal_bot/popes.json Normal file
View File

File diff suppressed because it is too large Load Diff

68
signal_bot/schema.py Normal file
View File

@ -0,0 +1,68 @@
# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence.
"""The database shape, kept apart from the code that queries it.
Split out of ``store.py`` so both stay under the 250-line cap, and because
"what the tables are" and "how the bot reads them" are two things.
"""
from __future__ import annotations
from typing import Final
# Bumped whenever the shape below changes. See Store._apply_schema for why
# a version number is the only safe way to notice.
#
# Deliberately NOT bumped when a whole new table is added: CREATE TABLE IF
# NOT EXISTS is additive and harmless, whereas a bump drops every table and
# would throw away the live tallies to gain nothing.
SCHEMA_VERSION: Final = 2
# Every table is keyed by group first. The bot watches several groups, and a
# tally, a cooldown or a vote that leaked between them would mean a test
# message throttling the production group -- so the scope is in the primary
# key rather than in the callers' discipline.
SCHEMA: Final = """
CREATE TABLE IF NOT EXISTS message_counts (
group_id TEXT NOT NULL,
uuid TEXT NOT NULL,
display_name TEXT NOT NULL,
count INTEGER NOT NULL DEFAULT 0,
PRIMARY KEY (group_id, uuid)
);
CREATE TABLE IF NOT EXISTS command_uses (
group_id TEXT NOT NULL,
user_uuid TEXT NOT NULL,
command TEXT NOT NULL,
used_at REAL NOT NULL,
PRIMARY KEY (group_id, user_uuid, command)
);
CREATE TABLE IF NOT EXISTS daily_uses (
group_id TEXT NOT NULL,
user_uuid TEXT NOT NULL,
command TEXT NOT NULL,
day TEXT NOT NULL,
uses INTEGER NOT NULL DEFAULT 0,
PRIMARY KEY (group_id, user_uuid, command, day)
);
CREATE TABLE IF NOT EXISTS votes (
group_id TEXT NOT NULL,
command TEXT NOT NULL,
voter_uuid TEXT NOT NULL,
voted_at REAL NOT NULL,
PRIMARY KEY (group_id, command, voter_uuid)
);
CREATE TABLE IF NOT EXISTS group_meta (
group_id TEXT NOT NULL,
key TEXT NOT NULL,
value TEXT NOT NULL,
PRIMARY KEY (group_id, key)
);
"""
TABLES: Final = (
"message_counts",
"command_uses",
"daily_uses",
"votes",
"group_meta",
)

68
signal_bot/store.py
View File

@ -12,52 +12,12 @@ import sqlite3
from dataclasses import dataclass
from typing import TYPE_CHECKING, Self
from signal_bot.schema import SCHEMA, SCHEMA_VERSION, TABLES
if TYPE_CHECKING:
from collections.abc import Iterator
from types import TracebackType
# Bumped whenever the shape below changes. See Store._apply_schema for why
# a version number is the only safe way to notice.
SCHEMA_VERSION = 2
# Every table is keyed by group first. The bot watches several groups, and a
# tally, a cooldown or a vote that leaked between them would mean a test
# message throttling the production group -- so the scope is in the primary
# key rather than in the callers' discipline.
_SCHEMA = """
CREATE TABLE IF NOT EXISTS message_counts (
group_id TEXT NOT NULL,
uuid TEXT NOT NULL,
display_name TEXT NOT NULL,
count INTEGER NOT NULL DEFAULT 0,
PRIMARY KEY (group_id, uuid)
);
CREATE TABLE IF NOT EXISTS command_uses (
group_id TEXT NOT NULL,
user_uuid TEXT NOT NULL,
command TEXT NOT NULL,
used_at REAL NOT NULL,
PRIMARY KEY (group_id, user_uuid, command)
);
CREATE TABLE IF NOT EXISTS daily_uses (
group_id TEXT NOT NULL,
user_uuid TEXT NOT NULL,
command TEXT NOT NULL,
day TEXT NOT NULL,
uses INTEGER NOT NULL DEFAULT 0,
PRIMARY KEY (group_id, user_uuid, command, day)
);
CREATE TABLE IF NOT EXISTS votes (
group_id TEXT NOT NULL,
command TEXT NOT NULL,
voter_uuid TEXT NOT NULL,
voted_at REAL NOT NULL,
PRIMARY KEY (group_id, command, voter_uuid)
);
"""
_TABLES = ("message_counts", "command_uses", "daily_uses", "votes")
@dataclass(frozen=True, slots=True)
class Tally:
@ -93,9 +53,9 @@ class Store:
"""
version = int(self._conn.execute("PRAGMA user_version").fetchone()[0])
if version < SCHEMA_VERSION:
for table in _TABLES:
for table in TABLES:
self._conn.execute(f"DROP TABLE IF EXISTS {table}")
self._conn.executescript(_SCHEMA)
self._conn.executescript(SCHEMA)
self._conn.execute(f"PRAGMA user_version = {SCHEMA_VERSION}")
self._conn.commit()
@ -131,6 +91,26 @@ class Store:
self._conn.execute("DELETE FROM command_uses WHERE group_id = ?", (group_id,))
self._conn.execute("DELETE FROM daily_uses WHERE group_id = ?", (group_id,))
self._conn.execute("DELETE FROM votes WHERE group_id = ?", (group_id,))
self._conn.execute("DELETE FROM group_meta WHERE group_id = ?", (group_id,))
self._conn.commit()
# -- per-group key/value state -----------------------------------------
def meta(self, group_id: str, key: str) -> str | None:
"""Return a stored value for one group, or None if never set."""
row = self._conn.execute(
"SELECT value FROM group_meta WHERE group_id = ? AND key = ?",
(group_id, key),
).fetchone()
return None if row is None else str(row["value"])
def set_meta(self, group_id: str, key: str, value: str) -> None:
"""Store a value for one group, replacing any previous one."""
self._conn.execute(
"INSERT INTO group_meta (group_id, key, value) VALUES (?, ?, ?) "
"ON CONFLICT(group_id, key) DO UPDATE SET value = excluded.value",
(group_id, key, value),
)
self._conn.commit()
# -- message tallies ---------------------------------------------------

103
signal_bot/versioning.py Normal file
View File

@ -0,0 +1,103 @@
# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence.
"""Release versions named after the popes, in order.
Release *n* is pope *n*: 0.0.0 is Święty Piotr Apostoł, and the version is a
pure function of the ordinal rather than a judgement about the size of a
change. The one fixed point is Jan Paweł II at 2.13.7, which is what picks
the two constants below -- eight patches per minor, and majors opening at
ordinals 33 and 153:
(264 - 153) = 111 = 13 * 8 + 7 -> 2.13.7
Both earlier majors happen to divide evenly by eight as well (32 = 4 * 8,
120 = 15 * 8), so no bracket needs a fudge factor. The anchors are not
arbitrary either: 33 is Sylwester I, pope through the Edict of Milan and
Nicaea, and 153 is Wiktor II, the first pope elected after the 1054 schism.
"""
from __future__ import annotations
import json
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Final
# Ordinals at which a new major version opens. Ascending, first entry 1.
MAJOR_STARTS: Final = (1, 33, 153)
PATCHES_PER_MINOR: Final = 8
# Which release this build is. Bump by one per release; everything else,
# including the version string in pyproject.toml, follows from it.
RELEASE_ORDINAL: Final = 2
_POPES_FILE: Final = Path(__file__).with_name("popes.json")
@dataclass(frozen=True, slots=True)
class Pope:
"""One entry of the papal list, as the version ladder needs it."""
ordinal: int
name: str
reign: str
portrait: str
def _load() -> tuple[Pope, ...]:
"""Read popes.json once, at import."""
document: dict[str, Any] = json.loads(_POPES_FILE.read_text(encoding="utf-8"))
return tuple(
Pope(
ordinal=int(entry["ordinal"]),
name=str(entry["name"]),
reign=str(entry["reign"]),
portrait=str(entry["portrait"]),
)
for entry in document["popes"]
)
POPES: Final = _load()
def version_for(ordinal: int) -> str:
"""Return the dotted version string for the given papal ordinal.
Raises:
ValueError: if the ordinal is outside the known list.
"""
if not 1 <= ordinal <= len(POPES):
message = f"no pope with ordinal {ordinal}; the list has {len(POPES)}"
raise ValueError(message)
major = max(index for index, start in enumerate(MAJOR_STARTS) if ordinal >= start)
offset = ordinal - MAJOR_STARTS[major]
return f"{major}.{offset // PATCHES_PER_MINOR}.{offset % PATCHES_PER_MINOR}"
def pope_for(ordinal: int) -> Pope:
"""Return the pope with the given ordinal.
Raises:
ValueError: if the ordinal is outside the known list.
"""
if not 1 <= ordinal <= len(POPES):
message = f"no pope with ordinal {ordinal}; the list has {len(POPES)}"
raise ValueError(message)
return POPES[ordinal - 1]
def ordinal_for(version: str) -> int:
"""Return the papal ordinal a version string denotes.
Raises:
ValueError: if no release carries that version.
"""
for pope in POPES:
if version_for(pope.ordinal) == version:
return pope.ordinal
message = f"no release carries version {version!r}"
raise ValueError(message)
__version__: Final = version_for(RELEASE_ORDINAL)
POPE: Final = pope_for(RELEASE_ORDINAL)

140
tests/test_announce.py Normal file
View File

@ -0,0 +1,140 @@
# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence.
"""Upgrade announcements: told once, never on the way in, never in a loop."""
from __future__ import annotations
from typing import TYPE_CHECKING
from signal_bot.announce import ANNOUNCED_KEY, announce_all, announce_to, render
from signal_bot.changelog import parse
from signal_bot.transport_fake import FakeTransport
if TYPE_CHECKING:
from signal_bot.store import Store
GROUP = "group.one"
DOCUMENT = """\
## 0.0.1 — Linus
- druga zmiana
## 0.0.0 — Święty Piotr Apostoł
- pierwsza zmiana
"""
RELEASES = parse(DOCUMENT)
async def _announce(store: Store, transport: FakeTransport, version: str = "0.0.1") -> None:
await announce_to(transport, store, GROUP, RELEASES, version, "Linus")
def test_a_single_release_is_rendered_flat() -> None:
"""The header already names it; a per-release block would just repeat it."""
message = render("0.0.1", "Linus", RELEASES[1:])
assert message == "Wersja 0.0.1 «Linus» wydana\nChangelog:\n- druga zmiana"
def test_a_long_backlog_is_capped_and_says_so() -> None:
"""A group added years from now must not be handed 200 releases."""
document = "\n\n".join(
f"## {v} — Pope{i}\n\n- zmiana {i}"
for i, v in enumerate(
[
"0.0.0",
"0.0.1",
"0.0.2",
"0.0.3",
"0.0.4",
"0.0.5",
"0.0.6",
"0.0.7",
"0.1.0",
"0.1.1",
"0.1.2",
"0.1.3",
]
)
)
message = render("0.1.3", "Soter", parse(document))
assert "(pominieto 2 starszych wersji)" in message
assert "zmiana 0" not in message
assert "zmiana 11" in message
def test_render_names_the_version_and_lists_every_pending_line() -> None:
message = render("0.0.1", "Linus", RELEASES)
assert message.startswith("Wersja 0.0.1 «Linus» wydana\nChangelog:")
assert "- pierwsza zmiana" in message
assert "- druga zmiana" in message
async def test_a_group_told_nothing_yet_hears_the_ladder_from_peter(
store: Store, transport: FakeTransport
) -> None:
"""kuhy asked for this explicitly: the joke only lands if Peter is posted."""
await _announce(store, transport)
assert len(transport.sent) == 1
message = transport.sent[0].text or ""
assert "0.0.0 «Święty Piotr Apostoł»" in message
assert "pierwsza zmiana" in message
assert "druga zmiana" in message
assert store.meta(GROUP, ANNOUNCED_KEY) == "0.0.1"
async def test_a_group_already_on_this_version_hears_nothing(
store: Store, transport: FakeTransport
) -> None:
store.set_meta(GROUP, ANNOUNCED_KEY, "0.0.1")
await _announce(store, transport)
assert transport.sent == []
async def test_an_upgraded_group_is_told_what_changed(
store: Store, transport: FakeTransport
) -> None:
store.set_meta(GROUP, ANNOUNCED_KEY, "0.0.0")
await _announce(store, transport)
assert len(transport.sent) == 1
assert "Wersja 0.0.1 «Linus» wydana" in (transport.sent[0].text or "")
assert "druga zmiana" in (transport.sent[0].text or "")
assert "pierwsza zmiana" not in (transport.sent[0].text or "")
assert store.meta(GROUP, ANNOUNCED_KEY) == "0.0.1"
async def test_a_version_with_no_changelog_entry_records_but_says_nothing(
store: Store, transport: FakeTransport
) -> None:
"""An empty announcement is worse than none."""
store.set_meta(GROUP, ANNOUNCED_KEY, "0.0.1")
await _announce(store, transport, version="0.0.2")
assert transport.sent == []
assert store.meta(GROUP, ANNOUNCED_KEY) == "0.0.2"
async def test_the_version_is_recorded_before_the_send(store: Store) -> None:
"""A crash between the two must not re-announce on every restart."""
class Exploding(FakeTransport):
async def send_text(self, text: str, recipient: str) -> None:
raise RuntimeError(text, recipient)
store.set_meta(GROUP, ANNOUNCED_KEY, "0.0.0")
await announce_all(Exploding(), store, [GROUP], RELEASES, "0.0.1", "Linus")
assert store.meta(GROUP, ANNOUNCED_KEY) == "0.0.1"
async def test_one_failing_group_does_not_silence_the_others(store: Store) -> None:
other = "group.two"
class RefusingFirst(FakeTransport):
async def send_text(self, text: str, recipient: str) -> None:
if recipient == GROUP:
raise RuntimeError(recipient)
await super().send_text(text, recipient)
for group in (GROUP, other):
store.set_meta(group, ANNOUNCED_KEY, "0.0.0")
transport = RefusingFirst()
await announce_all(transport, store, [GROUP, other], RELEASES, "0.0.1", "Linus")
assert [message.recipient for message in transport.sent] == [other]

10
tests/test_app.py
View File

@ -19,6 +19,7 @@ from signal_bot.membership import MembershipError
from signal_bot.transport_fake import FakeTransport
from signal_bot.transport_rest import TransportError
from signal_bot.trust import InviteTrust
from signal_bot.versioning import __version__
from tests.conftest import SECOND_GROUP, WATCHED_GROUP, envelope, group_entry, listing_client
if TYPE_CHECKING:
@ -121,8 +122,9 @@ async def test_runtime_start_and_stop_cancels_tasks(
client = _client(lambda r: httpx.Response(200))
runtime = Runtime(bot=_bot(config, store, transport, directory), client=client, store=store)
runtime.start()
# listen, the daily report, the group refresh loop, the deafness gate.
assert len(runtime.tasks) == 4
# listen, the daily report, the group refresh loop, the deafness gate,
# and the one-shot release announcement.
assert len(runtime.tasks) == 5
await asyncio.sleep(0)
await runtime.stop()
assert all(task.done() for task in runtime.tasks)
@ -138,7 +140,9 @@ def test_healthz_serves_and_startup_completes(monkeypatch: pytest.MonkeyPatch) -
monkeypatch.setattr(app_module.httpx, "AsyncClient", fake_client)
with TestClient(app_module.app) as client:
assert client.get("/healthz").json() == {"status": "ok"}
body = client.get("/healthz").json()
assert body["status"] == "ok"
assert body["version"] == __version__
async def test_lifespan_is_an_async_context_manager(

73
tests/test_changelog.py Normal file
View File

@ -0,0 +1,73 @@
# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence.
"""DOCS-changelog.md is the only copy, so the parser has to earn that."""
from __future__ import annotations
from pathlib import Path
from signal_bot.changelog import Release, load, parse, since
from signal_bot.versioning import __version__
DOCUMENT = """\
# Changelog
Prose between releases is ignored.
- a bullet before any heading belongs to no release
## 0.0.1 — Linus
- druga zmiana
- trzecia zmiana
Some more prose.
## 0.0.0 — Święty Piotr Apostoł
- pierwsza zmiana
"""
def test_parse_reads_headings_and_their_bullets() -> None:
releases = parse(DOCUMENT)
assert [r.version for r in releases] == ["0.0.0", "0.0.1"]
assert releases[0].pope == "Święty Piotr Apostoł"
assert releases[1].lines == ("druga zmiana", "trzecia zmiana")
def test_a_bullet_above_every_heading_belongs_to_nobody() -> None:
"""Otherwise the file's own format documentation becomes a changelog entry."""
assert all("belongs to no release" not in line for r in parse(DOCUMENT) for line in r.lines)
def test_parse_of_nothing_is_nothing() -> None:
assert parse("# Changelog\n\njust prose\n") == ()
def test_releases_come_back_oldest_first_whatever_the_file_order() -> None:
assert [r.ordinal for r in parse(DOCUMENT)] == [1, 2]
def test_since_returns_only_what_the_group_has_not_seen() -> None:
releases = parse(DOCUMENT)
assert [r.version for r in since(releases, "0.0.0")] == ["0.0.1"]
assert since(releases, "0.0.1") == ()
def test_since_an_unknown_version_announces_nothing() -> None:
"""A hand-edited or downgraded record must not dump the whole backlog."""
assert since(parse(DOCUMENT), "9.9.9") == ()
def test_a_missing_changelog_costs_an_announcement_not_the_process() -> None:
assert load(Path("/nonexistent/DOCS-changelog.md")) == ()
def test_the_real_changelog_covers_the_running_version() -> None:
"""A release with no entry would announce a version and then say nothing."""
versions = [release.version for release in load()]
assert __version__ in versions
def test_release_knows_its_ordinal() -> None:
assert Release("2.13.7", "Jan Paweł II", ()).ordinal == 264

20
tests/test_healthz.py
View File

@ -13,6 +13,7 @@ from fastapi import Response
from signal_bot import app as app_module
from signal_bot.app import healthz
from signal_bot.liveness import Liveness
from signal_bot.versioning import POPE, __version__
if TYPE_CHECKING:
from signal_bot.config import Config
@ -34,7 +35,8 @@ def _install(liveness: Liveness) -> None:
async def test_a_healthy_bot_answers_ok(config: Config) -> None:
_install(_gate(config, httpx.Response(200, json=[])))
response = Response()
assert await healthz(response) == {"status": "ok"}
body = await healthz(response)
assert body["status"] == "ok"
assert response.status_code == httpx.codes.OK
@ -50,3 +52,19 @@ async def test_a_deaf_bot_answers_503_and_says_why(config: Config) -> None:
assert response.status_code == httpx.codes.SERVICE_UNAVAILABLE
assert body["status"] == "deaf"
assert "Closed unexpectedly" in body["reason"]
async def test_healthz_names_the_build_either_way(config: Config) -> None:
"""Without this you cannot ask a running bot which version it is."""
_install(_gate(config, httpx.Response(200, json=[])))
healthy = await healthz(Response())
assert healthy["version"] == __version__
assert healthy["pope"] == POPE.name
deaf_gate = _gate(replace(config, silence_seconds=0), httpx.Response(400, json=DEAD_BODY))
for _ in range(3):
await deaf_gate.check()
_install(deaf_gate)
deaf = await healthz(Response())
assert deaf["version"] == __version__
assert deaf["pope"] == POPE.name

3
tests/test_store.py
View File

@ -6,7 +6,8 @@ from __future__ import annotations
import sqlite3
from typing import TYPE_CHECKING
from signal_bot.store import SCHEMA_VERSION, Store, open_store
from signal_bot.schema import SCHEMA_VERSION
from signal_bot.store import Store, open_store
if TYPE_CHECKING:
from pathlib import Path

94
tests/test_versioning.py Normal file
View File

@ -0,0 +1,94 @@
# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence.
"""The pope ladder, and the one fixed point the whole scheme exists for."""
from __future__ import annotations
import itertools
import json
import tomllib
from pathlib import Path
import pytest
from signal_bot.versioning import (
POPE,
POPES,
RELEASE_ORDINAL,
__version__,
ordinal_for,
pope_for,
version_for,
)
REPO_ROOT = Path(__file__).resolve().parent.parent
JOHN_PAUL_II = 264
def test_john_paul_ii_is_exactly_2_13_7() -> None:
"""The requirement the two ladder constants were solved for."""
assert version_for(JOHN_PAUL_II) == "2.13.7"
assert pope_for(JOHN_PAUL_II).name == "Jan Paweł II"
@pytest.mark.parametrize(
("ordinal", "version", "name"),
[
(1, "0.0.0", "Święty Piotr Apostoł"),
(2, "0.0.1", "Linus"),
(32, "0.3.7", "Milcjades"),
(33, "1.0.0", "Sylwester I"),
(152, "1.14.7", "Leon IX"),
(153, "2.0.0", "Wiktor II"),
(265, "2.14.0", "Benedykt XVI"),
(267, "2.14.2", "Leon XIV"),
],
)
def test_the_anchors_and_the_boundaries_either_side(ordinal: int, version: str, name: str) -> None:
assert version_for(ordinal) == version
assert pope_for(ordinal).name == name
def test_versions_never_go_backwards() -> None:
"""A ladder that is not monotonic is not a version scheme."""
ladder = [
tuple(int(part) for part in version_for(n).split(".")) for n in range(1, len(POPES) + 1)
]
assert all(earlier < later for earlier, later in itertools.pairwise(ladder))
@pytest.mark.parametrize("ordinal", [0, -1, 268, 9999])
def test_an_ordinal_off_the_list_is_refused(ordinal: int) -> None:
with pytest.raises(ValueError, match="no pope with ordinal"):
version_for(ordinal)
with pytest.raises(ValueError, match="no pope with ordinal"):
pope_for(ordinal)
def test_ordinal_for_is_the_inverse_of_version_for() -> None:
for ordinal in (1, 2, 33, 153, JOHN_PAUL_II, len(POPES)):
assert ordinal_for(version_for(ordinal)) == ordinal
def test_a_version_no_release_carries_is_refused() -> None:
with pytest.raises(ValueError, match="no release carries version"):
ordinal_for("9.9.9")
def test_this_build_names_itself() -> None:
assert __version__ == version_for(RELEASE_ORDINAL)
assert pope_for(RELEASE_ORDINAL) == POPE
def test_pyproject_agrees_with_the_ladder() -> None:
"""Two version sources with nothing reconciling them would drift silently."""
pyproject = tomllib.loads((REPO_ROOT / "pyproject.toml").read_text(encoding="utf-8"))
assert pyproject["project"]["version"] == version_for(RELEASE_ORDINAL)
def test_the_committed_pope_list_is_whole() -> None:
"""popes.json is generated; this is what stops a bad regeneration shipping."""
document = json.loads((REPO_ROOT / "signal_bot" / "popes.json").read_text(encoding="utf-8"))
entries = document["popes"]
assert len(entries) == 267
assert [entry["ordinal"] for entry in entries] == list(range(1, 268))
assert all(entry["name"] and entry["portrait"] and entry["reign"] for entry in entries)

120
tool/gen_popes.py Executable file
View File

@ -0,0 +1,120 @@
#!/usr/bin/env python3
# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence.
"""Regenerate ``signal_bot/popes.json`` from the Polish Wikipedia pope list.
Run this by hand when a new pope is elected; the output is committed, so the
bot never touches the network to learn its own version name.
The ``Lp.`` column of *Poczet papieży* is the authority for the ordinal. It is
the reason this is scraped rather than recalled: antipopes, Benedykt IX's three
separate reigns and the Stefan II question all shift the count, and being wrong
by one silently moves Jan Paweł II off 2.13.7.
"""
from __future__ import annotations
import json
import re
import sys
from pathlib import Path
from typing import Final
import httpx
API: Final = (
"https://pl.wikipedia.org/w/api.php?action=parse"
"&page=Poczet%20papie%C5%BCy&prop=wikitext&format=json&formatversion=2"
)
# Wikimedia's policy requires a descriptive agent with a contact URL; the
# generic client default is rate-limited hard and may be blocked outright.
USER_AGENT: Final = "signal-bot popes generator (https://github.com/kuhyx/signal-bot)"
# One row per pope: "|<n>." (later rows add a second, per-era numbering that
# is deliberately ignored), then the linked name, then the portrait file.
_ROW: Final = re.compile(r"\|(\d+)\.(?: */ *'''\d+\.''')?\s*\n\|\[?\[?([^\]|\n]+)")
_PORTRAIT: Final = re.compile(r"\[\[Plik:([^\]|]+)")
# A year is a wiki link whose whole target is digits. Matching bare numbers
# instead catches "100px" from an image and the "31" of "31 stycznia".
_YEARS: Final = re.compile(r"\[\[(\d{1,4})\]\]")
# The list spells the first entry "Piotr"; kuhy asked for his full title.
FIRST_POPE: Final = "Święty Piotr Apostoł"
EXPECTED_TOTAL: Final = 267
JOHN_PAUL_II: Final = 264
def _clean(name: str) -> str:
"""Strip the wiki disambiguation suffix from a linked page title."""
return re.sub(r"\s*\(papież\)", "", name).strip()
def _reign(row: str) -> str:
"""Return a row's reign years as ``start-end``, or ``start-`` if reigning.
The reign is the row's last cell -- but a row that ends a wikitable runs
on into the prose after ``|}``, which is how Jan Pawel II first came out
as "100" (an image width) and Leon XIV as "2005-069" (a footnote).
"""
cell = row.split("\n|}", maxsplit=1)[0].rsplit("\n|", 1)[-1]
years = _YEARS.findall(cell)
if not years:
return ""
if len(years) == 1:
return f"{years[0]}-" if cell.rstrip().endswith(("-", "\u2013")) else years[0]
return f"{years[0]}-{years[-1]}"
def parse(wikitext: str) -> list[dict[str, object]]:
"""Return one entry per pope, ordered by the list's own Lp. column."""
popes: dict[int, dict[str, object]] = {}
for row in re.split(r"\n\|-", wikitext):
match = _ROW.match(row.lstrip("\n"))
portrait = _PORTRAIT.search(row)
if match is None or portrait is None:
continue
ordinal = int(match.group(1))
popes.setdefault(
ordinal,
{
"ordinal": ordinal,
"name": _clean(match.group(2)),
"reign": _reign(row),
"portrait": portrait.group(1).strip(),
},
)
return [popes[n] for n in sorted(popes)]
def _rejection(popes: list[dict[str, object]]) -> str:
"""Return why this list must not be written, or an empty string."""
if [p["ordinal"] for p in popes] != list(range(1, len(popes) + 1)):
return "ordinals are not contiguous from 1"
if len(popes) != EXPECTED_TOTAL:
return f"got {len(popes)} popes, expected {EXPECTED_TOTAL}"
if popes[JOHN_PAUL_II - 1]["name"] != "Jan Paweł II":
return f"#{JOHN_PAUL_II} is not Jan Paweł II"
return ""
def main() -> int:
"""Write popes.json, refusing to write anything that fails its own checks."""
response = httpx.get(API, headers={"User-Agent": USER_AGENT}, timeout=60.0)
response.raise_for_status()
popes = parse(str(response.json()["parse"]["wikitext"]))
if popes:
popes[0]["name"] = FIRST_POPE
rejection = _rejection(popes)
if rejection:
sys.stderr.write(f"refusing to write: {rejection}\n")
return 1
out = Path(__file__).resolve().parent.parent / "signal_bot" / "popes.json"
document = {"source": API, "popes": popes}
out.write_text(json.dumps(document, ensure_ascii=False, indent=1) + "\n", encoding="utf-8")
sys.stdout.write(f"wrote {len(popes)} popes to {out}\n")
return 0
if __name__ == "__main__":
raise SystemExit(main())