136 lines
5.2 KiB
Python
136 lines
5.2 KiB
Python
import os
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path, PurePosixPath
|
|
from unittest.mock import patch
|
|
|
|
from archive_clients.config import ClientConfig, ConfigError, RootMapping
|
|
|
|
|
|
class ConfigTests(unittest.TestCase):
|
|
def test_strict_config_mode_override_secrets_and_mapping(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
for name in ("token", "qb-password", "syncthing-key"):
|
|
path = root / name
|
|
path.write_text(name, encoding="utf-8")
|
|
os.chmod(path, 0o600)
|
|
(root / "qb").mkdir()
|
|
(root / "sync").mkdir()
|
|
config_path = root / "client.toml"
|
|
config_path.write_text(
|
|
_config(root, role="cache").replace(
|
|
'username = "admin"', 'username = "${QB_USER}"'
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
with patch.dict(os.environ, {"QB_USER": "admin"}):
|
|
config = ClientConfig.load(config_path, "archive")
|
|
self.assertEqual(config.role, "archive")
|
|
self.assertEqual(config.qbittorrent.username, "admin")
|
|
self.assertEqual(config.read_shared_token(), "token")
|
|
self.assertEqual(
|
|
config.qbittorrent.roots.api_to_local("/downloads/a/b"),
|
|
root / "qb/a/b",
|
|
)
|
|
self.assertEqual(config.jobs.stall_after, 30 * 60)
|
|
self.assertEqual(config.jobs.free_space_reserve_bytes, 32 * 1024 * 1024)
|
|
self.assertEqual(
|
|
config.syncthing.advertised_addresses, ("dynamic",)
|
|
)
|
|
|
|
def test_unknown_key_and_unsafe_secret_fail(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
for name in ("token", "qb-password", "syncthing-key"):
|
|
path = root / name
|
|
path.write_text(name, encoding="utf-8")
|
|
os.chmod(path, 0o600)
|
|
config_path = root / "client.toml"
|
|
invalid = _config(root).replace(
|
|
"[qbittorrent]", "typo = true\n[qbittorrent]"
|
|
)
|
|
config_path.write_text(invalid, encoding="utf-8")
|
|
with self.assertRaisesRegex(ConfigError, "unknown root"):
|
|
ClientConfig.load(config_path)
|
|
os.chmod(root / "token", 0o640)
|
|
config_path.write_text(_config(root), encoding="utf-8")
|
|
with self.assertRaisesRegex(ConfigError, "permissions"):
|
|
ClientConfig.load(config_path).read_shared_token()
|
|
|
|
def test_mapping_rejects_escape(self):
|
|
mapping = RootMapping(Path("/api"), Path("/local"))
|
|
with self.assertRaises(ConfigError):
|
|
mapping.api_to_local("/elsewhere/file")
|
|
|
|
def test_mapping_can_override_one_syncthing_folder_locally(self):
|
|
mapping = RootMapping(
|
|
PurePosixPath("/sync"),
|
|
Path("/local/sync"),
|
|
((PurePosixPath("/sync/DownloadsSync"), Path("/local/qb/Sync")),),
|
|
)
|
|
self.assertEqual(
|
|
mapping.api_to_local("/sync/DownloadsSync/job/ready.json"),
|
|
Path("/local/qb/Sync/job/ready.json"),
|
|
)
|
|
self.assertEqual(
|
|
mapping.local_root_for_api("/sync/DownloadsSync"),
|
|
Path("/local/qb/Sync"),
|
|
)
|
|
|
|
def test_endpoint_scheme_and_job_keys_are_strict(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
invalid_endpoint = _config(root).replace(
|
|
'control_endpoint = "ws://control/archive_control"',
|
|
'control_endpoint = "http://control/archive_control"',
|
|
)
|
|
path = root / "client.toml"
|
|
path.write_text(invalid_endpoint, encoding="utf-8")
|
|
with self.assertRaisesRegex(ConfigError, "ws/wss"):
|
|
ClientConfig.load(path)
|
|
path.write_text(
|
|
_config(root) + "\n[jobs]\nunknown = true\n", encoding="utf-8"
|
|
)
|
|
with self.assertRaisesRegex(ConfigError, "unknown jobs"):
|
|
ClientConfig.load(path)
|
|
|
|
def test_state_and_backups_cannot_live_under_data_roots(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
root = Path(directory)
|
|
path = root / "client.toml"
|
|
nested = _config(root).replace(
|
|
f'state_db = "{root / "state.db"}"',
|
|
f'state_db = "{root / "qb/state.db"}"',
|
|
)
|
|
path.write_text(nested, encoding="utf-8")
|
|
with self.assertRaisesRegex(ConfigError, "outside data roots"):
|
|
ClientConfig.load(path)
|
|
|
|
|
|
def _config(root: Path, role: str = "cache") -> str:
|
|
return f'''client_id = "cache-1"
|
|
display_name = "Cache 1"
|
|
role = "{role}"
|
|
control_endpoint = "ws://control/archive_control"
|
|
shared_token_file = "{root / 'token'}"
|
|
state_db = "{root / 'state.db'}"
|
|
backup_dir = "{root / 'backups'}"
|
|
[qbittorrent]
|
|
endpoint = "http://qb"
|
|
username = "admin"
|
|
password_file = "{root / 'qb-password'}"
|
|
api_root = "/downloads"
|
|
local_root = "{root / 'qb'}"
|
|
[syncthing]
|
|
endpoint = "http://syncthing"
|
|
api_key_file = "{root / 'syncthing-key'}"
|
|
api_root = "/sync"
|
|
local_root = "{root / 'sync'}"
|
|
advertised_addresses = ["dynamic"]
|
|
'''
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|