bifurcado de antekov/mt5-manager
'''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()
|