ciaovolo/flight-comparator/tests/test_comprehensive_v3.py

#!/usr/bin/env python3
"""
Comprehensive test suite for fast-flights v3.0rc1 with SOCS cookie integration.
Tests multiple routes, dates, and edge cases.
"""

import sys
import logging
import asyncio
from datetime import date, timedelta

sys.path.insert(0, '..')

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)

try:
    from searcher_v3 import search_direct_flights, search_multiple_routes, SOCSCookieIntegration
    from fast_flights import FlightQuery, Passengers, get_flights, create_query
    HAS_V3 = True
except ImportError as e:
    logger.error(f"✗ Failed to import v3 modules: {e}")
    logger.error("  Install with: pip install --upgrade git+https://github.com/AWeirdDev/flights.git")
    HAS_V3 = False


class TestResults:
    """Track test results."""
    def __init__(self):
        self.total = 0
        self.passed = 0
        self.failed = 0
        self.errors = []

    def add_pass(self, test_name):
        self.total += 1
        self.passed += 1
        logger.info(f"✓ PASS: {test_name}")

    def add_fail(self, test_name, reason):
        self.total += 1
        self.failed += 1
        self.errors.append((test_name, reason))
        logger.error(f"✗ FAIL: {test_name} - {reason}")

    def summary(self):
        logger.info("\n" + "="*80)
        logger.info("TEST SUMMARY")
        logger.info("="*80)
        logger.info(f"Total: {self.total}")
        logger.info(f"Passed: {self.passed} ({self.passed/self.total*100:.1f}%)")
        logger.info(f"Failed: {self.failed}")

        if self.errors:
            logger.info("\nFailed Tests:")
            for name, reason in self.errors:
                logger.info(f"  • {name}: {reason}")

        return self.failed == 0


results = TestResults()


def test_socs_integration():
    """Test that SOCS cookie integration is properly configured."""
    if not HAS_V3:
        results.add_fail("SOCS Integration", "v3 not installed")
        return

    try:
        integration = SOCSCookieIntegration()
        assert hasattr(integration, 'SOCS_COOKIE')
        assert integration.SOCS_COOKIE.startswith('CAE')
        assert hasattr(integration, 'fetch_html')
        results.add_pass("SOCS Integration")
    except Exception as e:
        results.add_fail("SOCS Integration", str(e))


async def test_single_route_ber_bri():
    """Test BER to BRI route (known to work)."""
    if not HAS_V3:
        results.add_fail("BER→BRI Single Route", "v3 not installed")
        return

    try:
        test_date = (date.today() + timedelta(days=30)).strftime('%Y-%m-%d')
        flights = await search_direct_flights("BER", "BRI", test_date)

        if flights and len(flights) > 0:
            # Verify flight structure
            f = flights[0]
            assert 'origin' in f
            assert 'destination' in f
            assert 'price' in f
            assert 'airline' in f
            assert f['origin'] == 'BER'
            assert f['destination'] == 'BRI'
            assert f['price'] > 0

            logger.info(f"  Found {len(flights)} flight(s), cheapest: €{flights[0]['price']}")
            results.add_pass("BER→BRI Single Route")
        else:
            results.add_fail("BER→BRI Single Route", "No flights found")
    except Exception as e:
        results.add_fail("BER→BRI Single Route", str(e))


async def test_multiple_routes():
    """Test multiple routes in one batch."""
    if not HAS_V3:
        results.add_fail("Multiple Routes", "v3 not installed")
        return

    try:
        test_date = (date.today() + timedelta(days=30)).strftime('%Y-%m-%d')

        routes = [
            ("BER", "FCO", test_date),  # Berlin to Rome
            ("FRA", "MAD", test_date),  # Frankfurt to Madrid
            ("MUC", "BCN", test_date),  # Munich to Barcelona
        ]

        batch_results = await search_multiple_routes(
            routes,
            seat_class="economy",
            adults=1,
            max_workers=3,
        )

        # Check we got results for each route
        flights_found = sum(1 for flights in batch_results.values() if flights)

        if flights_found >= 2:  # At least 2 out of 3 should have flights
            logger.info(f"  Found flights for {flights_found}/3 routes")
            results.add_pass("Multiple Routes")
        else:
            results.add_fail("Multiple Routes", f"Only {flights_found}/3 routes had flights")

    except Exception as e:
        results.add_fail("Multiple Routes", str(e))


async def test_different_dates():
    """Test same route with different dates."""
    if not HAS_V3:
        results.add_fail("Different Dates", "v3 not installed")
        return

    try:
        dates = [
            (date.today() + timedelta(days=30)).strftime('%Y-%m-%d'),
            (date.today() + timedelta(days=60)).strftime('%Y-%m-%d'),
            (date.today() + timedelta(days=90)).strftime('%Y-%m-%d'),
        ]

        routes = [("BER", "BRI", d) for d in dates]

        batch_results = await search_multiple_routes(
            routes,
            seat_class="economy",
            adults=1,
            max_workers=2,
        )

        flights_found = sum(1 for flights in batch_results.values() if flights)

        if flights_found >= 2:
            logger.info(f"  Found flights for {flights_found}/3 dates")
            results.add_pass("Different Dates")
        else:
            results.add_fail("Different Dates", f"Only {flights_found}/3 dates had flights")

    except Exception as e:
        results.add_fail("Different Dates", str(e))


