fileflows/tests/test_sonarr_import_safety.py

@@ -0,0 +1,124 @@
+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_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],
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()