sabnzbd/tests/test_nzbqueue.py

#!/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_nzbqueue - Testing functions in nzbqueue.py
"""

import os
import shutil
import tempfile
from pathlib import Path
from types import SimpleNamespace
from typing import Optional
from unittest import mock

import pytest

import sabnzbd
from sabnzbd.constants import (
    JOB_ADMIN,
    ONDISK_VERSION,
    ONDISK_FILE,
    RENAMES_FILE,
    Status,
    LOW_PRIORITY,
    NORMAL_PRIORITY,
    HIGH_PRIORITY,
    FORCE_PRIORITY,
    REPAIR_PRIORITY,
)
from sabnzbd.database import HistoryDB
from sabnzbd.downloader import Server
from sabnzbd.filesystem import save_compressed, save_data
from sabnzbd.nzb import NzbFile, NzbObject
from sabnzbd.nzbqueue import NzbQueue
from tests.testhelper import (
    FakeHistoryDB,
    SAB_DATA_DIR,
    SAB_NEWSSERVER_HOST,
    SAB_NEWSSERVER_PORT,
    create_and_read_nzb_fp,
)


@pytest.fixture()
def nzbqueue_env(monkeypatch, mocker, tmp_path):
    sabnzbd.config.ConfigCat("*", {})

    sabnzbd.Scheduler = mocker.Mock()
    sabnzbd.Scheduler.analyse = mocker.Mock(return_value=False)
    sabnzbd.ArticleCache = mocker.Mock()
    sabnzbd.Assembler = mocker.Mock()
    sabnzbd.BPSMeter = mocker.Mock()
    sabnzbd.Downloader = SimpleNamespace(paused=False)
    sabnzbd.Downloader.servers = [
        Server(
            server_id="testserver1",
            displayname="testserver1",
            host=SAB_NEWSSERVER_HOST,
            port=SAB_NEWSSERVER_PORT,
            timeout=30,
            threads=8,
            priority=0,
            use_ssl=False,
            ssl_verify=3,
            ssl_ciphers="",
            pipelining_requests=mocker.Mock(return_value=1),
        )
    ]
    sabnzbd.NzbQueue = NzbQueue()
    sabnzbd.Downloader.disconnect = mocker.Mock()
    monkeypatch.setattr(sabnzbd.cfg.admin_dir, "get_path", lambda: str(tmp_path))
    monkeypatch.setattr(sabnzbd.cfg.download_dir, "get_path", lambda: str(tmp_path))

    yield

    del sabnzbd.NzbQueue
    del sabnzbd.Downloader
    del sabnzbd.BPSMeter
    del sabnzbd.Assembler
    del sabnzbd.ArticleCache
    del sabnzbd.Scheduler


def make_dummy_nzo(name: str, priority: int = NORMAL_PRIORITY, files: int = 50, articles: int = 200) -> NzbObject:
    work_name = f"job-{name}"

    article_size = 750_000
    nzo = NzbObject(work_name, priority=priority)
    nzo.files = [
        NzbFile(
            date=nzo.avg_date,
            subject=f"test-file-{file}",
            raw_article_db=[[f"{file}-article-{article}", article_size] for article in range(articles)],
            file_bytes=article_size * articles,
            nzo=nzo,
        )
        for file in range(files)
    ]

    return nzo


def make_dummy_postproc_nzo(name: str, download_path: str, status: str = Status.QUEUED, pp_active: bool = False):
    """Mock NzbObject in the post-processing queue, with all the attributes
    add_active_history() needs so it can pass through build_history()"""
    nzo = mock.Mock()
    nzo.nzo_id = f"SABnzbd_nzo_{name}"
    nzo.final_name = name
    nzo.filename = f"{name}.nzb"
    nzo.cat = "*"
    nzo.script = "none"
    nzo.url = ""
    nzo.status = status
    nzo.pp_active = pp_active
    nzo.repair = nzo.unpack = nzo.delete = True
    nzo.nzo_info = {}
    nzo.unpack_info = {}
    nzo.bytes_downloaded = 1024
    nzo.fail_msg = ""
    nzo.correct_password = ""
    nzo.action_line = ""
    nzo.duplicate_key = ""
    nzo.time_added = 0
    nzo.download_path = download_path
    return nzo


@pytest.fixture()
def make_nzb_workdir(nzbqueue_env):
    def _make_workdir(
        name: str,
        on_disk: Optional[dict[str, list[bool]]] = None,
        renames: Optional[dict[str, str]] = None,
    ) -> str:
        """Create a working directory for nzbqueue from tests/data"""
        wdir = tempfile.TemporaryDirectory(prefix=name, dir=sabnzbd.cfg.download_dir.get_path()).name
        admin_dir = os.path.join(wdir, JOB_ADMIN)
        os.makedirs(admin_dir)

        # Copy test data, create NZB-file and __ADMIN__
        nzb_fp = create_and_read_nzb_fp(name)
        save_compressed(admin_dir, name, nzb_fp)
        shutil.copytree(os.path.join(SAB_DATA_DIR, name), wdir, dirs_exist_ok=True)
        if on_disk:
            save_data((ONDISK_VERSION, on_disk), ONDISK_FILE, admin_dir)
        if renames:
            save_data(renames, RENAMES_FILE, admin_dir)
        return wdir

    yield _make_workdir


@pytest.mark.usefixtures("nzbqueue_env")
class TestNzbQueue:
    def test_save_and_restore_(self):
        q = NzbQueue()
        joba = make_dummy_nzo("a")
        jobb = make_dummy_nzo("b")
        q.add(joba)
        q.add(jobb)

        # Mark one of joba articles as tried
        article = next(iter(joba.files[0].articles))
        article.add_to_try_list(sabnzbd.Downloader.servers[0])
        q.save()

        # Both should be in the queue
        assert q.get_nzo(joba.nzo_id)
        assert q.get_nzo(jobb.nzo_id)

        # Reload the queue with no repair
        q = NzbQueue()
        q.read_queue(0)
        joba = q.get_nzo(joba.nzo_id)
        jobb = q.get_nzo(jobb.nzo_id)
        assert joba
        assert jobb

        # Try list restored
        assert sabnzbd.Downloader.servers[0] in next(iter(joba.files[0].articles)).try_list

    def test_stop_idle_jobs_no_crash_on_exhausted_articles(self):
        """Regression test: stop_idle_jobs must not raise RuntimeError when
        register_article removes an article from nzf.articles (a dict) while
        the same dict is being iterated.  Introduced by commit 44d94226e when
        nzf.articles was changed from list to dict but the protective [:] copy
        was dropped from the iteration in stop_idle_jobs."""
        server = sabnzbd.Downloader.servers[0]

        nzo = make_dummy_nzo("stall-test", files=1, articles=3)
        nzf = nzo.files[0]

        # Load all articles into memory (only first is loaded at NzbFile init)
        nzf.finish_import()

        q = NzbQueue()
        # add() resets try lists, so saturate them after adding
        q.add(nzo)
        sabnzbd.NzbQueue = q

        # Saturate all try-lists so stop_idle_jobs enters the article-removal branch
        nzo.add_to_try_list(server)
        nzf.add_to_try_list(server)
        for article in list(nzf.articles):
            article.add_to_try_list(server)

        # Must not raise RuntimeError: dictionary changed size during iteration
        q.stop_idle_jobs()

    @pytest.mark.skipif(not sabnzbd.WINDOWS, reason="Legacy 3.0.0 queue fixture contains Windows-specific paths")
    def test_restore_legacy_queue_format_3_0_0(self, tmp_path, monkeypatch):
        fixture_path = Path(SAB_DATA_DIR) / "test_3_0_0_data_format"
        shutil.copytree(fixture_path, tmp_path, dirs_exist_ok=True)

        nzbqueue = NzbQueue()
        nzbqueue.read_queue(repair=0)

        assert nzbqueue.actives() == 1
        nzo = nzbqueue.get_nzo("SABnzbd_nzo_7ormqfeg")

        # Check a number of basis statistics and parameters
        assert nzo.nzo_id == "SABnzbd_nzo_7ormqfeg"
        assert nzo.work_name == "test_download_100MB"
        assert nzo.filename == "test_download_100MB.nzb"
        assert nzo.cat == "*"
        assert nzo.priority == 1
        assert nzo.pp == 3

        assert len(nzo.files_table) == 13
        assert len(nzo.files) == 4
        assert len(nzo.finished_files) == 1
        assert len(nzo.extrapars["par-files"]) == 8
        assert len(nzo.saved_articles) == 22

        assert nzo.bytes == 114967316
        assert nzo.bytes_downloaded == 18532805
        assert nzo.bytes_missing == 0
        assert nzo.bytes_par2 == 42560
        assert nzo.bytes_tried == 30077349
        assert nzo.remaining == 84889967

        # Validate we can also load legacy NZF pickles from the restored NZO
        for nzf in nzo.files:
            nzf.finish_import()

        # Validate from specifc nzf some attributes
        cnzf = nzo.get_nzf_by_id("SABnzbd_nzf_799tdim0")
        assert cnzf.nzo == nzo
        assert cnzf.filename == "rar-files.part1.rar"
        assert cnzf.md5of16k == b"e\xc4^\x93\xc4\x92\xf9s\x057Oe!\x9dA\xe7"
        assert len(cnzf.articles) == 51
        assert len(cnzf.decodetable) == 73

        assert cnzf.bytes == 27049565
        assert cnzf.bytes_left == 10777869
        assert cnzf.crc32 is None

    def test_nzo_reuse(self, make_nzb_workdir):
        wdir = make_nzb_workdir("basic_rar5")

        nzo_id = sabnzbd.NzbQueue.repair_job(wdir, None, None)
        assert nzo_id
        nzo = sabnzbd.NzbQueue.get_nzo(nzo_id)
        # No files to download so goes straight to post-processing
        assert not nzo

    def test_nzo_reuse_failed_articles(self, make_nzb_workdir):
        wdir = make_nzb_workdir("basic_rar5", {"testfile.rar": [False]})

        nzo_id = sabnzbd.NzbQueue.repair_job(wdir, None, None)
        assert nzo_id
        nzo = sabnzbd.NzbQueue.get_nzo(nzo_id)
        assert nzo
        # testfile.rar is on disk, but it has missing articles
        assert nzo.files
        assert not nzo.finished_files

    def test_nzo_reuse_failed_articles_renamed(self, make_nzb_workdir):
        wdir = make_nzb_workdir(
            "basic_rar5",
            {"renamed.rar": [False]},
            {"renamed.rar": "testfile.rar"},
        )
        os.rename(os.path.join(wdir, "testfile.rar"), os.path.join(wdir, "renamed.rar"))

        nzo_id = sabnzbd.NzbQueue.repair_job(wdir, None, None)
        assert nzo_id
        nzo = sabnzbd.NzbQueue.get_nzo(nzo_id)
        assert nzo
        # renamed.rar is on disk, but it has missing articles
        assert nzo.files
        assert not nzo.finished_files

    def test_scan_jobs_known_jobs(self, mocker, monkeypatch, tmp_path):
        """Folders belonging to jobs in the download queue, the post-processing
        queue, or retryable from History must not be treated as orphans.
        Anything else in the incomplete folder is an orphan."""
        monkeypatch.setattr(HistoryDB, "db_path", str(tmp_path / "history1.db"))
        monkeypatch.setattr(HistoryDB, "startup_done", False)

        def make_job_folder(name: str) -> str:
            path = os.path.join(sabnzbd.cfg.download_dir.get_path(), name)
            os.makedirs(path, exist_ok=True)
            return path

        # Job in the download queue
        queued_nzo = make_dummy_nzo("queued", files=1, articles=1)
        sabnzbd.NzbQueue.add(queued_nzo, save=False)
        make_job_folder(queued_nzo.work_name)

        # Job waiting in the post-processing queue
        postproc_path = make_job_folder("job-postproc")
        pp_nzo = make_dummy_postproc_nzo("job-postproc", postproc_path)
        mocker.patch.object(sabnzbd, "PostProcessor", create=True)
        sabnzbd.PostProcessor.get_queue.return_value = [pp_nzo]

        with FakeHistoryDB(str(tmp_path / "history1.db")) as history_db:
            # The postproc job is also already in the history database, which briefly happens at the end of post-processing
            history_db.add_fake_history_job("job-postproc", Status.COMPLETED, path=postproc_path)
            # Failed job with the incomplete folder still on disk: retryable
            history_db.add_fake_history_job("job-failed", Status.FAILED, path=make_job_folder("job-failed"))
            # Failed job whose recorded incomplete folder no longer exists is not retryable, so a same-named folder is an orphan
            make_job_folder("job-gone")
            history_db.add_fake_history_job("job-gone", Status.FAILED, path=str(tmp_path / "elsewhere" / "job-gone"))
            # A failed URL-fetch (report = 'future') is always retryable, regardless of status or whether the folder exists
            history_db.add_fake_history_job(
                "job-future", Status.COMPLETED, path=make_job_folder("job-future"), futuretype=True
            )
            # Completed job: its folder should no longer exist, so treat leftovers as orphans
            history_db.add_fake_history_job("job-completed", Status.COMPLETED, path=make_job_folder("job-completed"))
            # Archived jobs are not visible to queue repair, even when they would
            # otherwise be retryable, so the folder is treated as an orphan
            history_db.add_fake_history_job(
                "job-archived", Status.FAILED, path=make_job_folder("job-archived"), archive=True
            )

        # Folder not known anywhere
        make_job_folder("job-orphan")

        orphans = sabnzbd.NzbQueue.scan_jobs(action=False)
        assert sorted(orphans) == ["job-archived", "job-completed", "job-gone", "job-orphan"]

        # With all_jobs=True the download queue itself is not considered registered
        orphans_all = sabnzbd.NzbQueue.scan_jobs(all_jobs=True, action=False)
        assert queued_nzo.work_name in orphans_all

    @pytest.fixture
    def queue(self, nzbqueue_env):
        return NzbQueue()

    def test_add_inserts_and_tracks_jobs(self, queue):
        a = make_dummy_nzo("a", priority=NORMAL_PRIORITY)
        b = make_dummy_nzo("b", priority=LOW_PRIORITY)
        c = make_dummy_nzo("c", priority=FORCE_PRIORITY)

        ida = queue.add(a, save=False, quiet=True)
        idb = queue.add(b, save=False, quiet=True)
        idc = queue.add(c, save=False, quiet=True)

        # All ids registered
        assert queue.get_nzo(ida) is a
        assert queue.get_nzo(idb) is b
        assert queue.get_nzo(idc) is c

        # queue_info returns all three, first should be the forced job
        _, _, _, nzo_list, _, count = queue.queue_info()
        assert count == 3
        assert [n.final_name for n in nzo_list] == [c.final_name, a.final_name, b.final_name]

    def test_remove_removes_from_queue_and_table(self, queue):
        a = make_dummy_nzo("a", priority=0)
        b = make_dummy_nzo("b", priority=0)
        c = make_dummy_nzo("c", priority=0)

        _ida = queue.add(a, save=False, quiet=True)
        idb = queue.add(b, save=False, quiet=True)
        _idc = queue.add(c, save=False, quiet=True)

        removed = queue.remove(idb, cleanup=False, delete_all_data=False)
        assert removed is b
        assert queue.get_nzo(idb) is None

        _, _, _, nzo_list, _, count = queue.queue_info()
        assert count == 2
        assert [nzo.final_name for nzo in nzo_list] == ["job-a", "job-c"]

    def test_remove_multiple_and_remove_all(self, queue):
        jobs = [make_dummy_nzo(f"job-{i}", priority=0) for i in range(5)]
        ids = [queue.add(nzo, save=False, quiet=True) for nzo in jobs]

        # Remove two specific jobs
        subset = ids[1:3]
        removed_ids = queue.remove_multiple(subset, delete_all_data=False)
        assert set(removed_ids) == set(subset)

        # Remaining ids still there
        remaining_ids = {nid for nid in ids if nid not in subset}
        assert {nzo.nzo_id for nzo in queue.queue_info()[3]} == remaining_ids

        # remove_all with search pattern should remove the rest
        removed_all = queue.remove_all(search="job-")
        assert set(removed_all) == remaining_ids
        assert queue.queue_info()[5] == 0  # nzos_matched

    def test_remove_multiple_writes_the_queue_admin_once(self, queue, monkeypatch, mocker):
        jobs = [make_dummy_nzo(f"job-{i}", priority=NORMAL_PRIORITY) for i in range(5)]
        ids = [queue.add(nzo, save=False, quiet=True) for nzo in jobs]

        save_admin = mocker.Mock()
        monkeypatch.setattr(sabnzbd.filesystem, "save_admin", save_admin)
        removed = queue.remove_multiple(ids, delete_all_data=False)

        assert set(removed) == set(ids)
        assert save_admin.call_count == 1

    def test_save_accepts_a_list_of_jobs(self, queue, monkeypatch, mocker):
        jobs = [make_dummy_nzo(f"job-{i}", priority=NORMAL_PRIORITY) for i in range(3)]
        for nzo in jobs:
            queue.add(nzo, save=False, quiet=True)
            mocker.patch.object(nzo, "save_to_disk")

        save_admin = mocker.Mock()
        monkeypatch.setattr(sabnzbd.filesystem, "save_admin", save_admin)
        queue.save(jobs[:2])

        # Only the listed jobs are written, the queue admin just once for the batch
        assert [nzo.save_to_disk.call_count for nzo in jobs] == [1, 1, 0]
        assert save_admin.call_count == 1

    def test_change_opts_sets_pp(self, queue):
        a = make_dummy_nzo("a", priority=LOW_PRIORITY)
        ida = queue.add(a, save=False, quiet=True)

        changed = queue.change_opts([ida], pp=3)
        assert changed == 1
        assert a.pp == 3

    def test_change_script_only_when_valid(self, queue, monkeypatch):
        from sabnzbd import nzbqueue as nzbqueue_mod

        # Always accept given script
        monkeypatch.setattr(nzbqueue_mod, "is_valid_script", lambda s: True)

        a = make_dummy_nzo("a", priority=0)
        b = make_dummy_nzo("b", priority=0)
        ida = queue.add(a, save=False, quiet=True)
        idb = queue.add(b, save=False, quiet=True)

        changed = queue.change_script([ida, idb], script="myscript.py")
        assert changed == 2
        assert a.script == "myscript.py"
        assert b.script == "myscript.py"

        # Now mark scripts invalid; no changes should be made
        monkeypatch.setattr(nzbqueue_mod, "is_valid_script", lambda s: False)
        changed = queue.change_script([ida, idb], script="other.py")
        assert changed == 0
        assert a.script == "myscript.py"
        assert b.script == "myscript.py"

    def test_change_cat_updates_cat_pp_script_and_priority(self, queue, monkeypatch):
        from sabnzbd import nzbqueue as nzbqueue_mod

        # Fake cat_to_opts: (cat, pp, script, prio)
        def fake_cat_to_opts(cat):
            return f"{cat}-cat", 2, "cat_script.py", FORCE_PRIORITY

        monkeypatch.setattr(nzbqueue_mod, "cat_to_opts", fake_cat_to_opts)

        a = make_dummy_nzo("a", priority=0)
        ida = queue.add(a, save=False, quiet=True)

        changed = queue.change_cat([ida], cat="movies")
        assert changed == 1
        assert a.cat == "movies-cat"
        assert a.script == "cat_script.py"
        assert a.priority == FORCE_PRIORITY

    def test_change_name_updates_final_name(self, queue):
        a = make_dummy_nzo("a", priority=0)
        ida = queue.add(a, save=False, quiet=True)

        ok = queue.change_name(ida, "renamed")
        assert ok is True
        assert a.final_name == "renamed"

    @staticmethod
    def get_queue_order(queue):
        return [n.final_name for n in queue.queue_info()[3]]

    def test_set_priority_moves_job_to_forced_top(self, queue):
        a = make_dummy_nzo("a", priority=0)
        b = make_dummy_nzo("b", priority=0)
        c = make_dummy_nzo("c", priority=0)

        _ida = queue.add(a, save=False, quiet=True)
        idb = queue.add(b, save=False, quiet=True)
        _idc = queue.add(c, save=False, quiet=True)

        assert self.get_queue_order(queue) == ["job-a", "job-b", "job-c"]

        # Set b to FORCE_PRIORITY, should go to the top
        _pos = queue.set_priority([idb], FORCE_PRIORITY)
        # pos is index; we just verify ordering
        assert self.get_queue_order(queue)[0] == "job-b"
        assert b.priority == FORCE_PRIORITY

    def test_switch_swaps_positions(self, queue):
        a = make_dummy_nzo("a", priority=NORMAL_PRIORITY)
        b = make_dummy_nzo("b", priority=NORMAL_PRIORITY)
        c = make_dummy_nzo("c", priority=NORMAL_PRIORITY)

        ida = queue.add(a, save=False, quiet=True)
        _idb = queue.add(b, save=False, quiet=True)
        idc = queue.add(c, save=False, quiet=True)

        assert self.get_queue_order(queue) == ["job-a", "job-b", "job-c"]

        # Move a to c
        new_pos, _prio = queue.switch(ida, idc)
        assert new_pos != -1
        assert self.get_queue_order(queue) == ["job-b", "job-c", "job-a"]

    def test_switch_swaps_positions_different_priority(self, queue):
        a = make_dummy_nzo("a", priority=HIGH_PRIORITY)
        b = make_dummy_nzo("b", priority=NORMAL_PRIORITY)
        c = make_dummy_nzo("c", priority=NORMAL_PRIORITY)

        ida = queue.add(a, save=False, quiet=True)
        _idb = queue.add(b, save=False, quiet=True)
        idc = queue.add(c, save=False, quiet=True)

        assert self.get_queue_order(queue) == ["job-a", "job-b", "job-c"]

        # Move a to c
        new_pos, _prio = queue.switch(ida, idc)
        assert new_pos != -1
        assert a.priority == NORMAL_PRIORITY
        assert b.priority == NORMAL_PRIORITY
        assert c.priority == NORMAL_PRIORITY
        assert self.get_queue_order(queue) == ["job-b", "job-c", "job-a"]

    def test_switch_swaps_positions_different_priority_2(self, queue):
        a = make_dummy_nzo("a", priority=HIGH_PRIORITY)
        b = make_dummy_nzo("b", priority=HIGH_PRIORITY)
        c = make_dummy_nzo("c", priority=NORMAL_PRIORITY)

        ida = queue.add(a, save=False, quiet=True)
        _idb = queue.add(b, save=False, quiet=True)
        idc = queue.add(c, save=False, quiet=True)

        assert self.get_queue_order(queue) == ["job-a", "job-b", "job-c"]

        # Move a to c
        new_pos, _prio = queue.switch(ida, idc)
        assert new_pos != -1
        assert b.priority == HIGH_PRIORITY
        assert c.priority == NORMAL_PRIORITY
        assert a.priority == NORMAL_PRIORITY
        assert self.get_queue_order(queue) == ["job-b", "job-c", "job-a"]

    def test_switch_swaps_positions_different_priority_3(self, queue):
        a = make_dummy_nzo("a", priority=HIGH_PRIORITY)
        b = make_dummy_nzo("b", priority=NORMAL_PRIORITY)
        c = make_dummy_nzo("c", priority=NORMAL_PRIORITY)

        ida = queue.add(a, save=False, quiet=True)
        _idb = queue.add(b, save=False, quiet=True)
        idc = queue.add(c, save=False, quiet=True)

        assert self.get_queue_order(queue) == ["job-a", "job-b", "job-c"]

        # Move c to a
        new_pos, _prio = queue.switch(idc, ida)
        assert new_pos != -1
        assert c.priority == HIGH_PRIORITY
        assert a.priority == HIGH_PRIORITY
        assert b.priority == NORMAL_PRIORITY
        assert self.get_queue_order(queue) == ["job-c", "job-a", "job-b"]

    def test_switch_moves_job_below_its_lower_priority_neighbour(self, queue):
        """Moving onto the first job of the next priority group must actually move it"""
        a = make_dummy_nzo("a", priority=HIGH_PRIORITY)
        b = make_dummy_nzo("b", priority=NORMAL_PRIORITY)
        queue.add(a, save=False, quiet=True)
        queue.add(b, save=False, quiet=True)

        pos, prio = queue.switch(a.nzo_id, b.nzo_id)

        assert self.get_queue_order(queue) == ["job-b", "job-a"]
        assert pos == 1
        assert prio == NORMAL_PRIORITY

    def test_send_back_keeps_job_removable_when_its_priority_changed(self, queue, monkeypatch):
        """The replacement job can resolve a different priority than the job it replaces"""
        old = make_dummy_nzo("old", priority=HIGH_PRIORITY)
        queue.add(old, save=False, quiet=True)
        monkeypatch.setattr(old, "download_path", "/tmp/old", raising=False)
        monkeypatch.setattr(sabnzbd.filesystem, "globber_full", lambda *a, **k: ["/tmp/old.nzb.gz"])
        monkeypatch.setattr("sabnzbd.nzbqueue.globber_full", lambda *a, **k: ["/tmp/old.nzb.gz"])

        def fake_process_single_nzb(*args, **kwargs):
            queue.remove(kwargs["nzo_id"], cleanup=False, delete_all_data=False)
            replacement = make_dummy_nzo("new", priority=NORMAL_PRIORITY)
            replacement.nzo_id = kwargs["nzo_id"]
            queue.add(replacement, save=False, quiet=True)
            return 0, [replacement.nzo_id]

        monkeypatch.setattr("sabnzbd.nzbqueue.process_single_nzb", fake_process_single_nzb)
        queue.send_back(old)

        # Removing must still find the job even though the replacement resolved another priority
        nzo_id = queue.queue_info()[3][0].nzo_id
        assert queue.remove(nzo_id, cleanup=False, delete_all_data=False)
        assert queue.queue_info()[3] == []

    @pytest.mark.parametrize(
        "value2, expected_order",
        [
            ("1", ["job-b", "job-a", "job-c"]),
            ("-1", ["job-b", "job-c", "job-a"]),
            ("-2", ["job-b", "job-a", "job-c"]),
            # Out of range, unparsable and non-decimal numerics all leave the queue alone
            ("99", ["job-a", "job-b", "job-c"]),
            ("-99", ["job-a", "job-b", "job-c"]),
            ("\u00b2", ["job-a", "job-b", "job-c"]),
            ("abc", ["job-a", "job-b", "job-c"]),
            ("", ["job-a", "job-b", "job-c"]),
        ],
    )
    def test_switch_accepts_an_index_as_second_parameter(self, queue, value2, expected_order):
        jobs = [make_dummy_nzo(name) for name in "abc"]
        for nzo in jobs:
            queue.add(nzo, save=False, quiet=True)

        queue.switch(jobs[0].nzo_id, value2)

        assert self.get_queue_order(queue) == expected_order

    @pytest.mark.parametrize("via_set_priority", [False, True])
    @pytest.mark.parametrize(
        "priority", [REPAIR_PRIORITY, FORCE_PRIORITY, HIGH_PRIORITY, NORMAL_PRIORITY, LOW_PRIORITY]
    )
    def test_jobs_queue_in_arrival_order_within_their_priority(self, queue, priority, via_set_priority):
        """Every group keeps the order its jobs arrived in, so a repair re-add does not starve"""
        for name in "abc":
            nzo = make_dummy_nzo(name, priority=NORMAL_PRIORITY if via_set_priority else priority)
            queue.add(nzo, save=False, quiet=True)
            if via_set_priority:
                queue.set_priority([nzo.nzo_id], priority)

        assert self.get_queue_order(queue) == ["job-a", "job-b", "job-c"]

    def test_set_priority_keeps_the_order_of_the_selection(self, queue):
        """Changing the priority of several jobs at once must not reverse them"""
        nzo_ids = [queue.add(make_dummy_nzo(name), save=False, quiet=True) for name in "abc"]
        queue.set_priority(nzo_ids, FORCE_PRIORITY)

        assert self.get_queue_order(queue) == ["job-a", "job-b", "job-c"]

    @pytest.mark.parametrize("add_order", [("forced", "repair"), ("repair", "forced")])
    def test_repair_outranks_forced(self, queue, add_order):
        """Whichever job arrives first, the repair job ends up above the forced one"""
        priorities = {"forced": FORCE_PRIORITY, "repair": REPAIR_PRIORITY}
        for name in add_order:
            queue.add(make_dummy_nzo(name, priority=priorities[name]), save=False, quiet=True)

        assert self.get_queue_order(queue) == ["job-repair", "job-forced"]

    def test_has_forced_jobs_true_when_forced_and_active(self, queue):
        forced = make_dummy_nzo("forced", priority=FORCE_PRIORITY)
        normal = make_dummy_nzo("normal", priority=0)

        queue.add(forced, save=False, quiet=True)
        queue.add(normal, save=False, quiet=True)

        assert queue.has_forced_jobs() is True

        # If forced job is paused, it should no longer count
        forced.status = Status.PAUSED
        assert queue.has_forced_jobs() is False

    def test_has_forced_jobs_false_when_no_forced(self, queue):
        a = make_dummy_nzo("a", priority=0)
        b = make_dummy_nzo("b", priority=LOW_PRIORITY)
        queue.add(a, save=False, quiet=True)
        queue.add(b, save=False, quiet=True)

        assert queue.has_forced_jobs() is False

    def test_add_future_job_saves_with_expected_prefix(self, queue):
        """Queue repair deletes anything in the future folder not named SABnzbd_nzo_*"""
        nzo = NzbObject("future-job", futuretype=True)
        queue.add(nzo)

        written = os.listdir(nzo.admin_path)
        assert written
        assert all(name.startswith("SABnzbd_nzo_") for name in written), written

    @pytest.mark.parametrize("field", ["name", "size", "bytes", "avg_age", "remaining", "remaining_bytes"])
    @pytest.mark.parametrize("direction", ["asc", "desc"])
    def test_sort_queue_every_field(self, queue, field, direction):
        for name, priority in (("a", NORMAL_PRIORITY), ("b", LOW_PRIORITY), ("c", HIGH_PRIORITY)):
            queue.add(make_dummy_nzo(name, priority=priority), save=False, quiet=True)

        queue.sort_queue(field, direction)

        # Whatever the field, priority grouping is preserved
        priorities = [nzo.priority for nzo in queue.queue_info()[3]]
        assert priorities == sorted(priorities, reverse=True)

    @pytest.mark.parametrize("auto_sort", ["name asc", "remaining asc", "remaining_bytes asc"])
    def test_update_sort_order(self, queue, monkeypatch, auto_sort):
        monkeypatch.setattr(sabnzbd.cfg.auto_sort, "get", lambda: auto_sort)
        queue.add(make_dummy_nzo("a"), save=False, quiet=True)

        queue.update_sort_order()