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