mirror of
https://github.com/NohamR/prometheus-qbittorrent-exporter.git
synced 2025-05-23 16:49:29 +00:00
291 lines
11 KiB
Python
291 lines
11 KiB
Python
import unittest
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
from prometheus_client.metrics_core import CounterMetricFamily, GaugeMetricFamily
|
|
|
|
from qbittorrent_exporter.exporter import (
|
|
Metric,
|
|
MetricType,
|
|
QbittorrentMetricsCollector,
|
|
)
|
|
from qbittorrentapi import TorrentStates
|
|
|
|
|
|
class TestQbittorrentMetricsCollector(unittest.TestCase):
|
|
@patch("qbittorrent_exporter.exporter.Client")
|
|
def setUp(self, mock_client):
|
|
self.mock_client = mock_client
|
|
self.config = {
|
|
"host": "localhost",
|
|
"port": "8080",
|
|
"username": "user",
|
|
"password": "pass",
|
|
"verify_webui_certificate": False,
|
|
"metrics_prefix": "qbittorrent",
|
|
}
|
|
self.torrentsState = [
|
|
{"name": "Torrent DOWNLOADING 1", "state": TorrentStates.DOWNLOADING},
|
|
{"name": "Torrent UPLOADING 1", "state": TorrentStates.UPLOADING},
|
|
{"name": "Torrent DOWNLOADING 2", "state": TorrentStates.DOWNLOADING},
|
|
{"name": "Torrent UPLOADING 2", "state": TorrentStates.UPLOADING},
|
|
]
|
|
self.torrentsCategories = [
|
|
{"name": "Torrent Movies 1", "category": "Movies"},
|
|
{"name": "Torrent Music 1", "category": "Music"},
|
|
{"name": "Torrent Movies 2", "category": "Movies"},
|
|
{"name": "Torrent unknown", "category": ""},
|
|
{"name": "Torrent Music 2", "category": "Music"},
|
|
{"name": "Torrent Uncategorized 1", "category": "Uncategorized"},
|
|
]
|
|
self.collector = QbittorrentMetricsCollector(self.config)
|
|
|
|
def test_init(self):
|
|
self.assertEqual(self.collector.config, self.config)
|
|
self.mock_client.assert_called_once_with(
|
|
host=self.config["host"],
|
|
port=self.config["port"],
|
|
username=self.config["username"],
|
|
password=self.config["password"],
|
|
VERIFY_WEBUI_CERTIFICATE=self.config["verify_webui_certificate"],
|
|
)
|
|
|
|
def test_collect_gauge(self):
|
|
mock_metric = Metric(
|
|
name="test_gauge",
|
|
metric_type=MetricType.GAUGE,
|
|
help_text="Test Gauge",
|
|
labels={"label1": "value1"},
|
|
value=10,
|
|
)
|
|
self.collector.get_qbittorrent_metrics = MagicMock(return_value=[mock_metric])
|
|
|
|
result = next(self.collector.collect())
|
|
|
|
self.assertIsInstance(result, GaugeMetricFamily)
|
|
self.assertEqual(result.name, "test_gauge")
|
|
self.assertEqual(result.documentation, "Test Gauge")
|
|
self.assertEqual(result.samples[0].labels, {"label1": "value1"})
|
|
self.assertEqual(result.samples[0].value, 10)
|
|
|
|
def test_collect_counter(self):
|
|
mock_metric = Metric(
|
|
name="test_counter",
|
|
metric_type=MetricType.COUNTER,
|
|
help_text="Test Counter",
|
|
labels={"label2": "value2"},
|
|
value=230,
|
|
)
|
|
self.collector.get_qbittorrent_metrics = MagicMock(return_value=[mock_metric])
|
|
|
|
result = next(self.collector.collect())
|
|
|
|
self.assertIsInstance(result, CounterMetricFamily)
|
|
self.assertEqual(result.name, "test_counter")
|
|
self.assertEqual(result.documentation, "Test Counter")
|
|
self.assertEqual(result.samples[0].labels, {"label2": "value2"})
|
|
self.assertEqual(result.samples[0].value, 230)
|
|
|
|
def test_get_qbittorrent_metrics(self):
|
|
metrics = self.collector.get_qbittorrent_metrics()
|
|
self.assertNotEqual(len(metrics), 0)
|
|
|
|
def test_fetch_categories(self):
|
|
# Mock the client.torrent_categories.categories attribute
|
|
self.collector.client.torrent_categories.categories = {
|
|
"category1": {"name": "Category 1"},
|
|
"category2": {"name": "Category 2"},
|
|
"category3": {"name": "Category 3"},
|
|
}
|
|
|
|
categories = self.collector._fetch_categories()
|
|
self.assertIsInstance(categories, dict)
|
|
self.assertNotEqual(len(categories), 0)
|
|
self.assertEqual(categories["category1"]["name"], "Category 1")
|
|
self.assertEqual(categories["category2"]["name"], "Category 2")
|
|
self.assertEqual(categories["category3"]["name"], "Category 3")
|
|
|
|
def test_fetch_categories_exception(self):
|
|
self.collector.client.torrent_categories.categories = Exception(
|
|
"Error fetching categories"
|
|
)
|
|
categories = self.collector._fetch_categories()
|
|
self.assertEqual(categories, {})
|
|
|
|
def test_fetch_torrents_success(self):
|
|
# Mock the return value of self.client.torrents.info()
|
|
self.collector.client.torrents.info.return_value = [
|
|
{"name": "Torrent 1", "size": 100},
|
|
{"name": "Torrent 2", "size": 200},
|
|
{"name": "Torrent 3", "size": 300},
|
|
]
|
|
|
|
expected_result = [
|
|
{"name": "Torrent 1", "size": 100},
|
|
{"name": "Torrent 2", "size": 200},
|
|
{"name": "Torrent 3", "size": 300},
|
|
]
|
|
|
|
result = self.collector._fetch_torrents()
|
|
self.assertEqual(result, expected_result)
|
|
|
|
def test_fetch_torrents_exception(self):
|
|
# Mock an exception being raised by self.client.torrents.info()
|
|
self.collector.client.torrents.info.side_effect = Exception("Connection error")
|
|
|
|
expected_result = []
|
|
|
|
result = self.collector._fetch_torrents()
|
|
self.assertEqual(result, expected_result)
|
|
|
|
def test_filter_torrents_by_state(self):
|
|
expected = [
|
|
{"name": "Torrent DOWNLOADING 1", "state": TorrentStates.DOWNLOADING},
|
|
{"name": "Torrent DOWNLOADING 2", "state": TorrentStates.DOWNLOADING},
|
|
]
|
|
result = self.collector._filter_torrents_by_state(
|
|
TorrentStates.DOWNLOADING, self.torrentsState
|
|
)
|
|
self.assertEqual(result, expected)
|
|
|
|
expected = [
|
|
{"name": "Torrent UPLOADING 1", "state": TorrentStates.UPLOADING},
|
|
{"name": "Torrent UPLOADING 2", "state": TorrentStates.UPLOADING},
|
|
]
|
|
result = self.collector._filter_torrents_by_state(
|
|
TorrentStates.UPLOADING, self.torrentsState
|
|
)
|
|
self.assertEqual(result, expected)
|
|
|
|
expected = []
|
|
result = self.collector._filter_torrents_by_state(
|
|
TorrentStates.ERROR, self.torrentsState
|
|
)
|
|
self.assertEqual(result, expected)
|
|
|
|
def test_filter_torrents_by_category(self):
|
|
expected_result = [
|
|
{"name": "Torrent Movies 1", "category": "Movies"},
|
|
{"name": "Torrent Movies 2", "category": "Movies"},
|
|
]
|
|
result = self.collector._filter_torrents_by_category(
|
|
"Movies", self.torrentsCategories
|
|
)
|
|
self.assertEqual(result, expected_result)
|
|
|
|
expected_result = [
|
|
{"name": "Torrent unknown", "category": ""},
|
|
{"name": "Torrent Uncategorized 1", "category": "Uncategorized"},
|
|
]
|
|
result = self.collector._filter_torrents_by_category(
|
|
"Uncategorized", self.torrentsCategories
|
|
)
|
|
self.assertEqual(result, expected_result)
|
|
|
|
expected_result = []
|
|
result = self.collector._filter_torrents_by_category(
|
|
"Books", self.torrentsCategories
|
|
)
|
|
self.assertEqual(result, expected_result)
|
|
|
|
def test_construct_metric_with_valid_state_and_category(self):
|
|
state = "downloading"
|
|
category = "movies"
|
|
count = 10
|
|
|
|
metric = self.collector._construct_metric(state, category, count)
|
|
|
|
self.assertEqual(metric.name, "qbittorrent_torrents_count")
|
|
self.assertEqual(metric.value, count)
|
|
self.assertEqual(metric.labels["status"], state)
|
|
self.assertEqual(metric.labels["category"], category)
|
|
self.assertEqual(
|
|
metric.help_text,
|
|
f"Number of torrents in status {state} under category {category}",
|
|
)
|
|
|
|
def test_construct_metric_with_empty_state_and_category(self):
|
|
state = ""
|
|
category = ""
|
|
count = 5
|
|
|
|
metric = self.collector._construct_metric(state, category, count)
|
|
|
|
self.assertEqual(metric.name, "qbittorrent_torrents_count")
|
|
self.assertEqual(metric.value, count)
|
|
self.assertEqual(metric.labels["status"], state)
|
|
self.assertEqual(metric.labels["category"], category)
|
|
self.assertEqual(
|
|
metric.help_text, "Number of torrents in status under category "
|
|
)
|
|
|
|
def test_get_qbittorrent_status_metrics(self):
|
|
self.collector.client.transfer.info = {"connection_status": "connected"}
|
|
self.collector.client.app.version = "1.2.3"
|
|
|
|
expected_metrics = [
|
|
Metric(
|
|
name="qbittorrent_up",
|
|
value=True,
|
|
labels={"version": "1.2.3"},
|
|
help_text=(
|
|
"Whether the qBittorrent server is answering requests from this"
|
|
" exporter. A `version` label with the server version is added."
|
|
),
|
|
),
|
|
Metric(
|
|
name="qbittorrent_connected",
|
|
value=True,
|
|
labels={},
|
|
help_text=(
|
|
"Whether the qBittorrent server is connected to the Bittorrent"
|
|
" network."
|
|
),
|
|
),
|
|
Metric(
|
|
name="qbittorrent_firewalled",
|
|
value=False,
|
|
labels={},
|
|
help_text=(
|
|
"Whether the qBittorrent server is connected to the Bittorrent"
|
|
" network but is behind a firewall."
|
|
),
|
|
),
|
|
Metric(
|
|
name="qbittorrent_dht_nodes",
|
|
value=0,
|
|
labels={},
|
|
help_text="Number of DHT nodes connected to.",
|
|
),
|
|
Metric(
|
|
name="qbittorrent_dl_info_data",
|
|
value=0,
|
|
labels={},
|
|
help_text="Data downloaded since the server started, in bytes.",
|
|
metric_type=MetricType.COUNTER,
|
|
),
|
|
Metric(
|
|
name="qbittorrent_up_info_data",
|
|
value=0,
|
|
labels={},
|
|
help_text="Data uploaded since the server started, in bytes.",
|
|
metric_type=MetricType.COUNTER,
|
|
),
|
|
Metric(
|
|
name="qbittorrent_alltime_dl",
|
|
value=0,
|
|
labels={}, # no labels in the example
|
|
help_text="Total data downloaded, in bytes.",
|
|
metric_type=MetricType.COUNTER,
|
|
),
|
|
Metric(
|
|
name="qbittorrent_alltime_ul",
|
|
value=0,
|
|
labels={}, # no labels in the example
|
|
help_text="Total data uploaded, in bytes.",
|
|
metric_type=MetricType.COUNTER,
|
|
),
|
|
]
|
|
|
|
metrics = self.collector._get_qbittorrent_status_metrics()
|
|
self.assertEqual(metrics, expected_metrics)
|