zwitschi
a4c92470b6
feat: implement tests for maintenance functions and dashboard interactions feat: add timezone and country tests for thctime module
89 lines
2.4 KiB
Python
89 lines
2.4 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_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
|