pyr0ball
828b69768a
Renames the app/ingest/ package to app/glean/ and updates all references across Python modules, shell scripts, Vue components, tests, and documentation. Intentionally preserved: - SQLite column name ingest_time (avoids schema migration) - RetrievedEntry.ingest_time field (maps to the column above) - Any public-facing JSON keys that reference ingest_time Changes by category: - app/ingest/ → app/glean/ (full package move, all parsers) - app/tasks/ingest_scheduler.py → app/tasks/glean_scheduler.py - scripts/ingest_corpus.py → scripts/glean_corpus.py - tests/test_ingest_*.py → tests/test_glean_*.py - Docstrings, log messages, comments: ingest → glean - Env var: TURNSTONE_INGEST_INTERVAL → TURNSTONE_GLEAN_INTERVAL - Shell scripts: glean.log, glean_corpus.py references - README.md: multi-source ingest → multi-source glean - .env.example: updated env var name - patterns/: new diagnostic patterns from 2026-05-20 SSH incident (service_crash_loop, pkg_daemon_restart, ssh_forward_conflict) - SourcesView.vue: pipeline label updated - All test import paths updated to app.glean.* 285 tests passing.
167 lines
6.5 KiB
Python
167 lines
6.5 KiB
Python
"""Tests for the qBittorrent log gleaner."""
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
|
|
from app.glean.qbittorrent import is_qbit_log, parse
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Classic format sample (pre-5.x GUI builds)
|
|
# ---------------------------------------------------------------------------
|
|
CLASSIC_LOG = """\
|
|
(2026/05/09 14:10:01) qBittorrent v5.0.3 started
|
|
(2026/05/09 14:10:02) [Warning] Tracker 'http://tracker.example.com/announce' is not working. Reason: Connection timed out
|
|
(2026/05/09 14:10:03) [Critical] Couldn't listen on any of the network interfaces. Aborting!
|
|
(2026/05/09 14:10:04) Download of 'ubuntu-24.04.iso' has finished.
|
|
(2026/05/09 14:10:05) [Warning] Hash check failed for piece 42 of 'ubuntu-24.04.iso'
|
|
(2026/05/09 14:10:06) Some long message
|
|
that continues on the next line
|
|
and a third line
|
|
(2026/05/09 14:10:07) Normal message without bracket level
|
|
"""
|
|
|
|
CLASSIC_DASH = "(2026-05-09 14:10:01) qBittorrent v4.6.2 started\n"
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Hotio format sample (5.x headless container, ghcr.io/hotio/qbittorrent)
|
|
# ---------------------------------------------------------------------------
|
|
HOTIO_LOG = """\
|
|
(N) 2026-04-26T03:32:59 - qBittorrent termination initiated
|
|
(N) 2026-04-26T03:32:59 - Saving resume data completed.
|
|
(N) 2026-04-26T03:33:02 - BitTorrent session successfully finished.
|
|
(N) 2026-04-26T03:33:11 - qBittorrent v5.1.4 started. Process ID: 309
|
|
(I) 2026-04-26T03:33:11 - Peer ID: "-qB5140-"
|
|
(I) 2026-04-26T03:33:11 - HTTP User-Agent: "qBittorrent/5.1.4"
|
|
(W) 2026-04-26T03:33:15 - Tracker is not working. Reason: Connection refused
|
|
(C) 2026-04-26T03:33:20 - Couldn't listen on any of the network interfaces. Aborting!
|
|
(N) 2026-04-26T03:33:25 - Restored torrent. Torrent: "wikipedia_en_all_maxi_2026-02.zim"
|
|
"""
|
|
|
|
|
|
class TestDetector:
|
|
def test_detects_classic_slash_format(self):
|
|
assert is_qbit_log("(2026/05/09 14:10:01) qBittorrent started")
|
|
|
|
def test_detects_classic_dash_format(self):
|
|
assert is_qbit_log(CLASSIC_DASH.strip())
|
|
|
|
def test_detects_hotio_normal(self):
|
|
assert is_qbit_log("(N) 2026-04-26T03:32:59 - qBittorrent termination initiated")
|
|
|
|
def test_detects_hotio_info(self):
|
|
assert is_qbit_log("(I) 2026-04-26T03:33:11 - Peer ID: \"-qB5140-\"")
|
|
|
|
def test_detects_hotio_warning(self):
|
|
assert is_qbit_log("(W) 2026-04-26T03:33:15 - Tracker is not working")
|
|
|
|
def test_detects_hotio_critical(self):
|
|
assert is_qbit_log("(C) 2026-04-26T03:33:20 - Couldn't listen")
|
|
|
|
def test_rejects_plex_format(self):
|
|
assert not is_qbit_log("Jan 01, 2026 12:00:00.000 [12345] DEBUG - message")
|
|
|
|
def test_rejects_journald_json(self):
|
|
assert not is_qbit_log('{"__REALTIME_TIMESTAMP": "12345", "MESSAGE": "hi"}')
|
|
|
|
def test_rejects_plaintext(self):
|
|
assert not is_qbit_log("2026-05-09 14:10:01 some syslog line")
|
|
|
|
def test_rejects_hotio_wrapper_log(self):
|
|
# hotio container wrapper uses [YYYY-MM-DD HH:MM:SS] [INF] [VPN] format — not qbit
|
|
assert not is_qbit_log("[2026-05-11 05:14:38] [INF] [VPN] Script found. Executing...")
|
|
|
|
|
|
class TestClassicParser:
|
|
def _parse(self, text: str) -> list:
|
|
return list(parse(iter(text.splitlines(keepends=True)), "qbit_test", []))
|
|
|
|
def test_entry_count(self):
|
|
assert len(self._parse(CLASSIC_LOG)) == 7
|
|
|
|
def test_startup_entry(self):
|
|
from datetime import datetime
|
|
e = self._parse(CLASSIC_LOG)[0]
|
|
assert "qBittorrent v5.0.3 started" in e.text
|
|
assert e.severity is None
|
|
local = datetime.fromisoformat(e.timestamp_iso).astimezone()
|
|
assert (local.hour, local.minute, local.second) == (14, 10, 1)
|
|
|
|
def test_warning_severity(self):
|
|
e = self._parse(CLASSIC_LOG)[1]
|
|
assert e.severity == "WARN"
|
|
assert "not working" in e.text
|
|
|
|
def test_critical_severity(self):
|
|
e = self._parse(CLASSIC_LOG)[2]
|
|
assert e.severity == "CRITICAL"
|
|
|
|
def test_multiline_continuation(self):
|
|
e = self._parse(CLASSIC_LOG)[5]
|
|
assert "continues on the next line" in e.text
|
|
assert "third line" in e.text
|
|
|
|
def test_no_level_bracket_falls_back_to_detect(self):
|
|
e = self._parse(CLASSIC_LOG)[6]
|
|
assert e.text == "Normal message without bracket level"
|
|
|
|
def test_dash_format_timestamp(self):
|
|
from datetime import datetime
|
|
entries = list(parse(iter(CLASSIC_DASH.splitlines(keepends=True)), "qbit", []))
|
|
local = datetime.fromisoformat(entries[0].timestamp_iso).astimezone()
|
|
assert (local.hour, local.minute, local.second) == (14, 10, 1)
|
|
|
|
def test_source_id_propagated(self):
|
|
assert all(e.source_id == "qbit_test" for e in self._parse(CLASSIC_LOG))
|
|
|
|
def test_sequence_is_monotonic(self):
|
|
entries = self._parse(CLASSIC_LOG)
|
|
seqs = [e.sequence for e in entries]
|
|
assert seqs == sorted(seqs) and len(set(seqs)) == len(seqs)
|
|
|
|
|
|
class TestHotioParser:
|
|
def _parse(self, text: str) -> list:
|
|
return list(parse(iter(text.splitlines(keepends=True)), "qbit_hotio", []))
|
|
|
|
def test_entry_count(self):
|
|
assert len(self._parse(HOTIO_LOG)) == 9
|
|
|
|
def test_normal_maps_to_info(self):
|
|
e = self._parse(HOTIO_LOG)[0]
|
|
assert e.severity == "INFO"
|
|
assert "termination initiated" in e.text
|
|
|
|
def test_info_severity(self):
|
|
e = self._parse(HOTIO_LOG)[4]
|
|
assert e.severity == "INFO"
|
|
assert "Peer ID" in e.text
|
|
|
|
def test_warning_severity(self):
|
|
e = self._parse(HOTIO_LOG)[6]
|
|
assert e.severity == "WARN"
|
|
assert "not working" in e.text
|
|
|
|
def test_critical_severity(self):
|
|
e = self._parse(HOTIO_LOG)[7]
|
|
assert e.severity == "CRITICAL"
|
|
|
|
def test_iso_timestamp_parsed(self):
|
|
from datetime import datetime
|
|
e = self._parse(HOTIO_LOG)[0]
|
|
ts = datetime.fromisoformat(e.timestamp_iso)
|
|
assert ts.utcoffset() is not None # stored as UTC-aware
|
|
local = ts.astimezone()
|
|
assert (local.hour, local.minute, local.second) == (3, 32, 59)
|
|
assert e.timestamp_raw == "2026-04-26T03:32:59"
|
|
|
|
def test_source_id_propagated(self):
|
|
assert all(e.source_id == "qbit_hotio" for e in self._parse(HOTIO_LOG))
|
|
|
|
def test_sequence_is_monotonic(self):
|
|
entries = self._parse(HOTIO_LOG)
|
|
seqs = [e.sequence for e in entries]
|
|
assert seqs == sorted(seqs) and len(set(seqs)) == len(seqs)
|
|
|
|
def test_detector_identifies_hotio_first_line(self):
|
|
first_line = HOTIO_LOG.splitlines()[0]
|
|
assert is_qbit_log(first_line)
|