#!/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())