279 lines
10 KiB
Python
279 lines
10 KiB
Python
import json
|
|
import subprocess
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
|
|
SCRIPT = Path(__file__).parents[1] / "scripts" / "Sonarr_-_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 FakeSonarr {
|
|
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 === 'history') return scenario.historyResponse || {records: []};
|
|
if (endpoint === 'parse') return scenario.parseResponse || {};
|
|
if (endpoint === 'manualimport') return scenario.candidates || [];
|
|
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 = {
|
|
Sonarr: FakeSonarr,
|
|
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}}},
|
|
encodeURIComponent,
|
|
console
|
|
};
|
|
vm.createContext(context);
|
|
vm.runInContext(source, context);
|
|
const result = context.__exports.Script(
|
|
'http://sonarr-fhd:8990',
|
|
'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 SonarrImportSafetyTests(unittest.TestCase):
|
|
def test_fhd_scope_does_not_match_uhd_sibling_prefix(self):
|
|
exact = (
|
|
"/tdarr/converted/serien4k/Marshals.2026.S01.GERMAN.DUBBED/"
|
|
"Marshals.2026.S01E05.GERMAN.DUBBED.mkv"
|
|
)
|
|
result = run_script(
|
|
{
|
|
"releaseTitle": "Marshals.2026.S01E05.GERMAN.DUBBED",
|
|
"exactPath": exact,
|
|
"scopeRoot": "/tdarr/converted/serien",
|
|
}
|
|
)
|
|
|
|
self.assertEqual(result["result"], 1)
|
|
self.assertEqual(result["calls"], [])
|
|
|
|
def test_season_pack_history_uses_exact_episode_filename_for_parse(self):
|
|
pack = "Marshals.2026.S01.GERMAN.DUBBED.DL.2160p.WEB.H265-TSCC"
|
|
episode = "Marshals.2026.S01E05.GERMAN.DUBBED.DL.2160p.WEB.H265-TSCC"
|
|
exact = f"/tdarr/converted/serien4k/{pack}/{episode}.mkv"
|
|
queue_records = [
|
|
{"title": pack, "seriesId": 600, "episodeId": 37500 + number}
|
|
for number in range(1, 14)
|
|
]
|
|
history_records = [
|
|
{
|
|
"sourceTitle": pack,
|
|
"seriesId": 600,
|
|
"episodeId": 37500 + number,
|
|
"eventType": "grabbed",
|
|
}
|
|
for number in range(1, 14)
|
|
]
|
|
result = run_script(
|
|
{
|
|
"releaseTitle": episode,
|
|
"exactPath": exact,
|
|
"scopeRoot": "/tdarr/converted/serien4k",
|
|
"queueResponse": {"records": queue_records},
|
|
"historyResponse": {"records": history_records},
|
|
"parseResponse": {
|
|
"series": {"id": 600, "title": "Marshals"},
|
|
"episodes": [{"id": 37505}],
|
|
},
|
|
"candidates": [
|
|
{
|
|
"path": exact,
|
|
"quality": {"quality": {"id": 19, "name": "WEBDL-2160p"}},
|
|
"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]["episodeIds"], [37505])
|
|
parse_call = next(
|
|
call for call in result["calls"]
|
|
if call["type"] == "fetch" and call["endpoint"] == "parse"
|
|
)
|
|
self.assertIn("s01e05", parse_call["query"].lower())
|
|
|
|
def test_percent_title_matches_queue_and_imports_exact_episode_file(self):
|
|
source_title = (
|
|
"100%.Wolf.S01E01.Die.Aufnahmeprufung.German.1080p."
|
|
"RTLP.WEB-DL.AAC2.0.H.264-GLOTZE"
|
|
)
|
|
queue_title = source_title.replace("100%.", "100.")
|
|
exact = f"/tdarr/converted/serien/{queue_title}/{source_title}.mkv"
|
|
result = run_script(
|
|
{
|
|
"releaseTitle": source_title,
|
|
"exactPath": exact,
|
|
"scopeRoot": "/tdarr/converted/serien",
|
|
"queueResponse": {
|
|
"records": [
|
|
{"title": queue_title, "seriesId": 586, "episodeId": 42301}
|
|
]
|
|
},
|
|
"parseResponse": {},
|
|
"candidates": [
|
|
{
|
|
"path": exact,
|
|
"releaseGroup": "GLOTZE",
|
|
"quality": {"quality": {"id": 3, "name": "WEBDL-1080p"}},
|
|
"languages": [{"id": 4, "name": "German"}],
|
|
}
|
|
],
|
|
}
|
|
)
|
|
|
|
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")
|
|
imported = commands[0]["body"]["files"][0]
|
|
self.assertEqual(imported["path"], exact)
|
|
self.assertEqual(imported["seriesId"], 586)
|
|
self.assertEqual(imported["episodeIds"], [42301])
|
|
self.assertNotIn(
|
|
"downloadedEpisodesScan",
|
|
[call.get("name") for call in commands],
|
|
)
|
|
|
|
def test_release_folder_matches_when_episode_filename_is_scene_shortened(self):
|
|
release = (
|
|
"Angie.Tribeca.S01E06.Frettchen.Royale.GERMAN.DL."
|
|
"1080p.HDTV.x264-SYNCTV"
|
|
)
|
|
exact = f"/tdarr/converted/serien/{release}/synctv-angie.s01e06.1080p.mkv"
|
|
result = run_script(
|
|
{
|
|
"releaseTitle": "synctv-angie.s01e06.1080p",
|
|
"exactPath": exact,
|
|
"scopeRoot": "/tdarr/converted/serien",
|
|
"queueResponse": {
|
|
"records": [
|
|
{"title": release, "seriesId": 853, "episodeId": 63336}
|
|
]
|
|
},
|
|
"historyResponse": {"records": []},
|
|
"parseResponse": {},
|
|
"candidates": [
|
|
{
|
|
"path": exact,
|
|
"quality": {"quality": {"id": 4, "name": "HDTV-1080p"}},
|
|
"languages": [{"id": 4, "name": "German"}],
|
|
}
|
|
],
|
|
}
|
|
)
|
|
|
|
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["seriesId"], 853)
|
|
self.assertEqual(imported["episodeIds"], [63336])
|
|
|
|
def test_grab_metadata_overrides_unknown_manualimport_metadata(self):
|
|
release = (
|
|
"The.Night.Agent.2025.S02E04.Desperate.Measures.2160p.NF."
|
|
"WEB-DL.MULTi.DDP5.1.Atmos.H.265-FUZEER"
|
|
)
|
|
exact = f"/tdarr/converted/serien4k/{release}/{release}.mkv"
|
|
grabbed_quality = {"quality": {"id": 18, "name": "WEBDL-2160p"}}
|
|
grabbed_languages = [
|
|
{"id": 1, "name": "English"},
|
|
{"id": 4, "name": "German"},
|
|
]
|
|
result = run_script(
|
|
{
|
|
"releaseTitle": release,
|
|
"exactPath": exact,
|
|
"scopeRoot": "/tdarr/converted/serien4k",
|
|
"queueResponse": {
|
|
"records": [
|
|
{"title": release, "seriesId": 602, "episodeId": 37528}
|
|
]
|
|
},
|
|
"historyResponse": {
|
|
"records": [
|
|
{
|
|
"sourceTitle": release,
|
|
"seriesId": 602,
|
|
"episodeId": 37528,
|
|
"eventType": "grabbed",
|
|
"quality": grabbed_quality,
|
|
"languages": grabbed_languages,
|
|
"data": {
|
|
"releaseGroup": "FUZEER",
|
|
"indexerFlags": "8",
|
|
"releaseType": "SingleEpisode",
|
|
},
|
|
}
|
|
]
|
|
},
|
|
"candidates": [
|
|
{
|
|
"path": exact,
|
|
"releaseGroup": "Sonarr",
|
|
"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"], "FUZEER")
|
|
self.assertEqual(imported["indexerFlags"], 8)
|
|
self.assertEqual(imported["releaseType"], "SingleEpisode")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|