wehub-resource-sync
e4dcfc49aa
Tests / Lint and Format (push) Waiting to run
Tests / Web Node Tests (push) Waiting to run
Tests / Import Check (Python 3.11) (push) Waiting to run
Tests / Import Check (Python 3.12) (push) Waiting to run
Tests / Import Check (Python 3.13) (push) Waiting to run
Tests / Import Check (Python 3.14) (push) Waiting to run
Tests / Python Tests (Python 3.11) (push) Blocked by required conditions
Tests / Python Tests (Python 3.12) (push) Blocked by required conditions
Tests / Python Tests (Python 3.13) (push) Blocked by required conditions
Tests / Python Tests (Python 3.14) (push) Blocked by required conditions
Tests / Test Summary (push) Blocked by required conditions
36 lines
1.2 KiB
Python
36 lines
1.2 KiB
Python
import logging
|
|
|
|
from deeptutor.logging import ProcessLogEvent, bind_log_context, capture_process_logs
|
|
|
|
|
|
def test_capture_process_logs_emits_structured_event_for_matching_task():
|
|
events: list[ProcessLogEvent] = []
|
|
logger = logging.getLogger("deeptutor.tests.process")
|
|
|
|
with bind_log_context(task_id="task-1", capability="knowledge", stage="indexing"):
|
|
with capture_process_logs(events.append, task_id="task-1"):
|
|
logger.info("Embedding batches: %s/%s", 2, 8)
|
|
|
|
assert len(events) == 1
|
|
event = events[0].to_dict()
|
|
assert event["type"] == "process_log"
|
|
assert event["level"] == "INFO"
|
|
assert event["message"] == "Embedding batches: 2/8"
|
|
assert event["logger"] == "deeptutor.tests.process"
|
|
assert event["context"] == {
|
|
"task_id": "task-1",
|
|
"capability": "knowledge",
|
|
"stage": "indexing",
|
|
}
|
|
|
|
|
|
def test_capture_process_logs_filters_other_tasks():
|
|
events: list[ProcessLogEvent] = []
|
|
logger = logging.getLogger("deeptutor.tests.process")
|
|
|
|
with capture_process_logs(events.append, task_id="task-1"):
|
|
with bind_log_context(task_id="task-2"):
|
|
logger.warning("wrong task")
|
|
|
|
assert events == []
|