mt5-manager/tests/test_http.py

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

import json
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):
        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.assertEqual(self.request('/api/files/servers', 'PUT', b'new-server-data', raw=True)[0], 200)
        self.assertEqual((self.fixture.template / 'config/servers.dat').read_bytes(), b'new-server-data')
        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()
        (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.assertEqual(settings['config']['mt5_installation'], '')
            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', 'service_ex5',
                       'startup_timeout_sec', 'disk_warning_mb')}
            values['mt5_installation'] = 'reference'
            self.assertEqual(self.request('/api/settings', 'PUT', values)[0], 200)
            _, body = self.request('/api/settings')
            self.assertEqual(json.loads(body)['config']['mt5_installation'],
                             str(self.fixture.root / 'reference'))

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