replypier/deploy/server/tests/test_updater.py

import sys
from pathlib import Path
from unittest.mock import Mock

import pytest

sys.path.insert(0, str(Path(__file__).parents[1] / 'updater'))
import cli
import deployment
import repository
import state

CURRENT = 'a' * 40
TARGET = 'b' * 40


@pytest.fixture
def cfg(tmp_path):
    cfg = {'state_dir': str(tmp_path / 'state'), 'project_dir': str(tmp_path / 'project'),
           'initial_revision': CURRENT, 'owner_ids': ['owner'], 'branch': 'main',
           'web_url': 'https://gitea.example/repo'}
    Path(cfg['project_dir']).mkdir()
    with state.transaction(cfg) as data:
        data['latest'] = {'revision': TARGET}
    return cfg


def test_check_never_starts_deployment(cfg, monkeypatch):
    monkeypatch.setattr(repository, 'fetch', lambda c: TARGET)
    monkeypatch.setattr(repository, 'ensure_forward', Mock())
    monkeypatch.setattr(repository, 'commit', lambda c, r: {'revision': r})
    launch = Mock()
    monkeypatch.setattr(cli, 'launch', launch)
    cli.check(cfg)
    assert state.read(cfg)['latest']['revision'] == TARGET
    assert not state.read(cfg)['checking']
    launch.assert_not_called()


def test_failed_check_is_visible_and_retryable(cfg, monkeypatch):
    monkeypatch.setattr(repository, 'fetch', Mock(side_effect=RuntimeError('unreachable')))
    with pytest.raises(RuntimeError):
        cli.check(cfg)
    data = state.read(cfg)
    assert data['check_error'] == 'unreachable'
    assert not data['checking']
    assert not state.public_status(cfg)['update_available']


def test_request_pins_the_confirmed_revision_and_rejects_duplicates(cfg, monkeypatch):
    launch = Mock()
    monkeypatch.setattr(cli, 'launch', launch)
    assert cli.request(cfg, TARGET, 'owner') == {'accepted': True}
    assert state.read(cfg)['update']['target_revision'] == TARGET
    launch.assert_called_once_with('replypier-update.service')
    with pytest.raises(ValueError, match='已有更新'):
        cli.request(cfg, TARGET, 'owner')


@pytest.mark.parametrize('revision,actor,error', [
    (TARGET, 'another-admin', PermissionError),
    ('main;rm -rf /', 'owner', ValueError),
    ('c' * 40, 'owner', ValueError),
    (CURRENT, 'owner', ValueError),
])
def test_unapproved_requests_do_not_start_a_service(cfg, monkeypatch, revision, actor, error):
    launch = Mock()
    monkeypatch.setattr(cli, 'launch', launch)
    with pytest.raises(error):
        cli.request(cfg, revision, actor)
    launch.assert_not_called()


def test_failed_service_start_releases_the_queued_state(cfg, monkeypatch):
    monkeypatch.setattr(cli, 'launch', Mock(side_effect=RuntimeError('systemd unavailable')))
    with pytest.raises(RuntimeError):
        cli.request(cfg, TARGET, 'owner')
    assert state.read(cfg)['update']['status'] == 'failed'


def test_build_failure_keeps_live_service_and_version(cfg, monkeypatch):
    d = deployment.Deployment(cfg)
    monkeypatch.setattr(deployment.shutil, 'disk_usage', lambda p: type('Usage', (), {'free': 10 * 1024**3})())
    monkeypatch.setattr(repository, 'git', Mock())
    monkeypatch.setattr(repository, 'ensure_forward', Mock())
    monkeypatch.setattr(repository, 'ensure_clean', Mock())
    monkeypatch.setattr(d, 'prepare', Mock(side_effect=RuntimeError('build failed')))
    services = Mock()
    monkeypatch.setattr(d, 'services', services)
    with pytest.raises(RuntimeError, match='build failed'):
        d.apply(TARGET)
    services.assert_not_called()
    assert state.read(cfg)['current']['revision'] == CURRENT
    assert state.read(cfg)['update']['status'] == 'failed'


