fix(transfer): avoid polluted history fallback at shared roots

Parent-path fallback should stop at shared download roots so an old root-level download cannot hijack unrelated manual transfers. Keep exact DownloadFiles matches allowed at the shared root to preserve valid no-subfolder lookups.

Closes #5716

Regression from afcd895f52

Follow-up to #5702
This commit is contained in:
jxxghp
2026-05-07 13:01:20 +08:00
parent 49a51cca25
commit 6d7b0733af
2 changed files with 240 additions and 12 deletions
+121 -3
View File
@@ -1,6 +1,7 @@
import unittest
from pathlib import Path
from types import SimpleNamespace
from unittest.mock import patch
from app.chain.transfer import TransferChain
@@ -32,6 +33,9 @@ class FakeDownloadHistoryOper:
class TransferDownloadHistoryLookupTest(unittest.TestCase):
def setUp(self):
self.chain = object.__new__(TransferChain)
def test_resolve_download_history_falls_back_to_parent_download_path(self):
expected = SimpleNamespace(download_hash="hash1", downloader="qb")
oper = FakeDownloadHistoryOper(
@@ -39,7 +43,7 @@ class TransferDownloadHistoryLookupTest(unittest.TestCase):
histories_by_path={"/downloads/season-pack": expected},
)
history = TransferChain._resolve_download_history(
history = self.chain._resolve_download_history(
downloadhis=oper,
file_path=Path("/downloads/season-pack/Test.Show.S01E01.mkv"),
)
@@ -58,7 +62,7 @@ class TransferDownloadHistoryLookupTest(unittest.TestCase):
},
)
history = TransferChain._resolve_download_history(
history = self.chain._resolve_download_history(
downloadhis=oper,
file_path=Path("/downloads/season-pack/subs/Test.Show.S01E01.zh.ass"),
)
@@ -79,13 +83,127 @@ class TransferDownloadHistoryLookupTest(unittest.TestCase):
},
)
history = TransferChain._resolve_download_history(
history = self.chain._resolve_download_history(
downloadhis=oper,
file_path=Path("/downloads/shared/Test.Show.S01E01.mkv"),
)
self.assertIsNone(history)
def test_resolve_download_history_stops_at_shared_download_root_path(self):
oper = FakeDownloadHistoryOper(
histories_by_path={
"/downloads": SimpleNamespace(download_hash="hash1", downloader="qb")
}
)
with patch(
"app.chain.transfer.DirectoryHelper.get_download_dirs",
return_value=[
SimpleNamespace(
download_path="/downloads",
media_type=None,
download_type_folder=False,
media_category=None,
download_category_folder=False,
)
],
):
history = self.chain._resolve_download_history(
downloadhis=oper,
file_path=Path("/downloads/Ghost.Concert.mkv"),
)
self.assertIsNone(history)
def test_resolve_download_history_stops_at_shared_download_root_savepath(self):
expected = SimpleNamespace(download_hash="hash1", downloader="qb")
oper = FakeDownloadHistoryOper(
histories_by_hash={"hash1": expected},
files_by_savepath={
"/downloads": [
SimpleNamespace(download_hash="hash1", filepath="Other.Show.mkv"),
]
},
)
with patch(
"app.chain.transfer.DirectoryHelper.get_download_dirs",
return_value=[
SimpleNamespace(
download_path="/downloads",
media_type=None,
download_type_folder=False,
media_category=None,
download_category_folder=False,
)
],
):
history = self.chain._resolve_download_history(
downloadhis=oper,
file_path=Path("/downloads/Ghost.Concert.mkv"),
)
self.assertIsNone(history)
def test_resolve_download_history_accepts_shared_root_savepath_for_exact_file(self):
expected = SimpleNamespace(download_hash="hash1", downloader="qb")
oper = FakeDownloadHistoryOper(
histories_by_hash={"hash1": expected},
files_by_savepath={
"/downloads": [
SimpleNamespace(download_hash="hash1", filepath="Ghost.Concert.mkv"),
]
},
)
with patch(
"app.chain.transfer.DirectoryHelper.get_download_dirs",
return_value=[
SimpleNamespace(
download_path="/downloads",
media_type=None,
download_type_folder=False,
media_category=None,
download_category_folder=False,
)
],
):
history = self.chain._resolve_download_history(
downloadhis=oper,
file_path=Path("/downloads/Ghost.Concert.mkv"),
)
self.assertIs(history, expected)
def test_resolve_download_history_stops_at_type_category_download_root(self):
oper = FakeDownloadHistoryOper(
histories_by_path={
"/downloads/电视剧/动漫": SimpleNamespace(
download_hash="hash1", downloader="qb"
)
}
)
with patch(
"app.chain.transfer.DirectoryHelper.get_download_dirs",
return_value=[
SimpleNamespace(
download_path="/downloads",
media_type=None,
download_type_folder=True,
media_category=None,
download_category_folder=True,
)
],
):
history = self.chain._resolve_download_history(
downloadhis=oper,
file_path=Path("/downloads/电视剧/动漫/Ghost.Concert.mkv"),
)
self.assertIsNone(history)
if __name__ == "__main__":
unittest.main()