async def test_no_direct_flights():
    """Test route with no direct flights (should return empty)."""
    if not HAS_V3:
        results.add_fail("No Direct Flights", "v3 not installed")
        return

    try:
        test_date = (date.today() + timedelta(days=30)).strftime('%Y-%m-%d')

        # BER to SYD probably has no direct flights
        flights = await search_direct_flights("BER", "SYD", test_date)

        # Should return empty list, not crash
        assert isinstance(flights, list)

        logger.info(f"  Correctly handled no-direct-flights case (found {len(flights)})")
        results.add_pass("No Direct Flights")

    except Exception as e:
        results.add_fail("No Direct Flights", str(e))


async def test_invalid_airport_code():
    """Test handling of invalid airport codes."""
    if not HAS_V3:
        results.add_fail("Invalid Airport", "v3 not installed")
        return

    try:
        test_date = (date.today() + timedelta(days=30)).strftime('%Y-%m-%d')

        # XXX is not a valid IATA code
        flights = await search_direct_flights("XXX", "BRI", test_date)

        # Should return empty or handle gracefully, not crash
        assert isinstance(flights, list)

        logger.info(f"  Gracefully handled invalid airport code")
        results.add_pass("Invalid Airport")

    except Exception as e:
        results.add_fail("Invalid Airport", str(e))


async def test_concurrent_requests():
    """Test that concurrent requests work properly."""
    if not HAS_V3:
        results.add_fail("Concurrent Requests", "v3 not installed")
        return

    try:
        test_date = (date.today() + timedelta(days=30)).strftime('%Y-%m-%d')

        # 10 concurrent requests
        routes = [
            ("BER", "BRI", test_date),
            ("FRA", "FCO", test_date),
            ("MUC", "VIE", test_date),
            ("BER", "CPH", test_date),
            ("FRA", "AMS", test_date),
            ("MUC", "ZRH", test_date),
            ("BER", "VIE", test_date),
            ("FRA", "BRU", test_date),
            ("MUC", "CDG", test_date),
            ("BER", "AMS", test_date),
        ]

        import time
        start = time.time()

        batch_results = await search_multiple_routes(
            routes,
            seat_class="economy",
            adults=1,
            max_workers=5,
        )

        elapsed = time.time() - start

        flights_found = sum(1 for flights in batch_results.values() if flights)

        # Should complete reasonably fast with concurrency
        if flights_found >= 5 and elapsed < 60:
            logger.info(f"  {flights_found}/10 routes successful in {elapsed:.1f}s")
            results.add_pass("Concurrent Requests")
        else:
            results.add_fail("Concurrent Requests", f"Only {flights_found}/10 in {elapsed:.1f}s")

    except Exception as e:
        results.add_fail("Concurrent Requests", str(e))


async def test_price_range():
    """Test that prices are reasonable."""
    if not HAS_V3:
        results.add_fail("Price Range", "v3 not installed")
        return

    try:
        test_date = (date.today() + timedelta(days=30)).strftime('%Y-%m-%d')
        flights = await search_direct_flights("BER", "BRI", test_date)

        if flights:
            prices = [f['price'] for f in flights if 'price' in f]

            if prices:
                min_price = min(prices)
                max_price = max(prices)

                # Sanity check: prices should be between 20 and 1000 EUR for EU routes
                if 20 <= min_price <= 1000 and 20 <= max_price <= 1000:
                    logger.info(f"  Price range: €{min_price} - €{max_price}")
                    results.add_pass("Price Range")
                else:
                    results.add_fail("Price Range", f"Unreasonable prices: €{min_price} - €{max_price}")
            else:
                results.add_fail("Price Range", "No prices found in results")
        else:
            results.add_fail("Price Range", "No flights to check prices")

    except Exception as e:
        results.add_fail("Price Range", str(e))


async def run_all_tests():
    """Run all tests."""
    logger.info("╔" + "="*78 + "╗")
    logger.info("║" + " "*15 + "COMPREHENSIVE TEST SUITE - fast-flights v3.0rc1" + " "*14 + "║")
    logger.info("╚" + "="*78 + "╝\n")

    if not HAS_V3:
        logger.error("fast-flights v3.0rc1 not installed!")
        logger.error("Install with: pip install --upgrade git+https://github.com/AWeirdDev/flights.git")
        return False

    # Unit tests
    logger.info("\n" + "-"*80)
    logger.info("UNIT TESTS")
    logger.info("-"*80)
    test_socs_integration()

    # Integration tests
    logger.info("\n" + "-"*80)
    logger.info("INTEGRATION TESTS")
    logger.info("-"*80)

    await test_single_route_ber_bri()
    await asyncio.sleep(2)  # Rate limiting

    await test_multiple_routes()
    await asyncio.sleep(2)

    await test_different_dates()
    await asyncio.sleep(2)

    await test_no_direct_flights()
    await asyncio.sleep(2)

    await test_invalid_airport_code()
    await asyncio.sleep(2)

    # Stress tests
    logger.info("\n" + "-"*80)
    logger.info("STRESS TESTS")
    logger.info("-"*80)

    await test_concurrent_requests()
    await asyncio.sleep(2)

    # Validation tests
    logger.info("\n" + "-"*80)
    logger.info("VALIDATION TESTS")
    logger.info("-"*80)

    await test_price_range()

    # Summary
    return results.summary()


if __name__ == "__main__":
    success = asyncio.run(run_all_tests())

    logger.info("\n" + "="*80)
    if success:
        logger.info("✅ ALL TESTS PASSED!")
    else:
        logger.info("⚠️  SOME TESTS FAILED - See summary above")
    logger.info("="*80)

    sys.exit(0 if success else 1)