archive-clients/tests/test_config.py

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_qbittorrent_can_override_a_nested_save_path(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 / "fast").mkdir()
            (root / "sync").mkdir()
            config_path = root / "client.toml"
            config_path.write_text(
                _config(root).replace(
                    f'local_root = "{root / "qb"}"',
                    f'local_root = "{root / "qb"}"\n'
                    "local_path_overrides = { \"/downloads/fast\" = "
                    f'"{root / "fast"}" }}',
                    1,
                ),
                encoding="utf-8",
            )
            config = ClientConfig.load(config_path)
            self.assertEqual(
                config.qbittorrent.roots.api_to_local(
                    "/downloads/fast/resource/file.bin"
                ),
                root / "fast/resource/file.bin",
            )

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