| '''Обновления проверяются на локальном 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()
|
