127 lines
3.5 KiB
Python
127 lines
3.5 KiB
Python
from datetime import datetime, timezone
|
|
|
|
import maintenance
|
|
|
|
|
|
class DummyResponse:
|
|
def __init__(self, headers=None, payload=None):
|
|
self.headers = headers or {}
|
|
self._payload = payload
|
|
|
|
def json(self):
|
|
if self._payload is None:
|
|
raise ValueError("no json")
|
|
return self._payload
|
|
|
|
|
|
def test_parse_float():
|
|
assert maintenance.parse_float("1.5") == 1.5
|
|
assert maintenance.parse_float(2) == 2.0
|
|
assert maintenance.parse_float(None) is None
|
|
assert maintenance.parse_float("nope") is None
|
|
|
|
|
|
def test_parse_message_timestamp_and_build_delete_entry():
|
|
msg = {"id": "42", "timestamp": "2026-01-01T10:00:00Z"}
|
|
parsed = maintenance.parse_message_timestamp(msg)
|
|
assert parsed == datetime(2026, 1, 1, 10, 0, 0)
|
|
|
|
entry = maintenance.build_delete_entry(msg)
|
|
assert entry["id"] == "42"
|
|
assert entry["timestamp"] == parsed
|
|
|
|
|
|
def test_should_delete_message():
|
|
ts = int(datetime(2026, 1, 1, 10, 0, 0, tzinfo=timezone.utc).timestamp())
|
|
message = {
|
|
"timestamp": "2026-01-01T10:00:00Z",
|
|
"webhook_id": "w",
|
|
"author": {"id": "a"},
|
|
}
|
|
|
|
assert maintenance.should_delete_message(
|
|
message,
|
|
webhook_id="w",
|
|
author_id="a",
|
|
cutoff=ts,
|
|
)
|
|
assert not maintenance.should_delete_message(
|
|
message,
|
|
webhook_id="x",
|
|
author_id="a",
|
|
cutoff=ts,
|
|
)
|
|
|
|
|
|
def test_message_matches_pattern_content_and_embeds():
|
|
message = {
|
|
"content": "This is a smoke test payload",
|
|
"embeds": [
|
|
{"title": "Reminder", "description": "Half-time in 5 minutes"}
|
|
],
|
|
}
|
|
|
|
assert maintenance.message_matches_pattern(message, r"smoke test")
|
|
assert maintenance.message_matches_pattern(message, r"half-time")
|
|
assert not maintenance.message_matches_pattern(message, r"does-not-match")
|
|
|
|
|
|
def test_should_delete_message_with_content_pattern():
|
|
ts = int(datetime(2026, 1, 1, 10, 0, 0, tzinfo=timezone.utc).timestamp())
|
|
message = {
|
|
"timestamp": "2026-01-01T10:00:00Z",
|
|
"webhook_id": "w",
|
|
"author": {"id": "a"},
|
|
"embeds": [{"description": "This is a test notification."}],
|
|
}
|
|
|
|
assert maintenance.should_delete_message(
|
|
message,
|
|
webhook_id="w",
|
|
author_id="a",
|
|
cutoff=ts,
|
|
content_pattern=r"test notification",
|
|
)
|
|
assert not maintenance.should_delete_message(
|
|
message,
|
|
webhook_id="w",
|
|
author_id="a",
|
|
cutoff=ts,
|
|
content_pattern=r"production-only",
|
|
)
|
|
|
|
|
|
def test_get_rate_limit_retry_after_header_priority():
|
|
response = DummyResponse(
|
|
headers={
|
|
"Retry-After": "2.5",
|
|
"X-RateLimit-Reset-After": "10",
|
|
},
|
|
payload={"retry_after": "20"},
|
|
)
|
|
assert maintenance.get_rate_limit_retry_after(response) == 2.5
|
|
|
|
|
|
def test_get_rate_limit_retry_after_json_fallback():
|
|
response = DummyResponse(payload={"retry_after": "3"})
|
|
assert maintenance.get_rate_limit_retry_after(response) == 3.0
|
|
|
|
|
|
def test_get_bucket_exhausted_delay():
|
|
response = DummyResponse(
|
|
headers={
|
|
"X-RateLimit-Remaining": "0",
|
|
"X-RateLimit-Reset-After": "1.25",
|
|
}
|
|
)
|
|
assert maintenance.get_bucket_exhausted_delay(response) == 1.25
|
|
|
|
response_not_exhausted = DummyResponse(
|
|
headers={
|
|
"X-RateLimit-Remaining": "1",
|
|
"X-RateLimit-Reset-After": "1.25",
|
|
}
|
|
)
|
|
assert maintenance.get_bucket_exhausted_delay(
|
|
response_not_exhausted) is None
|