130 lines
4.5 KiB
Python
130 lines
4.5 KiB
Python
import io
|
|
import os
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path, PurePosixPath
|
|
from unittest.mock import patch
|
|
|
|
from archive_clients.config import ServiceConfig
|
|
from archive_clients.services import (
|
|
_supported_qb_version,
|
|
probe_qbittorrent,
|
|
probe_syncthing,
|
|
)
|
|
from archive_control.v1 import common_pb2
|
|
|
|
|
|
class _Response:
|
|
def __init__(self, value: str, status: int = 200):
|
|
self._source = io.BytesIO(value.encode("utf-8"))
|
|
self.status = status
|
|
|
|
def read(self, size: int = -1) -> bytes:
|
|
return self._source.read(size)
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, *_):
|
|
return None
|
|
|
|
|
|
class _Opener:
|
|
def __init__(self, responses: list[str]):
|
|
self.responses = iter(responses)
|
|
self.requests = []
|
|
|
|
def open(self, call, timeout):
|
|
self.requests.append((call, timeout))
|
|
response = next(self.responses)
|
|
if isinstance(response, tuple):
|
|
return _Response(*response)
|
|
return _Response(response)
|
|
|
|
|
|
class ServiceProbeTests(unittest.TestCase):
|
|
def test_supported_qbittorrent_versions_include_live_4_4_api(self):
|
|
for version in ("v4.4.5", "v4.5.5", "v4.6.7", "v5.2.3"):
|
|
with self.subTest(version=version):
|
|
self.assertTrue(_supported_qb_version(version))
|
|
for version in ("v4.3.9", "v6.0.0", "invalid"):
|
|
with self.subTest(version=version):
|
|
self.assertFalse(_supported_qb_version(version))
|
|
|
|
def test_qbittorrent_versions_are_normalized(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
password = self._secret(root, "password", "private")
|
|
config = ServiceConfig(
|
|
"http://qb:8080", PurePosixPath("/downloads"), root,
|
|
username="admin", password_file=password,
|
|
)
|
|
opener = _Opener([
|
|
"Ok.", "v5.0.4", "2.11.4", '{"libtorrent":"2.0.11"}',
|
|
])
|
|
with patch(
|
|
"archive_clients.services.request.build_opener",
|
|
return_value=opener,
|
|
):
|
|
result = probe_qbittorrent(config)
|
|
self.assertEqual(result.state, common_pb2.HEALTH_STATE_HEALTHY)
|
|
self.assertEqual(result.version, "v5.0.4")
|
|
self.assertEqual(result.api_version, "2.11.4")
|
|
self.assertEqual(result.libtorrent_version, "2.0.11")
|
|
self.assertIn(b"password=private", opener.requests[0][0].data)
|
|
|
|
def test_qbittorrent_accepts_current_empty_204_login(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
password = self._secret(root, "password", "private")
|
|
config = ServiceConfig(
|
|
"http://qb:8080", PurePosixPath("/downloads"), root,
|
|
username="admin", password_file=password,
|
|
)
|
|
opener = _Opener([
|
|
("", 204), "v5.2.3", "2.11.4",
|
|
'{"libtorrent":"2.0.13"}',
|
|
])
|
|
with patch(
|
|
"archive_clients.services.request.build_opener",
|
|
return_value=opener,
|
|
):
|
|
result = probe_qbittorrent(config)
|
|
self.assertEqual(result.state, common_pb2.HEALTH_STATE_HEALTHY)
|
|
self.assertEqual(result.version, "v5.2.3")
|
|
|
|
def test_syncthing_identity_is_normalized(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
api_key = self._secret(root, "api-key", "private")
|
|
config = ServiceConfig(
|
|
"http://syncthing:8384", PurePosixPath("/sync"), root,
|
|
api_key_file=api_key,
|
|
)
|
|
opener = _Opener([
|
|
'{"version":"v2.0.1","longVersion":"syncthing v2.0.1"}',
|
|
'{"myID":"DEVICE-ID"}',
|
|
'{"folders":[]}',
|
|
])
|
|
with patch(
|
|
"archive_clients.services.request.build_opener",
|
|
return_value=opener,
|
|
):
|
|
result = probe_syncthing(config)
|
|
self.assertEqual(result.state, common_pb2.HEALTH_STATE_HEALTHY)
|
|
self.assertEqual(result.device_id, "DEVICE-ID")
|
|
self.assertEqual(
|
|
opener.requests[0][0].headers["X-api-key"], "private"
|
|
)
|
|
|
|
@staticmethod
|
|
def _secret(root: Path, name: str, value: str) -> Path:
|
|
path = root / name
|
|
path.write_text(value, encoding="utf-8")
|
|
os.chmod(path, 0o600)
|
|
return path
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|