mt5-manager/tests/test_project_updates.py

'''Обновления проверяются на локальном remote, без сетевого репозитория пользователя.'''

import os
import shutil
import subprocess
import tempfile
import unittest
from pathlib import Path
from unittest.mock import patch

from deploy_service import (SERVICE_EX5_REL, compile_reference, reference_terminal,
                            service_build_paths, services_ini_update, write_services_ini)
from project_updates import ProjectUpdates


class ProjectUpdateTests(unittest.TestCase):
    def setUp(self):
        self.temp = tempfile.TemporaryDirectory()
        self.root = Path(self.temp.name)
        self.remote = self.root / 'remote.git'
        self.writer = self.root / 'writer'
        self.reference = self.root / 'reference'
        self.project = self.reference / 'MQL5/Shared Projects/mt5-manager'
        self.project.parent.mkdir(parents=True)
        self.git(self.root, 'init', '--bare', str(self.remote))
        self.git(self.root, 'clone', str(self.remote), str(self.writer))
        self.git(self.writer, 'checkout', '-b', 'main')
        self.git(self.writer, 'config', 'user.name', 'Fixture')
        self.git(self.writer, 'config', 'user.email', 'fixture@example.invalid')
        (self.writer / 'ServiceStateWriter.mq5').write_text('first source', encoding='utf-8')
        self.copy_template(self.writer)
        self.git(self.writer, 'add', '.')
        self.git(self.writer, 'commit', '-m', 'Initial fixture')
        self.git(self.writer, 'push', '-u', 'origin', 'main')
        self.git(self.root, 'clone', '--branch', 'main', str(self.remote), str(self.project))
        self.updates = ProjectUpdates(self.project)

    @staticmethod
    def copy_template(project):
        (project / 'static').mkdir(parents=True, exist_ok=True)
        shutil.copyfile(Path(__file__).resolve().parent.parent / 'static/services.ini',
                        project / 'static/services.ini')

    def tearDown(self):
        # Git на Windows помечает объекты read-only.
        for path in self.root.rglob('*'):
            if path.is_file():
                path.chmod(0o666)
        self.temp.cleanup()

    @staticmethod
    def git(cwd, *args):
        return subprocess.run(['git', *args], cwd=cwd, capture_output=True, text=True, check=True).stdout.strip()

    def publish(self):
        (self.writer / 'ServiceStateWriter.mq5').write_text('second source', encoding='utf-8')
        self.git(self.writer, 'add', '.')
        self.git(self.writer, 'commit', '-m', 'Update fixture service')
        self.git(self.writer, 'push')

    def test_fetch_preview_and_fast_forward(self):
        self.publish()
        before = self.git(self.project, 'rev-parse', 'HEAD')
        status = self.updates.check()
        self.assertEqual(status['behind'], 1)
        self.assertEqual(self.git(self.project, 'rev-parse', 'HEAD'), before)
        with patch.dict(os.environ, {'MT5_MANAGER_RELOAD': '0'}):
            result = self.updates.update()
        self.assertTrue(result['updated'])
        self.assertEqual((self.project / 'ServiceStateWriter.mq5').read_text(), 'second source')
        self.assertEqual(self.updates.repository()['behind'], 0)

    def test_dirty_tree_and_divergence_preserved(self):
        self.publish()
        local = self.project / 'local.txt'
        local.write_text('keep this')
        with patch.dict(os.environ, {'MT5_MANAGER_RELOAD': '0'}):
            with self.assertRaisesRegex(RuntimeError, 'локальные изменения'):
                self.updates.update()
        self.assertEqual(local.read_text(), 'keep this')
        self.git(self.project, 'config', 'user.name', 'Fixture')
        self.git(self.project, 'config', 'user.email', 'fixture@example.invalid')
        self.git(self.project, 'add', '.')
        self.git(self.project, 'commit', '-m', 'Local fixture change')
        before = self.git(self.project, 'rev-parse', 'HEAD')
        with patch.dict(os.environ, {'MT5_MANAGER_RELOAD': '0'}):
            with self.assertRaisesRegex(RuntimeError, 'свои коммиты'):
                self.updates.update()
        self.assertEqual(self.git(self.project, 'rev-parse', 'HEAD'), before)

    def test_reload_mode_rejects_update(self):
        with patch.dict(os.environ, {'MT5_MANAGER_RELOAD': '1'}), \
             patch.object(self.updates, 'git') as git:
            with self.assertRaisesRegex(RuntimeError, '--reload'):
                self.updates.update()
            git.assert_not_called()

    def test_deploy_uses_base_folder_ignores_old_paths_and_reports_partial_failure(self):
        (self.reference / 'MetaEditor64.exe').write_bytes(b'fixture')
        source = self.reference / SERVICE_EX5_REL
        source.parent.mkdir(parents=True)
        source.write_bytes(b'old-ex5')
        base = self.root / 'instances'
        target = base / 'Custom'
        target.mkdir(parents=True)
        (target / 'terminal64.exe').write_bytes(b'fixture')
        old_target = self.root / 'old-base/Custom'
        (old_target / SERVICE_EX5_REL).parent.mkdir(parents=True)
        (old_target / SERVICE_EX5_REL).write_bytes(b'keep-old-location')
        data = {'mt5_folder': str(base), 'mt5_exe': 'terminal64.exe', 'database': 'custom.sqlite', 'terminals': {
            'Custom': {'path': str(old_target)}, 'Missing': {'path': str(self.root / 'missing')}}}
        ini_path, ini_text, _ = services_ini_update(target, self.project / 'static/services.ini', data['database'])
        write_services_ini(ini_path, ini_text)
        self.assertTrue(self.updates.service_status(data)['pending'])

        def compile_service(editor, mq5, output, *, force=False):
            self.assertTrue(force)
            source.write_bytes(b'new-ex5')
            return True

        with patch('project_updates.compile_service', side_effect=compile_service):
            result = self.updates.deploy(data, lambda folder: folder == 'Custom')
        self.assertEqual((target / SERVICE_EX5_REL).read_bytes(), b'new-ex5')
        self.assertEqual(result['copied'], ['Custom'])
        self.assertEqual(result['failed'], ['Missing'])
        self.assertIn(str(base / 'Missing/terminal64.exe'), result['errors']['Missing'])
        self.assertEqual(result['restart_required'], ['Custom'])
        self.assertFalse((base / 'Missing').exists())
        self.assertEqual((old_target / SERVICE_EX5_REL).read_bytes(), b'keep-old-location')

    def test_deploy_reports_windows_copy_error_and_preserves_previous_service(self):
        source = self.reference / SERVICE_EX5_REL
        source.parent.mkdir(parents=True)
        source.write_bytes(b'new-build')
        target = self.root / 'One'
        destination = target / SERVICE_EX5_REL
        destination.parent.mkdir(parents=True)
        destination.write_bytes(b'old-build')
        (target / 'terminal64.exe').write_bytes(b'terminal')
        data = {'mt5_folder': str(self.root), 'mt5_exe': 'terminal64.exe', 'database': 'custom.sqlite', 'terminals': {'One': {}}}
        error = PermissionError(13, 'Access is denied', str(destination))
        with patch('project_updates.compile_service', return_value=True), \
             patch('project_updates.os.replace', side_effect=error):
            result = self.updates.deploy(data, lambda _: False)
        self.assertEqual(result['failed'], ['One'])
        self.assertEqual(result['copied'], [])
        self.assertIn('Access is denied', result['errors']['One'])
        self.assertIn(str(destination), result['errors']['One'])
        self.assertIn('Замена EX5', result['errors']['One'])
        self.assertEqual(destination.read_bytes(), b'old-build')
        self.assertFalse(list(destination.parent.glob('*.tmp')))

    def test_failed_compilation_never_distributes_previous_ex5(self):
        target = self.root / 'One'
        destination = target / SERVICE_EX5_REL
        destination.parent.mkdir(parents=True)
        destination.write_bytes(b'keep-old-service')
        data = {'mt5_folder': str(self.root), 'mt5_exe': 'terminal64.exe', 'database': 'custom.sqlite', 'terminals': {'One': {}}}
        with patch('project_updates.compile_service', return_value=False):
            with self.assertRaisesRegex(RuntimeError, 'не подтверждена'):
                self.updates.deploy(data, lambda _: False)
        self.assertEqual(destination.read_bytes(), b'keep-old-service')

    def test_force_build_does_not_accept_metaeditor_no_op(self):
        (self.reference / 'MetaEditor64.exe').write_bytes(b'fixture')
        output = self.reference / SERVICE_EX5_REL
        output.parent.mkdir(parents=True)
        output.write_bytes(b'old-build')
        with patch('deploy_service.run') as run:
            self.assertFalse(compile_reference(self.reference, force=True))
            run.assert_called_once()

    def test_standalone_status_and_deploy_use_uploaded_editor_and_current_project(self):
        project = self.root / 'standalone'
        project.mkdir()
        self.copy_template(project)
        (project / 'ServiceStateWriter.mq5').write_text('standalone source')
        template = self.root / 'uploads'
        template.mkdir()
        (template / 'MetaEditor64.exe').write_bytes(b'editor')
        target = self.root / 'One'
        target.mkdir()
        (target / 'terminal64.exe').write_bytes(b'terminal')
        data = {'terminal_template': str(template), 'mt5_exe': 'terminal64.exe',
                'mt5_folder': str(self.root), 'database': 'custom.sqlite', 'terminals': {'One': {}}}
        updates = ProjectUpdates(project)
        self.assertTrue(updates.service_status(data)['available'])
        self.assertTrue(updates.service_status(data)['needs_compile'])

        def compile_service(editor, source, output, *, force=False):
            self.assertEqual(editor, template / 'MetaEditor64.exe')
            self.assertEqual(source.read_text(), 'standalone source')
            self.assertTrue(force)
            output.write_bytes(b'standalone build')
            return True

        with patch('project_updates.compile_service', side_effect=compile_service):
            result = updates.deploy(data, lambda _: False)
        self.assertEqual(result['copied'], ['One'])
        self.assertEqual((target / SERVICE_EX5_REL).read_bytes(), b'standalone build')
        self.assertIn('InpDatabase=custom.sqlite', (target / 'config/services.ini').read_text(encoding='utf-16'))
        self.assertFalse(updates.service_status(data)['pending'])

    def test_running_terminal_requires_stop_before_autostart_repair(self):
        source = self.reference / SERVICE_EX5_REL
        source.parent.mkdir(parents=True)
        source.write_bytes(b'compiled')
        target = self.root / 'One'
        target.mkdir()
        (target / 'terminal64.exe').write_bytes(b'terminal')
        data = {'mt5_folder': str(self.root), 'mt5_exe': 'terminal64.exe',
                'database': 'custom.sqlite', 'terminals': {'One': {}}}
        with patch('project_updates.compile_service', return_value=True):
            result = self.updates.deploy(data, lambda _: True)
        self.assertEqual(result['failed'], ['One'])
        self.assertIn('Остановите терминал', result['errors']['One'])
        self.assertFalse((target / 'config/services.ini').exists())

    def test_missing_compiler_at_filesystem_root_does_not_index_ancestors(self):
        project = Path(self.root.anchor)
        self.assertIsNone(reference_terminal(project))
        editor, source, output = service_build_paths(project, {})
        self.assertEqual(source, project / 'ServiceStateWriter.mq5')
        self.assertEqual(output, source.with_suffix('.ex5'))
        self.assertEqual(editor, project / 'terminal-template/MetaEditor64.exe')


if __name__ == '__main__':
    unittest.main()