simplefe

Log | Files | Refs

test_pipeline.py (4221B)

      1 from pathlib import Path
      2 
      3 from simplefe import pipeline
      4 from simplefe.settings import load_settings
      5 
      6 
      7 def _settings(tmp_path):
      8     return load_settings(
      9         env={
     10             "OUTPUT_DIR": str(tmp_path),
     11             "RIP_DB_PATH": str(tmp_path / "rip.db"),
     12         }
     13     )
     14 
     15 
     16 def _fresh_status():
     17     return {"running": False, "stage": None, "detail": None, "error": None}
     18 
     19 
     20 def test_run_success_updates_status_through_stages(tmp_path, monkeypatch):
     21     ripped_file = tmp_path / "title00.mkv"
     22 
     23     def fake_rip_title(disc, title, out_dir):
     24         ripped_file.write_bytes(b"ripped data")
     25         return ripped_file
     26 
     27     def fake_encode(input_path, output_path, settings):
     28         Path(output_path).write_bytes(b"encoded data")
     29         return True
     30 
     31     record_calls = []
     32     pushover_calls = []
     33     jellyfin_calls = []
     34     eject_calls = []
     35 
     36     monkeypatch.setattr(pipeline.makemkv, "rip_title", fake_rip_title)
     37     monkeypatch.setattr(pipeline.encode_module, "encode", fake_encode)
     38     monkeypatch.setattr(
     39         pipeline.makemkv,
     40         "record_rip",
     41         lambda *args, **kwargs: record_calls.append((args, kwargs)),
     42     )
     43     monkeypatch.setattr(
     44         pipeline.notify_module,
     45         "send_pushover",
     46         lambda *args, **kwargs: pushover_calls.append((args, kwargs)) or True,
     47     )
     48     monkeypatch.setattr(
     49         pipeline.notify_module,
     50         "trigger_jellyfin_scan",
     51         lambda *args, **kwargs: jellyfin_calls.append((args, kwargs)) or True,
     52     )
     53     monkeypatch.setattr(
     54         pipeline.subprocess, "run", lambda *args, **kwargs: eject_calls.append(args)
     55     )
     56 
     57     status = _fresh_status()
     58     pipeline.run(
     59         disc=0,
     60         title=1,
     61         movie_name="Test Movie",
     62         raw_id="TEST_RAW_ID",
     63         settings=_settings(tmp_path),
     64         status=status,
     65         jellyfin_token="jtok",
     66         pushover_token="ptok",
     67         pushover_user="puser",
     68     )
     69 
     70     assert status["error"] is None
     71     assert status["running"] is False
     72     assert status["stage"] == "done"
     73     assert len(record_calls) == 1
     74     assert len(pushover_calls) == 1
     75     assert len(jellyfin_calls) == 1
     76     assert len(eject_calls) == 1
     77     assert not ripped_file.exists()
     78     encoded_file = tmp_path / "Test Movie" / "Test Movie_x265.mkv"
     79     assert encoded_file.exists()
     80 
     81 
     82 def test_run_failure_sets_error_and_stops_at_that_stage(tmp_path, monkeypatch):
     83     def failing_rip_title(disc, title, out_dir):
     84         raise RuntimeError("drive not found")
     85 
     86     monkeypatch.setattr(pipeline.makemkv, "rip_title", failing_rip_title)
     87 
     88     status = _fresh_status()
     89     pipeline.run(
     90         disc=0,
     91         title=1,
     92         movie_name="Test Movie",
     93         raw_id="ID",
     94         settings=_settings(tmp_path),
     95         status=status,
     96     )
     97 
     98     assert status["running"] is False
     99     assert status["error"] == "drive not found"
    100     assert status["stage"] == "ripping"
    101 
    102 
    103 def test_run_skips_notifications_when_tokens_missing(tmp_path, monkeypatch):
    104     ripped_file = tmp_path / "title00.mkv"
    105 
    106     def fake_rip_title(disc, title, out_dir):
    107         ripped_file.write_bytes(b"ripped data")
    108         return ripped_file
    109 
    110     def fake_encode(input_path, output_path, settings):
    111         Path(output_path).write_bytes(b"encoded data")
    112         return True
    113 
    114     pushover_calls = []
    115     jellyfin_calls = []
    116 
    117     monkeypatch.setattr(pipeline.makemkv, "rip_title", fake_rip_title)
    118     monkeypatch.setattr(pipeline.encode_module, "encode", fake_encode)
    119     monkeypatch.setattr(pipeline.makemkv, "record_rip", lambda *a, **k: None)
    120     monkeypatch.setattr(
    121         pipeline.notify_module,
    122         "send_pushover",
    123         lambda *a, **k: pushover_calls.append(1) or True,
    124     )
    125     monkeypatch.setattr(
    126         pipeline.notify_module,
    127         "trigger_jellyfin_scan",
    128         lambda *a, **k: jellyfin_calls.append(1) or True,
    129     )
    130     monkeypatch.setattr(pipeline.subprocess, "run", lambda *a, **k: None)
    131 
    132     status = _fresh_status()
    133     pipeline.run(
    134         disc=0,
    135         title=1,
    136         movie_name="Test Movie",
    137         raw_id="ID",
    138         settings=_settings(tmp_path),
    139         status=status,
    140     )
    141 
    142     assert status["error"] is None
    143     assert status["stage"] == "done"
    144     assert pushover_calls == []
    145     assert jellyfin_calls == []