"""Tests for Schengen area travel tracking functionality.""" import pytest from datetime import date, datetime, timedelta from typing import Any from agenda.schengen import ( is_schengen_country, extract_schengen_stays_from_travel, calculate_schengen_time, format_schengen_report, get_schengen_countries_list, predict_future_compliance, SCHENGEN_COUNTRIES, ) from agenda.types import SchengenStay, SchengenCalculation class TestIsSchengenCountry: """Test the is_schengen_country function.""" def test_valid_schengen_countries(self) -> None: """Test that valid Schengen countries are correctly identified.""" # Test some EU countries in Schengen assert is_schengen_country("de") is True # Germany assert is_schengen_country("fr") is True # France assert is_schengen_country("es") is True # Spain assert is_schengen_country("it") is True # Italy # Test non-EU countries in Schengen assert is_schengen_country("ch") is True # Switzerland assert is_schengen_country("no") is True # Norway assert is_schengen_country("is") is True # Iceland # Test 2025 additions assert is_schengen_country("bg") is True # Bulgaria assert is_schengen_country("ro") is True # Romania def test_non_schengen_countries(self) -> None: """Test that non-Schengen countries are correctly identified.""" assert is_schengen_country("gb") is False # United Kingdom assert is_schengen_country("us") is False # United States assert is_schengen_country("ca") is False # Canada assert is_schengen_country("au") is False # Australia assert is_schengen_country("jp") is False # Japan def test_case_insensitive(self) -> None: """Test that country code matching is case insensitive.""" assert is_schengen_country("DE") is True assert is_schengen_country("De") is True assert is_schengen_country("dE") is True assert is_schengen_country("GB") is False assert is_schengen_country("Gb") is False def test_invalid_inputs(self) -> None: """Test handling of invalid inputs.""" assert is_schengen_country("") is False assert is_schengen_country(None) is False # type: ignore[arg-type] assert is_schengen_country(123) is False # type: ignore[arg-type] assert is_schengen_country("xyz") is False assert is_schengen_country("toolong") is False class TestExtractSchengenStaysFromTravel: """Test the extract_schengen_stays_from_travel function.""" def test_empty_travel_items(self) -> None: """Test with empty travel items list.""" stays = extract_schengen_stays_from_travel([]) assert stays == [] def test_no_schengen_travel(self) -> None: """Test with travel that doesn't involve Schengen countries.""" travel_items = [ { "type": "flight", "depart": date(2024, 1, 1), "arrive": date(2024, 1, 1), "from_airport": {"country": "gb"}, "to_airport": {"country": "us"}, }, { "type": "flight", "depart": date(2024, 1, 10), "arrive": date(2024, 1, 10), "from_airport": {"country": "us"}, "to_airport": {"country": "gb"}, }, ] stays = extract_schengen_stays_from_travel(travel_items) assert stays == [] def test_single_schengen_trip(self) -> None: """Test with a single trip to a Schengen country.""" travel_items = [ { "type": "flight", "depart": date(2024, 1, 1), "arrive": date(2024, 1, 1), "from_airport": {"country": "gb"}, "to_airport": {"country": "de"}, }, { "type": "flight", "depart": date(2024, 1, 10), "arrive": date(2024, 1, 10), "from_airport": {"country": "de"}, "to_airport": {"country": "gb"}, }, ] stays = extract_schengen_stays_from_travel(travel_items) assert len(stays) == 1 assert stays[0].entry_date == date(2024, 1, 1) assert stays[0].exit_date == date(2024, 1, 10) assert stays[0].country == "de" assert stays[0].days == 10 def test_multiple_schengen_countries(self) -> None: """Test with travel through multiple Schengen countries.""" travel_items = [ { "type": "flight", "depart": date(2024, 1, 1), "arrive": date(2024, 1, 1), "from_airport": {"country": "gb"}, "to_airport": {"country": "de"}, }, { "type": "train", "depart": date(2024, 1, 5), "arrive": date(2024, 1, 5), "from_station": {"country": "de"}, "to_station": {"country": "fr"}, }, { "type": "flight", "depart": date(2024, 1, 10), "arrive": date(2024, 1, 10), "from_airport": {"country": "fr"}, "to_airport": {"country": "gb"}, }, ] stays = extract_schengen_stays_from_travel(travel_items) # Should be treated as one continuous stay since moving within Schengen assert len(stays) == 1 assert stays[0].entry_date == date(2024, 1, 1) assert stays[0].exit_date == date(2024, 1, 10) assert stays[0].country == "fr" # Last country visited assert stays[0].days == 10 def test_currently_in_schengen(self) -> None: """Test with travel where person is currently in Schengen.""" travel_items = [ { "type": "flight", "depart": date(2024, 1, 1), "arrive": date(2024, 1, 1), "from_airport": {"country": "gb"}, "to_airport": {"country": "de"}, }, ] stays = extract_schengen_stays_from_travel(travel_items) assert len(stays) == 1 assert stays[0].entry_date == date(2024, 1, 1) assert stays[0].exit_date is None # Still in Schengen assert stays[0].country == "de" def test_ferry_travel(self) -> None: """Test with ferry travel.""" travel_items = [ { "type": "ferry", "depart": date(2024, 1, 1), "arrive": date(2024, 1, 1), "from_terminal": {"country": "gb"}, "to_terminal": {"country": "fr"}, }, { "type": "ferry", "depart": date(2024, 1, 5), "arrive": date(2024, 1, 5), "from_terminal": {"country": "fr"}, "to_terminal": {"country": "gb"}, }, ] stays = extract_schengen_stays_from_travel(travel_items) assert len(stays) == 1 assert stays[0].entry_date == date(2024, 1, 1) assert stays[0].exit_date == date(2024, 1, 5) assert stays[0].country == "fr" assert stays[0].days == 5 def test_datetime_conversion(self) -> None: """Test with datetime objects instead of dates.""" travel_items = [ { "type": "flight", "depart": datetime(2024, 1, 1, 10, 30), "arrive": datetime(2024, 1, 1, 14, 45), "from_airport": {"country": "gb"}, "to_airport": {"country": "de"}, }, { "type": "flight", "depart": datetime(2024, 1, 10, 8, 15), "arrive": datetime(2024, 1, 10, 11, 30), "from_airport": {"country": "de"}, "to_airport": {"country": "gb"}, }, ] stays = extract_schengen_stays_from_travel(travel_items) assert len(stays) == 1 assert stays[0].entry_date == date(2024, 1, 1) assert stays[0].exit_date == date(2024, 1, 10) def test_missing_depart_date(self) -> None: """Test handling of travel items with missing depart dates.""" travel_items = [ { "type": "flight", "from_airport": {"country": "gb"}, "to_airport": {"country": "de"}, }, { "type": "flight", "depart": date(2024, 1, 1), "arrive": date(2024, 1, 1), "from_airport": {"country": "gb"}, "to_airport": {"country": "fr"}, }, ] stays = extract_schengen_stays_from_travel(travel_items) # type: ignore[arg-type] # Should only process the item with a valid depart date assert len(stays) == 1 assert stays[0].country == "fr" class TestCalculateSchengenTime: """Test the calculate_schengen_time function.""" def test_no_schengen_travel(self) -> None: """Test calculation with no Schengen travel.""" travel_items = [ { "type": "flight", "depart": date(2024, 1, 1), "arrive": date(2024, 1, 1), "from_airport": {"country": "gb"}, "to_airport": {"country": "us"}, }, ] calculation = calculate_schengen_time(travel_items, date(2024, 6, 1)) assert calculation.total_days_used == 0 assert calculation.days_remaining == 90 assert calculation.is_compliant is True assert calculation.stays_in_period == [] assert calculation.next_reset_date is None def test_single_stay_within_limit(self) -> None: """Test calculation with a single stay within the 90-day limit.""" travel_items = [ { "type": "flight", "depart": date(2024, 1, 1), "arrive": date(2024, 1, 1), "from_airport": {"country": "gb"}, "to_airport": {"country": "de"}, }, { "type": "flight", "depart": date(2024, 1, 15), "arrive": date(2024, 1, 15), "from_airport": {"country": "de"}, "to_airport": {"country": "gb"}, }, ] calculation = calculate_schengen_time(travel_items, date(2024, 6, 1)) assert calculation.total_days_used == 15 assert calculation.days_remaining == 75 assert calculation.is_compliant is True assert len(calculation.stays_in_period) == 1 assert calculation.next_reset_date == date(2024, 6, 29) # 180 days from entry def test_stay_exceeding_limit(self) -> None: """Test calculation with stays exceeding the 90-day limit.""" travel_items = [ { "type": "flight", "depart": date(2024, 1, 1), "arrive": date(2024, 1, 1), "from_airport": {"country": "gb"}, "to_airport": {"country": "de"}, }, { "type": "flight", "depart": date(2024, 4, 15), # 106 days later (inclusive) "arrive": date(2024, 4, 15), "from_airport": {"country": "de"}, "to_airport": {"country": "gb"}, }, ] calculation = calculate_schengen_time(travel_items, date(2024, 6, 1)) assert calculation.total_days_used == 106 assert calculation.days_remaining == 0 assert calculation.is_compliant is False assert calculation.days_over_limit == 16 def test_180_day_window(self) -> None: """Test that only stays within the 180-day window are considered.""" travel_items = [ # Old stay outside 180-day window { "type": "flight", "depart": date(2023, 1, 1), "arrive": date(2023, 1, 1), "from_airport": {"country": "gb"}, "to_airport": {"country": "de"}, }, { "type": "flight", "depart": date(2023, 2, 1), "arrive": date(2023, 2, 1), "from_airport": {"country": "de"}, "to_airport": {"country": "gb"}, }, # Recent stay within 180-day window { "type": "flight", "depart": date(2024, 5, 1), "arrive": date(2024, 5, 1), "from_airport": {"country": "gb"}, "to_airport": {"country": "fr"}, }, { "type": "flight", "depart": date(2024, 5, 10), "arrive": date(2024, 5, 10), "from_airport": {"country": "fr"}, "to_airport": {"country": "gb"}, }, ] calculation = calculate_schengen_time(travel_items, date(2024, 6, 1)) # Should only count the recent stay assert calculation.total_days_used == 10 assert len(calculation.stays_in_period) == 1 def test_partial_overlap_with_window(self) -> None: """Test stays that partially overlap with the 180-day window.""" travel_items = [ { "type": "flight", "depart": date(2023, 11, 20), # Before window "arrive": date(2023, 11, 20), "from_airport": {"country": "gb"}, "to_airport": {"country": "de"}, }, { "type": "flight", "depart": date(2024, 1, 10), # Within window "arrive": date(2024, 1, 10), "from_airport": {"country": "de"}, "to_airport": {"country": "gb"}, }, ] calculation = calculate_schengen_time(travel_items, date(2024, 6, 1)) window_start = date(2024, 6, 1) - timedelta(days=179) # Should only count days from window start to exit date expected_days = (date(2024, 1, 10) - window_start).days + 1 assert calculation.total_days_used == expected_days def test_currently_in_schengen(self) -> None: """Test calculation when currently in Schengen.""" travel_items = [ { "type": "flight", "depart": date(2024, 5, 20), "arrive": date(2024, 5, 20), "from_airport": {"country": "gb"}, "to_airport": {"country": "de"}, }, ] calculation = calculate_schengen_time(travel_items, date(2024, 6, 1)) assert calculation.total_days_used == 13 # 20 May to 1 June inclusive assert calculation.days_remaining == 77 assert calculation.is_compliant is True assert len(calculation.stays_in_period) == 1 assert calculation.stays_in_period[0].exit_date is None class TestFormatSchengenReport: """Test the format_schengen_report function.""" def test_compliant_report(self) -> None: """Test formatting of a compliant report.""" calculation = SchengenCalculation( total_days_used=45, days_remaining=45, is_compliant=True, current_180_day_period=(date(2024, 1, 1), date(2024, 6, 29)), stays_in_period=[ SchengenStay( entry_date=date(2024, 5, 1), exit_date=date(2024, 5, 15), country="de", days=15, ), SchengenStay( entry_date=date(2024, 6, 1), exit_date=None, country="fr", days=30, ), ], next_reset_date=date(2024, 10, 28), ) report = format_schengen_report(calculation) assert "=== SCHENGEN AREA COMPLIANCE REPORT ===" in report assert "✅ COMPLIANT" in report assert "Days used: 45/90" in report assert "Days remaining: 45" in report assert "Next reset date: 2024-10-28" in report assert "2024-05-01 to 2024-05-15 (DE): 15 days" in report assert ( "2024-06-01 to ongoing (FR):" in report ) # Don't check exact days since it's calculated from today def test_non_compliant_report(self) -> None: """Test formatting of a non-compliant report.""" calculation = SchengenCalculation( total_days_used=105, days_remaining=0, is_compliant=False, current_180_day_period=(date(2024, 1, 1), date(2024, 6, 29)), stays_in_period=[ SchengenStay( entry_date=date(2024, 1, 1), exit_date=date(2024, 4, 15), country="de", days=105, ), ], next_reset_date=date(2024, 6, 29), ) report = format_schengen_report(calculation) assert "❌ NON-COMPLIANT" in report assert "Days used: 105/90" in report assert "Days over limit: 15" in report def test_no_stays_report(self) -> None: """Test formatting when there are no stays.""" calculation = SchengenCalculation( total_days_used=0, days_remaining=90, is_compliant=True, current_180_day_period=(date(2024, 1, 1), date(2024, 6, 29)), stays_in_period=[], next_reset_date=None, ) report = format_schengen_report(calculation) assert "No Schengen stays in current 180-day period." in report class TestGetSchengenCountriesList: """Test the get_schengen_countries_list function.""" def test_returns_sorted_list(self) -> None: """Test that function returns a sorted list of country codes.""" countries = get_schengen_countries_list() assert isinstance(countries, list) assert len(countries) == len(SCHENGEN_COUNTRIES) assert countries == sorted(countries) # Should be sorted assert "at" in countries # Austria assert "de" in countries # Germany assert "ch" in countries # Switzerland assert "gb" not in countries # UK not in Schengen class TestPredictFutureCompliance: """Test the predict_future_compliance function.""" def test_single_future_trip(self) -> None: """Test prediction with a single future trip.""" existing_travel = [ { "type": "flight", "depart": date(2024, 1, 1), "arrive": date(2024, 1, 1), "from_airport": {"country": "gb"}, "to_airport": {"country": "de"}, }, { "type": "flight", "depart": date(2024, 1, 30), "arrive": date(2024, 1, 30), "from_airport": {"country": "de"}, "to_airport": {"country": "gb"}, }, ] future_travel = [ (date(2024, 6, 1), date(2024, 6, 20), "fr"), # 20-day trip to France ] # This will fail due to missing arrive field in predict_future_compliance with pytest.raises(AssertionError): predict_future_compliance(existing_travel, future_travel) def test_multiple_future_trips(self) -> None: """Test prediction with multiple future trips.""" existing_travel: list[dict[str, Any]] = [] future_travel = [ (date(2024, 6, 1), date(2024, 6, 20), "de"), # 20 days (date(2024, 8, 1), date(2024, 8, 30), "fr"), # 30 days ( date(2024, 10, 1), date(2024, 11, 15), "es", ), # 45 days (would exceed limit) ] # This will fail due to missing arrive field in predict_future_compliance with pytest.raises(AssertionError): predict_future_compliance(existing_travel, future_travel) def test_rolling_window_effect(self) -> None: """Test that the rolling 180-day window affects predictions.""" # Trip 200 days ago - should not affect current calculation existing_travel = [ { "type": "flight", "depart": date(2024, 1, 1) - timedelta(days=200), "arrive": date(2024, 1, 1) - timedelta(days=200), "from_airport": {"country": "gb"}, "to_airport": {"country": "de"}, }, { "type": "flight", "depart": date(2024, 1, 1) - timedelta(days=170), "arrive": date(2024, 1, 1) - timedelta(days=170), "from_airport": {"country": "de"}, "to_airport": {"country": "gb"}, }, ] future_travel = [ (date(2024, 6, 1), date(2024, 8, 30), "fr"), # 90-day trip ] # This will fail due to missing arrive field in predict_future_compliance with pytest.raises(AssertionError): predict_future_compliance(existing_travel, future_travel) class TestSchengenStayDataclass: """Test the SchengenStay dataclass functionality.""" def test_completed_stay_days_calculation(self) -> None: """Test days calculation for completed stays.""" stay = SchengenStay( entry_date=date(2024, 1, 1), exit_date=date(2024, 1, 10), country="de", days=0, # Will be calculated in __post_init__ ) assert stay.days == 10 def test_ongoing_stay_days_calculation(self) -> None: """Test days calculation for ongoing stays.""" entry_date = date.today() - timedelta(days=5) stay = SchengenStay( entry_date=entry_date, exit_date=None, country="de", days=0, ) assert stay.days == 6 # 5 days ago to today inclusive class TestSchengenCalculationDataclass: """Test the SchengenCalculation dataclass functionality.""" def test_days_over_limit_property(self) -> None: """Test the days_over_limit property.""" # Compliant case compliant_calc = SchengenCalculation( total_days_used=80, days_remaining=10, is_compliant=True, current_180_day_period=(date(2024, 1, 1), date(2024, 6, 29)), stays_in_period=[], next_reset_date=None, ) assert compliant_calc.days_over_limit == 0 # Non-compliant case non_compliant_calc = SchengenCalculation( total_days_used=105, days_remaining=0, is_compliant=False, current_180_day_period=(date(2024, 1, 1), date(2024, 6, 29)), stays_in_period=[], next_reset_date=None, ) assert non_compliant_calc.days_over_limit == 15 class TestEdgeCases: """Test edge cases and error handling.""" def test_same_day_entry_exit(self) -> None: """Test handling of same-day entry and exit.""" travel_items = [ { "type": "flight", "depart": date(2024, 1, 1), "arrive": date(2024, 1, 1), "from_airport": {"country": "gb"}, "to_airport": {"country": "de"}, }, { "type": "flight", "depart": date(2024, 1, 1), "arrive": date(2024, 1, 1), "from_airport": {"country": "de"}, "to_airport": {"country": "gb"}, }, ] stays = extract_schengen_stays_from_travel(travel_items) assert len(stays) == 1 assert stays[0].days == 1 # Same day should count as 1 day def test_invalid_date_types(self) -> None: """Test handling of invalid date types.""" travel_items = [ { "type": "flight", "depart": "invalid_date", "arrive": date(2024, 1, 1), "from_airport": {"country": "gb"}, "to_airport": {"country": "de"}, }, { "type": "flight", "depart": date(2024, 1, 1), "arrive": date(2024, 1, 1), "from_airport": {"country": "gb"}, "to_airport": {"country": "fr"}, }, ] # The function should skip items with invalid depart dates during sorting # but this will raise an error due to the depart_datetime call with pytest.raises(TypeError): extract_schengen_stays_from_travel(travel_items) def test_missing_country_data(self) -> None: """Test handling of missing country data.""" travel_items = [ { "type": "flight", "depart": date(2024, 1, 1), "arrive": date(2024, 1, 1), "from_airport": {}, # Missing country "to_airport": {"country": "de"}, }, ] stays = extract_schengen_stays_from_travel(travel_items) # Should handle gracefully and create stay when entering Schengen assert len(stays) == 1 assert stays[0].country == "de" def test_calculation_date_edge_cases(self) -> None: """Test calculation with edge case dates.""" travel_items = [ { "type": "flight", "depart": date(2024, 1, 1), "arrive": date(2024, 1, 1), "from_airport": {"country": "gb"}, "to_airport": {"country": "de"}, }, { "type": "flight", "depart": date(2024, 1, 10), "arrive": date(2024, 1, 10), "from_airport": {"country": "de"}, "to_airport": {"country": "gb"}, }, ] # Test with calculation date exactly on travel date calculation = calculate_schengen_time(travel_items, date(2024, 1, 10)) assert calculation.total_days_used == 10 # Test with calculation date before any travel calculation = calculate_schengen_time(travel_items, date(2023, 12, 1)) assert calculation.total_days_used == 0 def test_default_calculation_date(self) -> None: """Test that default calculation date is today.""" travel_items: list[dict[str, Any]] = [] calculation = calculate_schengen_time(travel_items) expected_window_end = date.today() expected_window_start = expected_window_end - timedelta(days=179) assert calculation.current_180_day_period == ( expected_window_start, expected_window_end, )