"""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)