'''Обновления проверяются на локальном remote, без сетевого репозитория пользователя.''' import os import shutil import subprocess import tempfile import unittest from pathlib import Path from unittest.mock import patch from deploy_service import (SERVICE_EX5_REL, compile_reference, reference_terminal, service_build_paths, services_ini_update, write_services_ini) from project_updates import ProjectUpdates class ProjectUpdateTests(unittest.TestCase): def setUp(self): self.temp = tempfile.TemporaryDirectory() self.root = Path(self.temp.name) self.remote = self.root / 'remote.git' self.writer = self.root / 'writer' self.reference = self.root / 'reference' self.project = self.reference / 'MQL5/Shared Projects/mt5-manager' self.project.parent.mkdir(parents=True) self.git(self.root, 'init', '--bare', str(self.remote)) self.git(self.root, 'clone', str(self.remote), str(self.writer)) self.git(self.writer, 'checkout', '-b', 'main') self.git(self.writer, 'config', 'user.name', 'Fixture') self.git(self.writer, 'config', 'user.email', 'fixture@example.invalid') (self.writer / 'ServiceStateWriter.mq5').write_text('first source', encoding='utf-8') self.copy_template(self.writer) self.git(self.writer, 'add', '.') self.git(self.writer, 'commit', '-m', 'Initial fixture') self.git(self.writer, 'push', '-u', 'origin', 'main') self.git(self.root, 'clone', '--branch', 'main', str(self.remote), str(self.project)) self.updates = ProjectUpdates(self.project) @staticmethod def copy_template(project): (project / 'static').mkdir(parents=True, exist_ok=True) shutil.copyfile(Path(__file__).resolve().parent.parent / 'static/services.ini', project / 'static/services.ini') def tearDown(self): # Git на Windows помечает объекты read-only. for path in self.root.rglob('*'): if path.is_file(): path.chmod(0o666) self.temp.cleanup() @staticmethod def git(cwd, *args): return subprocess.run(['git', *args], cwd=cwd, capture_output=True, text=True, check=True).stdout.strip() def publish(self): (self.writer / 'ServiceStateWriter.mq5').write_text('second source', encoding='utf-8') self.git(self.writer, 'add', '.') self.git(self.writer, 'commit', '-m', 'Update fixture service') self.git(self.writer, 'push') def test_fetch_preview_and_fast_forward(self): self.publish() before = self.git(self.project, 'rev-parse', 'HEAD') status = self.updates.check() self.assertEqual(status['behind'], 1) self.assertEqual(self.git(self.project, 'rev-parse', 'HEAD'), before) with patch.dict(os.environ, {'MT5_MANAGER_RELOAD': '0'}): result = self.updates.update() self.assertTrue(result['updated']) self.assertEqual((self.project / 'ServiceStateWriter.mq5').read_text(), 'second source') self.assertEqual(self.updates.repository()['behind'], 0) def test_dirty_tree_and_divergence_preserved(self): self.publish() local = self.project / 'local.txt' local.write_text('keep this') with patch.dict(os.environ, {'MT5_MANAGER_RELOAD': '0'}): with self.assertRaisesRegex(RuntimeError, 'локальные изменения'): self.updates.update() self.assertEqual(local.read_text(), 'keep this') self.git(self.project, 'config', 'user.name', 'Fixture') self.git(self.project, 'config', 'user.email', 'fixture@example.invalid') self.git(self.project, 'add', '.') self.git(self.project, 'commit', '-m', 'Local fixture change') before = self.git(self.project, 'rev-parse', 'HEAD') with patch.dict(os.environ, {'MT5_MANAGER_RELOAD': '0'}): with self.assertRaisesRegex(RuntimeError, 'свои коммиты'): self.updates.update() self.assertEqual(self.git(self.project, 'rev-parse', 'HEAD'), before) def test_reload_mode_rejects_update(self): with patch.dict(os.environ, {'MT5_MANAGER_RELOAD': '1'}), \ patch.object(self.updates, 'git') as git: with self.assertRaisesRegex(RuntimeError, '--reload'): self.updates.update() git.assert_not_called() def test_deploy_uses_base_folder_ignores_old_paths_and_reports_partial_failure(self): (self.reference / 'MetaEditor64.exe').write_bytes(b'fixture') source = self.reference / SERVICE_EX5_REL source.parent.mkdir(parents=True) source.write_bytes(b'old-ex5') base = self.root / 'instances' target = base / 'Custom' target.mkdir(parents=True) (target / 'terminal64.exe').write_bytes(b'fixture') old_target = self.root / 'old-base/Custom' (old_target / SERVICE_EX5_REL).parent.mkdir(parents=True) (old_target / SERVICE_EX5_REL).write_bytes(b'keep-old-location') data = {'mt5_folder': str(base), 'mt5_exe': 'terminal64.exe', 'database': 'custom.sqlite', 'terminals': { 'Custom': {'path': str(old_target)}, 'Missing': {'path': str(self.root / 'missing')}}} ini_path, ini_text, _ = services_ini_update(target, self.project / 'static/services.ini', data['database']) write_services_ini(ini_path, ini_text) self.assertTrue(self.updates.service_status(data)['pending']) def compile_service(editor, mq5, output, *, force=False): self.assertTrue(force) source.write_bytes(b'new-ex5') return True with patch('project_updates.compile_service', side_effect=compile_service): result = self.updates.deploy(data, lambda folder: folder == 'Custom') self.assertEqual((target / SERVICE_EX5_REL).read_bytes(), b'new-ex5') self.assertEqual(result['copied'], ['Custom']) self.assertEqual(result['failed'], ['Missing']) self.assertIn(str(base / 'Missing/terminal64.exe'), result['errors']['Missing']) self.assertEqual(result['restart_required'], ['Custom']) self.assertFalse((base / 'Missing').exists()) self.assertEqual((old_target / SERVICE_EX5_REL).read_bytes(), b'keep-old-location') def test_deploy_reports_windows_copy_error_and_preserves_previous_service(self): source = self.reference / SERVICE_EX5_REL source.parent.mkdir(parents=True) source.write_bytes(b'new-build') target = self.root / 'One' destination = target / SERVICE_EX5_REL destination.parent.mkdir(parents=True) destination.write_bytes(b'old-build') (target / 'terminal64.exe').write_bytes(b'terminal') data = {'mt5_folder': str(self.root), 'mt5_exe': 'terminal64.exe', 'database': 'custom.sqlite', 'terminals': {'One': {}}} error = PermissionError(13, 'Access is denied', str(destination)) with patch('project_updates.compile_service', return_value=True), \ patch('project_updates.os.replace', side_effect=error): result = self.updates.deploy(data, lambda _: False) self.assertEqual(result['failed'], ['One']) self.assertEqual(result['copied'], []) self.assertIn('Access is denied', result['errors']['One']) self.assertIn(str(destination), result['errors']['One']) self.assertIn('Замена EX5', result['errors']['One']) self.assertEqual(destination.read_bytes(), b'old-build') self.assertFalse(list(destination.parent.glob('*.tmp'))) def test_failed_compilation_never_distributes_previous_ex5(self): target = self.root / 'One' destination = target / SERVICE_EX5_REL destination.parent.mkdir(parents=True) destination.write_bytes(b'keep-old-service') data = {'mt5_folder': str(self.root), 'mt5_exe': 'terminal64.exe', 'database': 'custom.sqlite', 'terminals': {'One': {}}} with patch('project_updates.compile_service', return_value=False): with self.assertRaisesRegex(RuntimeError, 'не подтверждена'): self.updates.deploy(data, lambda _: False) self.assertEqual(destination.read_bytes(), b'keep-old-service') def test_force_build_does_not_accept_metaeditor_no_op(self): (self.reference / 'MetaEditor64.exe').write_bytes(b'fixture') output = self.reference / SERVICE_EX5_REL output.parent.mkdir(parents=True) output.write_bytes(b'old-build') with patch('deploy_service.run') as run: self.assertFalse(compile_reference(self.reference, force=True)) run.assert_called_once() def test_standalone_status_and_deploy_use_uploaded_editor_and_current_project(self): project = self.root / 'standalone' project.mkdir() self.copy_template(project) (project / 'ServiceStateWriter.mq5').write_text('standalone source') template = self.root / 'uploads' template.mkdir() (template / 'MetaEditor64.exe').write_bytes(b'editor') target = self.root / 'One' target.mkdir() (target / 'terminal64.exe').write_bytes(b'terminal') data = {'terminal_template': str(template), 'mt5_exe': 'terminal64.exe', 'mt5_folder': str(self.root), 'database': 'custom.sqlite', 'terminals': {'One': {}}} updates = ProjectUpdates(project) self.assertTrue(updates.service_status(data)['available']) self.assertTrue(updates.service_status(data)['needs_compile']) def compile_service(editor, source, output, *, force=False): self.assertEqual(editor, template / 'MetaEditor64.exe') self.assertEqual(source.read_text(), 'standalone source') self.assertTrue(force) output.write_bytes(b'standalone build') return True with patch('project_updates.compile_service', side_effect=compile_service): result = updates.deploy(data, lambda _: False) self.assertEqual(result['copied'], ['One']) self.assertEqual((target / SERVICE_EX5_REL).read_bytes(), b'standalone build') self.assertIn('InpDatabase=custom.sqlite', (target / 'config/services.ini').read_text(encoding='utf-16')) self.assertFalse(updates.service_status(data)['pending']) def test_running_terminal_requires_stop_before_autostart_repair(self): source = self.reference / SERVICE_EX5_REL source.parent.mkdir(parents=True) source.write_bytes(b'compiled') target = self.root / 'One' target.mkdir() (target / 'terminal64.exe').write_bytes(b'terminal') data = {'mt5_folder': str(self.root), 'mt5_exe': 'terminal64.exe', 'database': 'custom.sqlite', 'terminals': {'One': {}}} with patch('project_updates.compile_service', return_value=True): result = self.updates.deploy(data, lambda _: True) self.assertEqual(result['failed'], ['One']) self.assertIn('Остановите терминал', result['errors']['One']) self.assertFalse((target / 'config/services.ini').exists()) def test_missing_compiler_at_filesystem_root_does_not_index_ancestors(self): project = Path(self.root.anchor) self.assertIsNone(reference_terminal(project)) editor, source, output = service_build_paths(project, {}) self.assertEqual(source, project / 'ServiceStateWriter.mq5') self.assertEqual(output, source.with_suffix('.ex5')) self.assertEqual(editor, project / 'terminal-template/MetaEditor64.exe') if __name__ == '__main__': unittest.main()