112 lines
3.5 KiB
Python
112 lines
3.5 KiB
Python
"""A target the account cannot post to must not spin the same post through the queue forever."""
|
|
import asyncio
|
|
import sys
|
|
|
|
sys.path.insert(0, "/app")
|
|
|
|
from telethon.errors import ChatAdminRequiredError
|
|
from db.models import TargetChannel
|
|
from services.publisher import PublisherService
|
|
|
|
|
|
class FakeQueue:
|
|
def __init__(self, payloads):
|
|
self.items = list(payloads)
|
|
self.pushes = 0
|
|
|
|
async def get_target_queue_size(self, target_id):
|
|
return len(self.items)
|
|
|
|
async def pop_target_post(self, target_id):
|
|
return self.items.pop(0) if self.items else None
|
|
|
|
async def push_target_post(self, target_id, payload):
|
|
self.pushes += 1
|
|
self.items.append(payload)
|
|
|
|
|
|
class FakeRepo:
|
|
def __init__(self, targets):
|
|
self._targets = targets
|
|
self.published = []
|
|
|
|
async def get_active_targets(self):
|
|
return self._targets
|
|
|
|
async def record_post_published_to_target(self, *a):
|
|
self.published.append(a)
|
|
|
|
async def update_target_last_post(self, *a):
|
|
pass
|
|
|
|
|
|
def make_target():
|
|
return TargetChannel(id=2, channel_id=-1003848540849, title="newsjoker", username=None,
|
|
post_interval_min=1, last_post_time=None)
|
|
|
|
|
|
class FailingClient:
|
|
def __init__(self, error):
|
|
self.error = error
|
|
self.attempts = 0
|
|
|
|
def is_connected(self):
|
|
return True
|
|
|
|
async def send_message(self, *a, **kw):
|
|
self.attempts += 1
|
|
raise self.error
|
|
|
|
async def send_file(self, *a, **kw):
|
|
self.attempts += 1
|
|
raise self.error
|
|
|
|
|
|
async def drive(error, cycles=3):
|
|
queue = FakeQueue([{"post_id": 117, "text": "hello", "media_path": None}])
|
|
repo = FakeRepo([make_target()])
|
|
notes = []
|
|
|
|
async def notify(text):
|
|
notes.append(text)
|
|
|
|
pub = PublisherService(repo=repo, queue=queue, client=FailingClient(error), notify_fn=notify)
|
|
for _ in range(cycles):
|
|
await pub._process_all_target_queues()
|
|
return queue, notes
|
|
|
|
|
|
async def _cleanup_error_rows():
|
|
"""log_exception writes to the real table even from fakes; remove those rows."""
|
|
from db.database import init_db, get_db_pool
|
|
await init_db()
|
|
pool = await get_db_pool()
|
|
async with pool.acquire() as conn:
|
|
await conn.execute(
|
|
"DELETE FROM error_logs WHERE service_name = 'publisher.publish' "
|
|
"AND context->>'target_id' = '2' AND context->>'post_id' = '117';")
|
|
|
|
|
|
async def run_tests():
|
|
# Permanent: drained after the first attempt, admins told exactly once.
|
|
queue, notes = await drive(ChatAdminRequiredError(request=None))
|
|
assert queue.pushes == 0, f"permanent failure must not requeue, got {queue.pushes} pushes"
|
|
assert len(queue.items) == 0, f"queue should be drained, still holds {queue.items}"
|
|
assert len(notes) == 1, f"admins should be warned exactly once, got {len(notes)}"
|
|
assert "newsjoker" in notes[0] and "-1003848540849" in notes[0]
|
|
|
|
# Transient: the post survives every failed cycle.
|
|
queue, notes = await drive(ConnectionError("network blip"))
|
|
assert queue.pushes == 3, f"transient failure must requeue each time, got {queue.pushes}"
|
|
assert len(queue.items) == 1, f"post must still be queued, got {queue.items}"
|
|
assert notes == [], "a transient blip must not raise a permanent-failure alarm"
|
|
|
|
await _cleanup_error_rows()
|
|
from db.database import close_db_pool
|
|
await close_db_pool()
|
|
print("All publisher failure-handling tests passed successfully!")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(run_tests())
|