| '''Проверки создания и авторизации без запуска реальных терминалов.'''
|
|
|
| import copy
|
| import json
|
| import os
|
| import shutil
|
| import tempfile
|
| import threading
|
| import time
|
| import unittest
|
| from pathlib import Path
|
| from unittest.mock import patch
|
|
|
| from config import ConfigStore
|
| from mt5_control import MT5_Control
|
| from terminal_manager import SERVICE_REL, TerminalManager, account_values, validate_folder
|
|
|
|
|
| class SimulatedManager(TerminalManager):
|
| '''Заменяет только границу с процессом и сервисом терминала.'''
|
|
|
| def __init__(self, store):
|
| super().__init__(store)
|
| self.running = set()
|
| self.saved_accounts = {}
|
| self.current_accounts = {}
|
| self.launches = []
|
| self.fail_wait = False
|
|
|
| def processes(self, folder):
|
| return [object()] if folder in self.running else []
|
|
|
| def launch(self, folder, config=None):
|
| self.running.add(folder)
|
| self.launches.append((folder, config is not None))
|
| if config:
|
| values = dict(line.split('=', 1) for line in config.read_text(encoding='utf-16').splitlines() if '=' in line)
|
| self.current_accounts[folder] = (int(values['Login']), values['Server'], values['Password'])
|
| else:
|
| self.current_accounts[folder] = self.saved_accounts.get(folder)
|
| # Без файла /config пароль не должен оставаться в папке запуска.
|
| directory = Path(self.control.instance_path(folder)).parent
|
| assert not list(directory.glob('manager-login-*.ini'))
|
|
|
| def stop(self, folder, *, graceful=False):
|
| if graceful and folder in self.current_accounts:
|
| self.saved_accounts[folder] = self.current_accounts[folder]
|
| self.running.discard(folder)
|
|
|
| def wait_state(self, folder, since, account=None):
|
| if self.fail_wait:
|
| raise RuntimeError('Имитирована ошибка запуска.')
|
| current = self.current_accounts.get(folder)
|
| if account and current != account:
|
| raise RuntimeError('Счёт не совпал.')
|
| return {'connected': bool(current), 'login': current[0] if current else 0,
|
| 'server': current[1] if current else ''}
|
|
|
|
|
| class ProvisioningTests(unittest.TestCase):
|
| def setUp(self):
|
| self.temporary = tempfile.TemporaryDirectory()
|
| self.root = Path(self.temporary.name)
|
| self.template = self.root / 'template'
|
| (self.template / 'config').mkdir(parents=True)
|
| for filename in ('terminal64.exe', 'MetaEditor64.exe', 'ServiceStateWriter.ex5'):
|
| (self.template / filename).write_bytes(b'MZ-test-binary')
|
| (self.template / 'config/servers.dat').write_bytes(b'test-server-directory')
|
| self.path = self.root / 'config.json'
|
| self.path.write_text(json.dumps({'terminals': {}, 'mt5_folder': str(self.root / 'instances'),
|
| 'mt5_exe': 'terminal64.exe', 'database': str(self.root / 'custom.sqlite'),
|
| 'terminal_template': str(self.template)}), encoding='utf-8')
|
| self.manager = SimulatedManager(ConfigStore(self.path))
|
|
|
| def tearDown(self):
|
| self.manager.close()
|
| self.temporary.cleanup()
|
|
|
| def run_job(self, action, name, record=None):
|
| task = self.manager.submit(action, name, record)
|
| deadline = time.monotonic() + 5
|
| while time.monotonic() < deadline:
|
| task = self.manager.job(task['id'])
|
| if task['status'] in ('done', 'error'):
|
| return task
|
| time.sleep(0.01)
|
| self.fail('Фоновая операция не завершилась.')
|
|
|
| @staticmethod
|
| def prepare_project(project):
|
| (project / 'static').mkdir(parents=True, exist_ok=True)
|
| shutil.copyfile(Path(__file__).resolve().parent.parent / 'static/services.ini',
|
| project / 'static/services.ini')
|
|
|
| def test_create_with_account_then_plain_restart(self):
|
| task = self.run_job('create', 'MetaTrader5.1', {'login': 123, 'server': 'Broker-Demo', 'password': 'secret'})
|
| self.assertEqual(task['status'], 'done', task)
|
| self.assertEqual(self.manager.launches, [('MetaTrader5.1', True), ('MetaTrader5.1', False)])
|
| record = self.manager.store.data['terminals']['MetaTrader5.1']
|
| target = Path(self.manager.control.instance_path('MetaTrader5.1')).parent
|
| self.assertNotIn('path', record)
|
| self.assertNotIn('path', json.loads(self.path.read_text())['terminals']['MetaTrader5.1'])
|
| self.assertTrue((target / 'config/servers.dat').is_file())
|
| self.assertTrue((target / 'MQL5/Services/Shared Projects/mt5-manager/ServiceStateWriter.ex5').is_file())
|
| self.assertIn('InpDatabase=custom.sqlite', (target / 'config/services.ini').read_text(encoding='utf-16'))
|
| self.assertFalse(list(target.glob('manager-login-*.ini')))
|
| self.assertNotIn('secret', json.dumps(self.manager.settings()))
|
| self.assertNotIn('secret', json.dumps(self.manager.snapshot()))
|
| self.manager.stop('MetaTrader5.1')
|
| self.assertEqual(self.run_job('start', 'MetaTrader5.1')['status'], 'done')
|
| self.assertFalse(self.manager.launches[-1][1])
|
|
|
| def test_create_without_account(self):
|
| task = self.run_job('create', 'Blank')
|
| self.assertEqual(task['status'], 'done', task)
|
| self.assertEqual(self.manager.launches, [('Blank', False)])
|
| record = self.manager.store.data['terminals']['Blank']
|
| self.assertEqual((record['login'], record['server'], record['password']), ('', '', ''))
|
|
|
| def test_project_operation_blocks_terminal_commands_until_finished(self):
|
| entered, release = threading.Event(), threading.Event()
|
|
|
| def check():
|
| entered.set()
|
| release.wait(5)
|
| return {'behind': 0}
|
|
|
| with patch.object(self.manager.updates, 'check', side_effect=check):
|
| self.manager.submit('check_updates', '')
|
| self.assertTrue(entered.wait(2))
|
| try:
|
| with self.assertRaises(ValueError):
|
| self.manager.submit('create', 'Blocked')
|
| with self.assertRaises(ValueError):
|
| self.manager.submit('deploy_service', '')
|
| self.assertEqual(self.manager.snapshot(), {})
|
| finally:
|
| release.set()
|
|
|
| def test_project_operation_waits_for_terminal_jobs(self):
|
| entered, release = threading.Event(), threading.Event()
|
|
|
| def wait_state(*args, **kwargs):
|
| entered.set()
|
| release.wait(5)
|
| return {}
|
|
|
| with patch.object(self.manager, 'wait_state', side_effect=wait_state):
|
| self.manager.submit('create', 'Pending')
|
| self.assertTrue(entered.wait(2))
|
| try:
|
| with self.assertRaises(ValueError):
|
| self.manager.submit('check_updates', '')
|
| finally:
|
| release.set()
|
|
|
| def test_deploy_job_exposes_copy_failure_details_without_account_password(self):
|
| result = {'copied': [], 'failed': ['One'], 'restart_required': [],
|
| 'errors': {'One': 'Access is denied: C:/instances/One/service.ex5'}}
|
| with patch.object(self.manager.updates, 'deploy', return_value=result):
|
| task = self.run_job('deploy_service', '')
|
| self.assertEqual(task['status'], 'error')
|
| self.assertIn('Access is denied', task['error'])
|
| self.assertIn('C:/instances/One/service.ex5', task['error'])
|
| self.assertEqual(task['result']['errors'], result['errors'])
|
|
|
| def test_missing_uploads_use_reference_terminal_and_compile_service(self):
|
| reference = self.root / 'reference'
|
| (reference / 'config').mkdir(parents=True)
|
| for filename in ('terminal64.exe', 'MetaEditor64.exe'):
|
| (reference / filename).write_bytes(b'MZ-reference-' + filename.encode())
|
| (self.template / filename).unlink()
|
| (reference / 'config/servers.dat').write_bytes(b'reference-servers')
|
| (self.template / 'ServiceStateWriter.ex5').unlink()
|
|
|
| project = reference / 'MQL5/Shared Projects/mt5-manager'
|
| self.prepare_project(project)
|
|
|
| def compile_service(editor, source, output):
|
| output.parent.mkdir(parents=True, exist_ok=True)
|
| output.write_bytes(b'compiled-service')
|
| return True
|
|
|
| with patch.object(self.manager, 'reference_terminal', return_value=reference), \
|
| patch('terminal_manager.PROJECT_DIR', project), \
|
| patch('terminal_manager.compile_service', side_effect=compile_service) as compile_call:
|
| # Просмотр настроек только показывает доступные файлы.
|
| self.assertTrue(self.manager.settings()['files']['terminal']['exists'])
|
| compile_call.assert_not_called()
|
| task = self.run_job('create', 'Fallback')
|
| self.assertEqual(task['status'], 'done', task)
|
| compile_call.assert_called_once_with(reference / 'MetaEditor64.exe',
|
| project / 'ServiceStateWriter.mq5', reference / SERVICE_REL)
|
| target = self.root / 'instances/Fallback'
|
| self.assertEqual((target / 'terminal64.exe').read_bytes(), (reference / 'terminal64.exe').read_bytes())
|
| self.assertEqual((target / 'MetaEditor64.exe').read_bytes(), (reference / 'MetaEditor64.exe').read_bytes())
|
| self.assertEqual((target / 'config/servers.dat').read_bytes(), b'test-server-directory')
|
| self.assertEqual((target / SERVICE_REL).read_bytes(), b'compiled-service')
|
|
|
| def test_missing_servers_file_uses_reference_directory(self):
|
| reference = self.root / 'reference'
|
| (reference / 'config').mkdir(parents=True)
|
| (reference / 'config/servers.dat').write_bytes(b'reference-servers')
|
| (self.template / 'config/servers.dat').unlink()
|
| with patch.object(self.manager, 'reference_terminal', return_value=reference), \
|
| patch('terminal_manager.compile_service') as compile_call:
|
| task = self.run_job('create', 'ReferenceServers')
|
| self.assertEqual(task['status'], 'done', task)
|
| compile_call.assert_not_called()
|
| self.assertEqual((self.root / 'instances/ReferenceServers/config/servers.dat').read_bytes(),
|
| b'reference-servers')
|
|
|
| def test_compile_failure_does_not_create_instance_or_copy_old_service(self):
|
| reference = self.root / 'reference'
|
| output = reference / SERVICE_REL
|
| output.parent.mkdir(parents=True)
|
| output.write_bytes(b'old-service')
|
| (self.template / 'ServiceStateWriter.ex5').unlink()
|
| with patch.object(self.manager, 'reference_terminal', return_value=reference), \
|
| patch('terminal_manager.compile_service', return_value=False):
|
| task = self.run_job('create', 'CompileFailed')
|
| self.assertEqual(task['status'], 'error', task)
|
| self.assertIn('скомпилировать', task['error'])
|
| self.assertFalse((self.root / 'instances/CompileFailed').exists())
|
| self.assertNotIn('CompileFailed', json.loads(self.path.read_text())['terminals'])
|
|
|
| def test_explicit_service_path_takes_priority_without_compilation(self):
|
| explicit = self.root / 'explicit.ex5'
|
| explicit.write_bytes(b'explicit-service')
|
| self.manager.save_settings({'service_ex5': str(explicit)})
|
| with patch('terminal_manager.compile_service') as compile_call:
|
| task = self.run_job('create', 'ExplicitService')
|
| self.assertEqual(task['status'], 'done', task)
|
| compile_call.assert_not_called()
|
| self.assertEqual((self.root / 'instances/ExplicitService' / SERVICE_REL).read_bytes(), b'explicit-service')
|
|
|
| def test_standalone_project_with_uploaded_files_creates_instance(self):
|
| project = self.root / 'standalone'
|
| self.prepare_project(project)
|
| with patch('terminal_manager.PROJECT_DIR', project), \
|
| patch('terminal_manager.compile_service') as compile_call:
|
| self.assertIsNone(self.manager.reference_terminal())
|
| self.assertTrue(all(file['exists'] for file in self.manager.settings()['files'].values()))
|
| task = self.run_job('create', 'Standalone')
|
| self.assertEqual(task['status'], 'done', task)
|
| compile_call.assert_not_called()
|
|
|
| def test_standalone_project_compiles_current_source_with_uploaded_editor(self):
|
| project = self.root / 'standalone'
|
| self.prepare_project(project)
|
| (project / 'ServiceStateWriter.mq5').write_text('current source')
|
| (self.template / 'ServiceStateWriter.ex5').unlink()
|
|
|
| def compile_service(editor, source, output):
|
| self.assertEqual(editor, self.template / 'MetaEditor64.exe')
|
| self.assertEqual(source.read_text(), 'current source')
|
| self.assertEqual(output, project / 'ServiceStateWriter.ex5')
|
| output.write_bytes(b'current compiled service')
|
| return True
|
|
|
| with patch('terminal_manager.PROJECT_DIR', project), \
|
| patch('terminal_manager.compile_service', side_effect=compile_service):
|
| task = self.run_job('create', 'CompiledStandalone')
|
| self.assertEqual(task['status'], 'done', task)
|
| self.assertEqual((self.root / 'instances/CompiledStandalone' / SERVICE_REL).read_bytes(),
|
| b'current compiled service')
|
|
|
| def test_standalone_missing_files_shows_settings_and_reports_missing_asset(self):
|
| # Родительская папка с похожим EXE не является установкой MT5.
|
| (self.root / 'terminal64.exe').write_bytes(b'unrelated executable')
|
| for path in self.template.rglob('*'):
|
| if path.is_file():
|
| path.unlink()
|
| with patch('terminal_manager.PROJECT_DIR', self.root / 'standalone'):
|
| self.assertIsNone(self.manager.reference_terminal())
|
| self.assertFalse(any(file['exists'] for file in self.manager.settings()['files'].values()))
|
| task = self.run_job('create', 'MissingAssets')
|
| self.assertEqual(task['status'], 'error')
|
| self.assertIn('terminal64.exe', task['error'])
|
| self.assertFalse((self.root / 'instances/MissingAssets').exists())
|
|
|
| def test_explicit_installation_is_resolved_relative_to_config_and_used(self):
|
| installation = self.root / 'installed-terminal'
|
| installation.mkdir()
|
| (installation / 'terminal64.exe').write_bytes(b'configured terminal')
|
| (self.template / 'terminal64.exe').unlink()
|
| self.manager.save_settings({'mt5_installation': 'installed-terminal'})
|
| self.prepare_project(self.root / 'standalone')
|
| with patch('terminal_manager.PROJECT_DIR', self.root / 'standalone'):
|
| self.assertEqual(self.manager.reference_terminal(), installation)
|
| task = self.run_job('create', 'ConfiguredSource')
|
| self.assertEqual(task['status'], 'done', task)
|
| self.assertEqual((self.root / 'instances/ConfiguredSource/terminal64.exe').read_bytes(),
|
| b'configured terminal')
|
|
|
| def test_failed_creation_rolls_back_folder_and_config(self):
|
| self.manager.fail_wait = True
|
| task = self.run_job('create', 'Failed', {'login': 123, 'server': 'Broker', 'password': 'secret'})
|
| self.assertEqual(task['status'], 'error')
|
| self.assertNotIn('Failed', self.manager.store.data['terminals'])
|
| self.assertNotIn('Failed', json.loads(self.path.read_text())['terminals'])
|
| self.assertFalse((self.root / 'instances/Failed').exists())
|
| self.assertNotIn('Failed', self.manager.running)
|
|
|
| def test_occupied_folder_preserved(self):
|
| target = self.root / 'instances/Occupied'
|
| target.mkdir(parents=True)
|
| (target / 'keep.txt').write_text('existing data')
|
| self.assertEqual(self.run_job('create', 'occupied')['status'], 'error')
|
| self.assertEqual((target / 'keep.txt').read_text(), 'existing data')
|
|
|
| def test_password_change_reauthenticates_existing_terminal(self):
|
| original = {'login': 123, 'server': 'Broker', 'password': 'old-secret'}
|
| self.assertEqual(self.run_job('create', 'Account', original)['status'], 'done')
|
| target = Path(self.manager.control.instance_path('Account')).parent
|
| (target / 'keep.txt').write_text('history')
|
| changed = dict(original, password='new-secret')
|
| self.assertEqual(self.run_job('account', 'Account', changed)['status'], 'done')
|
| self.assertEqual(self.manager.saved_accounts['Account'][2], 'new-secret')
|
| self.assertEqual((target / 'keep.txt').read_text(), 'history')
|
|
|
| def test_existing_instance_kept_after_account_error(self):
|
| self.assertEqual(self.run_job('create', 'Existing')['status'], 'done')
|
| self.manager.fail_wait = True
|
| task = self.run_job('account', 'Existing', {'login': 123, 'server': 'Broker', 'password': 'secret'})
|
| self.assertEqual(task['status'], 'error')
|
| self.assertTrue((self.root / 'instances/Existing').exists())
|
| self.assertIn('Existing', self.manager.store.data['terminals'])
|
|
|
| def test_atomic_config_write_failure_preserves_original(self):
|
| original = self.path.read_bytes()
|
| changed = copy.deepcopy(self.manager.store.data)
|
| changed['disk_warning_mb'] = 100
|
| with patch('config.os.replace', side_effect=OSError('simulated write failure')):
|
| with self.assertRaises(OSError):
|
| self.manager.store.save(changed)
|
| self.assertEqual(self.path.read_bytes(), original)
|
| self.assertEqual(self.manager.store.data['disk_warning_mb'], 1024)
|
| self.assertFalse(list(self.root.glob('*.tmp')))
|
|
|
| def test_account_defaults_preserve_options_and_omit_password(self):
|
| directory = self.root / 'prepared'
|
| (directory / 'config').mkdir(parents=True)
|
| path = directory / 'config/common.ini'
|
| path.write_text('[Common]\nProxyEnable=1\nPassword=old-secret\n[Charts]\nMaxBars=20000\n', encoding='utf-16')
|
| self.manager.prepare_account_defaults(directory, (123, 'Broker-Demo', 'new-secret'))
|
| text = path.read_text(encoding='utf-16')
|
| for expected in ('Login=123', 'Server=Broker-Demo', 'KeepPrivate=1', 'ProxyEnable=1', '[Charts]\nMaxBars=20000'):
|
| self.assertIn(expected, text)
|
| self.assertNotIn('Password=', text)
|
| self.assertNotIn('secret', text)
|
|
|
| def test_upload_does_not_change_existing_server_directory(self):
|
| self.assertEqual(self.run_job('create', 'Existing')['status'], 'done')
|
| target = self.root / 'instances/Existing/config/servers.dat'
|
| original = target.read_bytes()
|
| incoming = self.root / 'new.dat'
|
| incoming.write_bytes(b'updated server directory')
|
| self.manager.install_upload('servers', incoming)
|
| self.assertEqual(target.read_bytes(), original)
|
| self.assertEqual((self.template / 'config/servers.dat').read_bytes(), incoming.read_bytes())
|
|
|
| def test_changed_root_updates_existing_paths_without_moving_directories(self):
|
| self.assertEqual(self.run_job('create', 'Existing')['status'], 'done')
|
| path = self.manager.control.instance_path('Existing')
|
| self.manager.verified_accounts['Existing'] = (123, 'Broker', 'secret')
|
| self.manager.save_settings({'mt5_folder': str(self.root / 'another-root')})
|
| expected = str(self.root / 'another-root/Existing/terminal64.exe')
|
| self.assertEqual(self.manager.control.instance_path('Existing'), expected)
|
| self.assertEqual(self.manager.settings()['config']['terminals']['Existing']['path'],
|
| str(self.root / 'another-root/Existing'))
|
| self.assertTrue(Path(path).is_file())
|
| self.assertFalse((self.root / 'another-root').exists())
|
| self.assertFalse(self.manager.verified_accounts)
|
| self.assertNotIn('path', json.loads(self.path.read_text())['terminals']['Existing'])
|
|
|
| def test_legacy_config_paths_are_ignored_and_removed_on_save(self):
|
| original = json.loads(self.path.read_text())
|
| original['terminals']['Existing'] = {'path': str(self.root / 'old-location'),
|
| 'login': 123, 'server': 'Broker', 'password': 'secret'}
|
| self.path.write_text(json.dumps(original), encoding='utf-8')
|
| store = ConfigStore(self.path)
|
| self.assertNotIn('path', store.data['terminals']['Existing'])
|
| control = MT5_Control(original['terminals'], store.data['mt5_folder'], store.data['database'])
|
| self.assertEqual(control.instance_path('Existing'),
|
| str(self.root / 'instances/Existing/terminal64.exe'))
|
| store.save(store.data)
|
| saved = json.loads(self.path.read_text())['terminals']['Existing']
|
| self.assertNotIn('path', saved)
|
| self.assertEqual(saved['password'], 'secret')
|
|
|
| def test_configuration_rejects_instance_names_outside_base_folder(self):
|
| for name in ('../outside', 'C:/elsewhere', 'nested/child', '..'):
|
| original = json.loads(self.path.read_text())
|
| original['terminals'] = {name: {}}
|
| self.path.write_text(json.dumps(original), encoding='utf-8')
|
| with self.assertRaises(ValueError):
|
| ConfigStore(self.path)
|
|
|
| def test_process_detection_follows_changed_base_folder(self):
|
| original = {'Existing': {'path': str(self.root / 'old-base/Existing')}}
|
| new_base = self.root / 'new-base'
|
| control = MT5_Control(original, str(new_base), str(self.root / 'state.sqlite'))
|
| old_process = type('Process', (), {'info': {'pid': 111, 'exe': str(self.root / 'old-base/Existing/terminal64.exe')}})()
|
| new_process = type('Process', (), {'info': {'pid': 222, 'exe': str(new_base / 'Existing/terminal64.exe')}})()
|
| with patch('mt5_control.psutil.process_iter', return_value=[old_process]):
|
| control.find_instances()
|
| self.assertEqual(control.instances['Existing']['pid'], 0)
|
| with patch('mt5_control.psutil.process_iter', return_value=[old_process, new_process]):
|
| control.find_instances()
|
| self.assertEqual(control.instances['Existing']['pid'], 222)
|
|
|
| def test_state_from_old_base_is_not_used_for_terminal_in_new_base(self):
|
| control = MT5_Control({'Existing': {}}, str(self.root / 'new-base'), str(self.root / 'state.sqlite'))
|
| control.instances['Existing']['pid'] = 222
|
| row = {'terminal_path': str(self.root / 'old-base/Existing'), 'updated_at': time.time()}
|
| with patch.object(control, 'read_state', return_value=row):
|
| info = control.instance_info('Existing')
|
| self.assertEqual(info['status'], 'starting')
|
| self.assertNotIn('account', info)
|
|
|
| def test_wait_requires_fresh_matching_connected_account(self):
|
| self.assertEqual(self.run_job('create', 'Existing')['status'], 'done')
|
| path = str(self.root / 'instances/Existing')
|
| row = {'terminal_path': path, 'updated_at': time.time() - 100,
|
| 'connected': 1, 'login': 123, 'server': 'Broker'}
|
| with patch.object(self.manager.control, 'read_state', return_value=row), \
|
| patch.object(self.manager.closing, 'wait'), \
|
| patch('terminal_manager.time.monotonic', side_effect=[0, 0, 9999]):
|
| with self.assertRaises(RuntimeError):
|
| TerminalManager.wait_state(self.manager, 'Existing', time.time(), (123, 'Broker', 'secret'))
|
| row.update(updated_at=time.time() + 1, connected=0)
|
| with patch.object(self.manager.control, 'read_state', return_value=row), \
|
| patch.object(self.manager.closing, 'wait'), \
|
| patch('terminal_manager.time.monotonic', side_effect=[0, 0, 9999]):
|
| with self.assertRaises(RuntimeError):
|
| TerminalManager.wait_state(self.manager, 'Existing', time.time(), (123, 'Broker', 'secret'))
|
|
|
| def test_reject_bad_names_and_partial_new_account(self):
|
| for folder in ('../other', 'CON', 'name.', 'a/b', 'nul.txt', ' trailing '):
|
| with self.assertRaises(ValueError):
|
| validate_folder(folder)
|
| with self.assertRaises(ValueError):
|
| account_values({'login': 123, 'server': 'Broker'}, complete=True)
|
| with self.assertRaises(ValueError):
|
| account_values({'login': 123, 'server': 'Broker', 'password': 'secret\nKeepPrivate=0'})
|
|
|
|
|
| if __name__ == '__main__':
|
| unittest.main()
|
