"""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