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