mirror of
https://github.com/sabnzbd/sabnzbd.git
synced 2026-09-22 02:45:20 -04:00
mnightingale
59218a687d
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.
121 lines
5.0 KiB
Python
121 lines
5.0 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_clean - Tests for settings interaction when adding NZBs (clean SABnzbd instance)
|
|
"""
|
|
|
|
import os
|
|
from zipfile import ZipFile
|
|
|
|
import pytest
|
|
|
|
from sabnzbd.constants import STOP_PRIORITY
|
|
from tests.testhelper import AddingNZBsTestBase, get_api_result, wait_for
|
|
|
|
|
|
def get_slot(mode: str, nzo_id: str):
|
|
"""Return the queue or history slot of the job, if it has one"""
|
|
if slots := get_api_result(mode=mode, extra_arguments={"nzo_ids": nzo_id})[mode]["slots"]:
|
|
return slots[0]
|
|
return None
|
|
|
|
|
|
@pytest.mark.usefixtures("run_sabnzbd")
|
|
class TestAddingNZBsClean(AddingNZBsTestBase):
|
|
def test_adding_nzbs_nzoids(self):
|
|
"""Test if we return the right output"""
|
|
# Create NZB and zipped version
|
|
basenzbfile = self._create_random_nzb()
|
|
zipnzbfile = basenzbfile + ".zip"
|
|
with ZipFile(zipnzbfile, "w") as zipobj:
|
|
zipobj.write(basenzbfile)
|
|
assert os.path.exists(zipnzbfile)
|
|
|
|
# Test for both normal and zipped version
|
|
for nzbfile in (basenzbfile, zipnzbfile):
|
|
# Pause the queue at first
|
|
assert get_api_result(mode="pause")["status"] is True
|
|
|
|
# Add the job a first time
|
|
job1 = get_api_result(mode="addlocalfile", extra_arguments={"name": nzbfile})
|
|
assert job1["status"]
|
|
assert job1["nzo_ids"]
|
|
|
|
# 1=Discard
|
|
self._api_set_config("no_dupes", 1)
|
|
|
|
# Add the job a second time, it should be added with ALTERNATIVE label
|
|
job2 = get_api_result(mode="addlocalfile", extra_arguments={"name": nzbfile})
|
|
assert job2["status"]
|
|
assert job2["nzo_ids"]
|
|
queue = get_api_result(mode="queue", extra_arguments={"nzo_ids": job2["nzo_ids"][0]})
|
|
job_in_queue = queue["queue"]["slots"][0]
|
|
assert "ALTERNATIVE" in job_in_queue["labels"]
|
|
assert job_in_queue["status"] == "Paused"
|
|
|
|
# Stop the first job
|
|
get_api_result(
|
|
mode="queue", extra_arguments={"name": "priority", "value": job1["nzo_ids"][0], "value2": STOP_PRIORITY}
|
|
)
|
|
|
|
# Wait for the job to be removed and appear in the history
|
|
wait_for(
|
|
lambda: (slot := get_slot("history", job1["nzo_ids"][0])) and slot["status"] == "Failed",
|
|
timeout=10,
|
|
err_msg="Job did not appear in history",
|
|
)
|
|
|
|
def alternative_released():
|
|
"""Jobs show up in the history as soon as they enter post-processing, but their
|
|
alternatives are only released at the end of it, so we have to wait for that"""
|
|
if (slot := get_slot("queue", job2["nzo_ids"][0])) and "ALTERNATIVE" not in slot["labels"]:
|
|
return slot
|
|
|
|
# Now the second job should no longer be paused and labelled
|
|
job_in_queue = wait_for(alternative_released, timeout=10, err_msg="Duplicate alternative was not released")
|
|
assert job_in_queue["status"] == "Queued"
|
|
|
|
# Reset duplicate detection
|
|
self._api_set_config("no_dupes", 0)
|
|
|
|
# Test unwanted extensions Fail to history
|
|
self._api_set_config("unwanted_extensions", ["bin"])
|
|
self._api_set_config("action_on_unwanted_extensions", 2)
|
|
job = get_api_result(mode="addlocalfile", extra_arguments={"name": nzbfile})
|
|
assert job["status"]
|
|
assert job["nzo_ids"]
|
|
|
|
# Wait for the job to be removed and appear in the history
|
|
wait_for(
|
|
lambda: not get_slot("queue", job["nzo_ids"][0])
|
|
and (slot := get_slot("history", job["nzo_ids"][0]))
|
|
and slot["status"] == "Failed",
|
|
timeout=10,
|
|
err_msg="Job did not appear in history",
|
|
)
|
|
|
|
# Reset and clean up
|
|
get_api_result(
|
|
mode="set_config_default",
|
|
extra_arguments={"keyword": ["unwanted_extensions", "action_on_unwanted_extensions"]},
|
|
)
|
|
|
|
# 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})
|