mirror of
https://github.com/sabnzbd/sabnzbd.git
synced 2025-12-24 00:00:12 -05:00
Correct process_nzb_only_download and add tests
This commit is contained in:
@@ -518,10 +518,10 @@ def process_job(nzo: NzbObject) -> bool:
|
||||
|
||||
# Check if this is an NZB-only download, if so redirect to queue
|
||||
# except when PP was Download-only
|
||||
nzb_list = None
|
||||
if flag_repair:
|
||||
nzb_list = process_nzb_only_download(tmp_workdir_complete, nzo)
|
||||
else:
|
||||
nzb_list = None
|
||||
|
||||
if nzb_list:
|
||||
nzo.set_unpack_info("Download", T("Sent %s to queue") % nzb_list)
|
||||
cleanup_empty_directories(tmp_workdir_complete)
|
||||
@@ -1169,14 +1169,14 @@ def process_nzb_only_download(workdir: str, nzo: NzbObject) -> Optional[list[str
|
||||
if get_ext(nzb_file) != ".nzb":
|
||||
return None
|
||||
|
||||
# Determine name based on number of files
|
||||
nzb_filename = get_filename(nzb_file)
|
||||
nzbname = nzo.final_name
|
||||
if len(files) > 1:
|
||||
nzbname = f"{nzo.final_name} - {nzb_filename}"
|
||||
|
||||
# Process all NZB files
|
||||
new_nzbname = nzo.final_name
|
||||
for nzb_file in files:
|
||||
# Determine name based on number of files
|
||||
nzb_filename = get_filename(nzb_file)
|
||||
if len(files) > 1:
|
||||
new_nzbname = f"{nzo.final_name} - {nzb_filename}"
|
||||
|
||||
process_single_nzb(
|
||||
nzb_filename,
|
||||
nzb_file,
|
||||
@@ -1185,7 +1185,7 @@ def process_nzb_only_download(workdir: str, nzo: NzbObject) -> Optional[list[str
|
||||
cat=nzo.cat,
|
||||
url=nzo.url,
|
||||
priority=nzo.priority,
|
||||
nzbname=nzbname,
|
||||
nzbname=new_nzbname,
|
||||
dup_check=False,
|
||||
)
|
||||
return files
|
||||
|
||||
@@ -257,3 +257,137 @@ class TestPostProc:
|
||||
assert tmp_workdir_complete == workdir_complete
|
||||
|
||||
_func()
|
||||
|
||||
|
||||
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()
|
||||
|
||||
Reference in New Issue
Block a user