mt5-manager/tools/smoke_create_terminal.py

'''Проверка создания настоящего MT5 без счёта в изолированной папке.

Берёт файлы из первого терминала. После проверки удаляет только собственные
временные папки и отдельную тестовую базу Common/Files.
'''

import argparse
import getpass
import json
import shutil
import sys
import time
import uuid
from pathlib import Path

import psutil

PROJECT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(PROJECT))
from config import ConfigStore
from terminal_manager import TerminalManager


class DiagnosticManager(TerminalManager):
    def wait_state(self, folder, since, account=None):
        try:
            return super().wait_state(folder, since, account)
        except RuntimeError:
            row = self.control.read_state(folder)
            if row is not None:
                print('Last service state:', json.dumps({key: row[key] for key in
                      ('updated_at', 'connected', 'login', 'server', 'terminal_path')}, ensure_ascii=False), flush=True)
            log_dir = Path(self.control.instance_path(folder)).parent / 'Logs'
            for path in sorted(log_dir.glob('*.log')):
                if not path.stem.isdecimal():
                    continue
                for line in path.read_text(encoding='utf-16', errors='replace').splitlines():
                    if ('\tNetwork\t' in line or '\tStartup\t' in line) and 'previous successful' not in line:
                        print(line, flush=True)
            raise


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument('--with-account', action='store_true',
                        help='Запросить реквизиты счёта при запуске (пароль вводится скрыто)')
    parser.add_argument('--login', type=int, help='Логин для --with-account')
    parser.add_argument('--server', help='Торговый сервер для --with-account')
    parser.add_argument('--timeout', type=int, default=180)
    args = parser.parse_args()
    account = {}
    if args.with_account:
        try:
            login = args.login or int(input('Логин счёта: ').strip())
            server = args.server or input('Торговый сервер: ').strip()
            password = getpass.getpass('Пароль счёта (не отображается): ')
        except (ValueError, EOFError):
            parser.error('Не удалось прочитать реквизиты счёта.')
        if login <= 0 or not server or not password:
            parser.error('Укажите положительный логин, сервер и непустой пароль.')
        account = {'login': login, 'server': server, 'password': password}
    elif args.login is not None or args.server:
        parser.error('--login и --server используются вместе с --with-account.')
    token = uuid.uuid4().hex[:8]
    check_root = (PROJECT / '.checks' / token).resolve()
    check_root.mkdir(parents=True)
    common = Path(json.loads((PROJECT / 'config.json').read_text(encoding='utf-8-sig'))['database']).parent
    database = common / f'manager-article5-check-{token}.sqlite'
    path = check_root / 'config.json'
    path.write_text(json.dumps({'terminals': {}, 'mt5_folder': str(check_root / 'instances'),
                               'mt5_exe': 'terminal64.exe', 'database': str(database),
                               'terminal_template': r'C:\MT5\MetaTrader5.1',
                               'startup_timeout_sec': args.timeout}), encoding='utf-8')
    manager = DiagnosticManager(ConfigStore(path))
    try:
        job = manager.submit('create', f'Article5Check-{token}', account)
        previous = None
        while True:
            job = manager.job(job['id'])
            if job['stage'] != previous:
                print(job['stage'], flush=True)
                previous = job['stage']
            if job['status'] in ('done', 'error'):
                break
            time.sleep(1)
        print(json.dumps(job, ensure_ascii=False), flush=True)
        if job['status'] != 'done':
            return 1
        info = manager.info(job['folder'])
        assert info['pid'] and info['status'] == 'running', info
        print('Real terminal process and service heartbeat: OK', flush=True)
        manager.stop(job['folder'])
        return 0
    finally:
        manager.close()
        for process in psutil.process_iter(['exe']):
            try:
                if process.info['exe'] and Path(process.info['exe']).resolve().is_relative_to(check_root):
                    process.terminate()
                    process.wait(10)
            except (psutil.NoSuchProcess, psutil.AccessDenied):
                continue
        assert check_root.parent == (PROJECT / '.checks').resolve()
        shutil.rmtree(check_root)
        for suffix in ('', '-wal', '-shm'):
            Path(str(database) + suffix).unlink(missing_ok=True)


if __name__ == '__main__':
    raise SystemExit(main())