forked from antekov/mt5-manager
'''Проверки схемы сервиса и чтения SQLite без создания пустых баз менеджером.'''
| |
| |
import ast
| |
import io
| |
import re
| |
import sqlite3
| |
import tempfile
| |
import unittest
| |
from contextlib import closing, redirect_stdout
| |
from pathlib import Path
| |
| |
from mt5_control import MT5_Control
| |
| |
| |
class DatabaseTests(unittest.TestCase):
| |
def setUp(self):
| |
self.temp = tempfile.TemporaryDirectory()
| |
self.root = Path(self.temp.name)
| |
# Решётка и пробелы проверяют корректность URI файла SQLite.
| |
self.database = self.root / 'состояние #1.sqlite'
| |
self.control = MT5_Control({'One': {}}, str(self.root), str(self.database))
| |
| |
def tearDown(self):
| |
self.temp.cleanup()
| |
| |
def create_service_schema(self, connection):
| |
# Используем SQL из настоящего MQL5-исходника, чтобы проверять его
| |
# совместимость с STATE_QUERY, а не отдельную копию схемы в тестах.
| |
source = (Path(__file__).resolve().parent.parent / 'ServiceStateWriter.mq5').read_text(encoding='utf-8-sig')
| |
definitions = re.findall(r'const string sql_(?:terminals|states)\s*=\s*(.*?);', source, re.S)
| |
self.assertEqual(len(definitions), 2)
| |
for definition in definitions:
| |
sql = ''.join(ast.literal_eval(value) for value in re.findall(r'"(?:[^"\\]|\\.)*"', definition))
| |
connection.execute(sql)
| |
| |
def test_missing_database_is_not_created_by_reader(self):
| |
with redirect_stdout(io.StringIO()) as output:
| |
self.assertIsNone(self.control.read_state('One'))
| |
self.assertFalse(self.database.exists())
| |
self.assertIn(str(self.database), output.getvalue())
| |
| |
def test_existing_empty_database_is_reported_without_modifying_it(self):
| |
self.database.touch()
| |
with redirect_stdout(io.StringIO()) as output:
| |
self.assertIsNone(self.control.read_state('One'))
| |
self.assertIn('no such table: terminals', output.getvalue())
| |
self.assertIn(str(self.database), output.getvalue())
| |
self.assertEqual(self.database.stat().st_size, 0)
| |
| |
def test_service_schema_initializes_existing_file_and_matches_python_query(self):
| |
self.database.touch()
| |
with closing(sqlite3.connect(self.database)) as connection:
| |
self.create_service_schema(connection)
| |
connection.execute('INSERT INTO terminals(id, folder, path) VALUES (1, ?, ?)',
| |
('One', str(self.root / 'One')))
| |
connection.execute('INSERT INTO states(terminal_id, updated_at, equity) VALUES (1, 123, 50000)')
| |
connection.commit()
| |
# Повторное создание не удаляет уже записанные данные.
| |
self.create_service_schema(connection)
| |
connection.commit()
| |
row = self.control.read_state('One')
| |
self.assertEqual(row['equity'], 50000)
| |
self.assertEqual(row['terminal_path'], str(self.root / 'One'))
| |
self.assertIn('holder_name', row.keys())
| |
| |
def test_readonly_connection_observes_committed_wal_updates(self):
| |
with closing(sqlite3.connect(self.database)) as writer:
| |
writer.execute('PRAGMA journal_mode=WAL')
| |
self.create_service_schema(writer)
| |
writer.execute('INSERT INTO terminals(id, folder) VALUES (1, ?)', ('One',))
| |
writer.execute('INSERT INTO states(terminal_id, updated_at, equity) VALUES (1, 123, 50000)')
| |
writer.commit()
| |
self.assertEqual(self.control.read_state('One')['equity'], 50000)
| |
writer.execute('UPDATE states SET equity=51000 WHERE terminal_id=1')
| |
writer.commit()
| |
self.assertEqual(self.control.read_state('One')['equity'], 51000)
| |
| |
| |
if __name__ == '__main__':
| |
unittest.main()
|