268 lines
9.7 KiB
Python
268 lines
9.7 KiB
Python
import json
|
|
import subprocess
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
|
|
SCRIPT = Path(__file__).parents[1] / "scripts" / "Radarr_-_Trigger_Manual_Import.js"
|
|
|
|
NODE_HARNESS = r"""
|
|
const fs = require('fs');
|
|
const vm = require('vm');
|
|
const path = require('path').posix;
|
|
const scenario = JSON.parse(process.argv[1]);
|
|
let source = fs.readFileSync(process.argv[2], 'utf8')
|
|
.replace(/^\/\*\*[\s\S]*?\*\/\s*/, '')
|
|
.replace(/^import .*$/m, '');
|
|
source += '\n;globalThis.__exports = { Script };';
|
|
const calls = [];
|
|
class FakeRadarr {
|
|
constructor(url, apiKey) { this.url = url; this.apiKey = apiKey; }
|
|
fetchJson(endpoint, query) {
|
|
calls.push({type: 'fetch', endpoint, query});
|
|
if (endpoint === 'queue') return scenario.queueResponse || {records: []};
|
|
if (endpoint === 'parse') return scenario.parseResponse || {};
|
|
if (endpoint === 'manualimport') return scenario.candidates || [];
|
|
if (endpoint === 'history') return scenario.historyResponse || {records: []};
|
|
return null;
|
|
}
|
|
sendCommand(name, body) {
|
|
calls.push({type: 'command', name, body});
|
|
return {id: 42};
|
|
}
|
|
waitForCompletion(id, timeout) {
|
|
calls.push({type: 'wait', id, timeout});
|
|
return scenario.waitSuccess !== false;
|
|
}
|
|
}
|
|
const exactPath = scenario.exactPath;
|
|
const context = {
|
|
Radarr: FakeRadarr,
|
|
Variables: {
|
|
'file.FullName': exactPath,
|
|
file: {
|
|
FullName: exactPath,
|
|
Orig: {FileNameNoExtension: scenario.releaseTitle}
|
|
},
|
|
folder: {Name: path.basename(path.dirname(exactPath))}
|
|
},
|
|
Flow: {UnMapPath: value => value},
|
|
Logger: {ILog(){}, WLog(){}, ELog(){}, DLog(){}},
|
|
System: {IO: {Path: {GetFileName: path.basename, GetDirectoryName: path.dirname}}},
|
|
Sleep(){},
|
|
encodeURIComponent,
|
|
console
|
|
};
|
|
vm.createContext(context);
|
|
vm.runInContext(source, context);
|
|
const result = context.__exports.Script(
|
|
'http://radarr-fhd:7879',
|
|
'test-key',
|
|
scenario.scopeRoot,
|
|
false,
|
|
true,
|
|
300
|
|
);
|
|
process.stdout.write(JSON.stringify({result, calls}));
|
|
"""
|
|
|
|
|
|
def run_script(scenario):
|
|
completed = subprocess.run(
|
|
["node", "-e", NODE_HARNESS, json.dumps(scenario), str(SCRIPT)],
|
|
check=True,
|
|
text=True,
|
|
capture_output=True,
|
|
)
|
|
return json.loads(completed.stdout)
|
|
|
|
|
|
class RadarrImportSafetyTests(unittest.TestCase):
|
|
def test_queue_match_imports_only_the_exact_processed_file(self):
|
|
release = (
|
|
"Operation.Kabul.-.13.Days.13.Nights.-.13.jours.13.nuits.2025."
|
|
"German.DTSHD.DL.1080p.BluRay.AVC.Remux-MAMA"
|
|
)
|
|
exact = f"/tdarr/converted/video/{release}/{release}.mkv"
|
|
result = run_script(
|
|
{
|
|
"releaseTitle": release,
|
|
"exactPath": exact,
|
|
"scopeRoot": "/tdarr/converted/video/",
|
|
"queueResponse": {
|
|
"records": [
|
|
{"title": release, "movieId": 6658}
|
|
]
|
|
},
|
|
"parseResponse": {},
|
|
"candidates": [
|
|
{
|
|
"path": exact,
|
|
"quality": {"quality": {"id": 7, "name": "Bluray-1080p"}},
|
|
}
|
|
],
|
|
}
|
|
)
|
|
|
|
self.assertEqual(result["result"], 1)
|
|
commands = [call for call in result["calls"] if call["type"] == "command"]
|
|
self.assertEqual(len(commands), 1)
|
|
self.assertEqual(commands[0]["name"], "ManualImport")
|
|
self.assertEqual(commands[0]["body"]["files"][0]["path"], exact)
|
|
self.assertEqual(commands[0]["body"]["files"][0]["movieId"], 6658)
|
|
self.assertNotIn(
|
|
"downloadedMoviesScan",
|
|
[call.get("name") for call in commands],
|
|
)
|
|
|
|
def test_unknown_movie_fails_closed_without_scanning_the_category_root(self):
|
|
release = "Unknown.Multilingual.Release.2026.1080p"
|
|
exact = f"/tdarr/converted/video/{release}/{release}.mkv"
|
|
result = run_script(
|
|
{
|
|
"releaseTitle": release,
|
|
"exactPath": exact,
|
|
"scopeRoot": "/tdarr/converted/video/",
|
|
"queueResponse": {"records": []},
|
|
"parseResponse": {},
|
|
"candidates": [],
|
|
}
|
|
)
|
|
|
|
self.assertEqual(result["result"], -1)
|
|
commands = [call for call in result["calls"] if call["type"] == "command"]
|
|
self.assertEqual(commands, [])
|
|
|
|
def test_manual_import_rejects_a_candidate_for_a_different_file(self):
|
|
release = "Expected.Movie.2026.1080p"
|
|
exact = f"/tdarr/converted/video/{release}/{release}.mkv"
|
|
wrong = "/tdarr/converted/video/Other.Movie/Other.Movie.mkv"
|
|
result = run_script(
|
|
{
|
|
"releaseTitle": release,
|
|
"exactPath": exact,
|
|
"scopeRoot": "/tdarr/converted/video/",
|
|
"queueResponse": {"records": [{"title": release, "movieId": 7001}]},
|
|
"candidates": [
|
|
{
|
|
"path": wrong,
|
|
"quality": {"quality": {"id": 7, "name": "Bluray-1080p"}},
|
|
},
|
|
{
|
|
"path": exact,
|
|
"quality": {"quality": {"id": 7, "name": "Bluray-1080p"}},
|
|
},
|
|
],
|
|
}
|
|
)
|
|
|
|
command = next(call for call in result["calls"] if call["type"] == "command")
|
|
self.assertEqual(command["body"]["files"][0]["path"], exact)
|
|
|
|
def test_grab_history_match_recovers_when_download_is_no_longer_in_queue(self):
|
|
release = "Tracked.Movie.2026.German.1080p-TEST"
|
|
exact = f"/tdarr/converted/video/{release}/{release}.mkv"
|
|
result = run_script(
|
|
{
|
|
"releaseTitle": release,
|
|
"exactPath": exact,
|
|
"scopeRoot": "/tdarr/converted/video/",
|
|
"queueResponse": {"records": []},
|
|
"historyResponse": {
|
|
"records": [
|
|
{"sourceTitle": release, "movieId": 7111, "eventType": "grabbed"}
|
|
]
|
|
},
|
|
"parseResponse": {},
|
|
"candidates": [
|
|
{
|
|
"path": exact,
|
|
"quality": {"quality": {"id": 7, "name": "Bluray-1080p"}},
|
|
}
|
|
],
|
|
}
|
|
)
|
|
|
|
commands = [call for call in result["calls"] if call["type"] == "command"]
|
|
self.assertEqual(len(commands), 1)
|
|
self.assertEqual(commands[0]["body"]["files"][0]["movieId"], 7111)
|
|
history_fetch = next(
|
|
call
|
|
for call in result["calls"]
|
|
if call["type"] == "fetch" and call["endpoint"] == "history"
|
|
)
|
|
self.assertIn("eventType=1", history_fetch["query"])
|
|
|
|
def test_grab_metadata_overrides_unknown_manualimport_metadata(self):
|
|
release = "American.Sweatshop.2025.German.DTSHD.DL.1080p.BluRay.AVC.Remux-MAMA"
|
|
exact = f"/tdarr/converted/video/{release}/{release}.mkv"
|
|
grabbed_quality = {"quality": {"id": 30, "name": "Remux-1080p"}}
|
|
grabbed_languages = [
|
|
{"id": 1, "name": "English"},
|
|
{"id": 4, "name": "German"},
|
|
]
|
|
result = run_script(
|
|
{
|
|
"releaseTitle": release,
|
|
"exactPath": exact,
|
|
"scopeRoot": "/tdarr/converted/video/",
|
|
"queueResponse": {"records": [{"title": release, "movieId": 6425}]},
|
|
"historyResponse": {
|
|
"records": [
|
|
{
|
|
"sourceTitle": release,
|
|
"movieId": 6425,
|
|
"eventType": "grabbed",
|
|
"quality": grabbed_quality,
|
|
"languages": grabbed_languages,
|
|
"data": {"releaseGroup": "MAMA", "indexerFlags": "4"},
|
|
}
|
|
]
|
|
},
|
|
"candidates": [
|
|
{
|
|
"path": exact,
|
|
"releaseGroup": "Radarr",
|
|
"quality": {"quality": {"id": 0, "name": "Unknown"}},
|
|
"languages": [{"id": 0, "name": "Unknown"}],
|
|
}
|
|
],
|
|
}
|
|
)
|
|
|
|
self.assertEqual(result["result"], 1)
|
|
command = next(call for call in result["calls"] if call["type"] == "command")
|
|
imported = command["body"]["files"][0]
|
|
self.assertEqual(imported["quality"], grabbed_quality)
|
|
self.assertEqual(imported["languages"], grabbed_languages)
|
|
self.assertEqual(imported["releaseGroup"], "MAMA")
|
|
self.assertEqual(imported["indexerFlags"], 4)
|
|
|
|
def test_release_folder_matches_when_movie_filename_is_shortened(self):
|
|
release = "Example.Movie.2026.German.DL.1080p.WEB.H264-GROUP"
|
|
exact = f"/tdarr/converted/video/{release}/movie-final.mkv"
|
|
result = run_script(
|
|
{
|
|
"releaseTitle": "movie-final",
|
|
"exactPath": exact,
|
|
"scopeRoot": "/tdarr/converted/video/",
|
|
"queueResponse": {"records": [{"title": release, "movieId": 7331}]},
|
|
"historyResponse": {"records": []},
|
|
"parseResponse": {},
|
|
"candidates": [
|
|
{
|
|
"path": exact,
|
|
"quality": {"quality": {"id": 3, "name": "WEBDL-1080p"}},
|
|
"languages": [{"id": 4, "name": "German"}],
|
|
}
|
|
],
|
|
}
|
|
)
|
|
|
|
self.assertEqual(result["result"], 1)
|
|
command = next(call for call in result["calls"] if call["type"] == "command")
|
|
self.assertEqual(command["body"]["files"][0]["movieId"], 7331)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|