archive-clients/tests/test_qbittorrent.py

import hashlib
import io
import json
import os
import tempfile
import unittest
from pathlib import Path, PurePosixPath
from unittest.mock import patch
from urllib import error

from archive_clients.bencode import encode
from archive_clients.config import ServiceConfig
from archive_clients.qbittorrent import QBittorrentReader


class _Response:
    def __init__(self, value: bytes, status: int = 200):
        self.value = io.BytesIO(value)
        self.status = status

    def read(self, size=-1):
        return self.value.read(size)

    def __enter__(self):
        return self

    def __exit__(self, *_):
        return None


class _Opener:
    def __init__(self, values):
        self.values = iter(values)
        self.calls = []

    def open(self, call, timeout):
        self.calls.append(call)
        response = next(self.values)
        if isinstance(response, BaseException):
            raise response
        if isinstance(response, tuple):
            return _Response(*response)
        return _Response(response)


class QBittorrentReaderTests(unittest.TestCase):
    def test_hash_scoped_lookup_exports_and_normalizes(self):
        info = {
            b"length": 3, b"name": b"a.txt", b"piece length": 16384,
            b"pieces": b"x" * 20,
        }
        torrent_bytes = encode({b"info": info})
        torrent_hash = hashlib.sha1(encode(info)).hexdigest()
        responses = [
            b"Ok.",
            json.dumps([{
                "hash": torrent_hash, "name": "a.txt", "state": "uploading",
            }]).encode(),
            b'[{"index":0,"name":"a.txt","size":3,"progress":1,"priority":1}]',
            torrent_bytes,
        ]
        with tempfile.TemporaryDirectory() as directory:
            root = Path(directory)
            password = root / "password"
            password.write_text("secret", encoding="utf-8")
            os.chmod(password, 0o600)
            config = ServiceConfig(
                "http://qb", PurePosixPath("/downloads"), root,
                username="admin", password_file=password,
            )
            opener = _Opener(responses)
            with patch(
                "archive_clients.qbittorrent.request.build_opener",
                return_value=opener,
            ):
                resource = QBittorrentReader(config).get_resource(torrent_hash)
        self.assertIsNotNone(resource)
        self.assertEqual(resource.summary.resource_id.info_hash_v1_hex, torrent_hash)
        self.assertIn("hashes=", opener.calls[1])

    def test_current_empty_204_login_is_accepted(self):
        responses = [(b"", 204), b"[]"]
        with tempfile.TemporaryDirectory() as directory:
            root = Path(directory)
            password = root / "password"
            password.write_text("secret", encoding="utf-8")
            os.chmod(password, 0o600)
            config = ServiceConfig(
                "http://qb", PurePosixPath("/downloads"), root,
                username="admin", password_file=password,
            )
            opener = _Opener(responses)
            with patch(
                "archive_clients.qbittorrent.request.build_opener",
                return_value=opener,
            ):
                resources = QBittorrentReader(config).list_resources()
        self.assertEqual(resources, [])
        self.assertEqual(len(opener.calls), 2)

    def test_stopped_add_selection_recheck_and_entry_only_delete(self):
        torrent_hash = "a" * 40
        responses = [
            b"Ok.",  # login
            b"Ok.",  # multipart add
            b"",     # skip all files
            b"",     # select requested files
            b"",     # stop before recheck
            b'{"total_downloaded":0}',
            b"",     # recheck
            json.dumps([{
                "hash": torrent_hash, "state": "checkingUP",
            }]).encode(),
            b'{"total_downloaded":0}',
            b'[{"index":0,"progress":1},{"index":1,"progress":0}]',
            json.dumps([{
                "hash": torrent_hash, "state": "stoppedUP",
            }]).encode(),
            b'{"total_downloaded":0}',
            b'[{"index":0,"progress":1},{"index":1,"progress":0}]',
            b"",     # entry-only delete
        ]
        with tempfile.TemporaryDirectory() as directory:
            root = Path(directory)
            password = root / "password"
            password.write_text("secret", encoding="utf-8")
            os.chmod(password, 0o600)
            config = ServiceConfig(
                "http://qb", PurePosixPath("/downloads"), root,
                username="admin", password_file=password,
            )
            opener = _Opener(responses)
            with patch(
                "archive_clients.qbittorrent.request.build_opener",
                return_value=opener,
            ):
                adapter = QBittorrentReader(config)
                adapter.add_stopped(b"torrent", "/downloads/archive")
                adapter.set_selection(torrent_hash, [0], 2)
                result = adapter.recheck_and_wait(
                    torrent_hash, [0], timeout=1, poll_interval=0
                )
                adapter.delete_entry(torrent_hash)

        self.assertEqual(result.final_state, "stoppedUP")
        self.assertEqual(result.selected_file_indices, (0,))
        calls = [
            call for call in opener.calls
            if not isinstance(call, str)
        ]
        add = calls[1]
        self.assertIn(b'name="stopped"\r\n\r\ntrue', add.data)
        form_bodies = [
            call.data.decode()
            for call in calls[2:]
            if getattr(call, "data", None)
        ]
        self.assertIn("priority=0", form_bodies[0])
        self.assertIn("priority=1", form_bodies[1])
        self.assertIn("deleteFiles=false", form_bodies[-1])

    def test_stop_falls_back_to_qbittorrent_4_pause_endpoint(self):
        missing = error.HTTPError(
            "http://qb/api/v2/torrents/stop", 404, "not found", {}, None
        )
        responses = [b"Ok.", missing, b""]
        with tempfile.TemporaryDirectory() as directory:
            root = Path(directory)
            password = root / "password"
            password.write_text("secret", encoding="utf-8")
            os.chmod(password, 0o600)
            config = ServiceConfig(
                "http://qb", PurePosixPath("/downloads"), root,
                username="admin", password_file=password,
            )
            opener = _Opener(responses)
            with patch(
                "archive_clients.qbittorrent.request.build_opener",
                return_value=opener,
            ):
                QBittorrentReader(config).stop("a" * 40)

        self.assertTrue(opener.calls[-1].full_url.endswith("/torrents/pause"))


if __name__ == "__main__":
    unittest.main()