73 lines
1.8 KiB
Python
73 lines
1.8 KiB
Python
import asyncio
|
|
import os
|
|
import tempfile
|
|
import pytest
|
|
from log_watcher import LogWatcher
|
|
|
|
pytestmark = pytest.mark.asyncio
|
|
|
|
|
|
async def test_watcher_receives_new_lines():
|
|
with tempfile.NamedTemporaryFile(mode='w', suffix='.log', delete=False) as f:
|
|
path = f.name
|
|
try:
|
|
received = []
|
|
|
|
async def on_line(line):
|
|
received.append(line.rstrip('\n'))
|
|
|
|
watcher = LogWatcher(path, on_line)
|
|
watcher.start()
|
|
await asyncio.sleep(0.05) # let watcher open and seek to end
|
|
|
|
with open(path, 'a') as f:
|
|
f.write("line one\n")
|
|
f.write("line two\n")
|
|
|
|
await asyncio.sleep(0.3) # wait for poll cycles
|
|
watcher.stop()
|
|
|
|
assert received == ["line one", "line two"]
|
|
finally:
|
|
os.unlink(path)
|
|
|
|
|
|
async def test_watcher_handles_missing_file_then_appears():
|
|
path = tempfile.mktemp(suffix='.log')
|
|
received = []
|
|
|
|
async def on_line(line):
|
|
received.append(line.rstrip('\n'))
|
|
|
|
watcher = LogWatcher(path, on_line)
|
|
watcher.start()
|
|
await asyncio.sleep(0.15) # file doesn't exist yet
|
|
|
|
with open(path, 'w') as f:
|
|
f.write("hello\n")
|
|
|
|
await asyncio.sleep(0.3)
|
|
watcher.stop()
|
|
|
|
# File appeared after watcher started — lines written before watcher opens
|
|
# will be consumed. This test just checks it doesn't crash.
|
|
assert isinstance(received, list)
|
|
|
|
if os.path.exists(path):
|
|
os.unlink(path)
|
|
|
|
|
|
async def test_watcher_calls_on_offline_after_timeout():
|
|
path = tempfile.mktemp(suffix='.log') # never created
|
|
offline_called = []
|
|
|
|
async def on_offline():
|
|
offline_called.append(True)
|
|
|
|
watcher = LogWatcher(path, lambda _: None, on_offline=on_offline, offline_timeout=0.2)
|
|
watcher.start()
|
|
await asyncio.sleep(0.5)
|
|
watcher.stop()
|
|
|
|
assert offline_called, "on_offline should have been called"
|