mirror of
https://github.com/kuhyx/signal-bot.git
synced 2026-11-10 19:27:20 +01:00
Krzysztof kuhy Rudnicki
0c4b259c17
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
92 lines
3.3 KiB
Python
92 lines
3.3 KiB
Python
# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence.
|
|
"""Invited is not joined: the join attempt and its memory of failures."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import TYPE_CHECKING
|
|
from urllib.parse import quote
|
|
|
|
import httpx
|
|
import pytest
|
|
|
|
from signal_bot.membership import (
|
|
JoinAttempts,
|
|
attempt_join,
|
|
entry_matches,
|
|
find_entry,
|
|
)
|
|
from tests.conftest import SECOND_GROUP, WATCHED_GROUP, WATCHED_INTERNAL, group_entry
|
|
|
|
if TYPE_CHECKING:
|
|
from signal_bot.config import Config
|
|
|
|
|
|
def _client(handler: object) -> httpx.AsyncClient:
|
|
return httpx.AsyncClient(transport=httpx.MockTransport(handler))
|
|
|
|
|
|
def test_entry_matches_either_spelling() -> None:
|
|
"""Envelopes carry internal_id; the listing reports the prefixed form."""
|
|
assert entry_matches({"internal_id": WATCHED_INTERNAL}, WATCHED_GROUP)
|
|
assert entry_matches({"id": WATCHED_GROUP}, WATCHED_GROUP)
|
|
assert not entry_matches({"id": SECOND_GROUP}, WATCHED_GROUP)
|
|
|
|
|
|
def test_find_entry_returns_none_when_absent() -> None:
|
|
assert find_entry([group_entry(SECOND_GROUP)], WATCHED_GROUP) is None
|
|
assert find_entry([group_entry(WATCHED_GROUP)], WATCHED_GROUP) is not None
|
|
|
|
|
|
def test_join_attempts_remembers_and_forgets() -> None:
|
|
"""Without this, an impossible invite is retried every 30s forever."""
|
|
attempts = JoinAttempts()
|
|
assert not attempts.has_failed(WATCHED_GROUP)
|
|
assert attempts.reason(WATCHED_GROUP) == "not attempted"
|
|
attempts.record_failure(WATCHED_GROUP, "no service ID")
|
|
assert attempts.has_failed(WATCHED_GROUP)
|
|
assert attempts.reason(WATCHED_GROUP) == "no service ID"
|
|
attempts.forget(WATCHED_GROUP)
|
|
assert not attempts.has_failed(WATCHED_GROUP)
|
|
|
|
|
|
async def test_attempt_join_percent_encodes_the_group_id(config: Config) -> None:
|
|
"""The internal spelling carries '/' and '+', which are not path-safe."""
|
|
group_id = "Y29uZmlnLWdyb3VwLWlkLTMy"
|
|
seen: list[str] = []
|
|
|
|
def handler(request: httpx.Request) -> httpx.Response:
|
|
seen.append(request.url.raw_path.decode())
|
|
return httpx.Response(201, text="")
|
|
|
|
detail = await attempt_join(_client(handler), config, group_id)
|
|
assert seen == [f"/v1/groups/{config.phone_number}/{quote(group_id, safe='')}/join"]
|
|
assert "201" in detail
|
|
|
|
|
|
async def test_attempt_join_reports_a_refusal_instead_of_raising(config: Config) -> None:
|
|
"""The caller's membership re-check is the authority, not this response."""
|
|
|
|
def handler(request: httpx.Request) -> httpx.Response:
|
|
return httpx.Response(400, text="Cannot find service ID for self")
|
|
|
|
detail = await attempt_join(_client(handler), config, WATCHED_GROUP)
|
|
assert "400" in detail
|
|
assert "Cannot find service ID" in detail
|
|
|
|
|
|
async def test_attempt_join_survives_an_endpoint_that_is_not_there(config: Config) -> None:
|
|
def handler(request: httpx.Request) -> httpx.Response:
|
|
msg = "connection refused"
|
|
raise httpx.ConnectError(msg)
|
|
|
|
assert "join request failed" in await attempt_join(_client(handler), config, WATCHED_GROUP)
|
|
|
|
|
|
@pytest.mark.parametrize("body", ["", "x" * 400])
|
|
async def test_attempt_join_truncates_a_long_body(config: Config, body: str) -> None:
|
|
"""A whole HTML error page in a log line helps nobody."""
|
|
detail = await attempt_join(
|
|
_client(lambda r: httpx.Response(500, text=body)), config, WATCHED_GROUP
|
|
)
|
|
assert len(detail) < 260
|