"""Tests for the SSE event system.""" from __future__ import annotations import asyncio import json import sys from pathlib import Path import pytest sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) from ingestion.events import ( CallbackEmitter, FileCompleted, FileFailed, FileProgress, FileSkipped, FileStarted, FileEmpty, IngestionCompleted, IngestionProgress, IngestionStarted, LogEmitter, MultiEmitter, NullEmitter, QueueEmitter, event_to_dict, event_to_sse, ) # --------------------------------------------------------------------------- # Event dataclass tests # --------------------------------------------------------------------------- class TestEventDataclasses: def test_ingestion_started_defaults(self): e = IngestionStarted(total_files=10, skipped_files=[{"path": "a.exe", "reason": "unsupported"}]) assert e.event_type == "ingestion_started" assert e.total_files == 10 assert len(e.skipped_files) == 1 def test_file_started_fields(self): e = FileStarted(file_path="docs/a.pdf", file_index=0, total_files=5, modality="text", filetype="pdf") assert e.event_type == "file_started" assert e.file_path == "docs/a.pdf" def test_file_progress_optional_percent(self): e1 = FileProgress(file_path="video.mp4", detail="Chunk 2/5") assert e1.percent is None e2 = FileProgress(file_path="video.mp4", detail="Chunk 2/5", percent=40.0) assert e2.percent == 40.0 def test_file_completed(self): e = FileCompleted(file_path="a.txt", records_emitted=3, duration_sec=1.5) assert e.event_type == "file_completed" assert e.records_emitted == 3 def test_file_failed(self): e = FileFailed(file_path="bad.pdf", error="corrupt", modality="text") assert e.event_type == "file_failed" def test_file_skipped(self): e = FileSkipped(file_path="a.exe", reason="Unsupported file type: .exe") assert e.event_type == "file_skipped" def test_file_empty(self): e = FileEmpty(file_path="blank.png", modality="image", reason="No text found") assert e.event_type == "file_empty" def test_ingestion_progress(self): e = IngestionProgress(files_completed=3, total_files=10, records_emitted=50, elapsed_sec=12.5, eta_sec=30.0) assert e.event_type == "ingestion_progress" assert e.eta_sec == 30.0 def test_ingestion_completed(self): e = IngestionCompleted(stats={"counts": {"records_emitted": 100}}) assert e.event_type == "ingestion_completed" assert e.stats["counts"]["records_emitted"] == 100 # --------------------------------------------------------------------------- # Serialization tests # --------------------------------------------------------------------------- class TestSerialization: def test_event_to_dict(self): e = FileStarted(file_path="a.txt", file_index=0, total_files=1, modality="text", filetype="txt") d = event_to_dict(e) assert d["event_type"] == "file_started" assert d["file_path"] == "a.txt" assert isinstance(d, dict) def test_event_to_sse_format(self): e = FileCompleted(file_path="a.txt", records_emitted=1, duration_sec=0.5) sse = event_to_sse(e) assert sse.startswith("event: file_completed\n") assert "data: " in sse assert sse.endswith("\n\n") # Data should be valid JSON data_line = sse.split("data: ")[1].rstrip("\n") parsed = json.loads(data_line) assert parsed["records_emitted"] == 1 def test_all_event_types_serializable(self): events = [ IngestionStarted(total_files=1), FileStarted(file_path="a", file_index=0, total_files=1, modality="text", filetype="txt"), FileProgress(file_path="a", detail="working"), FileCompleted(file_path="a", records_emitted=1, duration_sec=0.1), FileFailed(file_path="a", error="oops", modality="text"), FileSkipped(file_path="a", reason="nope"), FileEmpty(file_path="a", modality="text", reason="empty"), IngestionProgress(files_completed=1, total_files=1, records_emitted=1, elapsed_sec=1.0), IngestionCompleted(stats={}), ] for event in events: sse = event_to_sse(event) assert f"event: {event.event_type}" in sse data = json.loads(sse.split("data: ")[1].rstrip("\n")) assert data["event_type"] == event.event_type # --------------------------------------------------------------------------- # Emitter tests # --------------------------------------------------------------------------- class TestNullEmitter: def test_emit_is_noop(self): emitter = NullEmitter() emitter.emit(FileStarted(file_path="a", file_index=0, total_files=1, modality="text", filetype="txt")) # Should not raise class TestLogEmitter: def test_emit_prints(self, capsys): emitter = LogEmitter() emitter.emit(IngestionStarted(total_files=5)) captured = capsys.readouterr() assert "5 files" in captured.out def test_all_event_types(self, capsys): emitter = LogEmitter() events = [ IngestionStarted(total_files=1, skipped_files=[{"path": "a.exe", "reason": "unsupported"}]), FileStarted(file_path="a.txt", file_index=0, total_files=1, modality="text", filetype="txt"), FileProgress(file_path="a.txt", detail="working", percent=50.0), FileCompleted(file_path="a.txt", records_emitted=1, duration_sec=0.1), FileFailed(file_path="b.pdf", error="corrupt", modality="text"), FileSkipped(file_path="c.exe", reason="unsupported"), FileEmpty(file_path="d.png", modality="image", reason="no text"), IngestionProgress(files_completed=1, total_files=1, records_emitted=1, elapsed_sec=1.0, eta_sec=0.0), IngestionCompleted(stats={"counts": {"records_emitted": 1, "files_succeeded": 1}}), ] for event in events: emitter.emit(event) captured = capsys.readouterr() assert "[ingest]" in captured.out class TestCallbackEmitter: def test_calls_callback(self): received = [] emitter = CallbackEmitter(received.append) event = FileStarted(file_path="a", file_index=0, total_files=1, modality="text", filetype="txt") emitter.emit(event) assert len(received) == 1 assert received[0] is event class TestMultiEmitter: def test_broadcasts_to_all(self): r1, r2 = [], [] e1 = CallbackEmitter(r1.append) e2 = CallbackEmitter(r2.append) multi = MultiEmitter(e1, e2) event = FileCompleted(file_path="a", records_emitted=1, duration_sec=0.1) multi.emit(event) assert len(r1) == 1 assert len(r2) == 1 # --------------------------------------------------------------------------- # QueueEmitter tests # --------------------------------------------------------------------------- class TestQueueEmitter: def test_emit_and_stream(self): emitter = QueueEmitter() event = FileCompleted(file_path="a.txt", records_emitted=1, duration_sec=0.1) emitter.emit(event) emitter.close() async def collect(): results = [] async for sse in emitter.stream(): results.append(sse) return results results = asyncio.run(collect()) assert len(results) == 1 assert "file_completed" in results[0] def test_multiple_events(self): emitter = QueueEmitter() emitter.emit(FileStarted(file_path="a", file_index=0, total_files=2, modality="text", filetype="txt")) emitter.emit(FileCompleted(file_path="a", records_emitted=1, duration_sec=0.1)) emitter.close() async def collect(): results = [] async for sse in emitter.stream(): results.append(sse) return results results = asyncio.run(collect()) assert len(results) == 2 def test_close_without_events(self): emitter = QueueEmitter() emitter.close() async def collect(): results = [] async for sse in emitter.stream(): results.append(sse) return results results = asyncio.run(collect()) assert len(results) == 0