Files
sabnzbd/tests/test_functional_adding_nzbs.py
mnightingale 59218a687d Fix race in TestAddingNZBsClean and share the NZB-adding helpers (#3571)
Jobs are listed in the history output as soon as they enter the
post-processing queue, and a stopped job has its status set to Failed at
the very start of process_job. The duplicate alternative is only released
near the end of it, so waiting for the job to show up in the history as
failed does not guarantee the alternative was already released. The gap
is tiny when running serially, but wide enough to fail the test under the
load of pytest-xdist. Wait for the label to be dropped instead.

Also move the helpers shared by both adding-NZB test classes into a base
class in testhelper, so the clean tests no longer have to import the
other test module to copy its methods.
2026-08-16 15:46:05 +02:00

583 lines
24 KiB
Python

#!/usr/bin/python3 -OO
# Copyright 2007-2026 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_adding_nzbs - Tests for settings interaction when adding NZBs
"""
import os
import stat
import sys
from functools import cached_property
from random import choice, randint, sample
import pytest
from tests.testhelper import (
MIN_FILESIZE,
SAB_CACHE_DIR,
AddingNZBsTestBase,
get_api_result,
)
from sabnzbd.constants import (
PAUSED_PRIORITY,
DEFAULT_PRIORITY,
LOW_PRIORITY,
NORMAL_PRIORITY,
HIGH_PRIORITY,
FORCE_PRIORITY,
REPAIR_PRIORITY,
PP_LOOKUP,
)
# Repair priority is out of scope for the purpose of these tests: it cannot be
# set as a default, upon adding a job, or from a pre-queue script.
# "None" is used to *not* set any particular priority at a given stage.
# Define valid options for various stages
PRIO_OPTS_ADD = [
DEFAULT_PRIORITY,
PAUSED_PRIORITY,
LOW_PRIORITY,
NORMAL_PRIORITY,
HIGH_PRIORITY,
FORCE_PRIORITY,
None,
]
PRIO_OPTS_PREQ = [
DEFAULT_PRIORITY,
PAUSED_PRIORITY,
LOW_PRIORITY,
NORMAL_PRIORITY,
HIGH_PRIORITY,
FORCE_PRIORITY,
None,
]
PRIO_OPTS_ADD_CAT = [
DEFAULT_PRIORITY,
PAUSED_PRIORITY,
LOW_PRIORITY,
NORMAL_PRIORITY,
HIGH_PRIORITY,
FORCE_PRIORITY,
None,
]
PRIO_OPTS_PREQ_CAT = [
DEFAULT_PRIORITY,
PAUSED_PRIORITY,
LOW_PRIORITY,
NORMAL_PRIORITY,
HIGH_PRIORITY,
FORCE_PRIORITY,
None,
]
PRIO_OPTS_META_CAT = [
DEFAULT_PRIORITY,
PAUSED_PRIORITY,
LOW_PRIORITY,
NORMAL_PRIORITY,
HIGH_PRIORITY,
FORCE_PRIORITY,
None,
]
# Valid priority values for the Default category (as determined by their availability from the interface)
VALID_DEFAULT_PRIORITIES = [PAUSED_PRIORITY, LOW_PRIORITY, NORMAL_PRIORITY, HIGH_PRIORITY, FORCE_PRIORITY]
# Priorities that do *not* set a job state
REGULAR_PRIOS = [LOW_PRIORITY, NORMAL_PRIORITY, HIGH_PRIORITY, FORCE_PRIORITY]
# Priorities that set job states
STATE_PRIOS = [PAUSED_PRIORITY]
# Needed for translating priority values to names
ALL_PRIOS = {
DEFAULT_PRIORITY: "Default",
PAUSED_PRIORITY: "Paused",
LOW_PRIORITY: "Low",
NORMAL_PRIORITY: "Normal",
HIGH_PRIORITY: "High",
FORCE_PRIORITY: "Force",
REPAIR_PRIORITY: "Repair",
}
# Tags to randomise category/script/nzb name
CAT_RANDOM = os.urandom(4).hex()
SCRIPT_RANDOM = os.urandom(4).hex()
NZB_RANDOM = os.urandom(4).hex()
class ModuleVars:
# Full path to script directory resp. nzb files, once in place/generated
SCRIPT_DIR = None
NZB_FILE = None
META_NZB_FILE = None
# Pre-queue script setup marker
PRE_QUEUE_SETUP_DONE = False
@pytest.fixture(scope="function")
def pause_and_clear():
# Pause the queue
assert get_api_result(mode="pause")["status"] is True
yield
# Delete all jobs from queue and history
for mode in ("queue", "history"):
get_api_result(mode=mode, extra_arguments={"name": "delete", "value": "all", "del_files": 1})
# Unpause the queue
assert get_api_result(mode="resume")["status"] is True
@pytest.mark.usefixtures("run_sabnzbd", "pause_and_clear")
class TestAddingNZBs(AddingNZBsTestBase):
@cached_property
def config(self):
return ModuleVars()
def _setup_script_dir(self):
self.config.SCRIPT_DIR = os.path.join(SAB_CACHE_DIR, "scripts" + SCRIPT_RANDOM)
try:
os.makedirs(self.config.SCRIPT_DIR, exist_ok=True)
except Exception:
pytest.fail("Cannot create script_dir %s" % self.config.SCRIPT_DIR)
self._api_set_config("script_dir", self.config.SCRIPT_DIR)
def _customize_pre_queue_script(self, priority, category):
"""Add a script that accepts the job and sets priority & category"""
script_name = "SCRIPT%s.py" % SCRIPT_RANDOM
try:
script_path = os.path.join(self.config.SCRIPT_DIR, script_name)
with open(script_path, "w") as f:
# line 1 = accept; 4 = category; 6 = priority
f.write(
"#!%s\n\nprint('1\\n\\n\\n%s\\n\\n%s\\n')"
% (
sys.executable,
(category if category else ""),
(str(priority) if priority is not None else ""),
)
)
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 %s" % (script_name, self.config.SCRIPT_DIR))
if not self.config.PRE_QUEUE_SETUP_DONE:
# Set as pre-queue script
self._api_set_config("pre_script", script_name)
self.config.PRE_QUEUE_SETUP_DONE = True
def _configure_cat(self, priority, tag):
category_name = "cat" + tag + CAT_RANDOM
category_config = {
"section": "categories",
"name": category_name,
"pp": choice(list(PP_LOOKUP.keys())),
"script": "None",
"priority": priority if priority is not None else DEFAULT_PRIORITY,
}
# Add the category
json = get_api_result(mode="set_config", extra_arguments=category_config)
assert json["config"]["categories"][0]["name"] == category_name
if priority is not None:
assert json["config"]["categories"][0]["priority"] == priority
return category_name
def _configure_default_category_priority(self, priority):
if priority not in VALID_DEFAULT_PRIORITIES:
priority = DEFAULT_PRIORITY
json = get_api_result(
mode="set_config",
extra_arguments={
"section": "categories",
"name": "*",
"priority": priority,
},
)
assert ("*", priority) == (json["config"]["categories"][0]["name"], json["config"]["categories"][0]["priority"])
def _create_meta_nzb(self, cat_meta):
return self._create_random_nzb(metadata={"category": cat_meta})
def _expected_results(self, STAGES, return_state=None):
"""Figure out what priority and state the job should end up with"""
# Define a bunch of helpers
def sanitize_stages(hit_stage, STAGES):
# Fallback is always category-based, so nix any explicit priorities (stages 1, 3).
# This is conditional only because explicit priority-upon-adding takes precedence
# over implicit-from-pre-queue, as discussed in #1703.
if not (hit_stage == 4 and STAGES[1] is not None):
STAGES[1] = None
STAGES[3] = None
# If the category was set from pre-queue, it replaces any category set earlier
if hit_stage == 4:
STAGES[2] = None
STAGES[5] = None
if hit_stage == 2:
STAGES[5] = None
return STAGES
def handle_state_prio(hit_stage, STAGES, return_state):
"""Find the priority that should to be set after changing the job state"""
# Keep record of the priority that caused the initial hit (for verification of the job state later on)
if not return_state:
return_state = STAGES[hit_stage]
# No point in trying to find a fallback
if hit_stage == 0:
return NORMAL_PRIORITY, return_state
STAGES = sanitize_stages(hit_stage, STAGES)
# Work forward to find the priority prior to the hit_stage
pre_state_prio = None
pre_state_stage = None
# default cat -> implicit meta -> implicit on add -> explicit on add -> implicit pre-q -> explicit pre-q
for stage in (0, 5, 2, 1, 4, 3):
if stage == hit_stage:
if hit_stage == 1 and STAGES[4] is not None:
# An explicit state-setting priority still has to deal with the category from pre-queue
# for fallback purposes, unlike non-state-setting priorities-on-add that override it.
continue
else:
break
if STAGES[stage] is not None:
pre_state_prio = STAGES[stage]
pre_state_stage = stage
if pre_state_prio is not None and LOW_PRIORITY <= pre_state_prio <= HIGH_PRIORITY:
return pre_state_prio, return_state
else:
# The next-in-line prio is unsuitable; recurse with relevant stages zero'ed out
STAGES[hit_stage] = None
if pre_state_stage:
if pre_state_prio == DEFAULT_PRIORITY:
handle_default_cat(pre_state_stage, STAGES, return_state)
else:
STAGES[pre_state_stage] = None
# Sanitize again, with 'pre_state_stage' as the new hit_stage. This is needed again
# in cases such as hit_stage 3 setting a job state, with a fallback from stage 4.
sanitize_stages(pre_state_stage, STAGES)
return self._expected_results(STAGES, return_state)
def handle_default_cat(hit_stage, STAGES, return_state):
"""Figure out the (category) default priority"""
STAGES = sanitize_stages(hit_stage, STAGES)
# Strip the current -100 hit before recursing
STAGES[hit_stage] = None
return self._expected_results(STAGES, return_state)
# Work backwards through all stages:
# explicit pre-q -> implicit pre-q -> explicit on add -> implicit on add -> implicit meta
for stage in (3, 4, 1, 2, 5):
if STAGES[stage] is not None:
if stage == 4 and STAGES[1] is not None:
# Explicit priority on add takes precedence over implicit-from-pre-queue
continue
if STAGES[stage] in REGULAR_PRIOS:
return STAGES[stage], return_state
if STAGES[stage] in STATE_PRIOS:
return handle_state_prio(stage, STAGES, return_state)
if STAGES[stage] == DEFAULT_PRIORITY:
return handle_default_cat(stage, STAGES, return_state)
# # ...and finally the Default category (stage 0)
if STAGES[0] not in (None, DEFAULT_PRIORITY):
if STAGES[0] in REGULAR_PRIOS:
# Avoid falling back to priority Force after setting a job state
if not (return_state in STATE_PRIOS and STAGES[0] == FORCE_PRIORITY):
return STAGES[0], return_state
else:
return NORMAL_PRIORITY, return_state
if STAGES[0] in STATE_PRIOS:
return handle_state_prio(0, STAGES, return_state)
# The default of defaults...
return NORMAL_PRIORITY, return_state
def _prep_priority_tester(self, prio_def_cat, prio_add, prio_add_cat, prio_preq, prio_preq_cat, prio_meta_cat):
if not self.config.SCRIPT_DIR:
self._setup_script_dir()
if not self.config.NZB_FILE:
self.config.NZB_FILE = self._create_random_nzb()
# Set the priority for the Default category
self._configure_default_category_priority(prio_def_cat)
# Setup categories
cat_meta = None
if prio_meta_cat is not None:
cat_meta = self._configure_cat(prio_meta_cat, "meta")
if not self.config.META_NZB_FILE:
self.config.META_NZB_FILE = self._create_meta_nzb(cat_meta)
cat_add = None
if prio_add_cat is not None:
cat_add = self._configure_cat(prio_add_cat, "add")
cat_preq = None
if prio_preq_cat is not None:
cat_preq = self._configure_cat(prio_preq_cat, "pre")
# Setup the pre-queue script
self._customize_pre_queue_script(prio_preq, cat_preq)
# Queue the job, store the nzo_id
extra = {"name": self.config.META_NZB_FILE if cat_meta else self.config.NZB_FILE}
if cat_add:
extra["cat"] = cat_add
if prio_add is not None:
extra["priority"] = prio_add
nzo_id = ",".join(get_api_result(mode="addlocalfile", extra_arguments=extra)["nzo_ids"])
# Fetch the queue output for the current job
return get_api_result(mode="queue", extra_arguments={"nzo_ids": nzo_id})["queue"]["slots"][0]
def _priority_tester(self, prio_def_cat, prio_add, prio_add_cat, prio_preq, prio_preq_cat, prio_meta_cat):
# Setup the current test job, and fetch its queue output
job = self._prep_priority_tester(prio_def_cat, prio_add, prio_add_cat, prio_preq, prio_preq_cat, prio_meta_cat)
# Determine the expected results
expected_prio, expected_state = self._expected_results(
[prio_def_cat, prio_add, prio_add_cat, prio_preq, prio_preq_cat, prio_meta_cat]
)
# Verify the results; queue output uses a string representation for the priority
assert ALL_PRIOS.get(expected_prio) == job["priority"]
if expected_state:
# Also check the correct state or label was set
if expected_state == PAUSED_PRIORITY:
assert "Paused" == job["status"]
# Caution: a full run is good for 90k+ tests
# @pytest.mark.parametrize("prio_meta_cat", PRIO_OPTS_META_CAT)
# @pytest.mark.parametrize("prio_def_cat", VALID_DEFAULT_PRIORITIES)
# @pytest.mark.parametrize("prio_add", PRIO_OPTS_ADD)
# @pytest.mark.parametrize("prio_add_cat", PRIO_OPTS_ADD_CAT)
# @pytest.mark.parametrize("prio_preq", PRIO_OPTS_PREQ)
# @pytest.mark.parametrize("prio_preq_cat", PRIO_OPTS_PREQ_CAT)
@pytest.mark.parametrize("prio_meta_cat", sample(PRIO_OPTS_META_CAT, 2))
@pytest.mark.parametrize("prio_def_cat", sample(VALID_DEFAULT_PRIORITIES, 2))
@pytest.mark.parametrize("prio_add", sample(PRIO_OPTS_ADD, 3))
@pytest.mark.parametrize("prio_add_cat", sample(PRIO_OPTS_ADD_CAT, 2))
@pytest.mark.parametrize("prio_preq", sample(PRIO_OPTS_PREQ, 2))
@pytest.mark.parametrize("prio_preq_cat", sample(PRIO_OPTS_PREQ_CAT, 2))
def test_adding_nzbs_priority_sample(
self, prio_def_cat, prio_add, prio_add_cat, prio_preq, prio_preq_cat, prio_meta_cat
):
self._priority_tester(prio_def_cat, prio_add, prio_add_cat, prio_preq, prio_preq_cat, prio_meta_cat)
@pytest.mark.parametrize(
"prio_def_cat, prio_add, prio_add_cat, prio_preq, prio_preq_cat, prio_meta_cat",
[
# Specific triggers for fixed bugs
(-1, -2, None, None, None, None), # State-setting priorities always fell back to Normal
(1, None, -2, None, None, None),
(2, None, None, -2, None, None),
(0, 2, None, None, 1, None), # Explicit priority on add was bested by implicit from pre-queue
(1, None, None, None, -1, None), # Category-based values from pre-queue didn't work at all
# Checks for test code regressions
(-2, -100, 2, None, None, None),
(-2, 0, 2, -100, None, None),
(1, 2, 0, -100, None, None),
(-2, None, -2, None, 2, None),
(-2, None, -1, None, 1, None),
(2, None, -1, None, -2, None),
(2, 2, None, -2, None, None),
(2, 1, None, -2, None, None),
(1, -2, 0, None, None, None),
(0, 2, None, -2, None, -1),
(1, -2, -100, None, None, -1),
(1, None, None, None, None, -1),
(-1, None, None, None, None, 1),
(0, None, None, None, None, None),
],
)
def test_adding_nzbs_priority_triggers(
self, prio_def_cat, prio_add, prio_add_cat, prio_preq, prio_preq_cat, prio_meta_cat
):
self._priority_tester(prio_def_cat, prio_add, prio_add_cat, prio_preq, prio_preq_cat, prio_meta_cat)
@pytest.mark.parametrize(
"prio_def_cat, prio_add, prio_add_cat, prio_preq, prio_preq_cat, prio_meta_cat, expected_priority, expected_status",
[
# Added category priority should apply when no explicit priority is set.
(LOW_PRIORITY, None, HIGH_PRIORITY, None, None, None, HIGH_PRIORITY, "Queued"),
# Pre-queue category should override category supplied during add.
(LOW_PRIORITY, None, LOW_PRIORITY, None, HIGH_PRIORITY, None, HIGH_PRIORITY, "Queued"),
# Explicit priority on add should beat implicit category from pre-queue.
(LOW_PRIORITY, NORMAL_PRIORITY, None, None, HIGH_PRIORITY, None, NORMAL_PRIORITY, "Queued"),
# NZB metadata category should be used if no category is provided later.
(LOW_PRIORITY, None, None, None, None, HIGH_PRIORITY, HIGH_PRIORITY, "Queued"),
],
)
def test_adding_nzbs_category_priority_precedence(
self,
prio_def_cat,
prio_add,
prio_add_cat,
prio_preq,
prio_preq_cat,
prio_meta_cat,
expected_priority,
expected_status,
):
job = self._prep_priority_tester(prio_def_cat, prio_add, prio_add_cat, prio_preq, prio_preq_cat, prio_meta_cat)
assert job["priority"] == ALL_PRIOS.get(expected_priority)
assert job["status"] == expected_status
def test_adding_nzbs_partial(self):
"""Test adding parts of an NZB file, cut off somewhere in the middle to simulate
the effects of an interrupted download or bad hardware. Should fail, of course."""
if not self.config.NZB_FILE:
self.config.NZB_FILE = self._create_random_nzb()
nzb_basedir, nzb_basename = os.path.split(self.config.NZB_FILE)
nzb_size = os.stat(self.config.NZB_FILE).st_size
part_size = round(randint(40, 70) / 100 * nzb_size)
first_part = os.path.join(nzb_basedir, "part1_of_" + nzb_basename)
second_part = os.path.join(nzb_basedir, "part2_of_" + nzb_basename)
with open(self.config.NZB_FILE, "rb") as nzb_in:
for nzb_part, chunk in (first_part, part_size), (second_part, -1):
with open(nzb_part, "wb") as nzb_out:
nzb_out.write(nzb_in.read(chunk))
for nzb_part in first_part, second_part:
json = get_api_result(mode="addlocalfile", extra_arguments={"name": nzb_part})
assert json["status"] is False
assert json["nzo_ids"] == []
os.remove(nzb_part)
@pytest.mark.parametrize(
"keep_first, keep_last, strip_first, strip_last, should_work",
[
# Keep parts
(6, 3, 0, 0, False), # Remove all segments content
(6, 0, 0, 0, False),
(5, 2, 0, 0, False), # Remove all segments
(5, 0, 0, 0, False),
(4, 2, 0, 0, False), # Remove all groups
(3, 1, 0, 0, False), # Remove all files
# Strip parts
(0, 0, 1, 0, True), # Strip '?xml' line (survivable)
(0, 0, 2, 0, True), # Also strip 'doctype' line (survivable)
(0, 0, 3, 0, False), # Also strip 'nzb xmlns' line
(0, 0, 0, 1, False), # Forget the 'nzb' closing tag
(0, 0, 0, 2, False), # Also forget the (last) 'file' closing tag
(0, 0, 0, 3, False), # Also forget the (last) 'segment' closing tag
],
)
def test_adding_nzbs_malformed(self, keep_first, keep_last, strip_first, strip_last, should_work):
"""Test adding broken, empty, or otherwise malformed NZB file"""
if not self.config.NZB_FILE:
self.config.NZB_FILE = self._create_random_nzb()
with open(self.config.NZB_FILE, "rt") as nzb_in:
nzb_lines = nzb_in.readlines()
assert len(nzb_lines) >= 9
broken_nzb_basename = "broken_" + os.urandom(4).hex() + ".nzb"
broken_nzb = os.path.join(SAB_CACHE_DIR, broken_nzb_basename)
with open(broken_nzb, "wt") as nzb_out:
# Keep only first x, last y lines
if keep_first:
nzb_out.write("".join(nzb_lines[:keep_first]))
elif strip_first:
nzb_out.write("".join(nzb_lines[strip_first:]))
if keep_last:
nzb_out.write("".join(nzb_lines[(-1 * keep_last) :]))
elif strip_last:
nzb_out.write("".join(nzb_lines[: (-1 * strip_last)]))
json = get_api_result(mode="warnings", extra_arguments={"name": "clear"})
json = get_api_result(mode="addlocalfile", extra_arguments={"name": broken_nzb})
assert json["status"] is should_work
assert len(json["nzo_ids"]) == int(should_work)
json = get_api_result(mode="warnings")
assert (len(json["warnings"]) == 0) is should_work
if not should_work:
for warning in range(0, len(json["warnings"])):
warn_text = json["warnings"][warning]["text"]
assert "Empty NZB" in warn_text or "Failed to import" in warn_text or "Invalid NZB" in warn_text
assert broken_nzb_basename in warn_text
os.remove(broken_nzb)
@pytest.mark.parametrize("prio_meta_cat", sample(PRIO_OPTS_META_CAT, 1))
@pytest.mark.parametrize("prio_def_cat", sample(VALID_DEFAULT_PRIORITIES, 1))
@pytest.mark.parametrize("prio_add", PRIO_OPTS_ADD)
def test_adding_nzbs_size_limit(self, prio_meta_cat, prio_def_cat, prio_add):
"""Verify state and priority of a job exceeding the size_limit"""
# Set size limit
self._api_set_config("size_limit", str(MIN_FILESIZE - 1))
job = self._prep_priority_tester(prio_def_cat, prio_add, None, None, None, prio_meta_cat)
# Verify job is paused and low priority, and correctly labeled
assert job["status"] == "Paused"
assert job["priority"] == ALL_PRIOS.get(-1)
assert "TOO LARGE" in job["labels"]
# Unset size limit
self._api_set_config("size_limit", "")
@pytest.mark.parametrize("prio_def_cat", sample(VALID_DEFAULT_PRIORITIES, 2))
@pytest.mark.parametrize("prio_add", PRIO_OPTS_ADD)
@pytest.mark.parametrize("prio_add_cat", sample(PRIO_OPTS_ADD_CAT, 1))
@pytest.mark.parametrize("prio_preq", sample(PRIO_OPTS_PREQ, 1))
@pytest.mark.parametrize("prio_preq_cat", sample(PRIO_OPTS_PREQ_CAT, 2))
def test_adding_nzbs_duplicate_pausing(self, prio_def_cat, prio_add, prio_add_cat, prio_preq, prio_preq_cat):
# Set an nzb backup directory
backup_dir = self._add_backup_directory()
# Add the job a first time
job = self._prep_priority_tester(None, None, None, None, None, None)
assert job["status"] == "Queued"
# Setup duplicate handling to 2 (Pause)
self._api_set_config("no_dupes", 2)
expected_prio, _ = self._expected_results(
[prio_def_cat, prio_add, prio_add_cat, prio_preq, prio_preq_cat, None]
)
job = self._prep_priority_tester(prio_def_cat, prio_add, prio_add_cat, prio_preq, prio_preq_cat, None)
# Verify job is paused and correctly labeled, and given the right (fallback) priority
assert job["priority"] == ALL_PRIOS.get(expected_prio)
if expected_prio == FORCE_PRIORITY:
assert "DUPLICATE" not in job["labels"]
assert "ALTERNATIVE" not in job["labels"]
assert job["status"] == "Downloading"
else:
assert "ALTERNATIVE" in job["labels"]
assert job["status"] == "Paused"
self._clear_and_reset_backup_directory(backup_dir)