generated from john/python-template
@@ -0,0 +1,34 @@
|
||||
"""Tests for the standalone worker service entrypoint."""
|
||||
|
||||
import pytest
|
||||
|
||||
from transcription import worker_service
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_main_runs_async_worker_once(monkeypatch):
|
||||
calls: list[str] = []
|
||||
|
||||
async def _fake_run() -> None:
|
||||
calls.append("run")
|
||||
|
||||
monkeypatch.setattr(worker_service, "_run", _fake_run)
|
||||
worker_service.main()
|
||||
|
||||
assert calls == ["run"]
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_main_handles_keyboard_interrupt(monkeypatch):
|
||||
calls: list[str] = []
|
||||
|
||||
def _raise_keyboard_interrupt(coro):
|
||||
coro.close()
|
||||
raise KeyboardInterrupt
|
||||
|
||||
monkeypatch.setattr(worker_service.asyncio, "run", _raise_keyboard_interrupt)
|
||||
monkeypatch.setattr(worker_service.logger, "info", lambda _msg: calls.append("logged"))
|
||||
|
||||
worker_service.main()
|
||||
|
||||
assert calls == ["logged"]
|
||||
Reference in New Issue
Block a user