def test_recovery_restores_the_previous_version_and_keeps_error(cfg, monkeypatch):
    d = deployment.Deployment(cfg)
    (state.directory(cfg) / 'maintenance').touch()
    with state.transaction(cfg) as data:
        data['current'] = {'revision': TARGET}
        data['update'] = {'status': 'migrating', 'backup_ready': True,
                          'backup_path': '/backup', 'previous_revision': CURRENT, 'error': 'migration failed'}
    restore = Mock()
    monkeypatch.setattr(d, 'restore', restore)
    monkeypatch.setattr(repository, 'commit', lambda c, r: {'revision': r})
    d.recover()
    restore.assert_called_once_with(Path('/backup'), boot=False)
    assert state.read(cfg)['current']['revision'] == CURRENT
    assert state.read(cfg)['update']['status'] == 'rolled_back'
    assert state.read(cfg)['update']['error'] == 'migration failed'
    assert not (state.directory(cfg) / 'maintenance').exists()


def test_boot_recovery_does_not_start_a_dependency_of_itself(cfg, monkeypatch):
    d = deployment.Deployment(cfg)
    (state.directory(cfg) / 'maintenance').touch()
    with state.transaction(cfg) as data:
        data['update'] = {'status': 'backing_up', 'backup_ready': False}
    services = Mock()
    monkeypatch.setattr(d, 'services', services)
    d.recover(boot=True)
    services.assert_not_called()
    assert not (state.directory(cfg) / 'maintenance').exists()


def test_recovery_failure_keeps_the_maintenance_gate(cfg, monkeypatch):
    d = deployment.Deployment(cfg)
    (state.directory(cfg) / 'maintenance').touch()
    with state.transaction(cfg) as data:
        data['update'] = {'status': 'migrating', 'backup_ready': True, 'backup_path': '/backup'}
    monkeypatch.setattr(d, 'restore', Mock(side_effect=RuntimeError('restore failed')))
    with pytest.raises(RuntimeError, match='restore failed'):
        d.recover()
    assert (state.directory(cfg) / 'maintenance').exists()


@pytest.mark.parametrize('path', ['backend/.env','frontend/.env.prod','backend/.venv/a','deploy/local/.state/a','backend/uploads/a','../../etc/passwd','key.pem'])
def test_runtime_files_are_never_deployed(path):
    assert repository.protected(path)


def test_env_templates_remain_deployable():
    assert not repository.protected('backend/.env.example')


def test_cannot_replace_failed_recovery_with_a_new_job(cfg, monkeypatch):
    (state.directory(cfg) / 'maintenance').touch()
    launch = Mock()
    monkeypatch.setattr(cli, 'launch', launch)
    with pytest.raises(ValueError, match='尚未恢复'):
        cli.request(cfg, TARGET, 'owner')
    launch.assert_not_called()


def test_interrupted_build_does_not_leave_a_permanently_busy_job(cfg):
    with state.transaction(cfg) as data:
        data['update'] = {'status': 'building', 'message': 'building'}
    deployment.Deployment(cfg).recover()
    assert state.read(cfg)['update']['status'] == 'failed'
    assert state.read(cfg)['current']['revision'] == CURRENT


def test_public_build_permissions_do_not_inherit_private_backup_umask(tmp_path):
    root = tmp_path / 'dashboard'
    root.mkdir(mode=0o700)
    (root / 'assets').mkdir(mode=0o700)
    (root / 'assets/main.js').write_text('export default {}')
    (root / 'assets/main.js').chmod(0o600)
    deployment.Deployment.publish_permissions(root)
    assert root.stat().st_mode & 0o777 == 0o755
    assert (root / 'assets').stat().st_mode & 0o777 == 0o755
    assert (root / 'assets/main.js').stat().st_mode & 0o777 == 0o644


def test_frontend_verification_rejects_html_instead_of_javascript(cfg, monkeypatch):
    web = Path(cfg['project_dir']) / 'deploy/local/.state/dashboard'
    web.mkdir(parents=True)
    html = b'<html><script type="module" src="/assets/main.js"></script></html>'
    (web / 'index.html').write_bytes(html)
    class Response:
        headers = {'Content-Type': 'text/html'}
        def __enter__(self): return self
        def __exit__(self, *_): return False
        def read(self): return html
    monkeypatch.setattr(deployment.urllib.request, 'urlopen', lambda *a, **k: Response())
    with pytest.raises(RuntimeError, match='静态资源响应异常'):
        deployment.Deployment(cfg).verify_frontend()