mt5-manager/tests/test_provisioning.py

'''Проверки создания и авторизации без запуска реальных терминалов.'''

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()