'''Проверки создания и авторизации без запуска реальных терминалов.''' 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()