From 2413cdd83206cc80bf748df95773de4b2f0407d2 Mon Sep 17 00:00:00 2001 From: moleculekayak Date: Tue, 6 Aug 2024 10:00:07 -0700 Subject: [PATCH] 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 | 44 +++++++++++++-------------- src/trackers.py | 6 ++-- tests/test_torrent.py | 68 +++++++++++++++++++++++++++--------------- tests/test_trackers.py | 6 ++-- 4 files changed, 71 insertions(+), 53 deletions(-) diff --git a/src/torrent.py b/src/torrent.py index b70c30e..0077a8c 100644 --- a/src/torrent.py +++ b/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}") diff --git a/src/trackers.py b/src/trackers.py index 24e81df..469c82e 100644 --- a/src/trackers.py +++ b/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(): diff --git a/tests/test_torrent.py b/tests/test_torrent.py index 21f75f4..61fd339 100644 --- a/tests/test_torrent.py +++ b/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) diff --git a/tests/test_trackers.py b/tests/test_trackers.py index ee635ce..97ed4c8 100644 --- a/tests/test_trackers.py +++ b/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"