mirror of
https://github.com/kuhyx/signal-bot.git
synced 2026-11-10 18:07:29 +01:00
Krzysztof kuhy Rudnicki
dc642b8a1f
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
209 lines
7.2 KiB
Python
209 lines
7.2 KiB
Python
# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence.
|
|
"""Upgrade announcements: told once, never on the way in, never in a loop."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import base64
|
|
from typing import TYPE_CHECKING
|
|
|
|
import httpx
|
|
|
|
from signal_bot.announce import ANNOUNCED_KEY, announce_all, announce_to
|
|
from signal_bot.changelog import parse
|
|
from signal_bot.http_client import ImageFetcher
|
|
from signal_bot.portraits import PortraitCache
|
|
from signal_bot.transport_fake import FakeTransport
|
|
from signal_bot.versioning import Pope, pope_for
|
|
|
|
if TYPE_CHECKING:
|
|
from signal_bot.quoting import Quote
|
|
from signal_bot.store import Store
|
|
|
|
GROUP = "group.one"
|
|
DOCUMENT = """\
|
|
## 0.0.1 — Linus
|
|
|
|
- druga zmiana
|
|
|
|
## 0.0.0 — Petrus
|
|
|
|
- pierwsza zmiana
|
|
"""
|
|
RELEASES = parse(DOCUMENT)
|
|
LINUS = pope_for(2)
|
|
|
|
|
|
IMAGE = b"portrait-bytes"
|
|
ENCODED = base64.b64encode(IMAGE).decode("ascii")
|
|
|
|
|
|
def _portraits(*, broken: bool = False, seen: list[str] | None = None) -> PortraitCache:
|
|
"""A real cache over a mocked Commons, the way test_portraits builds one.
|
|
|
|
Broken means Commons answers without the thumbnail the fetcher narrows to,
|
|
which is what an outage or an upstream change looks like from here.
|
|
"""
|
|
|
|
def handler(request: httpx.Request) -> httpx.Response:
|
|
if "commons" not in str(request.url):
|
|
return httpx.Response(200, content=IMAGE)
|
|
if seen is not None:
|
|
seen.append(str(request.url))
|
|
if broken:
|
|
return httpx.Response(200, json={})
|
|
page = {"imageinfo": [{"thumburl": "https://cdn/portrait.jpg"}]}
|
|
return httpx.Response(200, json={"query": {"pages": [page]}})
|
|
|
|
fetcher = ImageFetcher(httpx.AsyncClient(transport=httpx.MockTransport(handler)))
|
|
return PortraitCache(fetcher, "https://commons.example/w/api.php")
|
|
|
|
|
|
async def _announce(
|
|
store: Store,
|
|
transport: FakeTransport,
|
|
pope: Pope = LINUS,
|
|
portraits: PortraitCache | None = None,
|
|
) -> None:
|
|
cache = _portraits() if portraits is None else portraits
|
|
await announce_to(transport, store, GROUP, RELEASES, pope, cache)
|
|
|
|
|
|
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].caption
|
|
assert "0.0.0 «Petrus»" in message
|
|
assert "pierwsza zmiana" in message
|
|
assert "druga zmiana" in message
|
|
assert store.meta(GROUP, ANNOUNCED_KEY) == "0.0.1"
|
|
|
|
|
|
async def test_the_announcement_rides_on_the_pope_portrait(
|
|
store: Store, transport: FakeTransport
|
|
) -> None:
|
|
"""One captioned photo, the same shape !papież posts."""
|
|
seen: list[str] = []
|
|
await _announce(store, transport, portraits=_portraits(seen=seen))
|
|
assert len(transport.sent) == 1
|
|
sent = transport.sent[0]
|
|
assert sent.image_base64 == ENCODED
|
|
assert sent.text is None
|
|
assert sent.caption.startswith("Wersja 0.0.1 «Linus» wydana")
|
|
# The portrait asked for is the header pope's, not some other release's.
|
|
assert f"File%3A{LINUS.portrait}".replace(".", "%2E") in seen[0] or LINUS.portrait in seen[0]
|
|
|
|
|
|
async def test_an_unreachable_portrait_costs_the_photo_not_the_announcement(
|
|
store: Store, transport: FakeTransport
|
|
) -> None:
|
|
"""The marker is written first, so a Commons outage must not eat the notes."""
|
|
await _announce(store, transport, portraits=_portraits(broken=True))
|
|
assert len(transport.sent) == 1
|
|
assert transport.sent[0].image_base64 is None
|
|
assert "Wersja 0.0.1 «Linus» wydana" in (transport.sent[0].text or "")
|
|
assert store.meta(GROUP, ANNOUNCED_KEY) == "0.0.1"
|
|
|
|
|
|
async def test_a_refused_attachment_still_lands_the_notes_as_text(
|
|
store: Store, transport: FakeTransport
|
|
) -> None:
|
|
"""A megabyte of base64 is refusable in ways a 2 KB text post is not."""
|
|
|
|
class RefusingImages(FakeTransport):
|
|
async def send_image(
|
|
self,
|
|
image_base64: str,
|
|
recipient: str,
|
|
quote: Quote | None = None,
|
|
caption: str = "",
|
|
) -> None:
|
|
raise RuntimeError(recipient)
|
|
|
|
refusing = RefusingImages()
|
|
await _announce(store, refusing)
|
|
assert len(refusing.sent) == 1
|
|
assert refusing.sent[0].image_base64 is None
|
|
assert "Wersja 0.0.1 «Linus» wydana" in (refusing.sent[0].text or "")
|
|
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].caption
|
|
assert "druga zmiana" in transport.sent[0].caption
|
|
assert "pierwsza zmiana" not in transport.sent[0].caption
|
|
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, pope=pope_for(3))
|
|
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, quote: Quote | None = None) -> None:
|
|
raise RuntimeError(recipient, text)
|
|
|
|
async def send_image(
|
|
self,
|
|
image_base64: str,
|
|
recipient: str,
|
|
quote: Quote | None = None,
|
|
caption: str = "",
|
|
) -> None:
|
|
raise RuntimeError(recipient, caption)
|
|
|
|
store.set_meta(GROUP, ANNOUNCED_KEY, "0.0.0")
|
|
await announce_all(Exploding(), store, [GROUP], RELEASES, LINUS, _portraits())
|
|
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, quote: Quote | None = None) -> None:
|
|
if recipient == GROUP:
|
|
raise RuntimeError(recipient)
|
|
await super().send_text(text, recipient, quote)
|
|
|
|
async def send_image(
|
|
self,
|
|
image_base64: str,
|
|
recipient: str,
|
|
quote: Quote | None = None,
|
|
caption: str = "",
|
|
) -> None:
|
|
if recipient == GROUP:
|
|
raise RuntimeError(recipient)
|
|
await super().send_image(image_base64, recipient, quote, caption)
|
|
|
|
for group in (GROUP, other):
|
|
store.set_meta(group, ANNOUNCED_KEY, "0.0.0")
|
|
transport = RefusingFirst()
|
|
await announce_all(transport, store, [GROUP, other], RELEASES, LINUS, _portraits())
|
|
assert [message.recipient for message in transport.sent] == [other]
|