backend/tests/test_foreign_tracks_state.py

Rama master · réplica de lectura del repositorio original.

"""Tests for sweep state persistence and config-hash invalidation."""

import os

import pytest

from services.foreign_tracks.state import (
    PHASE_ENUMERATE,
    PHASE_IDLE,
    SweepState,
    config_hash,
    load_state,
    save_state,
)


@pytest.fixture()
def app(tmp_path):
    """Create a Flask app with an isolated SQLite DB for testing."""
    from app import create_app
    from config import reload_settings

    db_path = str(tmp_path / "test.db")
    os.environ["SUBLARR_DB_PATH"] = db_path
    os.environ["SUBLARR_API_KEY"] = ""
    os.environ["SUBLARR_LOG_LEVEL"] = "ERROR"

    reload_settings()

    application = create_app(testing=True)
    application.config["TESTING"] = True

    with application.app_context():
        yield application

    os.environ.pop("SUBLARR_DB_PATH", None)
    os.environ.pop("SUBLARR_API_KEY", None)
    os.environ.pop("SUBLARR_LOG_LEVEL", None)


def test_load_returns_a_fresh_state_when_nothing_is_stored(app):
    state = load_state()
    assert state.generation == 0
    assert state.phase == PHASE_IDLE
    assert state.enumeration_complete is False


def test_save_then_load_round_trips(app):
    save_state(
        SweepState(
            generation=3,
            phase=PHASE_ENUMERATE,
            enumeration_complete=False,
            config_hash="abc",
            started_at="2026-08-05T00:00:00+00:00",
            completed_at=None,
            paused_reason=None,
        )
    )
    state = load_state()
    assert state.generation == 3
    assert state.phase == PHASE_ENUMERATE
    assert state.config_hash == "abc"


def test_hash_is_stable_across_cosmetic_differences():
    a = config_hash(
        {"keep_languages": ["de", "en"], "keep_und": True, "include_paths": ["_Filme/"]},
        "/media",
    )
    b = config_hash(
        {"keep_languages": ["EN", "DE"], "keep_und": True, "include_paths": ["_Filme"]},
        "/media/",
    )
    assert a == b


def test_hash_changes_when_the_keep_list_narrows():
    a = config_hash({"keep_languages": ["de", "en"], "keep_und": True}, "/media")
    b = config_hash({"keep_languages": ["de"], "keep_und": True}, "/media")
    assert a != b


def test_hash_changes_with_keep_und():
    a = config_hash({"keep_languages": ["de"], "keep_und": True}, "/media")
    b = config_hash({"keep_languages": ["de"], "keep_und": False}, "/media")
    assert a != b


def test_hash_changes_with_the_media_root():
    a = config_hash({"keep_languages": ["de"]}, "/media")
    b = config_hash({"keep_languages": ["de"]}, "/other")
    assert a != b


def test_hash_changes_when_a_path_differs_only_by_case():
    # Sublarr runs on Linux/Docker, a case-sensitive filesystem: "_Filme" and
    # "_filme" are genuinely different scopes and must hash differently,
    # unlike language codes which are legitimately case-insensitive.
    a = config_hash({"keep_languages": ["de"], "include_paths": ["_Filme"]}, "/media")
    b = config_hash({"keep_languages": ["de"], "include_paths": ["_filme"]}, "/media")
    assert a != b

    c = config_hash({"keep_languages": ["de"]}, "/Media")
    d = config_hash({"keep_languages": ["de"]}, "/media")
    assert c != d


def test_load_returns_a_fresh_state_on_invalid_json(app):
    from db.repositories.config import ConfigRepository

    ConfigRepository().save_config_entry("foreign_track_sweep_state", "not json")

    state = load_state()

    assert state == SweepState()


def test_load_returns_a_fresh_state_on_non_object_json(app):
    from db.repositories.config import ConfigRepository

    ConfigRepository().save_config_entry("foreign_track_sweep_state", "[1, 2, 3]")

    state = load_state()

    assert state == SweepState()


def test_load_returns_a_fresh_state_on_unknown_field(app):
    from db.repositories.config import ConfigRepository

    # A legacy blob from an older SweepState shape must be discarded rather
    # than partially loaded — a fresh state is recoverable, a half-loaded one
    # is not.
    ConfigRepository().save_config_entry(
        "foreign_track_sweep_state", '{"generation": 1, "bogus_field": 1}'
    )

    state = load_state()

    assert state == SweepState()