139 lines
5.4 KiB
Python
139 lines
5.4 KiB
Python
"""Regression checks for host-side /xbox storage and safe upgrade migration."""
|
|
import os
|
|
import shutil
|
|
import subprocess
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from app import main
|
|
|
|
PROJECT = Path(__file__).resolve().parents[1]
|
|
|
|
|
|
def _copy_scripts(dst):
|
|
for name in ('xbox-paths.sh', 'migrate-to-xbox.sh', 'setup.sh', '.env.example'):
|
|
shutil.copy2(PROJECT / name, dst / name)
|
|
|
|
|
|
def _run_script(project, name, root, **extra):
|
|
env = os.environ.copy()
|
|
env.update(XBOX_HOST_DIR=str(root), XBOX_MIGRATION_OFFLINE='1', XBOX_SKIP_DOCKER='1', **extra)
|
|
return subprocess.run(['sh', name], cwd=project, env=env,
|
|
capture_output=True, text=True, timeout=30)
|
|
|
|
|
|
def test_config_uses_host_and_dedicated_downloads():
|
|
compose = (PROJECT / 'compose.yaml').read_text()
|
|
assert 'source: ${XBOX_HOST_DIR:-/xbox}' in compose
|
|
assert 'source: ${XBOX_HOST_DIR:-/xbox}/downloads' in compose
|
|
assert 'target: /data/raw' in compose
|
|
assert 'target: /data' in compose
|
|
assert 'create_host_path: false' in compose
|
|
assert 'TMPDIR: /data/work/tmp' in compose
|
|
assert './data:/data' not in compose
|
|
assert 'XBOX_HOST_DIR=/xbox' in (PROJECT / '.env.example').read_text()
|
|
assert '--dir=/data/raw' in (PROJECT / 'entrypoint.sh').read_text()
|
|
|
|
|
|
@pytest.mark.skipif(not shutil.which('rsync'), reason='rsync not installed on test host')
|
|
def test_migrate_preserves_original_data_and_queue(tmp_path):
|
|
project = tmp_path / 'old project'
|
|
project.mkdir()
|
|
_copy_scripts(project)
|
|
host = tmp_path / 'host volume'
|
|
host.mkdir()
|
|
data = project / 'data'
|
|
(data / 'raw' / 'task-1').mkdir(parents=True)
|
|
(data / 'raw' / 'task-1' / 'demo.iso.part').write_bytes(b'12345')
|
|
(data / 'work' / 'task-1').mkdir(parents=True)
|
|
(data / 'work' / 'task-1' / 'unpacked.iso').write_bytes(b'unpacked')
|
|
(data / 'ready' / 'xex' / 'demo').mkdir(parents=True)
|
|
(data / 'ready' / 'xex' / 'demo' / 'default.xex').write_bytes(b'xex')
|
|
(data / 'private').mkdir()
|
|
(data / 'private' / 'archive-session.json').write_text('{"session": "example"}')
|
|
(data / 'queue.sqlite3').write_bytes(b'queue')
|
|
(data / 'queue.sqlite3-wal').write_bytes(b'wal')
|
|
(data / 'aria2').mkdir()
|
|
(data / 'aria2' / 'session.txt').write_text('/data/raw/task-1/demo.iso')
|
|
|
|
result = _run_script(project, 'migrate-to-xbox.sh', host)
|
|
assert result.returncode == 0, result.stderr
|
|
assert (host / 'downloads' / 'task-1' / 'demo.iso.part').read_bytes() == b'12345'
|
|
assert (host / 'work' / 'task-1' / 'unpacked.iso').read_bytes() == b'unpacked'
|
|
assert (host / 'ready' / 'xex' / 'demo' / 'default.xex').read_bytes() == b'xex'
|
|
assert (host / 'queue.sqlite3').read_bytes() == b'queue'
|
|
assert (host / 'queue.sqlite3-wal').read_bytes() == b'wal'
|
|
assert (host / 'private' / 'archive-session.json').is_file()
|
|
assert (host / 'aria2' / 'session.txt').read_text() == '/data/raw/task-1/demo.iso'
|
|
assert not (host / 'raw').exists()
|
|
assert not (host / '.x360-migration-in-progress').exists()
|
|
assert (data / 'raw' / 'task-1' / 'demo.iso.part').read_bytes() == b'12345'
|
|
assert (data / 'queue.sqlite3').read_bytes() == b'queue'
|
|
|
|
|
|
@pytest.mark.skipif(not shutil.which('rsync'), reason='rsync not installed on test host')
|
|
def test_migration_does_not_replace_existing_queue(tmp_path):
|
|
project = tmp_path / 'project'
|
|
project.mkdir()
|
|
_copy_scripts(project)
|
|
host = tmp_path / 'xbox'
|
|
host.mkdir()
|
|
(project / 'data').mkdir()
|
|
(project / 'data' / 'queue.sqlite3').write_bytes(b'old')
|
|
(host / 'queue.sqlite3').write_bytes(b'new')
|
|
result = _run_script(project, 'migrate-to-xbox.sh', host)
|
|
assert result.returncode != 0
|
|
assert (host / 'queue.sqlite3').read_bytes() == b'new'
|
|
assert not (host / '.x360-migration-in-progress').exists()
|
|
|
|
|
|
def test_setup_does_not_silently_drop_existing_database(tmp_path):
|
|
project = tmp_path / 'project'
|
|
project.mkdir()
|
|
_copy_scripts(project)
|
|
host = tmp_path / 'xbox'
|
|
host.mkdir()
|
|
(project / 'data').mkdir()
|
|
(project / 'data' / 'queue.sqlite3').write_bytes(b'important')
|
|
result = _run_script(project, 'setup.sh', host)
|
|
assert result.returncode != 0
|
|
assert 'migrate-to-xbox.sh' in result.stderr
|
|
assert not (project / '.env').exists()
|
|
|
|
|
|
def test_setup_persists_custom_host_path(tmp_path):
|
|
project = tmp_path / 'project'
|
|
project.mkdir()
|
|
_copy_scripts(project)
|
|
host = tmp_path / 'external-host-share'
|
|
host.mkdir()
|
|
result = _run_script(project, 'setup.sh', host)
|
|
assert result.returncode == 0, result.stderr
|
|
assert f'XBOX_HOST_DIR={host}' in (project / '.env').read_text()
|
|
assert (host / 'downloads').is_dir()
|
|
assert (host / 'work' / 'tmp').is_dir()
|
|
assert (host / 'ready').is_dir()
|
|
|
|
|
|
def test_missing_host_directory_fails_without_creating(tmp_path):
|
|
project = tmp_path / 'project'
|
|
project.mkdir()
|
|
_copy_scripts(project)
|
|
host = tmp_path / 'not-mounted'
|
|
result = _run_script(project, 'setup.sh', host)
|
|
assert result.returncode != 0
|
|
assert not host.exists()
|
|
assert not (project / '.env').exists()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_system_shows_actual_host_path(tmp_path, monkeypatch):
|
|
from app import store
|
|
monkeypatch.setattr(store, 'DATA', tmp_path)
|
|
(tmp_path / 'raw').mkdir(parents=True, exist_ok=True)
|
|
monkeypatch.setenv('XBOX_HOST_DIR', '/xbox')
|
|
result = await main.system()
|
|
assert result['host_data_dir'] == '/xbox'
|
|
assert result['total'] > 0
|