mt5-manager/tests/test_database.py

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