bifurcado de antekov/mt5-manager
'''Проверки создания и авторизации без запуска реальных терминалов.'''
| |
| |
import copy
| |
import errno
| |
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.project = self.root / 'checkout'
| |
self.prepare_project(self.project)
| |
(self.project / 'ServiceStateWriter.mq5').write_text('fixture source')
| |
self.project_patch = patch('terminal_manager.PROJECT_DIR', self.project)
| |
self.compiler_patch = patch('terminal_manager.compile_service', side_effect=self.compile_service)
| |
self.project_patch.start()
| |
self.compiler_patch.start()
| |
self.manager = SimulatedManager(ConfigStore(self.path))
| |
| |
def tearDown(self):
| |
self.manager.close()
| |
self.compiler_patch.stop()
| |
self.project_patch.stop()
| |
self.temporary.cleanup()
| |
| |
def run_job(self, action, name, record=None):
| |
if action == 'create':
| |
record = dict(record or {})
| |
record.setdefault('_servers_dat', b'test-server-directory')
| |
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 compile_service(editor, source, output):
| |
output.parent.mkdir(parents=True, exist_ok=True)
| |
output.write_bytes(b'compiled-service')
| |
return True
| |
| |
@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', {'_servers_dat': b'pending servers'})
| |
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_source_installation_outside_base_is_used_for_both_executables(self):
| |
reference = self.root / 'separate-installation'
| |
reference.mkdir()
| |
for filename in ('terminal64.exe', 'MetaEditor64.exe'):
| |
(reference / filename).write_bytes(b'MZ-reference-' + filename.encode())
| |
self.manager.save_settings({'terminal_template': str(reference)})
| |
task = self.run_job('create', 'FromInstallation')
| |
self.assertEqual(task['status'], 'done', task)
| |
target = self.root / 'instances/FromInstallation'
| |
for filename in ('terminal64.exe', 'MetaEditor64.exe'):
| |
self.assertEqual((target / filename).read_bytes(), (reference / filename).read_bytes())
| |
self.assertEqual((target / SERVICE_REL).read_bytes(), b'compiled-service')
| |
| |
def test_shared_servers_file_is_ignored_and_each_creation_uses_its_own(self):
| |
(self.template / 'config/servers.dat').write_bytes(b'must-not-be-used')
| |
for name, servers in [('BrokerOne', b'first broker'), ('BrokerTwo', b'second broker')]:
| |
task = self.run_job('create', name, {'_servers_dat': servers})
| |
self.assertEqual(task['status'], 'done', task)
| |
self.assertEqual((self.root / 'instances' / name / 'config/servers.dat').read_bytes(), servers)
| |
self.assertEqual((self.template / 'config/servers.dat').read_bytes(), b'must-not-be-used')
| |
self.assertNotIn('_servers_dat', self.manager.store.data['terminals']['BrokerOne'])
| |
self.assertNotIn('first broker', json.dumps(self.manager.job(task['id'])))
| |
| |
def test_creation_accepts_omitted_servers_file(self):
| |
task = self.run_job('create', 'WithoutServers', {'_servers_dat': None})
| |
self.assertEqual(task['status'], 'done', task)
| |
target = self.root / 'instances/WithoutServers'
| |
self.assertTrue((target / 'terminal64.exe').is_file())
| |
self.assertTrue((target / 'config/services.ini').is_file())
| |
self.assertFalse((target / 'config/servers.dat').exists())
| |
self.assertIn('WithoutServers', self.manager.running)
| |
| |
def test_creation_rejects_invalid_selected_servers_file(self):
| |
for servers in (b'', 'not bytes', b'x' * (16 * 1024 * 1024 + 1)):
| |
with self.subTest(servers=servers), self.assertRaisesRegex(ValueError, 'servers.dat'):
| |
self.manager.submit('create', 'MissingServers', {'_servers_dat': servers})
| |
self.assertFalse((self.root / 'instances/MissingServers').exists())
| |
| |
def test_compile_failure_does_not_create_instance_or_copy_old_service(self):
| |
with 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_old_ex5_in_source_installation_is_not_used(self):
| |
(self.template / 'ServiceStateWriter.ex5').write_bytes(b'obsolete uploaded service')
| |
task = self.run_job('create', 'CurrentService')
| |
self.assertEqual(task['status'], 'done', task)
| |
self.assertEqual((self.root / 'instances/CurrentService' / SERVICE_REL).read_bytes(), b'compiled-service')
| |
| |
def test_standalone_project_compiles_current_source_with_selected_editor(self):
| |
project = self.root / 'standalone'
| |
self.prepare_project(project)
| |
(project / 'ServiceStateWriter.mq5').write_text('current source')
| |
| |
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_missing_source_files_do_not_fall_back_to_parent_installation(self):
| |
(self.root / 'terminal64.exe').write_bytes(b'unrelated executable')
| |
for filename in ('terminal64.exe', 'MetaEditor64.exe'):
| |
(self.template / filename).unlink()
| |
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_legacy_installation_setting_becomes_single_source(self):
| |
legacy = json.loads(self.path.read_text())
| |
legacy['mt5_installation'] = 'installed-terminal'
| |
legacy['service_ex5'] = 'old.ex5'
| |
self.path.write_text(json.dumps(legacy))
| |
store = ConfigStore(self.path)
| |
self.assertEqual(store.data['terminal_template'], str(self.root / 'installed-terminal'))
| |
self.assertNotIn('mt5_installation', store.data)
| |
self.assertNotIn('service_ex5', store.data)
| |
store.save(store.data)
| |
self.assertNotIn('mt5_installation', json.loads(self.path.read_text()))
| |
| |
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_is_saved_and_used_on_explicit_start(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], 'old-secret')
| |
self.assertEqual(self.manager.store.data['terminals']['Account']['password'], 'new-secret')
| |
self.assertEqual(self.run_job('start', 'Account')['status'], 'done')
| |
self.assertEqual(self.manager.saved_accounts['Account'][2], 'new-secret')
| |
self.assertEqual((target / 'keep.txt').read_text(), 'history')
| |
| |
def test_rename_stopped_instance_and_replace_its_servers(self):
| |
self.run_job('create', 'Original')
| |
self.manager.stop('Original')
| |
original = self.root / 'instances/Original'
| |
(original / 'history.txt').write_text('preserved')
| |
untouched = self.root / 'instances/Other/config'
| |
untouched.mkdir(parents=True)
| |
(untouched / 'servers.dat').write_bytes(b'other broker')
| |
self.manager.save_settings({'terminal_template': str(original)})
| |
task = self.run_job('account', 'Original', {'folder': 'Renamed', 'name': 'My terminal',
| |
'_servers_dat': b'new broker'})
| |
self.assertEqual(task['status'], 'done', task)
| |
self.assertEqual(task['result']['folder'], 'Renamed')
| |
target = self.root / 'instances/Renamed'
| |
self.assertFalse(original.exists())
| |
self.assertEqual((target / 'history.txt').read_text(), 'preserved')
| |
self.assertEqual((target / 'config/servers.dat').read_bytes(), b'new broker')
| |
self.assertEqual((untouched / 'servers.dat').read_bytes(), b'other broker')
| |
self.assertEqual(self.manager.store.data['terminal_template'], str(target))
| |
self.assertNotIn('Original', self.manager.store.data['terminals'])
| |
self.assertEqual(self.manager.store.data['terminals']['Renamed']['name'], 'My terminal')
| |
self.assertNotIn('_servers_dat', self.path.read_text())
| |
self.assertNotIn('Renamed', self.manager.running)
| |
| |
def test_default_display_name_follows_folder_and_custom_name_can_be_reset(self):
| |
self.run_job('create', 'Original')
| |
self.manager.stop('Original')
| |
task = self.run_job('account', 'Original', {'folder': 'Renamed'})
| |
self.assertEqual(task['status'], 'done', task)
| |
self.assertEqual(self.manager.store.data['terminals']['Renamed']['name'], 'Renamed')
| |
self.assertEqual(self.run_job('account', 'Renamed', {'name': 'My terminal'})['status'], 'done')
| |
self.manager.stop('Renamed')
| |
task = self.run_job('account', 'Renamed', {'folder': 'Final'})
| |
self.assertEqual(task['status'], 'done', task)
| |
self.assertEqual(self.manager.store.data['terminals']['Final']['name'], 'My terminal')
| |
self.assertEqual(self.run_job('account', 'Final', {'name': ''})['status'], 'done')
| |
self.assertEqual(self.manager.store.data['terminals']['Final']['name'], 'Final')
| |
| |
def test_running_instance_cannot_be_renamed(self):
| |
self.run_job('create', 'Running')
| |
with self.assertRaisesRegex(ValueError, 'Терминал запущен'):
| |
self.manager.submit('account', 'Running', {'folder': 'Renamed'})
| |
self.assertTrue(self.manager.settings()['config']['terminals']['Running']['running'])
| |
self.assertTrue((self.root / 'instances/Running').exists())
| |
| |
def test_rename_rechecks_process_before_changing_disk(self):
| |
self.run_job('create', 'Original')
| |
self.manager.stop('Original')
| |
with patch.object(self.manager, 'processes', side_effect=[[], [object()]]):
| |
task = self.run_job('account', 'Original', {'folder': 'Renamed'})
| |
self.assertEqual(task['status'], 'error')
| |
self.assertTrue((self.root / 'instances/Original').exists())
| |
self.assertNotIn('Renamed', self.manager.store.data['terminals'])
| |
| |
def test_rename_rejects_existing_destination_and_invalid_name(self):
| |
self.run_job('create', 'Original')
| |
self.manager.stop('Original')
| |
(self.root / 'instances/Occupied').mkdir()
| |
for name in ('../Outside', 'Occupied', ''):
| |
with self.subTest(name=name), self.assertRaises(ValueError):
| |
self.manager.submit('account', 'Original', {'folder': name})
| |
self.assertTrue((self.root / 'instances/Original').exists())
| |
| |
def test_failed_config_save_rolls_back_rename_and_server_replacement(self):
| |
self.run_job('create', 'Original')
| |
self.manager.stop('Original')
| |
original_config = self.path.read_bytes()
| |
with patch.object(self.manager.store, 'save', side_effect=OSError('save failed')):
| |
task = self.run_job('account', 'Original', {'folder': 'Renamed', '_servers_dat': b'new broker'})
| |
self.assertEqual(task['status'], 'error')
| |
self.assertEqual(self.path.read_bytes(), original_config)
| |
self.assertEqual((self.root / 'instances/Original/config/servers.dat').read_bytes(), b'test-server-directory')
| |
self.assertFalse((self.root / 'instances/Renamed').exists())
| |
| |
def test_servers_replacement_requires_stopped_instance_and_does_not_start_it(self):
| |
account = {'login': 123, 'server': 'Broker', 'password': 'secret'}
| |
self.run_job('create', 'Account', account)
| |
changes = {'login': 123, 'server': 'Broker', '_servers_dat': b'replacement broker'}
| |
with patch.object(self.manager, 'stop') as stop:
| |
with self.assertRaisesRegex(ValueError, 'Закройте терминал'):
| |
self.manager.submit('account', 'Account', changes)
| |
stop.assert_not_called()
| |
self.manager.stop('Account')
| |
task = self.run_job('account', 'Account', changes)
| |
self.assertEqual(task['status'], 'done', task)
| |
self.assertNotIn('Account', self.manager.running)
| |
record = self.manager.store.data['terminals']['Account']
| |
self.assertEqual(record['password'], 'secret')
| |
self.assertEqual((self.root / 'instances/Account/config/servers.dat').read_bytes(), b'replacement broker')
| |
| |
def test_saving_account_does_not_validate_connection_until_start(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'], 'done', task)
| |
task = self.run_job('start', 'Existing')
| |
self.assertEqual(task['status'], 'error')
| |
self.assertTrue((self.root / 'instances/Existing').exists())
| |
self.assertIn('Existing', self.manager.store.data['terminals'])
| |
| |
def test_name_and_account_edits_never_touch_processes(self):
| |
self.run_job('create', 'Existing')
| |
for running in (True, False):
| |
if not running:
| |
self.manager.stop('Existing')
| |
with self.subTest(running=running), \
| |
patch.object(self.manager, 'launch') as launch, \
| |
patch.object(self.manager, 'stop') as stop, \
| |
patch.object(self.manager, 'wait_state') as wait:
| |
task = self.run_job('account', 'Existing', {'name': 'Display name', 'login': 999, 'server': 'NotConnected'})
| |
self.assertEqual(task['status'], 'done', task)
| |
launch.assert_not_called()
| |
stop.assert_not_called()
| |
wait.assert_not_called()
| |
self.assertEqual('Existing' in self.manager.running, running)
| |
| |
def test_explicit_connection_test_uses_form_values_without_saving_them(self):
| |
self.run_job('create', 'Existing')
| |
before = self.path.read_bytes()
| |
task = self.run_job('test_connection', 'Existing', {'login': 321, 'server': 'Test-Broker', 'password': 'temporary-secret'})
| |
self.assertEqual(task['status'], 'done', task)
| |
self.assertTrue(task['result']['connected'])
| |
self.assertEqual(self.manager.current_accounts['Existing'][0], 321)
| |
self.assertEqual(self.path.read_bytes(), before)
| |
self.assertEqual(self.manager.store.data['terminals']['Existing']['login'], '')
| |
self.assertNotIn('temporary-secret', json.dumps(self.manager.job(task['id'])))
| |
| |
def test_failed_connection_test_does_not_save_candidate_credentials(self):
| |
self.run_job('create', 'Existing')
| |
before = self.path.read_bytes()
| |
self.manager.fail_wait = True
| |
task = self.run_job('test_connection', 'Existing', {'login': 321, 'server': 'Test-Broker', 'password': 'temporary-secret'})
| |
self.assertEqual(task['status'], 'error', task)
| |
self.assertEqual(self.path.read_bytes(), before)
| |
self.assertIn('Existing', self.manager.running)
| |
| |
def test_connection_test_restores_running_state_on_success_and_failure(self):
| |
self.run_job('create', 'Existing')
| |
for running in (False, True):
| |
for failed in (False, True):
| |
with self.subTest(running=running, failed=failed):
| |
if running:
| |
self.manager.launch('Existing')
| |
else:
| |
self.manager.stop('Existing')
| |
self.manager.verified_accounts.clear()
| |
self.manager.fail_wait = failed
| |
before = self.path.read_bytes()
| |
task = self.run_job('test_connection', 'Existing',
| |
{'login': 321, 'server': 'Test-Broker', 'password': 'temporary-secret'})
| |
self.assertEqual(task['status'], 'error' if failed else 'done', task)
| |
self.assertEqual('Existing' in self.manager.running, running)
| |
self.assertEqual(self.path.read_bytes(), before)
| |
target = Path(self.manager.control.instance_path('Existing')).parent
| |
self.assertFalse(list(target.glob('manager-login-*.ini')))
| |
| |
def test_test_connection_falls_back_to_stop_if_graceful_close_fails(self):
| |
self.run_job('create', 'Existing')
| |
self.manager.stop('Existing')
| |
stop = self.manager.stop
| |
| |
def close(folder, *, graceful=False):
| |
if graceful:
| |
raise RuntimeError('Window did not close')
| |
stop(folder)
| |
| |
with patch.object(self.manager, 'stop', side_effect=close):
| |
task = self.run_job('test_connection', 'Existing', {'login': 123, 'server': 'Broker'})
| |
self.assertEqual(task['status'], 'error', task)
| |
self.assertNotIn('Existing', self.manager.running)
| |
| |
def test_delete_stopped_instance_removes_only_its_folder_and_record(self):
| |
self.run_job('create', 'DeleteMe')
| |
self.run_job('create', 'KeepMe')
| |
self.manager.stop('DeleteMe')
| |
self.manager.verified_accounts['DeleteMe'] = (123, 'Broker', 'secret')
| |
task = self.run_job('delete', 'DeleteMe')
| |
self.assertEqual(task['status'], 'done', task)
| |
self.assertTrue(task['result']['deleted'])
| |
self.assertFalse((self.root / 'instances/DeleteMe').exists())
| |
self.assertNotIn('DeleteMe', self.manager.store.data['terminals'])
| |
self.assertNotIn('DeleteMe', self.manager.verified_accounts)
| |
self.assertTrue((self.root / 'instances/KeepMe/terminal64.exe').is_file())
| |
self.assertFalse(list((self.root / 'instances').glob('.mt5-manager-delete-*')))
| |
| |
def test_delete_running_instance_is_rejected_and_rechecked(self):
| |
self.run_job('create', 'DeleteMe')
| |
before = self.path.read_bytes()
| |
with self.assertRaisesRegex(ValueError, 'Терминал запущен'):
| |
self.manager.submit('delete', 'DeleteMe')
| |
self.manager.stop('DeleteMe')
| |
with patch.object(self.manager, 'processes', side_effect=[[], [object()]]):
| |
task = self.run_job('delete', 'DeleteMe')
| |
self.assertEqual(task['status'], 'error', task)
| |
self.assertTrue((self.root / 'instances/DeleteMe').exists())
| |
self.assertEqual(self.path.read_bytes(), before)
| |
| |
def test_delete_rolls_back_folder_when_config_cannot_be_saved(self):
| |
self.run_job('create', 'DeleteMe')
| |
self.manager.stop('DeleteMe')
| |
before = self.path.read_bytes()
| |
with patch.object(self.manager.store, 'save', side_effect=OSError('save failed')):
| |
task = self.run_job('delete', 'DeleteMe')
| |
self.assertEqual(task['status'], 'error', task)
| |
self.assertTrue((self.root / 'instances/DeleteMe/terminal64.exe').is_file())
| |
self.assertEqual(self.path.read_bytes(), before)
| |
self.assertFalse(list((self.root / 'instances').glob('.mt5-manager-delete-*')))
| |
| |
def test_delete_reports_remaining_files_after_cleanup_failure(self):
| |
self.run_job('create', 'DeleteMe')
| |
self.manager.stop('DeleteMe')
| |
with patch('terminal_manager.shutil.rmtree', side_effect=PermissionError('locked file')):
| |
task = self.run_job('delete', 'DeleteMe')
| |
self.assertEqual(task['status'], 'done', task)
| |
self.assertNotIn('DeleteMe', self.manager.store.data['terminals'])
| |
self.assertTrue(Path(task['result']['remaining_path']).is_dir())
| |
self.assertTrue(task['warnings'])
| |
| |
def test_delete_missing_folder_removes_only_config_record(self):
| |
self.run_job('create', 'DeleteMe')
| |
self.manager.stop('DeleteMe')
| |
target = (self.root / 'instances/DeleteMe').resolve()
| |
self.assertEqual(target.parent, (self.root / 'instances').resolve())
| |
shutil.rmtree(target)
| |
task = self.run_job('delete', 'DeleteMe')
| |
self.assertEqual(task['status'], 'done', task)
| |
self.assertNotIn('DeleteMe', self.manager.store.data['terminals'])
| |
self.assertTrue(task['warnings'])
| |
| |
def test_delete_protects_manager_configuration_database_and_source_installation(self):
| |
self.run_job('create', 'DeleteMe')
| |
self.manager.stop('DeleteMe')
| |
target = self.root / 'instances/DeleteMe'
| |
cases = (patch('terminal_manager.PROJECT_DIR', target / 'MQL5/project'),
| |
patch.object(self.manager.store, 'path', target / 'config.json'),
| |
patch.dict(self.manager.store.data, {'database': str(target / 'state.sqlite')}),
| |
patch.dict(self.manager.store.data, {'terminal_template': str(target)}))
| |
for guard in cases:
| |
with self.subTest(guard=guard), guard, self.assertRaisesRegex(ValueError, 'Нельзя удалить'):
| |
self.manager.submit('delete', 'DeleteMe')
| |
self.assertTrue((target / 'terminal64.exe').is_file())
| |
| |
def test_delete_rejects_link_and_path_outside_base(self):
| |
self.run_job('create', 'DeleteMe')
| |
self.manager.stop('DeleteMe')
| |
with patch.object(Path, 'is_symlink', return_value=True):
| |
with self.assertRaisesRegex(ValueError, 'ссылки'):
| |
self.manager.submit('delete', 'DeleteMe')
| |
with self.assertRaises(ValueError):
| |
self.manager.deletion_directory('../outside')
| |
self.assertTrue((self.root / 'instances/DeleteMe').is_dir())
| |
| |
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_executable_upload_only_fills_missing_files(self):
| |
incoming = self.root / 'incoming.exe'
| |
incoming.write_bytes(b'MZ-new-executable')
| |
with self.assertRaisesRegex(ValueError, 'уже существует'):
| |
self.manager.install_upload('terminal', incoming)
| |
(self.template / 'MetaEditor64.exe').unlink()
| |
self.manager.install_upload('editor', incoming)
| |
self.assertEqual((self.template / 'MetaEditor64.exe').read_bytes(), incoming.read_bytes())
| |
with self.assertRaises(ValueError):
| |
self.manager.install_upload('servers', incoming)
| |
| |
def move_root(self, target, *, stop_running=False):
| |
values = {'mt5_folder': str(target)}
| |
plan = self.manager.settings_move_plan(values)
| |
task = self.manager.save_settings(values, move_token=plan['token'], stop_running=stop_running)
| |
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('Перенос не завершился.')
| |
| |
def test_changed_root_moves_existing_directories_after_confirmation(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')
| |
target = self.root / 'another-root'
| |
target.mkdir()
| |
(target / 'other.txt').write_text('keep')
| |
(self.root / 'instances/other.txt').write_text('also keep')
| |
task = self.move_root(target, stop_running=True)
| |
self.assertEqual(task['status'], 'done', task)
| |
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.assertFalse(Path(path).exists())
| |
self.assertTrue(Path(expected).is_file())
| |
self.assertEqual((target / 'other.txt').read_text(), 'keep')
| |
self.assertEqual((self.root / 'instances/other.txt').read_text(), 'also keep')
| |
self.assertNotIn('Existing', self.manager.running)
| |
self.assertFalse(self.manager.verified_accounts)
| |
self.assertNotIn('path', json.loads(self.path.read_text())['terminals']['Existing'])
| |
| |
def test_root_move_requires_confirmation_and_permission_to_stop(self):
| |
self.run_job('create', 'Existing')
| |
values = {'mt5_folder': str(self.root / 'new')}
| |
original = self.path.read_bytes()
| |
plan = self.manager.settings_move_plan(values)
| |
self.assertEqual(plan['running'], ['Existing'])
| |
with self.assertRaisesRegex(ValueError, 'Подтвердите перенос'):
| |
self.manager.save_settings(values)
| |
with self.assertRaisesRegex(ValueError, 'остановить все терминалы'):
| |
self.manager.save_settings(values, move_token=plan['token'])
| |
self.assertEqual(self.path.read_bytes(), original)
| |
self.assertIn('Existing', self.manager.running)
| |
| |
def test_root_move_conflict_is_detected_before_stopping(self):
| |
self.run_job('create', 'Existing')
| |
target = self.root / 'new/Existing'
| |
target.mkdir(parents=True)
| |
(target / 'other.txt').write_text('keep')
| |
with self.assertRaisesRegex(ValueError, 'уже занята'):
| |
self.move_root(target.parent, stop_running=True)
| |
self.assertIn('Existing', self.manager.running)
| |
self.assertEqual((target / 'other.txt').read_text(), 'keep')
| |
| |
def test_root_move_rechecks_conflicts_after_confirmation(self):
| |
self.run_job('create', 'Existing')
| |
values = {'mt5_folder': str(self.root / 'new')}
| |
plan = self.manager.settings_move_plan(values)
| |
(self.root / 'new/Existing').mkdir(parents=True)
| |
task = self.run_job('move_settings', '', {'values': values, 'token': plan['token'], 'stop_running': True})
| |
self.assertEqual(task['status'], 'error', task)
| |
self.assertIn('Existing', self.manager.running)
| |
| |
def test_root_move_does_not_stop_a_terminal_started_after_confirmation(self):
| |
self.run_job('create', 'Existing')
| |
self.manager.stop('Existing')
| |
values = {'mt5_folder': str(self.root / 'new')}
| |
plan = self.manager.settings_move_plan(values)
| |
self.manager.launch('Existing')
| |
task = self.run_job('move_settings', '', {'values': values, 'token': plan['token'], 'stop_running': False})
| |
self.assertEqual(task['status'], 'error', task)
| |
self.assertIn('Existing', self.manager.running)
| |
| |
def test_root_move_rolls_back_all_directories_on_config_failure(self):
| |
for folder in ('One', 'Two'):
| |
self.run_job('create', folder)
| |
original = self.path.read_bytes()
| |
with patch.object(self.manager.store, 'save', side_effect=OSError('write failed')):
| |
task = self.move_root(self.root / 'new', stop_running=True)
| |
self.assertEqual(task['status'], 'error', task)
| |
self.assertEqual(self.path.read_bytes(), original)
| |
for folder in ('One', 'Two'):
| |
self.assertTrue((self.root / 'instances' / folder / 'terminal64.exe').exists())
| |
self.assertFalse((self.root / 'new' / folder).exists())
| |
| |
def test_root_move_aborts_if_graceful_stop_fails(self):
| |
self.run_job('create', 'Existing')
| |
original = self.path.read_bytes()
| |
with patch.object(self.manager, 'stop', side_effect=RuntimeError('still running')):
| |
task = self.move_root(self.root / 'new', stop_running=True)
| |
self.assertEqual(task['status'], 'error', task)
| |
self.assertEqual(self.path.read_bytes(), original)
| |
self.assertFalse((self.root / 'new').exists())
| |
| |
def test_root_move_protects_manager_and_rejects_nested_target(self):
| |
self.run_job('create', 'Existing')
| |
old = self.root / 'instances/Existing'
| |
for protected in ('project', 'config', 'database'):
| |
with self.subTest(protected=protected):
| |
if protected == 'project':
| |
guard = patch('terminal_manager.PROJECT_DIR', old / 'MQL5/project')
| |
elif protected == 'config':
| |
guard = patch.object(self.manager.store, 'path', old / 'config.json')
| |
else:
| |
guard = patch.dict(self.manager.store.data, {'database': str(old / 'state.sqlite')})
| |
with guard, self.assertRaisesRegex(ValueError, 'вне базовой папки|вне папок экземпляров'):
| |
self.manager.settings_move_plan({'mt5_folder': str(self.root / 'new')})
| |
with self.assertRaisesRegex(ValueError, 'внутри перемещаемого'):
| |
self.manager.settings_move_plan({'mt5_folder': str(old / 'new')})
| |
| |
def test_base_folder_is_locked_when_manager_is_inside_it(self):
| |
base = self.root / 'instances'
| |
with patch('terminal_manager.PROJECT_DIR', base / 'separate-manager'):
| |
self.assertIn('вне базовой папки', self.manager.settings()['base_folder_lock'])
| |
with self.assertRaisesRegex(ValueError, 'изменение базовой папки невозможно'):
| |
self.manager.save_settings({'mt5_folder': str(self.root / 'new')})
| |
self.manager.save_settings({'disk_warning_mb': 42})
| |
self.assertEqual(self.manager.store.data['disk_warning_mb'], 42)
| |
self.assertEqual(self.manager.settings()['base_folder_lock'], '')
| |
| |
def test_unchanged_base_preserves_verified_account(self):
| |
self.manager.verified_accounts['Existing'] = (123, 'Broker', 'secret')
| |
self.manager.save_settings({'disk_warning_mb': 42})
| |
self.assertIn('Existing', self.manager.verified_accounts)
| |
| |
def test_root_move_retargets_source_installation_inside_moved_terminal(self):
| |
self.run_job('create', 'Existing')
| |
self.manager.save_settings({'terminal_template': str(self.root / 'instances/Existing')})
| |
task = self.move_root(self.root / 'new', stop_running=True)
| |
self.assertEqual(task['status'], 'done', task)
| |
self.assertEqual(self.manager.store.data['terminal_template'], str(self.root / 'new/Existing'))
| |
| |
def test_root_move_is_exclusive_and_does_not_block_job_reads(self):
| |
self.run_job('create', 'Existing')
| |
entered, release = threading.Event(), threading.Event()
| |
rename = Path.rename
| |
| |
def delayed(path, target):
| |
if path.name == 'Existing':
| |
entered.set()
| |
release.wait(3)
| |
return rename(path, target)
| |
| |
values = {'mt5_folder': str(self.root / 'new')}
| |
plan = self.manager.settings_move_plan(values)
| |
with patch.object(Path, 'rename', delayed):
| |
task = self.manager.save_settings(values, move_token=plan['token'], stop_running=True)
| |
try:
| |
self.assertTrue(entered.wait(2))
| |
self.assertEqual(self.manager.job(task['id'])['status'], 'running')
| |
with self.assertRaises(ValueError):
| |
self.manager.submit('start', 'Existing')
| |
with self.assertRaises(ValueError):
| |
self.manager.save_settings({'disk_warning_mb': 20})
| |
finally:
| |
release.set()
| |
self.manager.pool.submit(lambda: None).result(timeout=3)
| |
self.assertEqual(self.manager.job(task['id'])['status'], 'done')
| |
| |
def test_cross_volume_move_keeps_original_until_config_commit(self):
| |
for fail_save in (False, True):
| |
with self.subTest(fail_save=fail_save):
| |
folder = 'Rollback' if fail_save else 'Success'
| |
self.run_job('create', folder)
| |
source = Path(self.manager.store.data['mt5_folder']) / folder
| |
target = self.root / ('new-' + folder)
| |
rename = Path.rename
| |
| |
def cross_volume(path, destination):
| |
if path == source and Path(destination) == target / folder:
| |
raise OSError(errno.EXDEV, 'different volume')
| |
return rename(path, destination)
| |
| |
original = self.path.read_bytes()
| |
save = self.manager.store.save
| |
| |
def save_config(data):
| |
self.assertTrue(list(source.parent.glob('.mt5-manager-move-*')))
| |
if fail_save:
| |
raise OSError('write failed')
| |
save(data)
| |
| |
with patch.object(Path, 'rename', cross_volume), patch.object(self.manager.store, 'save', save_config):
| |
task = self.move_root(target, stop_running=True)
| |
self.assertEqual(task['status'], 'error' if fail_save else 'done', task)
| |
if fail_save:
| |
self.assertEqual(self.path.read_bytes(), original)
| |
self.assertTrue(source.exists())
| |
self.assertFalse((target / folder).exists())
| |
else:
| |
self.assertFalse(source.exists())
| |
self.assertTrue((target / folder / 'terminal64.exe').exists())
| |
self.assertFalse(list(source.parent.glob('.mt5-manager-move-*')))
| |
self.assertFalse(list(target.glob('.mt5-manager-move-*')))
| |
| |
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()
|