41 lines
1.3 KiB
Python
41 lines
1.3 KiB
Python
import json
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
|
|
FLOW = Path(__file__).parents[1] / "flow" / "flow_auto_movie_pipeline.json"
|
|
|
|
|
|
class FlowImportFailureRoutingTests(unittest.TestCase):
|
|
def test_arr_import_failures_route_to_fail_flow(self):
|
|
flow = json.loads(FLOW.read_text())
|
|
fail_node = next(
|
|
part
|
|
for part in flow["Parts"]
|
|
if part.get("FlowElementUid") == "FileFlows.BasicNodes.Functions.FailFlow"
|
|
)
|
|
import_nodes = [
|
|
part
|
|
for part in flow["Parts"]
|
|
if part.get("Name")
|
|
in {"Sonarr - Trigger Manual Import", "Radarr - Trigger Manual Import"}
|
|
]
|
|
|
|
self.assertEqual(len(import_nodes), 4)
|
|
for node in import_nodes:
|
|
with self.subTest(node=node["Uid"]):
|
|
self.assertEqual(node["ErrorConnection"]["InputNode"], fail_node["Uid"])
|
|
|
|
def test_fhd_radarr_import_timeout_handles_large_nas_copy(self):
|
|
flow = json.loads(FLOW.read_text())
|
|
fhd_radarr = next(
|
|
part
|
|
for part in flow["Parts"]
|
|
if part.get("Name") == "Radarr - Trigger Manual Import"
|
|
and part.get("Model", {}).get("ImportPath") == "/tdarr/converted/video/"
|
|
)
|
|
self.assertGreaterEqual(fhd_radarr["Model"]["TimeOut"], 900)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|