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