ASE/test_ftp_send_migration.py

#!/usr/bin/env python3
"""
Test suite for AsyncFTPConnection class migration.

Tests the new async FTP implementation to ensure it correctly replaces
the blocking ftplib implementation.

Run this test:
    python3 test_ftp_send_migration.py
"""

import asyncio
import logging
import sys
from pathlib import Path

# Add src to path
sys.path.insert(0, str(Path(__file__).parent / "src"))

from utils.connect.send_data import AsyncFTPConnection, parse_ftp_parms

# Setup logging
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__)


class TestAsyncFTPConnection:
    """Test suite for AsyncFTPConnection class"""

    def __init__(self):
        self.passed = 0
        self.failed = 0
        self.test_results = []

    async def test_parse_ftp_parms_basic(self):
        """Test 1: Parse basic FTP parameters"""
        test_name = "Parse basic FTP parameters"
        try:
            ftp_parms_str = "port => 21, passive => true, timeout => 30"
            result = await parse_ftp_parms(ftp_parms_str)

            assert result["port"] == 21, f"Expected port=21, got {result['port']}"
            assert result["passive"] == "true", f"Expected passive='true', got {result['passive']}"
            assert result["timeout"] == 30, f"Expected timeout=30, got {result['timeout']}"

            self.passed += 1
            self.test_results.append((test_name, "✓ PASS", None))
            logger.info(f"✓ {test_name}: PASS")
        except Exception as e:
            self.failed += 1
            self.test_results.append((test_name, "✗ FAIL", str(e)))
            logger.error(f"✗ {test_name}: FAIL - {e}")

    async def test_parse_ftp_parms_with_ssl(self):
        """Test 2: Parse FTP parameters with SSL"""
        test_name = "Parse FTP parameters with SSL"
        try:
            ftp_parms_str = "port => 990, ssl_version => TLSv1.2, passive => true"
            result = await parse_ftp_parms(ftp_parms_str)

            assert result["port"] == 990, f"Expected port=990, got {result['port']}"
            assert "ssl_version" in result, "ssl_version key missing"
            assert result["ssl_version"] == "tlsv1.2", f"Expected ssl_version='tlsv1.2', got {result['ssl_version']}"

            self.passed += 1
            self.test_results.append((test_name, "✓ PASS", None))
            logger.info(f"✓ {test_name}: PASS")
        except Exception as e:
            self.failed += 1
            self.test_results.append((test_name, "✗ FAIL", str(e)))
            logger.error(f"✗ {test_name}: FAIL - {e}")

    async def test_async_ftp_connection_init(self):
        """Test 3: Initialize AsyncFTPConnection"""
        test_name = "Initialize AsyncFTPConnection"
        try:
            ftp = AsyncFTPConnection(
                host="ftp.example.com",
                port=21,
                use_tls=False,
                user="testuser",
                passwd="testpass",
                passive=True,
                timeout=30.0
            )

            assert ftp.host == "ftp.example.com", f"Expected host='ftp.example.com', got {ftp.host}"
            assert ftp.port == 21, f"Expected port=21, got {ftp.port}"
            assert ftp.use_tls is False, f"Expected use_tls=False, got {ftp.use_tls}"
            assert ftp.user == "testuser", f"Expected user='testuser', got {ftp.user}"
            assert ftp.passwd == "testpass", f"Expected passwd='testpass', got {ftp.passwd}"
            assert ftp.timeout == 30.0, f"Expected timeout=30.0, got {ftp.timeout}"

            self.passed += 1
            self.test_results.append((test_name, "✓ PASS", None))
            logger.info(f"✓ {test_name}: PASS")
        except Exception as e:
            self.failed += 1
            self.test_results.append((test_name, "✗ FAIL", str(e)))
            logger.error(f"✗ {test_name}: FAIL - {e}")

    async def test_async_ftp_connection_tls_init(self):
        """Test 4: Initialize AsyncFTPConnection with TLS"""
        test_name = "Initialize AsyncFTPConnection with TLS"
        try:
            ftp = AsyncFTPConnection(
                host="ftps.example.com",
                port=990,
                use_tls=True,
                user="testuser",
                passwd="testpass",
                passive=True,
                timeout=30.0
            )

            assert ftp.use_tls is True, f"Expected use_tls=True, got {ftp.use_tls}"
            assert ftp.port == 990, f"Expected port=990, got {ftp.port}"

            self.passed += 1
            self.test_results.append((test_name, "✓ PASS", None))
            logger.info(f"✓ {test_name}: PASS")
        except Exception as e:
            self.failed += 1
            self.test_results.append((test_name, "✗ FAIL", str(e)))
            logger.error(f"✗ {test_name}: FAIL - {e}")

    async def test_parse_ftp_parms_empty_values(self):
        """Test 5: Parse FTP parameters with empty values"""
        test_name = "Parse FTP parameters with empty values"
        try:
            ftp_parms_str = "port => 21, user => , passive => true"
            result = await parse_ftp_parms(ftp_parms_str)

            assert result["port"] == 21, f"Expected port=21, got {result['port']}"
            assert result["user"] is None, f"Expected user=None, got {result['user']}"
            assert result["passive"] == "true", f"Expected passive='true', got {result['passive']}"

            self.passed += 1
            self.test_results.append((test_name, "✓ PASS", None))
            logger.info(f"✓ {test_name}: PASS")
        except Exception as e:
            self.failed += 1
            self.test_results.append((test_name, "✗ FAIL", str(e)))
            logger.error(f"✗ {test_name}: FAIL - {e}")

    async def run_all_tests(self):
        """Run all tests"""
        logger.info("=" * 60)
        logger.info("Starting AsyncFTPConnection Migration Tests")
        logger.info("=" * 60)

        await self.test_parse_ftp_parms_basic()
        await self.test_parse_ftp_parms_with_ssl()
        await self.test_async_ftp_connection_init()
        await self.test_async_ftp_connection_tls_init()
        await self.test_parse_ftp_parms_empty_values()

        logger.info("=" * 60)
        logger.info(f"Test Results: {self.passed} passed, {self.failed} failed")
        logger.info("=" * 60)

        if self.failed > 0:
            logger.error("\n❌ Some tests failed:")
            for test_name, status, error in self.test_results:
                if status == "✗ FAIL":
                    logger.error(f"  - {test_name}: {error}")
            return False
        else:
            logger.info("\n✅ All tests passed!")
            return True


async def main():
    """Main test runner"""
    test_suite = TestAsyncFTPConnection()
    success = await test_suite.run_all_tests()

    if not success:
        sys.exit(1)


if __name__ == "__main__":
    asyncio.run(main())