mirror of
https://github.com/sabnzbd/sabnzbd.git
synced 2026-09-22 10:55:33 -04:00
Previous HTML stripping could allow malformed or unclosed tags to pass through, potentially leading to Cross-Site Scripting (XSS) vulnerabilities if completed by the browser. This change updates the regex to aggressively remove all HTML tags, even incomplete ones. The `knockout-extensions.js` code also now universally escapes HTML, removing a potential bypass. New tests confirm this behavior.
599 lines
25 KiB
Python
599 lines
25 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_postproc- Tests of various functions in newspack, among which rar_renamer()
|
|
"""
|
|
|
|
import os
|
|
import re
|
|
import shutil
|
|
from unittest import mock
|
|
|
|
import pytest
|
|
|
|
import sabnzbd
|
|
import sabnzbd.config
|
|
from sabnzbd import postproc
|
|
from sabnzbd.config import ConfigCat, ConfigSorter
|
|
from sabnzbd.filesystem import clip_path, globber_full
|
|
from sabnzbd.misc import sort_to_opts
|
|
from tests.testhelper import SAB_CACHE_DIR, SAB_DATA_DIR
|
|
|
|
|
|
@pytest.mark.usefixtures("clean_cache_dir")
|
|
class TestPostProc:
|
|
# Helper function for rar_renamer tests
|
|
def _deobfuscate_dir(self, sourcedir, expected_filename_matches):
|
|
"""Function to deobfuscate one directory with rar_renamer()"""
|
|
# We create a workingdir inside the sourcedir, because the filenames are really changed
|
|
workingdir = os.path.join(SAB_CACHE_DIR, "workingdir_test_rar_renamer")
|
|
|
|
# if workingdir is still there from previous run, remove it:
|
|
if os.path.isdir(workingdir):
|
|
try:
|
|
shutil.rmtree(workingdir)
|
|
except PermissionError:
|
|
pytest.fail("Could not remove existing workingdir %s for rar_renamer" % workingdir)
|
|
|
|
# create a fresh copy
|
|
try:
|
|
shutil.copytree(sourcedir, workingdir)
|
|
except Exception:
|
|
pytest.fail("Could not create copy of files for rar_renamer")
|
|
|
|
# And now let the magic happen:
|
|
nzo = mock.Mock()
|
|
nzo.final_name = "somedownloadname"
|
|
nzo.download_path = workingdir
|
|
number_renamed_files = postproc.rar_renamer(nzo)
|
|
|
|
# run check on the resulting files
|
|
if expected_filename_matches:
|
|
for filename_match in expected_filename_matches:
|
|
if len(globber_full(workingdir, filename_match)) != expected_filename_matches[filename_match]:
|
|
pytest.fail("Failed filename_match %s in %s" % (filename_match, workingdir))
|
|
|
|
# Remove workingdir again
|
|
try:
|
|
shutil.rmtree(workingdir)
|
|
except Exception:
|
|
pytest.fail("Could not remove existing workingdir %s for rar_renamer" % workingdir)
|
|
|
|
return number_renamed_files
|
|
|
|
def test_rar_renamer_obfuscated_single_rar_set(self):
|
|
"""Test rar_renamer with obfuscated single rar set"""
|
|
sourcedir = os.path.join(SAB_DATA_DIR, "obfuscated_single_rar_set")
|
|
# Now define the filematches we want to see, in which amount ("*-*-*-*-*" are the input files):
|
|
expected_filename_matches = {"*part007.rar": 1, "*-*-*-*-*": 0}
|
|
assert self._deobfuscate_dir(sourcedir, expected_filename_matches) == 7
|
|
|
|
def test_rar_renamer_obfuscated_two_rar_sets(self):
|
|
"""Test rar_renamer with obfuscated two rar sets"""
|
|
sourcedir = os.path.join(SAB_DATA_DIR, "obfuscated_two_rar_sets")
|
|
expected_filename_matches = {"*part007.rar": 2, "*part009.rar": 1, "*-*-*-*-*": 0}
|
|
assert self._deobfuscate_dir(sourcedir, expected_filename_matches) == 16
|
|
|
|
def test_rar_renamer_obfuscated_but_no_rar(self):
|
|
"""Test rar_renamer with obfuscated files that are not rar sets"""
|
|
sourcedir = os.path.join(SAB_DATA_DIR, "obfuscated_but_no_rar")
|
|
expected_filename_matches = {"*.rar": 0, "*-*-*-*-*": 6}
|
|
assert self._deobfuscate_dir(sourcedir, expected_filename_matches) == 0
|
|
|
|
def test_rar_renamer_single_rar_set_missing_first_rar(self):
|
|
"""Test rar_renamer with single rar set missing first rar"""
|
|
# One obfuscated rar set, but first rar (.part1.rar) is missing
|
|
sourcedir = os.path.join(SAB_DATA_DIR, "obfuscated_single_rar_set_missing_first_rar")
|
|
# single rar set (of 6 obfuscated rar files), so we expect renaming
|
|
# thus result must 6 rar files, and 0 obfuscated files
|
|
expected_filename_matches = {"*.rar": 6, "*-*-*-*-*": 0}
|
|
# 6 files should have been renamed
|
|
assert self._deobfuscate_dir(sourcedir, expected_filename_matches) == 6
|
|
|
|
def test_rar_renamer_double_rar_set_missing_rar(self):
|
|
"""Test rar_renamer with two rar sets where some rars are missing"""
|
|
# Two obfuscated rar sets, but some rars are missing
|
|
sourcedir = os.path.join(SAB_DATA_DIR, "obfuscated_double_rar_set_missing_rar")
|
|
# Two sets, missing rar, so we expect no renaming
|
|
# thus result should be 0 rar files, and still 8 obfuscated files
|
|
expected_filename_matches = {"*.rar": 0, "*-*-*-*-*": 8}
|
|
# 0 files should have been renamed
|
|
assert self._deobfuscate_dir(sourcedir, expected_filename_matches) == 0
|
|
|
|
def test_rar_renamer_fully_encrypted_and_obfuscated(self):
|
|
"""Test rar_renamer with fully encrypted and obfuscated rar set"""
|
|
# fully encrypted rar-set, and obfuscated rar names
|
|
sourcedir = os.path.join(SAB_DATA_DIR, "fully_encrypted_and_obfuscated_rars")
|
|
# SABnzbd cannot do anything with this, so we expect no renaming
|
|
expected_filename_matches = {"*.rar": 0, "*-*-*-*-*": 6}
|
|
# 0 files should have been renamed
|
|
assert self._deobfuscate_dir(sourcedir, expected_filename_matches) == 0
|
|
|
|
@pytest.mark.parametrize("category", ["testcat", "Default", None])
|
|
@pytest.mark.parametrize("has_jobdir", [True, False]) # With or without a job dir
|
|
@pytest.mark.parametrize("has_catdir", [True, False]) # Complete directory is defined at category level
|
|
@pytest.mark.parametrize("has_active_sorter", [True, False]) # Sorter active for the fake nzo
|
|
@pytest.mark.parametrize("sort_string", ["%sn (%r)", "%sn (%r)/file.%ext", ""]) # Identical path result
|
|
@pytest.mark.parametrize("marker_file", [None, ".marker"])
|
|
@pytest.mark.parametrize("do_folder_rename", [True, False])
|
|
@pytest.mark.config(
|
|
lambda params: {
|
|
"marker_file": params["marker_file"],
|
|
"folder_rename": params["do_folder_rename"],
|
|
"download_dir": os.path.join(SAB_CACHE_DIR, "incomplete"),
|
|
"complete_dir": os.path.join(SAB_CACHE_DIR, "complete"),
|
|
}
|
|
)
|
|
def test_prepare_extraction_path(
|
|
self, category, has_jobdir, has_catdir, has_active_sorter, sort_string, marker_file, do_folder_rename
|
|
):
|
|
# Every parametrization asserts on the exact directory name that comes out, but that
|
|
# name depends on what's already on disk. Best effort cleanup.
|
|
shutil.rmtree(SAB_CACHE_DIR, ignore_errors=True)
|
|
os.makedirs(SAB_CACHE_DIR, exist_ok=True)
|
|
|
|
# Ensure global CFG_ vars are initialised
|
|
sabnzbd.config.read_config(os.devnull)
|
|
|
|
# Define a sorter and a category (as @pytest.mark.config cannot handle those)
|
|
ConfigSorter(
|
|
"sorter__test_prepare_extraction_path",
|
|
{
|
|
"order": 0,
|
|
"min_size": 42,
|
|
"multipart_label": "",
|
|
"sort_string": sort_string,
|
|
"sort_cats": [category if category else "no_such_category"],
|
|
"sort_type": [
|
|
sort_to_opts("all"),
|
|
],
|
|
"is_active": int(has_active_sorter),
|
|
},
|
|
)
|
|
assert sabnzbd.config.CONFIG.database["sorters"]["sorter__test_prepare_extraction_path"]
|
|
|
|
if category:
|
|
ConfigCat(
|
|
category,
|
|
{
|
|
"order": 0,
|
|
"pp": None,
|
|
"script": None,
|
|
"dir": (
|
|
os.path.join(SAB_CACHE_DIR, ("category_dir_for_" + category + ("*" if not has_jobdir else "")))
|
|
if has_catdir
|
|
else None
|
|
),
|
|
"newzbin": "",
|
|
"priority": 0,
|
|
},
|
|
)
|
|
assert sabnzbd.config.CONFIG.database["categories"][category]
|
|
|
|
# Mock a minimal nzo, required as function input
|
|
fake_nzo = mock.Mock()
|
|
fake_nzo.final_name = "FOSS.Rules.S23E06.2160p-SABnzbd"
|
|
fake_nzo.cat = category
|
|
fake_nzo.nzo_info = {} # Placeholder to prevent a crash in sorting.get_titles()
|
|
|
|
def _func():
|
|
(
|
|
tmp_workdir_complete,
|
|
workdir_complete,
|
|
file_sorter,
|
|
not_create_job_dir,
|
|
marker_file_result,
|
|
) = postproc.prepare_extraction_path(fake_nzo)
|
|
|
|
tmp_workdir_complete = clip_path(tmp_workdir_complete)
|
|
workdir_complete = clip_path(workdir_complete)
|
|
|
|
# Verify marker file
|
|
if marker_file and not not_create_job_dir:
|
|
assert marker_file_result
|
|
else:
|
|
assert not marker_file_result
|
|
|
|
# Verify sorter
|
|
assert file_sorter
|
|
if has_active_sorter and category and sort_string:
|
|
assert file_sorter.sorter_active
|
|
else:
|
|
assert not file_sorter.sorter_active
|
|
|
|
# Verify not_create_job_dir
|
|
if category and has_catdir and not has_jobdir and not file_sorter.sorter_active:
|
|
assert not_create_job_dir
|
|
else:
|
|
# Double negatives ftw
|
|
assert not not_create_job_dir
|
|
|
|
# Verify workdir_complete
|
|
if not category or not has_catdir:
|
|
# Using standard Complete directory as base
|
|
assert workdir_complete.startswith(os.path.join(SAB_CACHE_DIR, "complete"))
|
|
elif category and has_catdir:
|
|
# Based on the category directory
|
|
assert workdir_complete.startswith(os.path.join(SAB_CACHE_DIR, "category_dir_for_" + category))
|
|
# Check the job directory part (or the lack thereof) as well
|
|
if has_active_sorter and category and sort_string:
|
|
# Sorter path, with an extra job name work directory inside
|
|
assert re.fullmatch(
|
|
re.escape(SAB_CACHE_DIR)
|
|
+ r".*"
|
|
+ re.escape(os.sep)
|
|
+ r"Foss Rules \(2160p\)"
|
|
+ re.escape(os.sep)
|
|
+ fake_nzo.final_name
|
|
+ r"(\.\d+)?",
|
|
workdir_complete,
|
|
)
|
|
elif has_jobdir or not (category and has_catdir):
|
|
# Standard job name directory
|
|
assert re.fullmatch(
|
|
re.escape(SAB_CACHE_DIR) + r".*" + re.escape(os.sep) + r"FOSS.Rules.S23E06.2160p-SABnzbd(\.\d+)?",
|
|
workdir_complete,
|
|
)
|
|
else:
|
|
# No job directory at all
|
|
assert re.fullmatch(
|
|
re.escape(SAB_CACHE_DIR) + r".*" + re.escape(os.sep) + r"category_dir_for_([a-zA-Z]+)",
|
|
workdir_complete,
|
|
)
|
|
|
|
# Verify tmp_workdir_complete
|
|
if do_folder_rename:
|
|
if not not_create_job_dir:
|
|
assert tmp_workdir_complete != workdir_complete
|
|
assert tmp_workdir_complete.replace("_UNPACK_", "") == workdir_complete
|
|
else:
|
|
assert tmp_workdir_complete == workdir_complete
|
|
|
|
_func()
|
|
|
|
|
|
@pytest.mark.usefixtures("clean_cache_dir")
|
|
class TestGetLastLine:
|
|
@pytest.mark.parametrize(
|
|
"txt, expected",
|
|
[
|
|
("", ""),
|
|
("plain text", "plain text"),
|
|
("first\nsecond\nlast", "last"),
|
|
("last\n\n\n", "last"),
|
|
("<b>bold</b>", "bold"),
|
|
("<a href='x'>link</a>", "link"),
|
|
# Unclosed tags must not survive to be completed by later markup
|
|
("<img src=x onerror=alert(apiKey)", ""),
|
|
("text <script>evil", "text evil"),
|
|
("<img src=x onerror=alert(1)\nafter", "after"),
|
|
("<div\n<span", ""),
|
|
],
|
|
)
|
|
def test_get_last_line(self, txt, expected):
|
|
result = postproc.get_last_line(txt)
|
|
assert result == expected
|
|
assert "<" not in result
|
|
|
|
def test_get_last_line_truncates(self):
|
|
result = postproc.get_last_line("x" * 200)
|
|
assert len(result) == 150
|
|
assert result.endswith("...")
|
|
|
|
|
|
class TestCleanupList:
|
|
@staticmethod
|
|
def _create_file(path):
|
|
os.makedirs(os.path.dirname(path), exist_ok=True)
|
|
with open(path, "wb") as f:
|
|
f.write(b"data")
|
|
assert os.path.exists(path)
|
|
return path
|
|
|
|
@pytest.mark.config({"cleanup_list": ["exe", "nzb"]})
|
|
def test_cleanup_list_only_tracked_files(self):
|
|
"""Only files of the job may be removed, not other files in the same folder"""
|
|
base_dir = os.path.join(SAB_CACHE_DIR, "complete_cleanup")
|
|
job_files = [self._create_file(os.path.join(base_dir, f)) for f in ("job.mkv", "job.exe", "job.nzb")]
|
|
# Files matching the cleanup list, but not part of the job
|
|
unrelated_file = self._create_file(os.path.join(base_dir, "MyApplication", "fake.exe"))
|
|
# Tracked files that no longer exist shouldn't cause problems
|
|
gone_file = os.path.join(base_dir, "gone.exe")
|
|
|
|
remaining_files = postproc.cleanup_list([*job_files, gone_file], base_dir, skip_nzb=False)
|
|
|
|
assert remaining_files == [job_files[0], gone_file]
|
|
assert os.path.exists(job_files[0])
|
|
assert not os.path.exists(job_files[1])
|
|
assert not os.path.exists(job_files[2])
|
|
assert os.path.exists(unrelated_file)
|
|
|
|
@pytest.mark.config({"cleanup_list": ["exe", "nzb"]})
|
|
def test_cleanup_list_skip_nzb(self):
|
|
"""The nzb extension should be ignored when skip_nzb is set"""
|
|
base_dir = os.path.join(SAB_CACHE_DIR, "complete_cleanup")
|
|
job_files = [self._create_file(os.path.join(base_dir, f)) for f in ("job.exe", "job.nzb")]
|
|
|
|
remaining_files = postproc.cleanup_list(job_files, base_dir, skip_nzb=True)
|
|
|
|
assert remaining_files == [job_files[1]]
|
|
assert not os.path.exists(job_files[0])
|
|
assert os.path.exists(job_files[1])
|
|
|
|
@pytest.mark.config({"cleanup_list": ["nfo"]})
|
|
def test_cleanup_list_removes_empty_parent_dirs(self):
|
|
"""Directories left empty by the removed files are pruned, but never the base directory"""
|
|
base_dir = os.path.join(SAB_CACHE_DIR, "complete_cleanup")
|
|
job_file = self._create_file(os.path.join(base_dir, "sub", "deeper", "job.nfo"))
|
|
# Unrelated empty directory should be left alone
|
|
unrelated_dir = os.path.join(base_dir, "empty_dir")
|
|
os.makedirs(unrelated_dir, exist_ok=True)
|
|
|
|
assert postproc.cleanup_list([job_file], base_dir, skip_nzb=False) == []
|
|
assert not os.path.exists(os.path.join(base_dir, "sub"))
|
|
assert os.path.exists(unrelated_dir)
|
|
assert os.path.exists(base_dir)
|
|
|
|
@pytest.mark.config({"cleanup_list": ["images/*"]})
|
|
def test_cleanup_list_path_patterns(self):
|
|
"""Path patterns are matched relative to the base directory"""
|
|
base_dir = os.path.join(SAB_CACHE_DIR, "complete_cleanup")
|
|
job_files = [
|
|
self._create_file(os.path.join(base_dir, f)) for f in (os.path.join("images", "pic.jpg"), "pic.jpg")
|
|
]
|
|
|
|
remaining_files = postproc.cleanup_list(job_files, base_dir, skip_nzb=False)
|
|
|
|
assert remaining_files == [job_files[1]]
|
|
assert not os.path.exists(job_files[0])
|
|
assert os.path.exists(job_files[1])
|
|
|
|
@pytest.mark.config({"cleanup_list": []})
|
|
def test_cleanup_list_disabled(self):
|
|
"""Without a cleanup list, nothing is removed"""
|
|
base_dir = os.path.join(SAB_CACHE_DIR, "complete_cleanup")
|
|
job_files = [self._create_file(os.path.join(base_dir, "job.exe"))]
|
|
|
|
assert postproc.cleanup_list(job_files, base_dir, skip_nzb=False) == job_files
|
|
assert os.path.exists(job_files[0])
|
|
|
|
|
|
@pytest.mark.usefixtures("clean_cache_dir")
|
|
class TestRemoveUnwantedFiles:
|
|
@staticmethod
|
|
def _create_file(path):
|
|
os.makedirs(os.path.dirname(path), exist_ok=True)
|
|
with open(path, "wb") as f:
|
|
f.write(b"data")
|
|
assert os.path.exists(path)
|
|
return path
|
|
|
|
@staticmethod
|
|
def _fake_nzo(unwanted_ext=0):
|
|
fake_nzo = mock.Mock()
|
|
fake_nzo.final_name = "TestDownload"
|
|
fake_nzo.unwanted_ext = unwanted_ext
|
|
return fake_nzo
|
|
|
|
@pytest.mark.config({"unwanted_extensions": ["exe"], "action_on_unwanted_extensions": 2})
|
|
def test_remove_unwanted_files_only_tracked_files(self):
|
|
"""Only files of the job may be removed, not other files in the same folder"""
|
|
base_dir = os.path.join(SAB_CACHE_DIR, "complete_unwanted")
|
|
job_files = [self._create_file(os.path.join(base_dir, f)) for f in ("job.mkv", "job.exe")]
|
|
# File matching the unwanted extensions, but not part of the job
|
|
unrelated_file = self._create_file(os.path.join(base_dir, "MyApplication", "fake.exe"))
|
|
# Tracked files that no longer exist shouldn't cause problems
|
|
gone_file = os.path.join(base_dir, "gone.exe")
|
|
fake_nzo = self._fake_nzo()
|
|
|
|
remaining_files = postproc.remove_unwanted_files(fake_nzo, [*job_files, gone_file], base_dir)
|
|
|
|
assert remaining_files == [job_files[0], gone_file]
|
|
assert os.path.exists(job_files[0])
|
|
assert not os.path.exists(job_files[1])
|
|
assert os.path.exists(unrelated_file)
|
|
fake_nzo.set_unpack_info.assert_called_once()
|
|
|
|
@pytest.mark.config({"unwanted_extensions": ["exe"], "action_on_unwanted_extensions": 1})
|
|
def test_remove_unwanted_files_removes_empty_parent_dirs(self):
|
|
"""Directories left empty by the removed files are pruned, but never the base directory"""
|
|
base_dir = os.path.join(SAB_CACHE_DIR, "complete_unwanted")
|
|
job_file = self._create_file(os.path.join(base_dir, "sub", "deeper", "job.exe"))
|
|
|
|
assert postproc.remove_unwanted_files(self._fake_nzo(), [job_file], base_dir) == []
|
|
assert not os.path.exists(os.path.join(base_dir, "sub"))
|
|
assert os.path.exists(base_dir)
|
|
|
|
@pytest.mark.config(
|
|
{"unwanted_extensions": ["mkv", "srt"], "unwanted_extensions_mode": 1, "action_on_unwanted_extensions": 2}
|
|
)
|
|
def test_remove_unwanted_files_whitelist_mode(self):
|
|
"""In whitelist mode anything not listed is removed, except files without extension"""
|
|
base_dir = os.path.join(SAB_CACHE_DIR, "complete_unwanted")
|
|
job_files = [self._create_file(os.path.join(base_dir, f)) for f in ("job.mkv", "job.exe", "no_extension")]
|
|
|
|
remaining_files = postproc.remove_unwanted_files(self._fake_nzo(), job_files, base_dir)
|
|
|
|
assert remaining_files == [job_files[0], job_files[2]]
|
|
assert os.path.exists(job_files[0])
|
|
assert not os.path.exists(job_files[1])
|
|
assert os.path.exists(job_files[2])
|
|
|
|
@pytest.mark.config({"unwanted_extensions": ["exe"], "action_on_unwanted_extensions": 0})
|
|
def test_remove_unwanted_files_no_action_configured(self):
|
|
"""Without an action configured, nothing is removed"""
|
|
base_dir = os.path.join(SAB_CACHE_DIR, "complete_unwanted")
|
|
job_files = [self._create_file(os.path.join(base_dir, "job.exe"))]
|
|
fake_nzo = self._fake_nzo()
|
|
|
|
assert postproc.remove_unwanted_files(fake_nzo, job_files, base_dir) == job_files
|
|
assert os.path.exists(job_files[0])
|
|
fake_nzo.set_unpack_info.assert_not_called()
|
|
|
|
@pytest.mark.config({"unwanted_extensions": [], "action_on_unwanted_extensions": 2})
|
|
def test_remove_unwanted_files_no_extensions_configured(self):
|
|
"""Without unwanted extensions configured, nothing is removed"""
|
|
base_dir = os.path.join(SAB_CACHE_DIR, "complete_unwanted")
|
|
job_files = [self._create_file(os.path.join(base_dir, "job.exe"))]
|
|
|
|
assert postproc.remove_unwanted_files(self._fake_nzo(), job_files, base_dir) == job_files
|
|
assert os.path.exists(job_files[0])
|
|
|
|
@pytest.mark.config({"unwanted_extensions": ["exe"], "action_on_unwanted_extensions": 2})
|
|
def test_remove_unwanted_files_user_override(self):
|
|
"""Skip the check after the user resumed a job paused for an unwanted extension"""
|
|
base_dir = os.path.join(SAB_CACHE_DIR, "complete_unwanted")
|
|
job_files = [self._create_file(os.path.join(base_dir, "job.exe"))]
|
|
|
|
assert postproc.remove_unwanted_files(self._fake_nzo(unwanted_ext=2), job_files, base_dir) == job_files
|
|
assert os.path.exists(job_files[0])
|
|
|
|
|
|
class TestNzbOnlyDownload:
|
|
@mock.patch("sabnzbd.postproc.process_single_nzb")
|
|
@mock.patch("sabnzbd.postproc.listdir_full")
|
|
def test_process_nzb_only_download_single_nzb(self, mock_listdir, mock_process_single_nzb):
|
|
"""Test process_nzb_only_download with a single NZB file"""
|
|
# Setup mock NZO
|
|
fake_nzo = mock.Mock()
|
|
fake_nzo.final_name = "TestDownload"
|
|
fake_nzo.pp = 3
|
|
fake_nzo.script = "test_script.py"
|
|
fake_nzo.cat = "movies"
|
|
fake_nzo.url = "http://example.com/test.nzb"
|
|
fake_nzo.priority = 0
|
|
|
|
# Mock single NZB file
|
|
workdir = os.path.join(SAB_CACHE_DIR, "test_workdir")
|
|
nzb_file = os.path.join(workdir, "test.nzb")
|
|
mock_listdir.return_value = [nzb_file]
|
|
|
|
# Call the function
|
|
result = postproc.process_nzb_only_download(workdir, fake_nzo)
|
|
|
|
# Verify result
|
|
assert result == [nzb_file]
|
|
|
|
# Verify process_single_nzb was called with correct arguments
|
|
mock_process_single_nzb.assert_called_once_with(
|
|
"test.nzb",
|
|
nzb_file,
|
|
pp=3,
|
|
script="test_script.py",
|
|
cat="movies",
|
|
url="http://example.com/test.nzb",
|
|
priority=0,
|
|
nzbname="TestDownload",
|
|
dup_check=False,
|
|
)
|
|
|
|
@mock.patch("sabnzbd.postproc.process_single_nzb")
|
|
@mock.patch("sabnzbd.postproc.listdir_full")
|
|
def test_process_nzb_only_download_multiple_nzbs(self, mock_listdir, mock_process_single_nzb):
|
|
"""Test process_nzb_only_download with multiple NZB files"""
|
|
# Setup mock NZO
|
|
fake_nzo = mock.Mock()
|
|
fake_nzo.final_name = "TestDownload"
|
|
fake_nzo.pp = 2
|
|
fake_nzo.script = None
|
|
fake_nzo.cat = "tv"
|
|
fake_nzo.url = "http://example.com/test.nzb"
|
|
fake_nzo.priority = 1
|
|
|
|
# Mock multiple NZB files
|
|
workdir = os.path.join(SAB_CACHE_DIR, "test_workdir")
|
|
first_nzb = os.path.join(workdir, "first.nzb")
|
|
second_nzb = os.path.join(workdir, "second.nzb")
|
|
mock_listdir.return_value = [first_nzb, second_nzb]
|
|
|
|
# Call the function
|
|
result = postproc.process_nzb_only_download(workdir, fake_nzo)
|
|
|
|
# Verify result
|
|
assert result == [first_nzb, second_nzb]
|
|
|
|
# Verify process_single_nzb was called twice with correct arguments
|
|
assert mock_process_single_nzb.call_count == 2
|
|
mock_process_single_nzb.assert_any_call(
|
|
"first.nzb",
|
|
first_nzb,
|
|
pp=2,
|
|
script=None,
|
|
cat="tv",
|
|
url="http://example.com/test.nzb",
|
|
priority=1,
|
|
nzbname="TestDownload - first.nzb",
|
|
dup_check=False,
|
|
)
|
|
mock_process_single_nzb.assert_any_call(
|
|
"second.nzb",
|
|
second_nzb,
|
|
pp=2,
|
|
script=None,
|
|
cat="tv",
|
|
url="http://example.com/test.nzb",
|
|
priority=1,
|
|
nzbname="TestDownload - second.nzb",
|
|
dup_check=False,
|
|
)
|
|
|
|
@mock.patch("sabnzbd.postproc.process_single_nzb")
|
|
@mock.patch("sabnzbd.postproc.listdir_full")
|
|
def test_process_nzb_only_download_mixed_files(self, mock_listdir, mock_process_single_nzb):
|
|
"""Test process_nzb_only_download with mixed file types returns None"""
|
|
# Setup mock NZO
|
|
fake_nzo = mock.Mock()
|
|
fake_nzo.final_name = "TestDownload"
|
|
|
|
# Mock mixed files (NZB and non-NZB)
|
|
workdir = os.path.join(SAB_CACHE_DIR, "test_workdir")
|
|
mock_listdir.return_value = [
|
|
os.path.join(workdir, "test.nzb"),
|
|
os.path.join(workdir, "readme.txt"),
|
|
]
|
|
|
|
# Call the function
|
|
result = postproc.process_nzb_only_download(workdir, fake_nzo)
|
|
|
|
# Verify result is None (not NZB-only)
|
|
assert result is None
|
|
|
|
# Verify process_single_nzb was NOT called
|
|
mock_process_single_nzb.assert_not_called()
|
|
|
|
@mock.patch("sabnzbd.postproc.process_single_nzb")
|
|
@mock.patch("sabnzbd.postproc.listdir_full")
|
|
def test_process_nzb_only_download_empty_directory(self, mock_listdir, mock_process_single_nzb):
|
|
"""Test process_nzb_only_download with empty directory returns None"""
|
|
# Setup mock NZO
|
|
fake_nzo = mock.Mock()
|
|
fake_nzo.final_name = "TestDownload"
|
|
|
|
# Mock empty directory
|
|
workdir = os.path.join(SAB_CACHE_DIR, "test_workdir")
|
|
mock_listdir.return_value = []
|
|
|
|
# Call the function
|
|
result = postproc.process_nzb_only_download(workdir, fake_nzo)
|
|
|
|
# Verify result is None (no files)
|
|
assert result is None
|
|
|
|
# Verify process_single_nzb was NOT called
|
|
mock_process_single_nzb.assert_not_called()
|