simplefe

Log | Files | Refs

test_app.py (7517B)

      1 import os
      2 
      3 import simplefe.app as app_module
      4 
      5 
      6 class ImmediateThread:
      7     def __init__(self, target=None, kwargs=None, daemon=None):
      8         self._target = target
      9         self._kwargs = kwargs or {}
     10 
     11     def start(self):
     12         self._target(**self._kwargs)
     13 
     14 
     15 def test_rip_index_shows_disc_info(client, auth_headers, monkeypatch):
     16     monkeypatch.setattr(
     17         app_module.makemkv,
     18         "query_disc",
     19         lambda: app_module.makemkv.DiscInfo(
     20             raw_id="ARRIVAL_2016", guessed_name="Arrival 2016", main_title=1
     21         ),
     22     )
     23     monkeypatch.setattr(
     24         app_module.makemkv, "find_existing_rip", lambda db, raw_id: None
     25     )
     26 
     27     response = client.get("/rip", headers=auth_headers)
     28     assert response.status_code == 200
     29     assert b"Arrival 2016" in response.data
     30     assert b"already been ripped" not in response.data
     31 
     32 
     33 def test_rip_index_skips_disc_query_while_job_running(client, auth_headers, monkeypatch):
     34     def fail_if_called():
     35         raise AssertionError("query_disc should not be called while a job is running")
     36 
     37     monkeypatch.setattr(app_module.makemkv, "query_disc", fail_if_called)
     38     client.application.config["JOB"]["running"] = True
     39 
     40     response = client.get("/rip", headers=auth_headers)
     41     assert response.status_code == 200
     42     assert b"already in progress" in response.data
     43 
     44 
     45 def test_rip_index_shows_duplicate_warning(client, auth_headers, monkeypatch):
     46     monkeypatch.setattr(
     47         app_module.makemkv,
     48         "query_disc",
     49         lambda: app_module.makemkv.DiscInfo(
     50             raw_id="ARRIVAL_2016", guessed_name="Arrival 2016", main_title=1
     51         ),
     52     )
     53     monkeypatch.setattr(
     54         app_module.makemkv,
     55         "find_existing_rip",
     56         lambda db, raw_id: {
     57             "title": "Arrival 2016",
     58             "output_file": "/x.mkv",
     59             "ripped_at": "now",
     60         },
     61     )
     62 
     63     response = client.get("/rip", headers=auth_headers)
     64     assert response.status_code == 200
     65     assert b"already been ripped" in response.data
     66 
     67 
     68 def test_rip_start_requires_movie_name(client, auth_headers, monkeypatch):
     69     monkeypatch.setattr(
     70         app_module.makemkv, "find_existing_rip", lambda db, raw_id: None
     71     )
     72     response = client.post(
     73         "/rip/start",
     74         data={"movie_name": "", "raw_id": "X", "title": "1"},
     75         headers=auth_headers,
     76     )
     77     assert response.status_code == 200
     78     assert b"Movie name is required" in response.data
     79 
     80 
     81 def test_rip_start_blocks_duplicate_without_override(client, auth_headers, monkeypatch):
     82     monkeypatch.setattr(
     83         app_module.makemkv,
     84         "find_existing_rip",
     85         lambda db, raw_id: {"title": "X", "output_file": "/x.mkv", "ripped_at": "now"},
     86     )
     87     response = client.post(
     88         "/rip/start",
     89         data={"movie_name": "Arrival 2016", "raw_id": "X", "title": "1"},
     90         headers=auth_headers,
     91     )
     92     assert response.status_code == 200
     93     assert b"already been ripped" in response.data
     94 
     95 
     96 def test_rip_start_allows_duplicate_with_override(client, auth_headers, monkeypatch):
     97     calls = []
     98     monkeypatch.setattr(
     99         app_module.makemkv,
    100         "find_existing_rip",
    101         lambda db, raw_id: {"title": "X", "output_file": "/x.mkv", "ripped_at": "now"},
    102     )
    103     monkeypatch.setattr(app_module.pipeline, "run", lambda **kwargs: calls.append(kwargs))
    104     monkeypatch.setattr(app_module.threading, "Thread", ImmediateThread)
    105 
    106     response = client.post(
    107         "/rip/start",
    108         data={
    109             "movie_name": "Arrival 2016",
    110             "raw_id": "X",
    111             "title": "1",
    112             "override_duplicate": "on",
    113         },
    114         headers=auth_headers,
    115     )
    116     assert response.status_code == 302
    117     assert len(calls) == 1
    118     assert calls[0]["movie_name"] == "Arrival 2016"
    119 
    120 
    121 def test_rip_start_rejects_when_already_running(client, auth_headers):
    122     client.application.config["JOB"]["running"] = True
    123     response = client.post(
    124         "/rip/start",
    125         data={"movie_name": "X", "raw_id": "Y", "title": "0"},
    126         headers=auth_headers,
    127     )
    128     assert response.status_code == 409
    129 
    130 
    131 def test_rip_status_shows_idle_message(client, auth_headers):
    132     response = client.get("/rip/status", headers=auth_headers)
    133     assert response.status_code == 200
    134     assert b"No rip has been started" in response.data
    135 
    136 
    137 def test_rip_status_shows_progress(client, auth_headers):
    138     client.application.config["JOB"].update(
    139         {
    140             "running": True,
    141             "stage": "encoding",
    142             "detail": "Encoding movie.mkv...",
    143             "error": None,
    144         }
    145     )
    146     response = client.get("/rip/status", headers=auth_headers)
    147     assert b"encoding" in response.data
    148     assert b"Encoding movie.mkv" in response.data
    149 
    150 
    151 def test_routes_require_auth(client):
    152     response = client.get("/rip")
    153     assert response.status_code == 401
    154 
    155 
    156 def _clear_setup_keys(monkeypatch):
    157     for key in app_module.SETUP_KEYS:
    158         monkeypatch.delenv(key, raising=False)
    159 
    160 
    161 def test_rip_index_shows_setup_banner_when_unconfigured(client, auth_headers, monkeypatch):
    162     _clear_setup_keys(monkeypatch)
    163     monkeypatch.setattr(
    164         app_module.makemkv,
    165         "query_disc",
    166         lambda: app_module.makemkv.DiscInfo(raw_id="", guessed_name="", main_title=0),
    167     )
    168     monkeypatch.setattr(
    169         app_module.makemkv, "find_existing_rip", lambda db, raw_id: None
    170     )
    171 
    172     response = client.get("/rip", headers=auth_headers)
    173     assert b"Set up now" in response.data
    174 
    175 
    176 def test_rip_index_hides_setup_banner_when_configured(client, auth_headers, monkeypatch):
    177     _clear_setup_keys(monkeypatch)
    178     monkeypatch.setenv("JELLYFIN_TOKEN", "jtok")
    179     monkeypatch.setenv("PUSHOVER_TOKEN", "ptok")
    180     monkeypatch.setenv("PUSHOVER_USER", "puser")
    181     monkeypatch.setattr(
    182         app_module.makemkv,
    183         "query_disc",
    184         lambda: app_module.makemkv.DiscInfo(raw_id="", guessed_name="", main_title=0),
    185     )
    186     monkeypatch.setattr(
    187         app_module.makemkv, "find_existing_rip", lambda db, raw_id: None
    188     )
    189 
    190     response = client.get("/rip", headers=auth_headers)
    191     assert b"Set up now" not in response.data
    192 
    193 
    194 def test_setup_get_shows_current_values(client, auth_headers, monkeypatch):
    195     _clear_setup_keys(monkeypatch)
    196     monkeypatch.setenv("JELLYFIN_TOKEN", "existing-jtok")
    197 
    198     response = client.get("/setup", headers=auth_headers)
    199     assert response.status_code == 200
    200     assert b"existing-jtok" in response.data
    201 
    202 
    203 def test_setup_post_saves_values_and_applies_immediately(client, auth_headers, monkeypatch):
    204     _clear_setup_keys(monkeypatch)
    205 
    206     response = client.post(
    207         "/setup",
    208         data={
    209             "JELLYFIN_TOKEN": "new-jtok",
    210             "PUSHOVER_TOKEN": "new-ptok",
    211             "PUSHOVER_USER": "new-puser",
    212         },
    213         headers=auth_headers,
    214     )
    215     assert response.status_code == 302
    216 
    217     assert os.environ["JELLYFIN_TOKEN"] == "new-jtok"
    218     assert os.environ["PUSHOVER_TOKEN"] == "new-ptok"
    219     assert os.environ["PUSHOVER_USER"] == "new-puser"
    220 
    221     dotenv_path = client.application.config["DOTENV_PATH"]
    222     content = open(dotenv_path).read()
    223     assert "new-jtok" in content
    224 
    225 
    226 def test_setup_post_blank_field_does_not_overwrite_existing_value(
    227     client, auth_headers, monkeypatch
    228 ):
    229     _clear_setup_keys(monkeypatch)
    230     monkeypatch.setenv("JELLYFIN_TOKEN", "keep-me")
    231 
    232     client.post(
    233         "/setup",
    234         data={"JELLYFIN_TOKEN": "", "PUSHOVER_TOKEN": "ptok", "PUSHOVER_USER": "puser"},
    235         headers=auth_headers,
    236     )
    237 
    238     assert os.environ["JELLYFIN_TOKEN"] == "keep-me"