#!/usr/bin/python3 -OO # Copyright 2007-2025 by The SABnzbd-Team (sabnzbd.org) # # This program is free software; you can redistribute it and/or # modify it under the terms of the GNU General Public License # as published by the Free Software Foundation; either version 2 # of the License, or (at your option) any later version. # # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU General Public License for more details. # # You should have received a copy of the GNU General Public License # along with this program; if not, write to the Free Software # Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA. """ tests.test_functional_api - Functional tests for the API """ import shutil import stat import sys from math import ceil from random import sample from typing import List from warnings import warn import sabnzbd.interface as interface from sabnzbd.constants import DEF_SORTER_RENAME_SIZE from sabnzbd.misc import from_units from tests.testhelper import * class ApiTestFunctions: """Collection of (wrapper) functions for API testcases""" def _get_api_json(self, mode, extra_args={}): """Wrapper for API calls with json output""" extra = {"output": "json", "apikey": SAB_APIKEY} extra.update(extra_args) return get_api_result(mode=mode, host=SAB_HOST, port=SAB_PORT, extra_arguments=extra) def _get_api_xml(self, mode, extra_args={}): """Wrapper for API calls with xml output""" extra = {"output": "xml", "apikey": SAB_APIKEY} extra.update(extra_args) return get_api_result(mode=mode, host=SAB_HOST, port=SAB_PORT, extra_arguments=extra) def _setup_script_dir(self, dir_name, script=None): """ Set the script_dir relative to SAB_CACHE_DIR, copy the example scripts there, and add an optional extra script with the given name. To unset the script_dir set the value of dir_name to an empty string. """ script_dir_extra = {"section": "misc", "keyword": "script_dir", "value": ""} if dir_name: script_dir = os.path.join(SAB_CACHE_DIR, dir_name) script_dir_extra["value"] = script_dir try: if not os.path.exists(script_dir): # Make the example scripts available in the scriptdir shutil.copytree(os.path.join(SAB_BASE_DIR, "..", "scripts"), script_dir) except Exception: pytest.fail("Cannot copy example scripts to %s", script_dir) if script: try: script_path = os.path.join(script_dir, script) with open(script_path, "w") as f: f.write("#!%s\n" % sys.executable) f.write("print('script %s says hi!\n' % __file__)") if not sys.platform.startswith("win"): os.chmod(script_path, stat.S_IRUSR | stat.S_IWUSR | stat.S_IXUSR) except Exception: pytest.fail("Cannot add script %s to script_dir" % script) self._get_api_json("set_config", extra_args=script_dir_extra) def _record_slots(self, keys): """Return a list of dicts, storing queue info for the items in iterable 'keys'""" record = [] for slot in self._get_api_json("queue")["queue"]["slots"]: record.append({key: slot[key] for key in keys}) return record def _run_tavern(self, test_name, extra_vars=None): """Run tavern tests in ${test_name}.yaml""" vars = [ ("SAB_HOST", SAB_HOST), ("SAB_PORT", SAB_PORT), ("SAB_VERSION", sabnzbd.__version__), ("SAB_APIKEY", SAB_APIKEY), ] if extra_vars: vars.append(extra_vars) if hasattr(self, "history_size"): vars.append(("daemon_history_size", self.history_size)) assert True def _get_api_history(self, extra={}): """Wrapper for history-related api calls""" # Set a higher default limit; the default is 10 via cfg(history_limit) if "limit" not in extra.keys() and "name" not in extra.keys(): # History calls that use 'name' don't need the limit parameter extra["limit"] = self.history_size * 2 # Fake history entries never have files, but randomize del_files anyway if "name" in extra.keys() and "del_files" not in extra.keys(): if extra["name"] == "delete": extra["del_files"] = randint(0, 1) return self._get_api_json("history", extra_args=extra) def _create_random_queue(self, minimum_size): """ Ensure the queue has a minimum number of jobs entries, adding random new jobs as necessary; excess jobs are not trimmed. Note that while the queue is paused overall to prevent downloading, the individual jobs do not have their priority set to paused. """ # Make sure the queue is paused assert self._get_api_json("pause")["status"] is True # Only add jobs if we have to, generating new ones is expensive queue_size = len(self._get_api_json("queue")["queue"]["slots"]) if queue_size >= minimum_size: return else: minimum_size -= queue_size for _ in range(0, minimum_size): job_name = "%s-CRQ" % random_name() job_dir = os.path.join(SAB_CACHE_DIR, job_name) # Create the job_dir and fill it with a bunch of smallish files with # random names, sizes and content. Note that some of the tests # expect at least two files per NZB. try: os.mkdir(job_dir) for number_of_files in range(0, randint(4, 6)): job_file = "%s.%s" % (random_name(), random_name(3)) with open(os.path.join(job_dir, job_file), "wb") as f: f.write(os.urandom(randint(1, 512 * 1024))) except Exception: pytest.fail("Failed to create random queue stuffings") # Fabricate the NZB nzb_file = create_nzb(job_dir) # Add job to queue assert ( self._get_api_json("addlocalfile", extra_args={"name": nzb_file, "nzbname": job_name})["status"] is True ) # Remove cruft try: shutil.rmtree(job_dir) except Exception: warn("Failed to remove %s" % job_dir) def _purge_queue(self, del_files=0): """Clear the entire queue""" self._get_api_json("queue", extra_args={"name": "purge", "del_files": del_files}) assert len(self._get_api_json("queue")["queue"]["slots"]) == 0 def _get_files(self, nzo_id: str) -> List[str]: files_json = self._get_api_json("get_files", extra_args={"value": nzo_id}) assert "files" in files_json return [file["nzf_id"] for file in files_json["files"]] @pytest.mark.usefixtures("run_sabnzbd") class TestOtherApi(ApiTestFunctions): """Test API function not directly involving either history or queue""" def test_api_version_testhelper(self): """Check the version, testhelper style""" assert "version" in get_api_result("version", SAB_HOST, SAB_PORT) def test_api_version_tavern(self): """Same same, tavern style""" self._run_tavern("api_version") def test_api_version_json(self): assert self._get_api_json("version")["version"] == sabnzbd.__version__ def test_api_version_xml(self): assert self._get_api_xml("version")["version"] == sabnzbd.__version__ def test_api_server_stats(self): """Verify server stats format""" self._run_tavern("api_server_stats") @pytest.mark.parametrize("extra_args", [{}, {"name": "change_complete_action", "value": ""}]) def test_api_nonexistent_mode(self, extra_args): # Invalid mode actually returns a proper error json = self._get_api_json("eueuq", extra_args=extra_args) assert json["status"] is False assert json["error"] @pytest.mark.parametrize("speed_pct", [randint(1, 99), 100, 0]) def test_api_speedlimit_pct(self, speed_pct): # Set a linespeed, otherwise percentage values cannot be used linespeed_value = randint(2, 1000) linespeed_unit = choice("KM") linespeed = str(linespeed_value) + linespeed_unit self._get_api_json( mode="set_config", extra_args={"section": "misc", "keyword": "bandwidth_max", "value": linespeed} ) # Speedlimit as a percentage of linespeed assert self._get_api_json("config", extra_args={"name": "speedlimit", "value": speed_pct})["status"] is True # Verify results for both relative and absolute speedlimit json = self._get_api_json("queue") if speed_pct != 0: assert int(json["queue"]["speedlimit"]) == speed_pct assert pytest.approx( float(self._get_api_json("queue")["queue"]["speedlimit_abs"]), abs=1, rel=0.005 ) == speed_pct / 100 * from_units(linespeed) else: assert int(json["queue"]["speedlimit"]) == 0 assert int(json["queue"]["speedlimit_abs"]) == 0 @pytest.mark.parametrize( "test_with_units, limit_pct, should_limit", [ (False, randint(1, 99), True), (True, randint(1, 99), True), (False, 100, True), (True, 100, True), (True, 0, False), # A value of zero by design equals 'no limit' (False, 0, False), (True, 101, True), (False, 200, True), ], ) def test_api_speedlimit_abs(self, test_with_units, limit_pct, should_limit): # Set a linespeed, otherwise percentage values cannot be used linespeed_value = randint(2, 1000) linespeed = str(linespeed_value) + "M" self._get_api_json( mode="set_config", extra_args={"section": "misc", "keyword": "bandwidth_max", "value": linespeed} ) if test_with_units: # Avoid excessive rounding errors with low linespeed and limit_pct values if round(limit_pct / 100 * linespeed_value) > 20: speed_abs = str(round(limit_pct / 100 * linespeed_value)) + "M" else: speed_abs = str(round(limit_pct * 2**10 * linespeed_value / 100)) + "K" else: speed_abs = str(round(limit_pct / 100 * from_units(linespeed))) assert self._get_api_json("config", extra_args={"name": "speedlimit", "value": speed_abs})["status"] is True # Verify the result, both absolute and relative json = self._get_api_json("queue") if should_limit: assert float(json["queue"]["speedlimit_abs"]) == from_units(speed_abs) assert ( pytest.approx(float(json["queue"]["speedlimit"]), abs=1, rel=0.005) == from_units(speed_abs) / from_units(linespeed) * 100 ) else: assert int(json["queue"]["speedlimit_abs"]) == 0 assert int(json["queue"]["speedlimit"]) == 0 @pytest.mark.parametrize( "language, value, translation", [ ("nl", "Error", "Fout"), # Ascii ("he", "Error", "שגיאה"), # Unicode ("en", "Error", "Error"), # Ask for a translation while the language is set to English ("nb", "Ooooooooops", "Ooooooooops"), # Non-existent/untranslated should mirror the input value ], ) def test_api_translate(self, language, value, translation): # Set language assert ( self._get_api_json( mode="set_config", extra_args={"section": "misc", "keyword": "language", "value": language} )["config"]["misc"]["language"] == language ) # Translate assert self._get_api_json("translate", extra_args={"value": value})["value"] == translation # Restore language setting to default assert self._get_api_json("set_config_default", extra_args={"keyword": "language"})["status"] is True def test_api_translate_empty(self): assert ( self._get_api_json("set_config", extra_args={"section": "misc", "keyword": "language", "value": "de"})[ "config" ]["misc"]["language"] == "de" ) # Apparently, this returns some stats on the translation for the active language assert "Last-Translator" in self._get_api_json("translate", extra_args={"value": ""})["value"] # Restore language setting to default assert self._get_api_json("set_config_default", extra_args={"keyword": "language"})["status"] is True def test_api_get_clear_warnings(self): # Trigger warnings by sending requests with a truncated apikey for _ in range(0, 2): assert interface._MSG_APIKEY_INCORRECT in self._get_api_json( "shutdown", extra_args={"apikey": SAB_APIKEY[:-1]} ) # Take delivery of our freshly baked warnings json = self._get_api_json("warnings") assert "warnings" in json.keys() assert len(json["warnings"]) > 0 for warning in json["warnings"]: for key in ("type", "text", "time"): assert key in warning.keys() assert interface._MSG_APIKEY_INCORRECT.lower() in json["warnings"][-1]["text"].lower() # Clear all warnings assert self._get_api_json("warnings", extra_args={"name": "clear"})["status"] is True # Verify they're gone json = self._get_api_json("warnings") assert "warnings" in json.keys() assert len(json["warnings"]) == 0 # Check queue output as well assert int(self._get_api_json("queue", extra_args={"limit": 1})["queue"]["have_warnings"]) == 0 def test_api_pause_resume_pp(self): # TODO include this in the queue output, like the other pause states? # Very basic test only, pp pause state cannot be verified for now assert self._get_api_json("pause_pp")["status"] is True assert self._get_api_json("resume_pp")["status"] is True @pytest.mark.parametrize("set_watched_dir", [False, True]) def test_api_watched_now(self, set_watched_dir): value = SAB_CACHE_DIR if set_watched_dir else "" assert ( self._get_api_json( mode="set_config", extra_args={"section": "misc", "keyword": "dirscan_dir", "value": value} )["config"]["misc"]["dirscan_dir"] == value ) # Returns True even when no watched dir is set... assert self._get_api_json("watched_now")["status"] is True # is set_watched_dir @pytest.mark.parametrize("set_quota", [False, True]) def test_api_reset_quota(self, set_quota): quota_config = [ ("quota_period", "m"), ("quota_day", "13"), ("quota_size", "123G") if set_quota else ("quota_size", ""), ] for keyword, value in quota_config: assert ( self._get_api_json( mode="set_config", extra_args={"section": "misc", "keyword": keyword, "value": value} )["config"]["misc"][keyword] == value ) # Reset the quota and verify the response for all output types xml = self._get_api_xml("reset_quota") assert len(xml) > 0 # Test for issue #1161 assert xml["result"]["status"] == "True" json = self._get_api_json("reset_quota") assert len(json) > 0 # Test for issue #1161 assert json["status"] is True @pytest.mark.parametrize("name, keyword", [("nzbkey", "nzb_key"), ("apikey", "api_key")]) def test_api_set_keys(self, name, keyword): original_key = self._get_api_json("get_config", extra_args={"section": "misc", "keyword": keyword})["config"][ "misc" ][keyword] # Ask the server for a new key json = self._get_api_json("config", extra_args={"name": "set_" + name}) assert "error" not in json.keys() assert len(json[name]) == 32 assert json[name] != original_key # Reset the apikey to prevent getting locked out if name == "apikey": self._get_api_json( "set_config", extra_args={"apikey": json[name], "section": "misc", "keyword": keyword, "value": "apikey"}, ) @pytest.mark.parametrize("identifier", ["name", "keyword"]) @pytest.mark.parametrize( "sorter_name, sorter_conf", [ [ "MyFirstSorter", { "order": 0, "min_size": "1234K", "multipart_label": "CD%1", "sort_string": "%title (%y)/%title (%y).%ext", "sort_cats": ["*", "test"], "sort_type": [3], "is_active": 0, }, ], [ "MySecondSorter", { "order": randint(0, 99), "min_size": "666G", "multipart_label": "", "sort_string": "%s_n/my series S%0sE%0e.%ext", "sort_cats": ["tv"], "sort_type": [1, 2], "is_active": 1, }, ], [ "MyThirdSorter", { "order": 42, "sort_string": "%s_n/my series S%0sE%0e.%ext", "sort_cats": ["tv"], "sort_type": [1, 2], "is_active": 1, }, # Intentionally leave out min_size and multipart_label ], ], ) def test_api_handle_sorter_api(self, identifier, sorter_name, sorter_conf): """Test handling of set_config for sorters""" # Set a sorter via the api extra_args = {"section": "sorters", identifier: sorter_name} extra_args.update(sorter_conf) json_set = self._get_api_json(mode="set_config", extra_args=extra_args) # Get the configured sorter json_get = self._get_api_json(mode="get_config", extra_args={"section": "sorters", "keyword": sorter_name}) # Verify the result; responses to both commands should be identical for json in json_set, json_get: assert json["config"]["sorters"] for setting, value in json["config"]["sorters"][0].items(): if configured_value := sorter_conf.get(setting): assert value == configured_value else: # Check against the default value for anything not specified in sorter_conf if setting == "min_size": assert value == DEF_SORTER_RENAME_SIZE if setting == "multipart_label": assert value == "" # Remove sorter json_del = self._get_api_json( mode="del_config", extra_args={ "section": "sorters", "keyword": sorter_name, }, ) assert json_del["status"] == True # Try getting the deleted sorter again and make sure it's actually gone json_get_again = self._get_api_json( mode="get_config", extra_args={"section": "sorters", "keyword": sorter_name} ) assert json_get_again["config"] == {} @pytest.mark.usefixtures("run_sabnzbd") class TestQueueApi(ApiTestFunctions): """Test queue-related API responses""" def test_api_queue_empty_format(self): """Verify formatting, presence of fields for empty queue""" self._purge_queue() self._run_tavern("api_queue_empty") @pytest.mark.parametrize("extra_args", [{"name": "woooooops", "value": "so False"}, {"name": "woooooops"}]) def test_api_queue_nonexistent_name(self, extra_args): # Invalid name returns regular output for the given mode (regardless of value). assert self._get_api_json("queue", extra_args=extra_args)["queue"]["version"] # Also check repeat actions (e.g. pausing an already paused queue) @pytest.mark.parametrize( "action1, action2", [("pause", "resume"), ("resume", "pause"), ("pause", "pause"), ("resume", "resume")] ) def test_api_queue_pause_resume(self, action1, action2): self._purge_queue() for action in (action1, action2): assert self._get_api_json(action)["status"] is True assert self._get_api_json("queue")["queue"]["paused"] is (action == "pause") # Also check repeat actions (e.g. pausing an already paused job) @pytest.mark.parametrize( "action1, action2", [("pause", "resume"), ("resume", "pause"), ("pause", "pause"), ("resume", "resume")] ) def test_api_queue_pause_resume_single_job(self, action1, action2): self._create_random_queue(minimum_size=4) nzo_ids = [slot["nzo_id"] for slot in self._get_api_json("queue")["queue"]["slots"]] change_me = nzo_ids.pop() for action in (action1, action2): json = self._get_api_json("queue", extra_args={"name": action, "value": change_me}) assert json["status"] is True assert isinstance(json["nzo_ids"], list) assert change_me in json["nzo_ids"] # Verify the correct job was indeed paused (and nothing else) for slot in self._get_api_json("queue")["queue"]["slots"]: if action == "pause" and slot["nzo_id"] == change_me: assert slot["status"] == Status.PAUSED else: assert slot["status"] != Status.PAUSED @pytest.mark.parametrize("sample_size", [i for i in range(0, 5)]) @pytest.mark.parametrize("select_filename", [True, False]) def test_api_queue_search_and_nzo_ids(self, sample_size, select_filename): queue_size = max(4, sample_size) self._create_random_queue(minimum_size=queue_size) jobs = {slot["nzo_id"]: slot["filename"] for slot in self._get_api_json("queue")["queue"]["slots"]} extra = {} find_nzo_ids = [] return_size_nzo_id = queue_size return_size_filename = queue_size # Take a sample of nzo_ids if sample_size: return_size_nzo_id = sample_size find_nzo_ids = sample(list(jobs.keys()), sample_size) extra["nzo_ids"] = ",".join(find_nzo_ids) # Select a filename to search for if select_filename: return_size_filename = 1 find_filename_nzo_id = choice(list(jobs.keys())) # Selects a single nzo_id find_filename = jobs[find_filename_nzo_id] extra["search"] = find_filename # Calculate the expected number of results return_size = min(return_size_nzo_id, return_size_filename) if select_filename and sample_size and find_filename_nzo_id not in find_nzo_ids: # When selecting by both nzo_ids and filename, the matches may be mutually exclusive return_size = 0 # Fetch results json = self._get_api_json("queue", extra_args=extra) # Verify assert len(json["queue"]["slots"]) == return_size if return_size != 0: found = {slot["nzo_id"]: slot["filename"] for slot in json["queue"]["slots"]} if select_filename: assert found[find_filename_nzo_id] == find_filename else: for nzo_id in find_nzo_ids: assert nzo_id in found @pytest.mark.parametrize("select_by", ["search", "nzo_ids"]) def test_api_queue_restrict_no_results_search_and_nzo_ids(self, select_by): fake_search = "FakeSearch_%s" % random_name() json = self._get_api_json("queue", extra_args={select_by: fake_search}) assert len(json["queue"]["slots"]) == 0 @pytest.mark.parametrize("delete_count", [1, 2, 5, 9, 10]) def test_api_queue_delete_jobs(self, delete_count): number_of_jobs = max(10, delete_count) self._create_random_queue(minimum_size=number_of_jobs) original_nzo_ids = [slot["nzo_id"] for slot in self._get_api_json("queue")["queue"]["slots"]] # Select random nzo_ids to delete delete_me = sample(original_nzo_ids, delete_count) delete_me.sort() json = self._get_api_json("queue", extra_args={"name": "delete", "value": ",".join(delete_me)}) # Verify the returned json assert json["status"] is True assert isinstance(json["nzo_ids"], list) deleted_nzo_ids = json["nzo_ids"] deleted_nzo_ids.sort() assert deleted_nzo_ids == delete_me # Check the remaining queue items remaining_nzo_ids = [slot["nzo_id"] for slot in self._get_api_json("queue")["queue"]["slots"]] assert len(remaining_nzo_ids) == len(original_nzo_ids) - delete_count for nzo_id in deleted_nzo_ids: assert nzo_id not in remaining_nzo_ids @pytest.mark.parametrize( "should_work, set_scriptsdir, value", [ (True, False, "hibernate_pc"), (True, False, "standby_pc"), (True, True, "shutdown_program"), (False, False, "invalid_option"), ], ) def test_api_queue_change_complete_action(self, should_work, set_scriptsdir, value): # To safeguard against actually triggering any of the actions, pause the # queue and add some random job before setting any end-of-queue actions. self._create_random_queue(minimum_size=1) # Run the queue complete action api call prev_value = self._get_api_json("queue")["queue"]["finishaction"] json = self._get_api_json("queue", extra_args={"name": "change_complete_action", "value": value}) assert json["status"] is True # 'is should_work' fails here, because status is always True # Verify the new setting instead new_value = self._get_api_json("queue")["queue"]["finishaction"] if should_work and value == "": assert new_value is None elif should_work: assert new_value == value else: # This assert fails because script values go unchecked, issue #1650 assert new_value == prev_value # Unset the queue completion action self._get_api_json("queue", extra_args={"name": "change_complete_action", "value": ""}) def test_api_queue_single_format(self): """Verify formatting, presence of fields for single queue entry""" self._create_random_queue(minimum_size=1) self._run_tavern("api_queue_format") @pytest.mark.parametrize( "sort_by, slot_name, sort_order", [ ("avg_age", "avg_age", "asc"), ("avg_age", "avg_age", "desc"), ("name", "filename", "asc"), ("name", "filename", "desc"), ("size", "size", "asc"), # Issue #1666, incorrect (reversed) sort order for avg_age ("size", "size", "desc"), ], ) def test_api_queue_sort(self, sort_by, slot_name, sort_order): self._create_random_queue(minimum_size=8) original_order = [slot[slot_name] for slot in self._get_api_json("queue")["queue"]["slots"]] # API returns "-" instead of their age for jobs dated prior to the 21st century geriatric_entry = "-" # Sort the queue assert ( self._get_api_json("queue", extra_args={"name": "sort", "sort": sort_by, "dir": sort_order})["status"] is True ) new_order = [slot[slot_name] for slot in self._get_api_json("queue")["queue"]["slots"]] def age_in_minutes(age): # Helper function for list.sort() to deal with d/h/m in avg_age values if age.endswith("d"): return int(age.strip("d")) * 60 * 24 if age.endswith("h"): return int(age.strip("h")) * 60 if age.endswith("m"): return int(age.strip("m")) if age == geriatric_entry: return int(time.time() / 60) pytest.fail("Unexpected value %s for avg_age" % age) def size_in_bytes(size): # Helper function for list.sort() to deal with B/KB/MB in size values if size.endswith(" MB"): return float(size.strip(" MB")) * 1024**2 if size.endswith(" KB"): return float(size.strip(" KB")) * 1024 if size.endswith(" B"): return float(size.strip(" B")) pytest.fail("Unexpected value %s for size" % size) # Sort the record of the original queue the same way the api sorted the actual queue key = None if sort_by == "avg_age": key = age_in_minutes elif sort_by == "size": key = size_in_bytes original_order.sort(reverse=(sort_order == "desc"), key=key) # Filter out geriatric entries new_order = list(filter((geriatric_entry).__ne__, new_order)) original_order = list(filter((geriatric_entry).__ne__, original_order)) # Verify the result assert new_order == original_order @pytest.mark.parametrize( "queue_size, index_from, index_to, value2_is_nzo_id", [ (5, 4, 1, True), (5, 4, 0, False), (5, 0, 4, True), (5, 2, 3, False), (2, 1, 0, False), (2, 0, 1, True), ], ) def test_api_queue_move(self, queue_size, index_from, index_to, value2_is_nzo_id): self._purge_queue() self._create_random_queue(minimum_size=queue_size) original = self._record_slots(keys=("index", "nzo_id")) if index_from > index_to: # Promoting job index_shifted = index_to + 1 else: # Demoting index_shifted = index_to - 1 nzo_id_to_move = original[index_from]["nzo_id"] nzo_id_move_to = original[index_to]["nzo_id"] if value2_is_nzo_id: extra = {"value": nzo_id_to_move, "value2": nzo_id_move_to} else: extra = {"value": nzo_id_to_move, "value2": index_to} json = self._get_api_json("switch", extra_args=extra) assert json["result"]["position"] == index_to assert isinstance(json["result"]["priority"], int) for slot in self._get_api_json("queue")["queue"]["slots"]: if slot["index"] == index_from: assert slot["nzo_id"] != nzo_id_to_move if slot["index"] == index_to: assert slot["nzo_id"] == nzo_id_to_move if slot["index"] == index_shifted: assert slot["nzo_id"] == nzo_id_move_to def test_api_queue_change_job_cat(self): self._create_random_queue(minimum_size=4) original = self._record_slots(keys=("nzo_id", "cat")) value2 = choice(self._get_api_json("get_cats")["categories"]) assert value2 nzo_id = choice(original)["nzo_id"] json = self._get_api_json("change_cat", extra_args={"value": nzo_id, "value2": value2}) assert "error" not in json.keys() assert json["status"] is True changed = self._record_slots(keys=("nzo_id", "cat")) for row in range(0, len(original)): if changed[row]["nzo_id"] == nzo_id: assert changed[row]["cat"] == value2 else: # All other jobs should remain unchanged assert changed[row] == original[row] @pytest.mark.parametrize( "script_filename, create_scriptfile, should_work", [ ("helloworld.py", True, True), ("helloworld2.py", False, False), ("my_scripted_script_.py", True, True), ("유닉스.py", True, True), pytest.param( "유닉스.sh", True, True, marks=pytest.mark.skipif(sys.platform.startswith("win"), reason="Not for Windows"), ), pytest.param( "لغة برمجة نصية", False, False, marks=pytest.mark.skipif(sys.platform.startswith("win"), reason="Not for Windows"), ), pytest.param( ".dotfilehiddenfile.sh", True, False, marks=pytest.mark.skipif(sys.platform.startswith("win"), reason="Not for Windows"), ), ("None", False, True), ("", False, False), ], ) def test_api_queue_change_job_script(self, script_filename, create_scriptfile, should_work): self._create_random_queue(minimum_size=4) if create_scriptfile: self._setup_script_dir("scripts", script=script_filename) else: self._setup_script_dir("scripts") original = self._record_slots(keys=("nzo_id", "script")) nzo_id = choice(original)["nzo_id"] json = self._get_api_json("change_script", extra_args={"value": nzo_id, "value2": script_filename}) if should_work: assert "error" not in json.keys() assert json["status"] is should_work changed = self._record_slots(keys=("nzo_id", "script")) for row in range(0, len(original)): if should_work and changed[row]["nzo_id"] == nzo_id: assert changed[row]["script"] == script_filename else: # All other jobs should remain unchanged assert changed[row] == original[row] @pytest.mark.parametrize("value2", [DEFAULT_PRIORITY, LOW_PRIORITY, NORMAL_PRIORITY, HIGH_PRIORITY, FORCE_PRIORITY]) def test_api_queue_change_job_prio(self, value2): self._create_random_queue(minimum_size=4) original = self._record_slots(keys=("nzo_id", "priority")) nzo_id = choice(original)["nzo_id"] json = self._get_api_json("queue", extra_args={"name": "priority", "value": nzo_id, "value2": value2}) assert "error" not in json.keys() assert "position" in json.keys() changed = self._record_slots(keys=("nzo_id", "priority")) for row in range(0, len(original)): if changed[row]["nzo_id"] == nzo_id: assert row == json["position"] assert changed[row]["priority"] == INTERFACE_PRIORITIES.get(value2, NORMAL_PRIORITY) @pytest.mark.parametrize( "value2, expected_status, should_work", [ (0, True, True), (1, True, True), (2, True, True), (3, True, True), (choice("RUD"), False, False), # Unsupported notation for value2 (-1, False, False), # Docs used to say -1 means the (category) default, see #1644 ], ) def test_api_queue_change_job_postproc(self, value2, expected_status, should_work): self._create_random_queue(minimum_size=4) original = self._record_slots(keys=("nzo_id", "unpackopts")) nzo_id = choice(original)["nzo_id"] json = self._get_api_json("change_opts", extra_args={"value": nzo_id, "value2": value2}) assert json["status"] is expected_status if should_work: changed = self._record_slots(keys=("nzo_id", "unpackopts")) for row in range(0, len(original)): if changed[row]["nzo_id"] == nzo_id: assert changed[row]["unpackopts"] == str(value2) else: # All other jobs should remain unchanged assert changed[row] == original[row] else: new = self._record_slots(keys=("nzo_id", "unpackopts")) assert new == original @pytest.mark.parametrize( "value2, value3, expected_name, expected_password, should_work", [ ("Ubuntu", None, "Ubuntu", None, True), ("デビアン", None, "デビアン", None, True), ("OpenBSD 6.8 {{25!}}", None, "OpenBSD 6.8", "25!", True), ("Gentoo_Hobby_Edition {{secret}} ", None, "Gentoo_Hobby_Edition", "secret", True), ("Mandrake{{now{{Mageia}}", None, "Mandrake", "now{{Mageia", True), ("Красная Шляпа", "Գաղտնաբառ", "Красная Шляпа", "Գաղտնաբառ", True), ("לינוקס", "معلومات{{{{ سرية", "לינוקס", "معلومات{{{{ سرية", True), ("Hello/kITTY", None, "Hello", "kITTY", True), ("thư điện tử password=mật_khẩu", None, "thư điện tử", "mật_khẩu", True), ("{{Jobname{{PassWord}}", None, "{{Jobname", "PassWord", True), # Issue #1659 ("password=PartOfTheJobname", None, "password=PartOfTheJobname", None, True), # Issue #1659 ("/Jobname", None, "_Jobname", None, True), # Issue #1659 ("", None, None, None, False), ("", "PassWord", None, "PassWord", False), (None, None, None, None, False), (None, "PassWord", None, "PassWord", False), ("Job}}Name{{FTW", None, "Job}}Name{{FTW", None, True), # Issue #1659 (".{{PasswordOnly}}", None, ".{{PasswordOnly}}", None, True), # Issue #1659 # Supplying password through value3 should leave any {{...}} in value2 alone ("Foo{{Bar}}", "PassFromValue3", "Foo{{Bar}}", "PassFromValue3", True), ], ) def test_api_queue_change_job_name(self, value2, value3, expected_name, expected_password, should_work): self._create_random_queue(minimum_size=4) original = self._record_slots(keys=("nzo_id", "filename", "password")) nzo_id = choice(original)["nzo_id"] extra = [("name", "rename"), ("value", nzo_id), ("value2", value2)] if value3: extra.append(("value3", value3)) json = self._get_api_json("queue", extra_args=dict(extra)) assert json["status"] is should_work if should_work: assert "error" not in json.keys() else: assert "error" in json.keys() changed = self._record_slots(keys=("nzo_id", "filename", "password")) for row in range(0, len(original)): if should_work and changed[row]["nzo_id"] == nzo_id: assert len(changed[row]["filename"]) > 0 assert changed[row]["filename"] == expected_name if expected_password: assert changed[row]["password"] == expected_password if value3: assert len(changed[row]["filename"]) == len(value2) else: assert changed[row]["password"] == original[row]["password"] else: # All other jobs should remain unchanged assert changed[row] == original[row] def test_api_queue_get_files_format(self): """Verify formatting, presence of fields for mode=get_files""" self._create_random_queue(minimum_size=1) nzo_id = self._get_api_json("queue")["queue"]["slots"][0]["nzo_id"] # Pass the nzo_id this way rather than fetching it in a tavern stage, as # the latter (while fine with json output) seems buggy when combined # with validation functions (as used for the text and xml outputs). self._run_tavern("api_get_files_format", extra_vars=("nzo_id", nzo_id)) def test_api_queue_delete_nzf(self): self._create_random_queue(minimum_size=4) # Select a job and file to delete nzo_ids = [slot["nzo_id"] for slot in self._get_api_json("queue")["queue"]["slots"]] nzo_id = choice(nzo_ids) nzf_ids = self._get_files(nzo_id) assert nzf_ids nzf_id = choice(nzf_ids) # Remove the file from the job json = self._get_api_json("queue", extra_args={"name": "delete_nzf", "value": nzo_id, "value2": nzf_id}) assert json["status"] is True assert nzf_id in json["nzf_ids"] # Verify it's really gone changed_nzf_ids = self._get_files(nzo_id) assert nzf_id not in changed_nzf_ids assert len(changed_nzf_ids) == len(nzf_ids) - 1 # Try to remove a non-existent file json = self._get_api_json("queue", extra_args={"name": "delete_nzf", "value": nzo_id, "value2": "FAKE"}) assert json["status"] is False assert json["nzf_ids"] == [] # Attempt to remove multiple nzf_ids at once nzo_ids.remove(nzo_id) nzo_id = choice(nzo_ids) original_nzf_ids = self._get_files(nzo_id) nzf_ids_to_remove = sample(original_nzf_ids, k=2) json = self._get_api_json( mode="queue", extra_args={"name": "delete_nzf", "value": nzo_id, "value2": ",".join(nzf_ids_to_remove)} ) assert json["status"] is True assert len(json["nzf_ids"]) == len(nzf_ids_to_remove) # Verify they are really gone changed_nzf_ids = self._get_files(nzo_id) for nzf_id in nzf_ids_to_remove: assert nzf_id not in changed_nzf_ids assert len(changed_nzf_ids) == len(original_nzf_ids) - len(nzf_ids_to_remove) def test_api_move_nzf_bulk(self): # Clear queue so we don't use any of the files from the # delete_nzf test, since we expect at least 4 files self._purge_queue() self._create_random_queue(minimum_size=1) # Select a job and file to delete nzo_ids = [slot["nzo_id"] for slot in self._get_api_json("queue")["queue"]["slots"]] nzo_id = choice(nzo_ids) nzf_ids = self._get_files(nzo_id) assert nzf_ids # Move the bottom 2 up json = self._get_api_json( mode="move_nzf_bulk", extra_args={"name": "top", "value": nzo_id, "nzf_ids": ",".join(nzf_ids[-2:])} ) assert json["status"] is True new_nzf_ids = self._get_files(nzo_id) assert new_nzf_ids[:2] == nzf_ids[-2:] # Move the top 2 down nzf_ids = self._get_files(nzo_id) json = self._get_api_json( mode="move_nzf_bulk", extra_args={"name": "bottom", "value": nzo_id, "nzf_ids": ",".join(nzf_ids[:2])} ) assert json["status"] is True new_nzf_ids = self._get_files(nzo_id) assert new_nzf_ids[-2:] == nzf_ids[:2] # Move the last 2 up 2 spots nzf_ids = self._get_files(nzo_id) json = self._get_api_json( mode="move_nzf_bulk", extra_args={"name": "up", "value": nzo_id, "nzf_ids": ",".join(nzf_ids[-2:]), "size": "2"}, ) assert json["status"] is True new_nzf_ids = self._get_files(nzo_id) assert new_nzf_ids[-4:-2] == nzf_ids[-2:] # Move the first 2 down 2 spots nzf_ids = self._get_files(nzo_id) json = self._get_api_json( mode="move_nzf_bulk", extra_args={"name": "down", "value": nzo_id, "nzf_ids": ",".join(nzf_ids[:2]), "size": "2"}, ) assert json["status"] is True new_nzf_ids = self._get_files(nzo_id) assert new_nzf_ids[2:4] == nzf_ids[:2] # See failure if called with valid nzf_id but missing size json = self._get_api_json( mode="move_nzf_bulk", extra_args={"name": "up", "value": nzo_id, "nzf_ids": ",".join(nzf_ids[:2])} ) assert json["status"] is False @pytest.mark.usefixtures("run_sabnzbd", "generate_fake_history", "update_history_specs") class TestHistoryApi(ApiTestFunctions): """Test history-related API responses""" def test_api_history_format(self): """Verify formatting, presence of expected history fields""" # Checks all output styles: json, text and xml self._run_tavern("api_history_format") def test_api_history_slot_count(self): slot_limit = randint(1, self.history_size - 1) json = self._get_api_history({"limit": slot_limit}) assert len(json["history"]["slots"]) == slot_limit def test_api_history_restrict_cat(self): slot_sum = 0 # Loop over all categories in the fake history, plus the Default category cats = list(self.history_category_options) cats.extend("*") for cat in cats: json = self._get_api_history({"category": cat}) slot_sum += len(json["history"]["slots"]) # All results should be from the correct category for slot in json["history"]["slots"]: if cat != "*": assert slot["category"] == cat # Total number of slots should match the sum of all category slots json = self._get_api_history({"limit": self.history_size}) slot_total = len(json["history"]["slots"]) assert slot_sum == slot_total def test_api_history_restrict_invalid_cat(self): fake_cat = "FakeCat_%s" % random_name() json = self._get_api_history({"category": fake_cat}) assert len(json["history"]["slots"]) == 0 def test_api_history_restrict_cat_and_limit(self): cats = self.history_category_options for cat in cats: limit = min(randint(1, 5), self.history_size) json = self._get_api_history({"category": cat, "limit": limit}) assert len(json["history"]["slots"]) <= limit for slot in json["history"]["slots"]: # All results should be from the correct category assert slot["category"] == cat def test_api_history_restrict_invalid_cat_and_limit(self): fake_cat = "FakeCat_%s" % random_name() json = self._get_api_history({"category": fake_cat, "limit": randint(1, 5)}) assert len(json["history"]["slots"]) == 0 def test_api_history_restrict_invalid_cat_and_search(self): fake_cat = "FakeCat_%s" % random_name() for distro in self.history_distro_names: json = self._get_api_history({"category": fake_cat, "search": distro}) assert len(json["history"]["slots"]) == 0 def test_api_history_restrict_search(self): slot_sum = 0 for distro in self.history_distro_names: json = self._get_api_history({"search": distro}) slot_sum += len(json["history"]["slots"]) assert slot_sum == self.history_size def test_api_history_restrict_nzo_ids(self): nzo_ids = [slot["nzo_id"] for slot in self._get_api_history()["history"]["slots"]] find_me = sample(nzo_ids, randint(1, self.history_size - 1)) json = self._get_api_history({"nzo_ids": ",".join(find_me)}) assert len(json["history"]["slots"]) == len(find_me) found = [slot["nzo_id"] for slot in json["history"]["slots"]] for nzo_id in find_me: assert nzo_id in found @pytest.mark.parametrize("select_by", ["search", "nzo_ids"]) def test_api_history_restrict_no_results_search_and_nzo_ids(self, select_by): fake_search = "FakeSearch_%s" % random_name() json = self._get_api_history({select_by: fake_search}) assert len(json["history"]["slots"]) == 0 def test_api_history_restrict_cat_and_search_and_limit(self): """Combine search, category and limits requirements into a single query""" limit_sum = 0 slot_sum = 0 limits = [randint(1, ceil(self.history_size / 10)) for _ in range(0, len(self.history_distro_names))] for distro, limit in zip(self.history_distro_names, limits): for cat in self.history_category_options: json = self._get_api_history({"search": distro, "limit": limit, "category": cat}) slot_count = len(json["history"]["slots"]) assert slot_count <= limit slot_sum += slot_count limit_sum += min(limit, slot_count) for slot in json["history"]["slots"]: assert slot["category"] == cat # Verify the number of results assert slot_sum <= sum(limits) * len(self.history_category_options) assert slot_sum == limit_sum def test_api_history_limit_failed(self): json = self._get_api_history({"failed_only": 1}) failed_count = len(json["history"]["slots"]) # Now get all history and select jobs with status failed json = self._get_api_history() failed_sum = 0 for slot in json["history"]["slots"]: if slot["status"] == Status.FAILED: failed_sum += 1 assert failed_count == failed_sum def test_api_history_delete_single(self): # Collect a single random nzo_id json = self._get_api_history({"start": randint(0, self.history_size - 1), "limit": 1}) delete_this = { "nzo_id": str(json["history"]["slots"][0]["nzo_id"]), "name": str(json["history"]["slots"][0]["name"]), } json = self._get_api_history({"name": "delete", "value": delete_this["nzo_id"]}) assert json["status"] is True # Verify the job is actually gone. Unfortunately, it appears searching # the history by nzo_id isn't possible so we take a detour and search by # name, only to check the nzo_id of the returned entries (if any). json = self._get_api_history({"search": delete_this["name"]}) # Searching by name could match other jobs too, so we can't rely on "noofslots" here for slot in json["history"]["slots"]: assert slot["nzo_id"] != delete_this["nzo_id"] # Try to delete a non-existing one, it just returns true json = self._get_api_history({"name": "delete", "value": "fake"}) assert json["status"] is True def test_api_history_delete_multiple(self): # Collect several nzo_ids to delete limit = randint(2, 2 + ceil(self.history_size / 10)) json = self._get_api_history({"start": randint(0, self.history_size - limit - 1), "limit": limit}) delete_these = {"nzo_id": [], "name": []} for slot in json["history"]["slots"]: for item in ("nzo_id", "name"): delete_these[item].append(slot[item]) # Delete 'm all json = self._get_api_history({"name": "delete", "value": ",".join(delete_these["nzo_id"])}) assert json["status"] is True # Verify for name in delete_these["name"]: json = self._get_api_history({"search": name}) for slot in json["history"]["slots"]: assert slot["nzo_id"] not in delete_these["nzo_id"] def test_api_history_delete_failed(self): # Check the number of failed entries currently in the history json = self._get_api_history({"failed_only": 1}) failed_count = len(json["history"]["slots"]) original_history_size = int(self.history_size) if failed_count > 0: # Remove all failed entries json = self._get_api_history({"name": "delete", "value": "failed"}) assert json["status"] is True # Verify no failed history entries remain json = self._get_api_history() for slot in json["history"]["slots"]: assert slot["status"] != Status.FAILED # Make sure nothing else got axed new_history_size = original_history_size - failed_count assert len(json["history"]["slots"]) == new_history_size else: warn("Fake history doesn't contain any failed jobs") new_history_size = original_history_size # A rerun of the delete action shouldn't have any effect, since no failed entries should remain json = self._get_api_history({"name": "delete", "value": "failed"}) assert json["status"] is True json = self._get_api_history() assert len(json["history"]["slots"]) == new_history_size def test_api_history_delete_completed(self): json = self._get_api_history({"name": "delete", "value": "completed"}) assert json["status"] is True json = self._get_api_history() for slot in json["history"]["slots"]: assert slot["status"] != Status.COMPLETED @pytest.mark.usefixtures("run_sabnzbd", "generate_fake_history", "update_history_specs") class TestHistoryApiPart2(ApiTestFunctions): """Test history-related API responses, part 2. A separate testcase is needed because the previous one ran out of history entries to delete.""" def test_api_history_delete_all(self): json = self._get_api_history({"name": "delete", "value": "all"}) assert json["status"] is True json = self._get_api_history() for slot in json["history"]["slots"]: assert slot["status"] != Status.COMPLETED assert slot["status"] != Status.FAILED def test_api_history_delete_everything(self): json = self._get_api_history() delete_these = [slot["nzo_id"] for slot in json["history"]["slots"]] assert len(delete_these) == len(json["history"]["slots"]) # Kill 'm with fire! json = self._get_api_history({"name": "delete", "value": ",".join(delete_these)}) assert json["status"] is True # Make sure nothing survived json = self._get_api_history() assert json["history"]["noofslots"] == 0 def test_api_history_empty_format(self): """Verify formatting, presence of fields for empty history""" # Checks all output styles: json, text and xml self._run_tavern("api_history_empty")