MQL5-Google-Onedrive/scripts/test_telegram_async.py

import sys
import unittest
import asyncio
from unittest.mock import MagicMock, patch, AsyncMock

# Mock telegram modules BEFORE importing the bot script
# This prevents the script from exiting due to missing dependencies
sys.modules['telegram'] = MagicMock()
sys.modules['telegram.ext'] = MagicMock()

# Import the module to test
# We rely on the script being in the same directory or python path
import scripts.telegram_deploy_bot as bot

class TestTelegramBotAsync(unittest.TestCase):
    def test_check_fly_status_success(self):
        """Test check_fly_status returns success message on 0 return code"""
        async def run_test():
            # Mock the subprocess process object
            mock_proc = MagicMock()
            mock_proc.returncode = 0
            # communicate() is a coroutine, so the mock needs to return an awaitable
            # that resolves to (stdout, stderr)
            f = asyncio.Future()
            f.set_result((b'App is running', b''))
            mock_proc.communicate.return_value = f

            # create_subprocess_exec is an async function, so mock it with AsyncMock
            with patch('asyncio.create_subprocess_exec', new_callable=AsyncMock) as mock_exec:
                mock_exec.return_value = mock_proc

                result = await bot.check_fly_status()

                # Check that create_subprocess_exec was called with correct args
                mock_exec.assert_called_with(
                    'flyctl', 'status',
                    cwd=bot.REPO_ROOT,
                    stdout=asyncio.subprocess.PIPE,
                    stderr=asyncio.subprocess.PIPE
                )

                # Check output format
                self.assertIn("Fly.io Status", result)
                self.assertIn("App is running", result)
                self.assertNotIn("failed", result.lower())

        asyncio.run(run_test())

    def test_check_fly_status_failure(self):
        """Test check_fly_status returns failure message on non-0 return code"""
        async def run_test():
            mock_proc = MagicMock()
            mock_proc.returncode = 1

            f = asyncio.Future()
            f.set_result((b'', b'Error: App not found'))
            mock_proc.communicate.return_value = f

            with patch('asyncio.create_subprocess_exec', new_callable=AsyncMock) as mock_exec:
                mock_exec.return_value = mock_proc

                result = await bot.check_fly_status()

                self.assertIn("failed", result.lower())
                self.assertIn("Error: App not found", result)

        asyncio.run(run_test())

    def test_check_fly_status_timeout(self):
        """Test check_fly_status handles timeout"""
        async def run_test():
            mock_proc = MagicMock()

            # Simulate create_subprocess_exec returning our mock process
            with patch('asyncio.create_subprocess_exec', new_callable=AsyncMock) as mock_exec:
                mock_exec.return_value = mock_proc

                # Mock wait_for to raise TimeoutError
                with patch('asyncio.wait_for', side_effect=asyncio.TimeoutError):
                    result = await bot.check_fly_status()

                    self.assertIn("timed out", result.lower())
                    # Ensure kill was called
                    mock_proc.kill.assert_called_once()

        asyncio.run(run_test())

    def test_file_not_found(self):
        """Test check_fly_status handles missing executable"""
        async def run_test():
            with patch('asyncio.create_subprocess_exec', side_effect=FileNotFoundError):
                result = await bot.check_fly_status()
                self.assertIn("flyctl not found", result)

        asyncio.run(run_test())

    def test_check_fly_status_special_chars(self):
        """Test check_fly_status escapes HTML special characters"""
        async def run_test():
            mock_proc = MagicMock()
            mock_proc.returncode = 0

            f = asyncio.Future()
            # Output with special characters
            f.set_result((b'Status: <running> & happy', b''))
            mock_proc.communicate.return_value = f

            with patch('asyncio.create_subprocess_exec', new_callable=AsyncMock) as mock_exec:
                mock_exec.return_value = mock_proc

                result = await bot.check_fly_status()

                # Check that output is escaped
                self.assertIn("&lt;running&gt; &amp; happy", result)
                self.assertNotIn("<running>", result)

        asyncio.run(run_test())

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