Add alternate source names for OPS and RED (#10)

* [WIP] refactoring torrent module to accept other sources

* Improved torrent naming; updated tests

src/torrent.py

@@ -46,6 +46,7 @@ def generate_new_torrent_from_file(
   new_torrent_data = copy.deepcopy(source_torrent_data)
   new_tracker = source_tracker.reciprocal_tracker()
   new_tracker_api = __get_reciprocal_tracker_api(new_tracker, red_api, ops_api)
+  stored_api_response = None
 
   for new_source in new_tracker.source_flags_for_creation():
     new_hash = recalculate_hash_for_new_source(source_torrent_data, new_source)
@@ -55,46 +56,43 @@ def generate_new_torrent_from_file(
     if new_hash in output_infohashes:
       raise TorrentAlreadyExistsError(f"Torrent already exists in output directory as {output_infohashes[new_hash]}")
 
-    api_response = new_tracker_api.find_torrent(new_hash)
+    stored_api_response = new_tracker_api.find_torrent(new_hash)
 
-    if api_response["status"] == "success":
-      new_torrent_filepath = generate_torrent_output_filepath(
-        api_response,
+    if stored_api_response["status"] == "success":
+      new_torrent_filepath = __generate_torrent_output_filepath(
+        stored_api_response,
+        new_tracker,
         new_source.decode("utf-8"),
         output_directory,
       )
 
       if new_torrent_filepath:
-        torrent_id = __get_torrent_id(api_response)
+        torrent_id = __get_torrent_id(stored_api_response)
 
         new_torrent_data[b"info"][b"source"] = new_source  # This is already bytes rather than str
         new_torrent_data[b"announce"] = new_tracker_api.announce_url.encode()
         new_torrent_data[b"comment"] = __generate_torrent_url(new_tracker_api.site_url, torrent_id).encode()
 
         return (new_tracker, save_bencoded_data(new_torrent_filepath, new_torrent_data))
-    elif api_response["error"] in ("bad hash parameter", "bad parameters"):
-      raise TorrentNotFoundError(f"Torrent could not be found on {new_tracker.site_shortname()}")
-    else:
-      raise Exception(f"An unknown error occurred in the API response from {new_tracker.site_shortname()}")
 
+  if stored_api_response["error"] in ("bad hash parameter", "bad parameters"):
+    raise TorrentNotFoundError(f"Torrent could not be found on {new_tracker.site_shortname()}")
 
-def generate_torrent_output_filepath(api_response: dict, new_source: str, output_directory: str) -> str:
-  """
-  Generates the output filepath for the new torrent file. Does not create the file.
+  raise Exception(f"An unknown error occurred in the API response from {new_tracker.site_shortname()}")
 
-  Args:
-    `api_response` (`dict`): The response from the tracker API.
-    `new_source` (`str`): The source of the new torrent file (`"RED"` or `"OPS"`).
-    `output_directory` (`str`): The directory to save the new torrent file.
-  Returns:
-    The path to the new torrent file.
-  Raises:
-    `TorrentAlreadyExistsError`: if the new torrent file already exists in the output directory.
-  """
+
+def __generate_torrent_output_filepath(
+  api_response: dict,
+  new_tracker: OpsTracker | RedTracker,
+  new_source: str,
+  output_directory: str,
+) -> str:
+  tracker_name = new_tracker.site_shortname()
+  source_name = f" [{new_source}]" if new_source else ""
 
   filepath_from_api_response = unescape(api_response["response"]["torrent"]["filePath"])
-  filename = f"{filepath_from_api_response} [{new_source}].torrent"
-  torrent_filepath = os.path.join(output_directory, new_source, filename)
+  filename = f"{filepath_from_api_response}{source_name}.torrent"
+  torrent_filepath = os.path.join(output_directory, tracker_name, filename)
 
   if os.path.isfile(torrent_filepath):
     raise TorrentAlreadyExistsError(f"Torrent file already exists at {torrent_filepath}")

src/trackers.py

@@ -23,11 +23,11 @@ def reciprocal_tracker():
 class OpsTracker(Tracker):
   @staticmethod
   def source_flags_for_search():
-    return [b"OPS"]
+    return [b"OPS", b"APL"]
 
   @staticmethod
   def source_flags_for_creation():
-    return [b"OPS"]
+    return [b"OPS", b"APL", b""]
 
   @staticmethod
   def announce_url():
@@ -49,7 +49,7 @@ def source_flags_for_search():
 
   @staticmethod
   def source_flags_for_creation():
-    return [b"RED"]
+    return [b"RED", b"PTH", b""]
 
   @staticmethod
   def announce_url():

tests/test_torrent.py

@@ -8,7 +8,7 @@
 from src.trackers import RedTracker
 from src.parser import get_bencoded_data
 from src.errors import TorrentAlreadyExistsError, TorrentDecodingError, UnknownTrackerError, TorrentNotFoundError
-from src.torrent import generate_new_torrent_from_file, generate_torrent_output_filepath
+from src.torrent import generate_new_torrent_from_file
 
 
 class TestGenerateNewTorrentFromFile(SetupTeardown):
@@ -72,6 +72,48 @@ def test_returns_new_tracker_instance_and_filepath(self, red_api, ops_api):
 
       os.remove(filepath)
 
+  def test_works_with_alternate_sources_for_creation(self, red_api, ops_api):
+    with requests_mock.Mocker() as m:
+      m.get(
+        re.compile("action=torrent"),
+        [{"json": self.TORRENT_KNOWN_BAD_RESPONSE}, {"json": self.TORRENT_SUCCESS_RESPONSE}],
+      )
+      m.get(re.compile("action=index"), json=self.ANNOUNCE_SUCCESS_RESPONSE)
+
+      torrent_path = get_torrent_path("ops_source")
+      _, filepath = generate_new_torrent_from_file(torrent_path, "/tmp", red_api, ops_api)
+      parsed_torrent = get_bencoded_data(filepath)
+
+      assert filepath == "/tmp/RED/foo [PTH].torrent"
+      assert parsed_torrent[b"announce"] == b"https://flacsfor.me/bar/announce"
+      assert parsed_torrent[b"comment"] == b"https://redacted.ch/torrents.php?torrentid=123"
+      assert parsed_torrent[b"info"][b"source"] == b"PTH"
+
+      os.remove(filepath)
+
+  def test_works_with_blank_source_for_creation(self, red_api, ops_api):
+    with requests_mock.Mocker() as m:
+      m.get(
+        re.compile("action=torrent"),
+        [
+          {"json": self.TORRENT_KNOWN_BAD_RESPONSE},
+          {"json": self.TORRENT_KNOWN_BAD_RESPONSE},
+          {"json": self.TORRENT_SUCCESS_RESPONSE},
+        ],
+      )
+      m.get(re.compile("action=index"), json=self.ANNOUNCE_SUCCESS_RESPONSE)
+
+      torrent_path = get_torrent_path("ops_source")
+      _, filepath = generate_new_torrent_from_file(torrent_path, "/tmp", red_api, ops_api)
+      parsed_torrent = get_bencoded_data(filepath)
+
+      assert filepath == "/tmp/RED/foo.torrent"
+      assert parsed_torrent[b"announce"] == b"https://flacsfor.me/bar/announce"
+      assert parsed_torrent[b"comment"] == b"https://redacted.ch/torrents.php?torrentid=123"
+      assert parsed_torrent[b"info"][b"source"] == b""
+
+      os.remove(filepath)
+
   def test_raises_error_if_cannot_decode_torrent(self, red_api, ops_api):
     with pytest.raises(TorrentDecodingError) as excinfo:
       torrent_path = get_torrent_path("broken")
@@ -105,7 +147,7 @@ def test_raises_error_if_infohash_found_in_output(self, red_api, ops_api):
     assert str(excinfo.value) == "Torrent already exists in output directory as bar"
 
   def test_raises_error_if_torrent_already_exists(self, red_api, ops_api):
-    filepath = generate_torrent_output_filepath(self.TORRENT_SUCCESS_RESPONSE, "OPS", "/tmp")
+    filepath = "/tmp/OPS/foo [OPS].torrent"
 
     os.makedirs(os.path.dirname(filepath), exist_ok=True)
     with open(filepath, "w") as f:
@@ -143,25 +185,3 @@ def test_raises_error_if_api_response_unknown(self, red_api, ops_api):
         generate_new_torrent_from_file(torrent_path, "/tmp", red_api, ops_api)
 
     assert str(excinfo.value) == "An unknown error occurred in the API response from OPS"
-
-
-class TestGenerateTorrentOutputFilepath(SetupTeardown):
-  API_RESPONSE = {"response": {"torrent": {"filePath": "foo"}}}
-
-  def test_constructs_a_path_from_response_and_source(self):
-    filepath = generate_torrent_output_filepath(self.API_RESPONSE, "OPS", "base/dir")
-
-    assert filepath == "base/dir/OPS/foo [OPS].torrent"
-
-  def test_raises_error_if_file_exists(self):
-    filepath = generate_torrent_output_filepath(self.API_RESPONSE, "OPS", "/tmp")
-
-    os.makedirs(os.path.dirname(filepath), exist_ok=True)
-    with open(filepath, "w") as f:
-      f.write("")
-
-    with pytest.raises(TorrentAlreadyExistsError) as excinfo:
-      generate_torrent_output_filepath(self.API_RESPONSE, "OPS", "/tmp")
-
-    assert str(excinfo.value) == f"Torrent file already exists at {filepath}"
-    os.remove(filepath)

tests/test_trackers.py

@@ -6,11 +6,11 @@
 class TestTrackerMethods(SetupTeardown):
   def test_source_flags_for_search(self):
     assert RedTracker.source_flags_for_search() == [b"RED", b"PTH"]
-    assert OpsTracker.source_flags_for_search() == [b"OPS"]
+    assert OpsTracker.source_flags_for_search() == [b"OPS", b"APL"]
 
   def test_source_flags_for_creation(self):
-    assert RedTracker.source_flags_for_creation() == [b"RED"]
-    assert OpsTracker.source_flags_for_creation() == [b"OPS"]
+    assert RedTracker.source_flags_for_creation() == [b"RED", b"PTH", b""]
+    assert OpsTracker.source_flags_for_creation() == [b"OPS", b"APL", b""]
 
   def test_announce_url(self):
     assert RedTracker.announce_url() == b"flacsfor.me"