| '''Проверки схемы сервиса и чтения 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()
|
