mt5-manager/tests/test_http.py

'''HTTP-проверки интерфейса и контрактов фоновых операций.'''

import json
import base64
import os
import socket
import threading
import time
import unittest
import urllib.error
import urllib.request
from unittest.mock import patch

import uvicorn

import main
import test_provisioning
from project_updates import ProjectUpdates


class HttpTests(unittest.TestCase):
    def setUp(self):
        # Используем тот же изолированный набор файлов, но обращаемся через HTTP.
        self.fixture = test_provisioning.ProvisioningTests(methodName='test_create_without_account')
        self.fixture.setUp()
        self.env = patch.dict(os.environ, {'MT5_MANAGER_CONFIG_FILE': str(self.fixture.path)})
        self.factory = patch.object(main, 'TerminalManager', return_value=self.fixture.manager)
        self.env.start()
        self.factory.start()
        self.socket = socket.socket()
        self.socket.bind(('127.0.0.1', 0))
        self.port = self.socket.getsockname()[1]
        self.server = uvicorn.Server(uvicorn.Config(main.app, log_level='error'))
        self.thread = threading.Thread(target=self.server.run, kwargs={'sockets': [self.socket]}, daemon=True)
        self.thread.start()
        deadline = time.monotonic() + 5
        while not self.server.started and self.thread.is_alive() and time.monotonic() < deadline:
            time.sleep(0.01)
        if not self.server.started:
            self.fail('HTTP-сервер не запустился.')

    def tearDown(self):
        self.server.should_exit = True
        self.thread.join(5)
        self.socket.close()
        self.factory.stop()
        self.env.stop()
        self.fixture.tearDown()

    def request(self, path, method='GET', data=None, raw=False, default_servers=True):
        if default_servers and path == '/api/terminals' and method == 'POST' and isinstance(data, dict):
            data = dict(data)
            data.setdefault('servers_dat', base64.b64encode(b'http-server-directory').decode('ascii'))
        body = data if raw else json.dumps(data).encode() if data is not None else None
        headers = {'Content-Type': 'application/octet-stream' if raw else 'application/json'}
        request = urllib.request.Request(f'http://127.0.0.1:{self.port}' + path,
                                         data=body, headers=headers, method=method)
        with urllib.request.urlopen(request, timeout=5) as response:
            return response.status, response.read()

    def test_create_job_and_secret_redaction(self):
        status, body = self.request('/api/terminals', 'POST',
                                    {'folder': 'HttpCheck', 'login': 123, 'server': 'Broker', 'password': 'fixture-secret'})
        self.assertEqual(status, 202)
        self.assertNotIn(b'fixture-secret', body)
        job = json.loads(body)
        deadline = time.monotonic() + 5
        while job['status'] in ('queued', 'running') and time.monotonic() < deadline:
            time.sleep(0.01)
            _, body = self.request('/api/jobs/' + job['id'])
            job = json.loads(body)
        self.assertEqual(job['status'], 'done', job)
        _, settings = self.request('/api/settings')
        self.assertNotIn(b'fixture-secret', settings)
        record = json.loads(settings)['config']['terminals']['HttpCheck']
        self.assertNotIn('password', record)
        self.assertTrue(record['password_set'])
        _, info = self.request('/instances/HttpCheck', 'POST')
        self.assertNotIn(b'fixture-secret', info)
        self.assertEqual(self.request('/')[0], 200)
        self.assertEqual(self.request('/settings')[0], 200)

    def test_upload_success_and_invalid_executable(self):
        (self.fixture.template / 'MetaEditor64.exe').unlink()
        self.assertEqual(self.request('/api/files/editor', 'PUT', b'MZ-new-editor', raw=True)[0], 200)
        self.assertEqual((self.fixture.template / 'MetaEditor64.exe').read_bytes(), b'MZ-new-editor')
        original = (self.fixture.template / 'terminal64.exe').read_bytes()
        with self.assertRaises(urllib.error.HTTPError) as caught:
            self.request('/api/files/terminal', 'PUT', b'not-an-exe', raw=True)
        self.assertEqual(caught.exception.code, 400)
        caught.exception.close()
        self.assertEqual((self.fixture.template / 'terminal64.exe').read_bytes(), original)

    def test_project_check_runs_as_background_job(self):
        with patch.object(self.fixture.manager.updates, 'check', return_value={'behind': 1}):
            status, body = self.request('/api/project/check', 'POST')
            self.assertEqual(status, 202)
            job = json.loads(body)
            deadline = time.monotonic() + 5
            while job['status'] in ('queued', 'running') and time.monotonic() < deadline:
                time.sleep(0.01)
                _, body = self.request('/api/jobs/' + job['id'])
                job = json.loads(body)
        self.assertEqual(job['status'], 'done', job)
        self.assertEqual(job['result']['behind'], 1)
        self.assertEqual(self.request('/api/project')[0], 200)
        with self.assertRaises(urllib.error.HTTPError) as caught:
            self.request('/api/project/unknown', 'POST')
        self.assertEqual(caught.exception.code, 404)
        caught.exception.close()

    def test_standalone_settings_and_project_endpoints(self):
        project = self.fixture.root / 'checkout'
        project.mkdir(exist_ok=True)
        (project / 'ServiceStateWriter.mq5').write_text('fixture source')
        self.fixture.manager.updates = ProjectUpdates(project)
        with patch('terminal_manager.PROJECT_DIR', project):
            self.assertEqual(self.request('/settings')[0], 200)
            _, body = self.request('/api/settings')
            settings = json.loads(body)
            self.assertNotIn('mt5_installation', settings['config'])
            self.assertEqual(set(settings['files']), {'terminal', 'editor'})
            self.assertTrue(settings['files']['editor']['exists'])
            _, body = self.request('/api/project')
            self.assertTrue(json.loads(body)['service']['available'])
            values = {key: settings['config'][key] for key in
                      ('mt5_folder', 'terminal_template',
                       'startup_timeout_sec', 'disk_warning_mb')}
            values['terminal_template'] = 'reference'
            self.assertEqual(self.request('/api/settings', 'PUT', values)[0], 200)
            _, body = self.request('/api/settings')
            self.assertEqual(json.loads(body)['config']['terminal_template'],
                             str(self.fixture.root / 'reference'))

    def test_create_copies_request_servers_file_and_rejects_invalid_data(self):
        _, body = self.request('/api/terminals', 'POST', {'folder': 'OwnServers'})
        job = json.loads(body)
        deadline = time.monotonic() + 5
        while job['status'] in ('queued', 'running') and time.monotonic() < deadline:
            time.sleep(0.01)
            _, body = self.request('/api/jobs/' + job['id'])
            job = json.loads(body)
        self.assertEqual(job['status'], 'done', job)
        target = self.fixture.root / 'instances/OwnServers/config/servers.dat'
        self.assertEqual(target.read_bytes(), b'http-server-directory')
        for value in ('not valid base64!',):
            with self.subTest(value=value), self.assertRaises(urllib.error.HTTPError) as caught:
                self.request('/api/terminals', 'POST', {'folder': 'BadServers', 'servers_dat': value})
            self.assertEqual(caught.exception.code, 400)
            caught.exception.close()
        self.assertFalse((self.fixture.root / 'instances/BadServers').exists())

    def test_create_without_servers_file(self):
        for folder, extra in (('NoServersField', {}), ('EmptyServersField', {'servers_dat': ''})):
            with self.subTest(folder=folder):
                _, body = self.request('/api/terminals', 'POST', {'folder': folder, **extra}, default_servers=False)
                job = json.loads(body)
                deadline = time.monotonic() + 5
                while job['status'] in ('queued', 'running') and time.monotonic() < deadline:
                    time.sleep(0.01)
                    _, body = self.request('/api/jobs/' + job['id'])
                    job = json.loads(body)
                self.assertEqual(job['status'], 'done', job)
                self.assertFalse((self.fixture.root / 'instances' / folder / 'config/servers.dat').exists())

    def test_account_edit_renames_instance_and_uploads_servers(self):
        self.fixture.run_job('create', 'Original')
        self.fixture.manager.stop('Original')
        _, body = self.request('/api/terminals/Original/account', 'PUT',
                               {'folder': 'Renamed', 'name': 'Renamed terminal',
                                'servers_dat': base64.b64encode(b'new http broker').decode('ascii')})
        job = json.loads(body)
        deadline = time.monotonic() + 5
        while job['status'] in ('queued', 'running') and time.monotonic() < deadline:
            time.sleep(0.01)
            _, body = self.request('/api/jobs/' + job['id'])
            job = json.loads(body)
        self.assertEqual(job['status'], 'done', job)
        self.assertEqual(job['result']['folder'], 'Renamed')
        target = self.fixture.root / 'instances/Renamed/config/servers.dat'
        self.assertEqual(target.read_bytes(), b'new http broker')
        _, body = self.request('/api/settings')
        records = json.loads(body)['config']['terminals']
        self.assertNotIn('Original', records)
        self.assertFalse(records['Renamed']['running'])
        self.assertNotIn('running', self.fixture.path.read_text())

    def test_account_edit_rejects_running_rename_and_invalid_servers(self):
        self.fixture.run_job('create', 'Running')
        for values in ({'folder': 'Renamed'}, {'servers_dat': 'invalid base64!'}):
            with self.subTest(values=values), self.assertRaises(urllib.error.HTTPError) as caught:
                self.request('/api/terminals/Running/account', 'PUT', values)
            self.assertEqual(caught.exception.code, 400)
            caught.exception.close()
        self.assertTrue((self.fixture.root / 'instances/Running').exists())

    def test_save_does_not_launch_and_explicit_test_does_not_save(self):
        self.fixture.run_job('create', 'Existing')
        self.fixture.manager.stop('Existing')
        launches = list(self.fixture.manager.launches)

        def finish(body):
            job = json.loads(body)
            deadline = time.monotonic() + 5
            while job['status'] in ('queued', 'running') and time.monotonic() < deadline:
                time.sleep(0.01)
                _, response = self.request('/api/jobs/' + job['id'])
                job = json.loads(response)
            self.assertEqual(job['status'], 'done', job)
            return job

        _, body = self.request('/api/terminals/Existing/account', 'PUT',
                               {'name': 'Display', 'login': 111, 'server': 'Saved-Broker'})
        finish(body)
        self.assertEqual(self.fixture.manager.launches, launches)
        self.assertNotIn('Existing', self.fixture.manager.running)
        before = self.fixture.path.read_bytes()
        _, body = self.request('/api/terminals/Existing/test-connection', 'POST',
                               {'login': 222, 'server': 'Test-Broker', 'password': 'test-only-secret'})
        self.assertNotIn(b'test-only-secret', body)
        self.assertTrue(finish(body)['result']['connected'])
        self.assertEqual(self.fixture.path.read_bytes(), before)
        self.assertEqual(self.fixture.manager.current_accounts['Existing'][0], 222)
        self.assertNotIn('Existing', self.fixture.manager.running)
        with self.assertRaises(urllib.error.HTTPError) as caught:
            self.request('/api/terminals/Existing/test-connection', 'POST',
                         {'folder': 'UnsavedRename', 'login': 222, 'server': 'Test-Broker'})
        self.assertEqual(caught.exception.code, 400)
        caught.exception.close()

    def test_delete_instance_rejects_running_then_removes_stopped(self):
        self.fixture.run_job('create', 'DeleteMe')
        with self.assertRaises(urllib.error.HTTPError) as caught:
            self.request('/api/terminals/DeleteMe', 'DELETE')
        self.assertEqual(caught.exception.code, 400)
        caught.exception.close()
        self.fixture.manager.stop('DeleteMe')
        status, body = self.request('/api/terminals/DeleteMe', 'DELETE')
        self.assertEqual(status, 202)
        job = json.loads(body)
        deadline = time.monotonic() + 5
        while job['status'] in ('queued', 'running') and time.monotonic() < deadline:
            time.sleep(0.01)
            _, body = self.request('/api/jobs/' + job['id'])
            job = json.loads(body)
        self.assertEqual(job['status'], 'done', job)
        _, body = self.request('/api/settings')
        self.assertNotIn('DeleteMe', json.loads(body)['config']['terminals'])
        self.assertFalse((self.fixture.root / 'instances/DeleteMe').exists())

    def test_saving_two_folders_keeps_advanced_settings(self):
        self.fixture.manager.save_settings({'startup_timeout_sec': 240, 'disk_warning_mb': 333})
        self.request('/api/settings', 'PUT', {'mt5_folder': str(self.fixture.root / 'instances'),
                                           'terminal_template': str(self.fixture.template)})
        self.assertEqual(self.fixture.manager.store.data['startup_timeout_sec'], 240)
        self.assertEqual(self.fixture.manager.store.data['disk_warning_mb'], 333)

    def test_bad_folder_and_unknown_job(self):
        with self.assertRaises(urllib.error.HTTPError) as caught:
            self.request('/api/terminals', 'POST', {'folder': '../outside'})
        self.assertEqual(caught.exception.code, 400)
        caught.exception.close()
        with self.assertRaises(urllib.error.HTTPError) as caught:
            self.request('/api/jobs/unknown')
        self.assertEqual(caught.exception.code, 404)
        caught.exception.close()

    def test_settings_move_plan_confirmation_and_background_job(self):
        self.fixture.run_job('create', 'MoveMe')
        original = self.fixture.path.read_bytes()
        values = {'mt5_folder': str(self.fixture.root / 'moved'),
                  'terminal_template': str(self.fixture.template)}
        status, body = self.request('/api/settings/plan', 'POST', values)
        plan = json.loads(body)
        self.assertEqual(status, 200)
        self.assertEqual(plan['running'], ['MoveMe'])
        self.assertEqual(self.fixture.path.read_bytes(), original)
        with self.assertRaises(urllib.error.HTTPError) as caught:
            self.request('/api/settings', 'PUT', values)
        self.assertEqual(caught.exception.code, 400)
        caught.exception.close()
        values.update(move_token=plan['token'], stop_running=True)
        status, body = self.request('/api/settings', 'PUT', values)
        self.assertEqual(status, 202)
        job = json.loads(body)
        deadline = time.monotonic() + 5
        while job['status'] in ('queued', 'running') and time.monotonic() < deadline:
            time.sleep(.01)
            _, body = self.request('/api/jobs/' + job['id'])
            job = json.loads(body)
        self.assertEqual(job['status'], 'done', job)
        self.assertTrue((self.fixture.root / 'moved/MoveMe/terminal64.exe').exists())
        self.assertFalse((self.fixture.root / 'instances/MoveMe').exists())
        self.assertNotIn('MoveMe', self.fixture.manager.running)

    def test_validation_error_does_not_echo_password(self):
        with self.assertRaises(urllib.error.HTTPError) as caught:
            self.request('/api/terminals', 'POST', {'folder': {'password': 'fixture-secret'}, 'password': 'fixture-secret'})
        self.assertEqual(caught.exception.code, 422)
        with caught.exception as response:
            self.assertNotIn(b'fixture-secret', response.read())


if __name__ == '__main__':
    unittest.main()