mirror of
https://github.com/kuhyx/signal-bot.git
synced 2026-11-10 20:37:03 +01:00
Krzysztof kuhy Rudnicki
f92502d18b
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
165 lines
5.8 KiB
Python
165 lines
5.8 KiB
Python
# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence.
|
|
"""SQLite persistence, including survival across a restart."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import sqlite3
|
|
from typing import TYPE_CHECKING
|
|
|
|
from signal_bot.schema import SCHEMA_VERSION
|
|
from signal_bot.store import Store, open_store
|
|
|
|
if TYPE_CHECKING:
|
|
from pathlib import Path
|
|
|
|
GROUP = "group-a=="
|
|
OTHER = "group-b=="
|
|
|
|
|
|
def test_increment_creates_and_accumulates(store: Store) -> None:
|
|
store.increment(GROUP, "u1", "Alice")
|
|
store.increment(GROUP, "u1", "Alice")
|
|
store.increment(GROUP, "u2", "Bob")
|
|
assert [(t.display_name, t.count) for t in store.tallies(GROUP)] == [("Alice", 2), ("Bob", 1)]
|
|
|
|
|
|
def test_display_name_is_refreshed(store: Store) -> None:
|
|
store.increment(GROUP, "u1", "Alice")
|
|
store.increment(GROUP, "u1", "Alicja")
|
|
assert store.tallies(GROUP)[0].display_name == "Alicja"
|
|
|
|
|
|
def test_tallies_sorted_by_count_then_name(store: Store) -> None:
|
|
store.increment(GROUP, "u1", "Zoe")
|
|
store.increment(GROUP, "u2", "Adam")
|
|
assert [t.display_name for t in store.tallies(GROUP)] == ["Adam", "Zoe"]
|
|
|
|
|
|
def test_reset_clears_everything(store: Store) -> None:
|
|
store.increment(GROUP, "u1", "Alice")
|
|
store.reset_tallies(GROUP)
|
|
assert store.tallies(GROUP) == []
|
|
|
|
|
|
def test_tallies_survive_a_restart(tmp_path: Path) -> None:
|
|
"""The old bot kept the day's tally in a module global."""
|
|
db = str(tmp_path / "bot.sqlite3")
|
|
with open_store(db) as first:
|
|
first.increment(GROUP, "u1", "Alice")
|
|
with open_store(db) as second:
|
|
assert second.tallies(GROUP)[0].count == 1
|
|
|
|
|
|
def test_last_used_is_none_before_first_use(store: Store) -> None:
|
|
assert store.last_used(GROUP, "u1", "cat") is None
|
|
|
|
|
|
def test_record_use_tracks_cooldown_and_daily(store: Store) -> None:
|
|
store.record_use(GROUP, "u1", "cat", 100.0, "2026-08-29")
|
|
assert store.last_used(GROUP, "u1", "cat") == 100.0
|
|
assert store.uses_today(GROUP, "u1", "cat", "2026-08-29") == 1
|
|
store.record_use(GROUP, "u1", "cat", 200.0, "2026-08-29")
|
|
assert store.last_used(GROUP, "u1", "cat") == 200.0
|
|
assert store.uses_today(GROUP, "u1", "cat", "2026-08-29") == 2
|
|
|
|
|
|
def test_daily_counter_is_per_day(store: Store) -> None:
|
|
store.record_use(GROUP, "u1", "cat", 100.0, "2026-08-29")
|
|
assert store.uses_today(GROUP, "u1", "cat", "2026-08-30") == 0
|
|
|
|
|
|
def test_uses_today_is_per_user_and_command(store: Store) -> None:
|
|
store.record_use(GROUP, "u1", "cat", 100.0, "2026-08-29")
|
|
assert store.uses_today(GROUP, "u2", "cat", "2026-08-29") == 0
|
|
assert store.uses_today(GROUP, "u1", "dog", "2026-08-29") == 0
|
|
|
|
|
|
def test_votes_are_deduplicated_per_person(store: Store) -> None:
|
|
store.add_vote(GROUP, "kot", "u1", 100.0)
|
|
store.add_vote(GROUP, "kot", "u1", 101.0)
|
|
store.add_vote(GROUP, "kot", "u2", 102.0)
|
|
assert store.count_votes(GROUP, "kot", 0.0) == 2
|
|
|
|
|
|
def test_votes_outside_the_window_are_ignored(store: Store) -> None:
|
|
store.add_vote(GROUP, "cmd", "u1", 100.0)
|
|
assert store.count_votes(GROUP, "cmd", 150.0) == 0
|
|
|
|
|
|
def test_clear_votes(store: Store) -> None:
|
|
store.add_vote(GROUP, "cmd", "u1", 100.0)
|
|
store.clear_votes(GROUP, "cmd")
|
|
assert store.count_votes(GROUP, "cmd", 0.0) == 0
|
|
|
|
|
|
def test_store_is_a_context_manager(tmp_path: Path) -> None:
|
|
with Store(str(tmp_path / "x.sqlite3")) as opened:
|
|
opened.increment(GROUP, "u1", "Alice")
|
|
assert opened.tallies.__self__ is opened
|
|
|
|
|
|
def test_state_is_scoped_per_group(store: Store) -> None:
|
|
"""The whole point of the group key: two groups never share a counter."""
|
|
store.increment(GROUP, "u1", "Alice")
|
|
store.increment(OTHER, "u1", "Alice")
|
|
store.increment(OTHER, "u1", "Alice")
|
|
assert store.tallies(GROUP)[0].count == 1
|
|
assert store.tallies(OTHER)[0].count == 2
|
|
|
|
|
|
def test_reset_clears_only_its_own_group(store: Store) -> None:
|
|
store.increment(GROUP, "u1", "Alice")
|
|
store.increment(OTHER, "u1", "Alice")
|
|
store.reset_tallies(GROUP)
|
|
assert store.tallies(GROUP) == []
|
|
assert store.tallies(OTHER)[0].count == 1
|
|
|
|
|
|
def test_rate_limits_are_scoped_per_group(store: Store) -> None:
|
|
store.record_use(GROUP, "u1", "cat", 100.0, "2026-08-29")
|
|
assert store.last_used(OTHER, "u1", "cat") is None
|
|
assert store.uses_today(OTHER, "u1", "cat", "2026-08-29") == 0
|
|
|
|
|
|
def test_votes_are_scoped_per_group(store: Store) -> None:
|
|
store.add_vote(GROUP, "cmd", "u1", 100.0)
|
|
store.add_vote(OTHER, "cmd", "u2", 100.0)
|
|
store.clear_votes(GROUP, "cmd")
|
|
assert store.count_votes(GROUP, "cmd", 0.0) == 0
|
|
assert store.count_votes(OTHER, "cmd", 0.0) == 1
|
|
|
|
|
|
def test_a_pre_group_database_is_rebuilt(tmp_path: Path) -> None:
|
|
"""CREATE TABLE IF NOT EXISTS is a no-op against the wrong columns.
|
|
|
|
A database written before the group key would survive startup untouched
|
|
and then fail on every INSERT: a bot that boots clean and answers
|
|
nothing. PRAGMA user_version is the only part of the file that knows.
|
|
"""
|
|
db = str(tmp_path / "old.sqlite3")
|
|
legacy = sqlite3.connect(db)
|
|
legacy.execute(
|
|
"CREATE TABLE message_counts (uuid TEXT PRIMARY KEY, "
|
|
"display_name TEXT NOT NULL, count INTEGER NOT NULL DEFAULT 0)"
|
|
)
|
|
legacy.execute("INSERT INTO message_counts VALUES ('u1', 'Alice', 7)")
|
|
legacy.commit()
|
|
legacy.close()
|
|
|
|
with open_store(db) as store:
|
|
store.increment(GROUP, "u1", "Alice")
|
|
assert [(t.display_name, t.count) for t in store.tallies(GROUP)] == [("Alice", 1)]
|
|
|
|
check = sqlite3.connect(db)
|
|
assert int(check.execute("PRAGMA user_version").fetchone()[0]) == SCHEMA_VERSION
|
|
check.close()
|
|
|
|
|
|
def test_a_current_database_is_left_alone(tmp_path: Path) -> None:
|
|
"""Reopening must not wipe the day's tally on every restart."""
|
|
db = str(tmp_path / "current.sqlite3")
|
|
with open_store(db) as first:
|
|
first.increment(GROUP, "u1", "Alice")
|
|
with open_store(db) as second:
|
|
assert second.tallies(GROUP)[0].count == 1
|