129 lines
5.6 KiB
Python
129 lines
5.6 KiB
Python
"""Tests for the Transmission torrents client."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
|
|
import pytest
|
|
import respx
|
|
from httpx import Response
|
|
|
|
from conduit_client import AsyncConduitClient, ConduitClient
|
|
from conduit_client.models.torrents import (
|
|
AddTorrentRequest,
|
|
MoveQueueRequest,
|
|
MoveTorrentRequest,
|
|
PatchTorrentRequest,
|
|
RenamePathRequest,
|
|
)
|
|
|
|
|
|
@respx.mock
|
|
def test_add_torrent(api_key: str) -> None:
|
|
route = respx.post("https://api.example.com/torrents").mock(
|
|
return_value=Response(200, json={"id": 7, "name": "file.iso"})
|
|
)
|
|
with ConduitClient(api_key, base_url="https://api.example.com") as client:
|
|
result = client.torrents.add(
|
|
AddTorrentRequest(url="magnet:?xt=urn:btih:abc", paused=True)
|
|
)
|
|
assert result["id"] == 7
|
|
assert json.loads(route.calls[0].request.content)["paused"] is True
|
|
|
|
|
|
@respx.mock
|
|
def test_list_torrents(api_key: str) -> None:
|
|
route = respx.get("https://api.example.com/torrents").mock(
|
|
return_value=Response(200, json={"torrents": []})
|
|
)
|
|
with ConduitClient(api_key, base_url="https://api.example.com") as client:
|
|
result = client.torrents.list(fields=["id", "name"], ids=[1, 2])
|
|
assert result == {"torrents": []}
|
|
assert route.calls[0].request.url.params["fields"] == "id,name"
|
|
assert route.calls[0].request.url.params["ids"] == "1,2"
|
|
|
|
|
|
@respx.mock
|
|
def test_session_and_stats(api_key: str) -> None:
|
|
respx.get("https://api.example.com/torrents/session").mock(
|
|
return_value=Response(200, json={"download-dir": "/tmp"})
|
|
)
|
|
patch_route = respx.patch("https://api.example.com/torrents/session").mock(
|
|
return_value=Response(204)
|
|
)
|
|
respx.get("https://api.example.com/torrents/session/stats").mock(
|
|
return_value=Response(200, json={"cumulative-stats": {}})
|
|
)
|
|
respx.post("https://api.example.com/torrents/session/close").mock(
|
|
return_value=Response(204)
|
|
)
|
|
with ConduitClient(api_key, base_url="https://api.example.com") as client:
|
|
session = client.torrents.get_session()
|
|
assert session["download-dir"] == "/tmp"
|
|
client.torrents.patch_session({"download-dir": "/new"})
|
|
stats = client.torrents.get_session_stats()
|
|
assert stats == {"cumulative-stats": {}}
|
|
client.torrents.close_session()
|
|
assert json.loads(patch_route.calls[0].request.content) == {"download-dir": "/new"}
|
|
|
|
|
|
@respx.mock
|
|
def test_free_space(api_key: str) -> None:
|
|
route = respx.get("https://api.example.com/torrents/free-space").mock(
|
|
return_value=Response(200, json={"path": "/tmp", "size-bytes": 1000, "total_size": 2000})
|
|
)
|
|
with ConduitClient(api_key, base_url="https://api.example.com") as client:
|
|
result = client.torrents.free_space("/tmp")
|
|
assert result.path == "/tmp"
|
|
assert result.size_bytes == 1000
|
|
assert route.calls[0].request.url.params["path"] == "/tmp"
|
|
|
|
|
|
@respx.mock
|
|
def test_torrent_actions(api_key: str) -> None:
|
|
respx.post("https://api.example.com/torrents/queue/move").mock(return_value=Response(204))
|
|
respx.get("https://api.example.com/torrents/groups").mock(return_value=Response(200, json={}))
|
|
respx.patch("https://api.example.com/torrents/groups/fast").mock(return_value=Response(204))
|
|
respx.get("https://api.example.com/torrents/7/download-link").mock(
|
|
return_value=Response(200, json={"torrent_link": "https://x.com/t", "file_links": []})
|
|
)
|
|
respx.get("https://api.example.com/torrents/7").mock(return_value=Response(200, json={"id": 7}))
|
|
respx.patch("https://api.example.com/torrents/7").mock(return_value=Response(204))
|
|
respx.post("https://api.example.com/torrents/7/start").mock(return_value=Response(204))
|
|
respx.post("https://api.example.com/torrents/7/start-now").mock(return_value=Response(204))
|
|
respx.post("https://api.example.com/torrents/7/stop").mock(return_value=Response(204))
|
|
respx.post("https://api.example.com/torrents/7/verify").mock(return_value=Response(204))
|
|
respx.post("https://api.example.com/torrents/7/reannounce").mock(return_value=Response(204))
|
|
respx.post("https://api.example.com/torrents/7/move").mock(return_value=Response(204))
|
|
respx.post("https://api.example.com/torrents/7/rename-path").mock(return_value=Response(204))
|
|
respx.delete("https://api.example.com/torrents/7").mock(return_value=Response(204))
|
|
with ConduitClient(api_key, base_url="https://api.example.com") as client:
|
|
client.torrents.move_queue(MoveQueueRequest(direction="top"))
|
|
client.torrents.list_groups()
|
|
client.torrents.patch_group("fast", {"speed_limit_down": 100})
|
|
link = client.torrents.download_link(7)
|
|
assert link.torrent_link == "https://x.com/t"
|
|
assert client.torrents.get(7)["id"] == 7
|
|
client.torrents.patch(7, PatchTorrentRequest(queue_position=1))
|
|
client.torrents.start(7)
|
|
client.torrents.start_now(7)
|
|
client.torrents.stop(7)
|
|
client.torrents.verify(7)
|
|
client.torrents.reannounce(7)
|
|
client.torrents.move(7, MoveTorrentRequest(location="/new", move=True))
|
|
client.torrents.rename_path(7, RenamePathRequest(path="old", name="new"))
|
|
client.torrents.remove(7, delete_local_data=True)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@respx.mock
|
|
async def test_async_add_torrent(api_key: str) -> None:
|
|
respx.post("https://api.example.com/torrents").mock(
|
|
return_value=Response(200, json={"id": 7, "name": "file.iso"})
|
|
)
|
|
async with AsyncConduitClient(api_key, base_url="https://api.example.com") as client:
|
|
result = await client.torrents.add(
|
|
AddTorrentRequest(url="magnet:?xt=urn:btih:abc")
|
|
)
|
|
assert result["id"] == 7
|