tdarr/fileflows/tests/test_radarr_import_safety.py

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"])


if __name__ == "__main__":
    unittest.main()