pyr0ball
0b3d95cd26
All five parsers (plex, syslog, servarr, qbittorrent, plaintext) were using .replace(tzinfo=timezone.utc) on naive datetimes parsed from log files, which slaps a UTC label on what is actually local-time data. On a UTC-7 system a 2pm entry was stored as 14:00Z instead of 21:00Z, causing time-window searches to return zero results. Fix: use .astimezone(timezone.utc) instead, which treats the naive datetime as local time and converts correctly. Tests updated to round-trip back to local time for assertion so they pass on any timezone, not just UTC.
167 lines
6.5 KiB
Python
167 lines
6.5 KiB
Python
"""Tests for the qBittorrent log ingestor."""
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
|
|
from app.ingest.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)
|