Add tests for Transmission RPC (see #59)

tests/rpc/conftest.py

@@ -0,0 +1,33 @@
+import pytest
+
+from torrt.toolbox import TorrentData
+from torrt.utils import Torrent
+
+
+@pytest.fixture
+def torrent_data():
+
+    return TorrentData(parsed=Torrent({
+        'info': {
+            'name': 'mytorr',
+            'pieces': 1,
+            'piece length': 1,
+            'files': [
+                {'length': 10, 'path': ['one.avi']},
+                {'length': 20, 'path': ['two.avi']},
+                {'length': 30, 'path': ['three.avi']},
+                {'length': 40, 'path': ['four.avi']},
+            ],
+        }
+    }))
+
+
+@pytest.fixture
+def torrent_params():
+    return {
+        'files': {
+            'mytorr/one.avi': {'name': 'mytorr/one.avi', 'exclude': True, 'priority': 0},
+            'mytorr/two.avi': {'name': 'mytorr/two.avi', 'exclude': False, 'priority': 0},
+            'mytorr/three.avi': {'name': 'mytorr/three.avi', 'exclude': True, 'priority': 0}
+        }
+    }

tests/rpc/datafixtures/transm_gettorents.json

@@ -0,0 +1,48 @@
+{
+  "arguments": {
+    "torrents": [
+      {
+        "comment": "somecomment",
+        "downloadDir": "/home/idle",
+        "fileStats": [
+          {
+            "bytesCompleted": 0,
+            "priority": 0,
+            "wanted": false
+          },
+          {
+            "bytesCompleted": 0,
+            "priority": 0,
+            "wanted": true
+          },
+          {
+            "bytesCompleted": 0,
+            "priority": 0,
+            "wanted": false
+          }
+        ],
+        "files": [
+          {
+            "bytesCompleted": 0,
+            "length": 10,
+            "name": "mytorr/one.avi"
+          },
+          {
+            "bytesCompleted": 0,
+            "length": 20,
+            "name": "mytorr/two.avi"
+          },
+          {
+            "bytesCompleted": 0,
+            "length": 30,
+            "name": "mytorr/three.avi"
+          }
+        ],
+        "hashString": "xxxxx",
+        "id": 1,
+        "name": "mytorr"
+      }
+    ]
+  },
+  "result": "success"
+}

tests/rpc/test_transmission.py

@@ -0,0 +1,77 @@
+import pytest
+
+from torrt.rpc.transmission import TransmissionRPC, TransmissionRPCException
+
+
+@pytest.fixture
+def transmission():
+    rpc = TransmissionRPC()
+    return rpc
+
+
+def test_auth(response_mock):
+
+    transmission = TransmissionRPC(user='wrong', password='wrongwrong')
+
+    with response_mock(
+        f'POST {transmission.url} -> 401:<h1>401: Unauthorized</h1>Unauthorized User',
+        bypass=False
+    ):
+        with pytest.raises(TransmissionRPCException):
+            transmission.method_get_version()
+
+
+def test_get_version(response_mock, transmission):
+
+    with response_mock(
+        f'POST {transmission.url} -> 200:'
+        '{"arguments":{"rpc-version":15,"rpc-version-minimum":1,"version":"2.94 (d8e60ee44f)"},"result":"success"}\n',
+        bypass=False
+    ):
+        version = transmission.method_get_version()
+        assert version == 15
+
+
+def test_get_torrents(response_mock, transmission, datafix_read, torrent_params):
+
+    with response_mock(
+        f"POST {transmission.url} -> 200:{datafix_read('transm_gettorents.json')}",
+        bypass=False
+    ):
+        response = transmission.method_get_torrents(hashes=['xxxxx'])
+        assert response == [{
+            'comment': 'somecomment',
+            'downloadDir': '/home/idle',
+            'hashString': 'xxxxx',
+            'id': 1,
+            'name': 'mytorr',
+            'hash': 'xxxxx',
+            'download_to': '/home/idle',
+            'params': torrent_params,
+        }]
+
+
+def test_remove_torrent(response_mock, transmission):
+
+    with response_mock(
+        f'POST {transmission.url} -> 200:'
+        '{"arguments":{},"result":"success"}',
+        bypass=False
+    ):
+        response = transmission.method_remove_torrent(hash_str='xxxxx')
+        assert response == {}
+
+
+def test_add_torrent(response_mock, transmission, torrent_params, torrent_data):
+
+    with response_mock(
+        f'POST {transmission.url} -> 200:'
+        '{"arguments":{},"result":"success"}',
+        bypass=False
+    ):
+        response = transmission.method_add_torrent(
+            torrent=torrent_data,
+            download_to='/here/',
+            params=torrent_params,
+        )
+        assert response == {}

torrt/rpc/transmission.py

@@ -6,8 +6,10 @@
 
 
 class TransmissionRPC(BaseRPC):
-    """See https://trac.transmissionbt.com/browser/trunk/extras/rpc-spec.txt for protocol spec details"""
+    """See https://github.com/transmission/transmission/blob/master/extras/rpc-spec.txt
+    for protocol spec details.
 
+    """
     csrf_header: str = 'X-Transmission-Session-Id'
 
     alias: str = 'transmission'
@@ -43,32 +45,39 @@ def __init__(
 
     def query_(self, data: dict) -> dict:
 
-        response = self.client.request(
+        json_data = self.client.request(
             url=self.url,
             data=data,
             auth=(self.user, self.password),
             headers={self.csrf_header: self.session_id},
+            json=True,
+            silence_exceptions=True,
         )
 
-        if response is None:
+        if json_data is None:
             raise TransmissionRPCException(self.client.last_error)
 
-        return response
+        response = self.client.last_response
+        status_code = response.status_code
+
+        if status_code == 409:
+            self.session_id = response.headers[self.csrf_header]
+            json_data = self.query_(data)
+
+        else:
+            if not json_data and not response.ok:
+                raise TransmissionRPCException(response.text)
+
+        return json_data
 
     def query(self, data: dict) -> dict:
 
         self.log_debug(f"RPC method `{data['method']}` ...")
 
         json_data = self.query_(data)
 
-        response = self.client.last_response
-
-        if response.status_code == 409:
-            self.session_id = response.headers[self.csrf_header]
-            json_data = self.query_(data)
-
-        if json_data['result'] != 'success':
-            raise TransmissionRPCException(json_data['result'])
+        if json_data.get('result', '') != 'success':
+            raise TransmissionRPCException(json_data)
 
         return json_data['arguments']
 
@@ -155,7 +164,7 @@ def method_remove_torrent(self, hash_str: str, with_data: bool = False) -> Any:
 
     def method_get_version(self) -> str:
         result = self.query(self.build_request_payload('session-get', ['rpc-version-minimum']))
-        return result['rpc-version-minimum']
+        return result['rpc-version']
 
 
 class TransmissionRPCException(TorrtRPCException):