Spaces:
Running on Zero
Running on Zero
File size: 5,229 Bytes
2de147f bfd6783 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 | from finbot.telegram import SeenUpdates, chunk_message, parse_update, redact
class TestRedact:
"""Space logs are world-readable on a public Space; tokens must not reach them."""
TOKEN = "8984845302:AAEPEL6yzFjMxJcVKT5e_-D_4n3DGUnofzI"
def test_removes_the_configured_token(self):
text = f"GET https://api.telegram.org/bot{self.TOKEN}/getMe failed"
assert self.TOKEN not in redact(text, self.TOKEN)
def test_removes_a_token_shape_even_when_unknown(self):
# e.g. a stale token still embedded in a cached exception string.
text = "https://api.telegram.org/bot1234567:AAbbCC-_dd/sendMessage"
assert "AAbbCC" not in redact(text, "")
def test_keeps_the_useful_part(self):
out = redact(f"bot{self.TOKEN}/getMe returned 401 Unauthorized", self.TOKEN)
assert "401 Unauthorized" in out
def test_handles_exceptions_and_none(self):
assert redact(RuntimeError("boom"), "") == "boom"
assert "None" in redact(None, "")
def test_no_token_configured_is_safe(self):
assert redact("nothing secret here", "") == "nothing secret here"
def message_update(update_id=1, **message):
base = {"from": {"id": 42}, "chat": {"id": 99}}
base.update(message)
return {"update_id": update_id, "message": base}
class TestParseUpdate:
def test_plain_text(self):
parsed = parse_update(message_update(text="coffee 250"))
assert parsed.user_id == 42
assert parsed.chat_id == 99
assert parsed.text == "coffee 250"
assert parsed.command == ""
def test_command_with_args(self):
parsed = parse_update(message_update(text="/report last_month"))
assert parsed.command == "report"
assert parsed.args == "last_month"
def test_bare_command(self):
parsed = parse_update(message_update(text="/month"))
assert parsed.command == "month"
assert parsed.args == ""
def test_command_is_lowercased(self):
assert parse_update(message_update(text="/MONTH")).command == "month"
def test_group_mention_is_stripped(self):
parsed = parse_update(message_update(text="/month@FinBot"), bot_username="FinBot")
assert parsed.command == "month"
def test_mention_of_another_bot_is_ignored(self):
assert parse_update(message_update(text="/month@OtherBot"), bot_username="FinBot") is None
def test_photo_picks_the_largest_size(self):
parsed = parse_update(
message_update(
photo=[
{"file_id": "small", "file_size": 100},
{"file_id": "large", "file_size": 9000},
],
caption="lunch",
)
)
assert parsed.file_id == "large"
assert parsed.file_kind == "photo"
assert parsed.mime_type == "image/jpeg"
assert parsed.caption == "lunch"
assert parsed.has_file
def test_document(self):
parsed = parse_update(
message_update(
document={
"file_id": "doc1",
"file_name": "statement.pdf",
"mime_type": "application/pdf",
"file_size": 2048,
}
)
)
assert parsed.file_kind == "document"
assert parsed.file_name == "statement.pdf"
assert parsed.file_size == 2048
def test_edited_message_is_handled(self):
update = {"update_id": 5, "edited_message": {"from": {"id": 42}, "chat": {"id": 99}, "text": "x"}}
assert parse_update(update).text == "x"
def test_non_message_updates_are_skipped(self):
assert parse_update({"update_id": 1, "poll": {}}) is None
def test_malformed_update_is_skipped(self):
assert parse_update({"update_id": 1, "message": {"text": "hi"}}) is None
assert parse_update({}) is None
def test_no_file_means_no_file(self):
assert not parse_update(message_update(text="hi")).has_file
class TestChunkMessage:
def test_short_message_is_untouched(self):
assert chunk_message("hello") == ["hello"]
def test_splits_on_line_boundaries(self):
text = "\n".join(f"line {i}" for i in range(500))
chunks = chunk_message(text, limit=200)
assert len(chunks) > 1
assert all(len(c) <= 200 for c in chunks)
# No content lost.
assert "\n".join(chunks) == text
def test_never_splits_mid_line(self):
chunks = chunk_message("aaa\nbbb\nccc", limit=8)
for chunk in chunks:
for line in chunk.split("\n"):
assert line in ("aaa", "bbb", "ccc")
class TestSeenUpdates:
def test_first_time_is_new(self):
seen = SeenUpdates()
assert seen.add_if_new(1) is True
def test_redelivery_is_rejected(self):
seen = SeenUpdates()
seen.add_if_new(1)
assert seen.add_if_new(1) is False
def test_evicts_oldest_beyond_capacity(self):
seen = SeenUpdates(capacity=3)
for i in range(5):
seen.add_if_new(i)
# 0 and 1 were evicted, so they read as new again.
assert seen.add_if_new(0) is True
assert seen.add_if_new(4) is False
|