wrolpi/modules/zim/test/test_restart_kiwix.py

"""Tests for the restart_kiwix switch handler triggered by model_zim()."""
from unittest import mock

import pytest

from modules.zim import lib
from modules.zim.models import Zim
from wrolpi.api_utils import api_app
from wrolpi.files.models import FileGroup
from wrolpi.files.worker import file_worker


@pytest.mark.asyncio
async def test_restart_kiwix_systemctl_command_shape():
    """restart_kiwix() must call systemctl with `restart` and the unit name as
    separate argv entries.  Adjacent-string-literal typos (missing comma in
    a Python tuple) silently produce 'restartwrolpi-kiwix.service' as a single
    argument, which systemctl rejects."""
    captured = {}

    async def fake_run_command(cmd):
        captured['cmd'] = cmd
        return mock.MagicMock(return_code=0, stderr=b'')

    with mock.patch.object(lib, 'PYTEST', False), \
            mock.patch.object(lib, 'DOCKERIZED', False), \
            mock.patch.object(lib, 'run_command', side_effect=fake_run_command):
        await lib.restart_kiwix()

    cmd = captured['cmd']
    assert 'restart' in cmd
    assert 'wrolpi-kiwix.service' in cmd
    # 'restart' and the unit must NOT be concatenated into one argv entry.
    assert 'restartwrolpi-kiwix.service' not in cmd


@pytest.mark.asyncio
async def test_model_zim_activates_restart_switch(async_client, test_session, zim_path_factory):
    """model_zim() should activate the restart_kiwix switch when creating a new Zim."""
    # Create a zim file and its FileGroup
    zim_path = zim_path_factory('test_restart.zim')
    file_group = FileGroup.from_paths(test_session, zim_path)
    test_session.commit()

    # Verify no switches are pending
    switches = dict(api_app.shared_ctx.switches) if api_app.shared_ctx.switches else {}
    assert 'restart_kiwix' not in switches

    # Call model_zim which should activate the switch
    lib.model_zim(file_group, test_session)
    test_session.commit()

    # Verify the switch was activated
    switches = dict(api_app.shared_ctx.switches) if api_app.shared_ctx.switches else {}
    assert 'restart_kiwix' in switches


@pytest.mark.asyncio
async def test_existing_zim_does_not_trigger_restart(async_client, test_session, zim_factory, await_switches):
    """model_zim() should NOT activate restart switch for existing Zims."""
    # Create a zim using the factory (this calls model_zim internally via Zim.from_paths)
    zim = zim_factory('existing.zim')
    test_session.commit()

    # Clear any pending switches
    await await_switches()

    # Verify no switches are pending
    switches = dict(api_app.shared_ctx.switches) if api_app.shared_ctx.switches else {}
    assert 'restart_kiwix' not in switches

    # Call model_zim again on the same FileGroup - should NOT trigger restart
    lib.model_zim(zim.file_group, test_session)
    test_session.commit()

    # Verify the switch was NOT activated (Zim already exists)
    switches = dict(api_app.shared_ctx.switches) if api_app.shared_ctx.switches else {}
    assert 'restart_kiwix' not in switches


@pytest.mark.asyncio
async def test_multiple_zims_trigger_single_restart(async_client, test_session, zim_path_factory):
    """Multiple new Zims should result in only one restart (debouncing via switch)."""
    # Create multiple zim files
    paths = [zim_path_factory(f'test_debounce_{i}.zim') for i in range(3)]

    # Create FileGroups and model each one
    for path in paths:
        file_group = FileGroup.from_paths(test_session, path)
        test_session.commit()
        lib.model_zim(file_group, test_session)
        test_session.commit()

    # Verify switch was activated (only once, due to debouncing)
    switches = dict(api_app.shared_ctx.switches) if api_app.shared_ctx.switches else {}
    assert 'restart_kiwix' in switches
    # The switch dict contains only one entry for restart_kiwix, proving debouncing works
    assert list(switches.keys()).count('restart_kiwix') == 1


@pytest.mark.asyncio
async def test_remove_outdated_zim_files_triggers_restart(async_client, test_session, test_directory, zim_path_factory,
                                                          await_switches):
    """remove_outdated_zim_files() should trigger the restart_kiwix switch."""
    # Create outdated and current zim files
    zim_dir = test_directory / 'zims'
    zim_dir.mkdir(exist_ok=True)

    # Create an "outdated" zim and a "current" zim (based on date in filename)
    outdated_path = zim_dir / 'wikipedia_en_all_maxi_2020-01.zim'
    current_path = zim_dir / 'wikipedia_en_all_maxi_2020-02.zim'

    # Copy test zim bytes to both paths
    test_zim_bytes = zim_path_factory().read_bytes()
    outdated_path.write_bytes(test_zim_bytes)
    current_path.write_bytes(test_zim_bytes)

    # Clear any pending switches first
    await await_switches()

    # Call remove_outdated_zim_files
    deleted_count = await lib.remove_outdated_zim_files(zim_dir)

    # Verify the outdated file was deleted
    assert deleted_count == 1
    assert not outdated_path.exists()
    assert current_path.exists()

    # Verify the switch was activated
    switches = dict(api_app.shared_ctx.switches) if api_app.shared_ctx.switches else {}
    assert 'restart_kiwix' in switches


@pytest.mark.asyncio
async def test_remove_outdated_zim_files_no_restart_when_nothing_deleted(async_client, test_session, test_directory,
                                                                        zim_path_factory, await_switches):
    """Clearing a stale flag must not restart Kiwix when no file was deleted."""
    zim_dir = test_directory / 'zims'
    zim_dir.mkdir(exist_ok=True)
    (zim_dir / 'wikipedia_en_all_maxi_2020-02.zim').write_bytes(zim_path_factory().read_bytes())

    await await_switches()

    deleted_count = await lib.remove_outdated_zim_files(zim_dir)

    assert deleted_count == 0
    switches = dict(api_app.shared_ctx.switches) if api_app.shared_ctx.switches else {}
    assert 'restart_kiwix' not in switches


@pytest.mark.asyncio
async def test_remove_outdated_zim_files_skips_post_processing(async_client, test_session, test_directory,
                                                               zim_path_factory):
    """Deleting outdated Zims must not run the global modelers inside the request.

    `DELETE /api/zim/outdated` awaits this; modeling the whole unindexed backlog there exceeded the
    response timeout and the cancelled request stranded the file worker status."""
    zim_dir = test_directory / 'zims'
    zim_dir.mkdir(exist_ok=True)
    test_zim_bytes = zim_path_factory().read_bytes()
    paths = [zim_dir / 'wikipedia_en_all_maxi_2020-01.zim', zim_dir / 'wikipedia_en_all_maxi_2020-02.zim']
    for path in paths:
        path.write_bytes(test_zim_bytes)
    await file_worker.refresh_sync(paths)
    assert test_session.query(FileGroup).count() == 2

    with mock.patch('wrolpi.files.worker.apply_modelers') as apply_modelers:
        deleted_count = await lib.remove_outdated_zim_files(zim_dir)

    assert deleted_count == 1
    apply_modelers.assert_not_called()
    # The deleted Zim is still removed from the DB.
    assert test_session.query(FileGroup).count() == 1