"""Property-based tests for the PAT Manager alert sender module using Hypothesis.""" from __future__ import annotations import datetime from hypothesis import given, settings from hypothesis import strategies as st from scripts.pat_manager.alerter import AlertSender from scripts.pat_manager.models import ( AlertChannel, CheckResult, PatEntry, PatStatus, ) # --- Strategies --- SERVICES = ["GitLab", "Jira", "Confluence", "OrgMyLife"] RENEWAL_METHODS = ["auto", "manual"] ALL_STATUSES = [PatStatus.HEALTHY, PatStatus.EXPIRING_SOON, PatStatus.EXPIRED, PatStatus.CHECK_FAILED] # Services that require manual renewal (used for alert testing) manual_services = st.sampled_from(["Jira", "Confluence", "OrgMyLife"]) # Statuses that trigger alerts alert_statuses = st.sampled_from([PatStatus.EXPIRING_SOON, PatStatus.EXPIRED]) # Token names: non-empty strings up to 128 chars (printable, no control chars) token_names = st.text( alphabet=st.characters(whitelist_categories=("L", "N", "P", "S"), min_codepoint=32), min_size=1, max_size=128, ).filter(lambda s: s.strip() != "") # Days remaining: 0 for expired, 1-14 for expiring soon days_remaining_values = st.integers(min_value=0, max_value=14) # Expiry dates in ISO 8601 format expiry_dates = st.dates( min_value=datetime.date(2020, 1, 1), max_value=datetime.date(2030, 12, 31), ).map(lambda d: d.isoformat()) def pat_entries() -> st.SearchStrategy[PatEntry]: """Generate valid PatEntry instances with random services and token names.""" return st.builds( PatEntry, service=st.sampled_from(SERVICES), token_name=st.text( alphabet=st.characters(whitelist_categories=("L", "N", "P")), min_size=1, max_size=64, ), expiry_date=st.one_of( st.none(), st.dates().map(lambda d: d.isoformat()), ), renewal_method=st.sampled_from(RENEWAL_METHODS), ) # --- Additional imports for Property 9 --- from datetime import date from scripts.pat_manager.checker import classify_by_date # --- Strategies for Property 9 --- # Generate ISO 8601 date strings in range 2020-2099 prop9_expiry_dates = st.dates( min_value=date(2020, 1, 1), max_value=date(2099, 12, 31), ).map(lambda d: d.isoformat()) # Generate reference dates in range 2020-2099 prop9_reference_dates = st.dates( min_value=date(2020, 1, 1), max_value=date(2099, 12, 31), ) # --- Property 9 Test --- # Feature: pat-renewal, Property 9: Days remaining calculation class TestDaysRemainingCalculation: """ **Validates: Requirements 4.4** Property 9: For any expiry date and reference date, the computed days_remaining SHALL equal max(0, (expiry_date - reference_date).days) — yielding 0 for already-expired tokens and the correct positive integer for future dates. """ @given(expiry_date=prop9_expiry_dates, reference_date=prop9_reference_dates) @settings(max_examples=100) def test_days_remaining_equals_max_zero_delta( self, expiry_date: str, reference_date: date ) -> None: """Days remaining is always max(0, (expiry - ref).days).""" _status, days_remaining = classify_by_date( expiry_date, reference_date, window=14 ) parsed_expiry = date.fromisoformat(expiry_date) expected = max(0, (parsed_expiry - reference_date).days) assert days_remaining == expected def healthy_check_results() -> st.SearchStrategy[CheckResult]: """Generate CheckResult instances that are all HEALTHY.""" return st.builds( CheckResult, entry=pat_entries(), status=st.just(PatStatus.HEALTHY), days_remaining=st.integers(min_value=15, max_value=365), error_message=st.none(), stale_warning=st.just(False), ) def check_results_any_status() -> st.SearchStrategy[CheckResult]: """Generate CheckResult instances with any valid status.""" return st.builds( CheckResult, entry=pat_entries(), status=st.sampled_from(ALL_STATUSES), days_remaining=st.one_of( st.none(), st.integers(min_value=0, max_value=365), ), error_message=st.one_of(st.none(), st.text(min_size=1, max_size=50)), stale_warning=st.booleans(), ) @st.composite def check_results_requiring_alert(draw: st.DrawFn) -> CheckResult: """Generate CheckResult objects that require alerts (manual renewal, expiring/expired).""" service = draw(manual_services) token_name = draw(token_names) status = draw(alert_statuses) days = draw(days_remaining_values) expiry_date = draw(expiry_dates) # Align days_remaining with status if status == PatStatus.EXPIRED: days = 0 else: days = max(1, days) # expiring_soon should have at least 1 day entry = PatEntry( service=service, token_name=token_name, expiry_date=expiry_date, renewal_method="manual", ) return CheckResult( entry=entry, status=status, days_remaining=days, error_message=None, ) # --- Helper: Alert decision logic --- def needs_alert(result: CheckResult) -> bool: """Determine if a CheckResult should trigger an alert. A PAT needs an alert if: - Its status is EXPIRING_SOON or EXPIRED - AND its renewal_method is "manual" GitLab tokens with auto renewal that failed rotation would also need alerts, but that is handled by the orchestrator flow (rotation failure falls back to alert). For this property test, we focus on the direct filtering logic based on status and renewal_method. """ if result.status not in (PatStatus.EXPIRING_SOON, PatStatus.EXPIRED): return False if result.entry.renewal_method == "manual": return True return False # --- Property 7 Tests --- # Feature: pat-renewal, Property 7: Alert triggering correctness class TestAlertTriggeringCorrectness: """ **Validates: Requirements 4.1, 5.4** Property 7: Alert triggering correctness - If ALL PATs have status "healthy", THEN zero alerts SHALL be sent - For each manual-renewal PAT with status "expiring soon" or "expired", exactly one alert SHALL be triggered """ @given(results=st.lists(healthy_check_results(), min_size=0, max_size=20)) @settings(max_examples=100) def test_all_healthy_means_zero_alerts( self, results: list[CheckResult] ) -> None: """If all PATs are healthy, zero alerts should be triggered. This verifies Requirement 5.4: IF all PATs are classified as "healthy", THEN the PAT_Manager SHALL complete without creating OrgMyLife tasks or sending email alerts. """ alerts_needed = [r for r in results if needs_alert(r)] assert len(alerts_needed) == 0, ( f"Expected zero alerts for all-healthy set, got {len(alerts_needed)}" ) @given(results=st.lists(check_results_any_status(), min_size=1, max_size=20)) @settings(max_examples=100) def test_one_alert_per_expiring_expired_manual_token( self, results: list[CheckResult] ) -> None: """Each manual-renewal PAT with status expiring/expired triggers exactly one alert. This verifies Requirement 4.1: WHEN a Manual_Renewal_Token has 14 or fewer days until expiration or is already expired, THE PAT_Manager SHALL send an alert via the configured Alert_Channel. The count of alerts needed must equal the count of results that are manual-renewal AND have status EXPIRING_SOON or EXPIRED. """ alerts_needed = [r for r in results if needs_alert(r)] # Count expected: manual tokens with expiring_soon or expired status expected_count = sum( 1 for r in results if r.entry.renewal_method == "manual" and r.status in (PatStatus.EXPIRING_SOON, PatStatus.EXPIRED) ) assert len(alerts_needed) == expected_count, ( f"Expected {expected_count} alerts, got {len(alerts_needed)}. " f"Results: {[(r.entry.renewal_method, r.status.value) for r in results]}" ) @given(results=st.lists(check_results_any_status(), min_size=1, max_size=20)) @settings(max_examples=100) def test_healthy_tokens_never_trigger_alerts( self, results: list[CheckResult] ) -> None: """No token with HEALTHY status should ever trigger an alert, regardless of its renewal_method.""" alerts_needed = [r for r in results if needs_alert(r)] for alert in alerts_needed: assert alert.status != PatStatus.HEALTHY, ( f"Healthy token should not trigger alert: " f"{alert.entry.service}/{alert.entry.token_name}" ) @given(results=st.lists(check_results_any_status(), min_size=1, max_size=20)) @settings(max_examples=100) def test_auto_renewal_tokens_do_not_trigger_alerts( self, results: list[CheckResult] ) -> None: """Tokens with renewal_method "auto" should not trigger alerts through the standard filtering logic (they go through rotation first).""" alerts_needed = [r for r in results if needs_alert(r)] for alert in alerts_needed: assert alert.entry.renewal_method != "auto", ( f"Auto-renewal token should not trigger alert via filtering: " f"{alert.entry.service}/{alert.entry.token_name}" ) @given(results=st.lists(check_results_any_status(), min_size=1, max_size=20)) @settings(max_examples=100) def test_check_failed_tokens_do_not_trigger_alerts( self, results: list[CheckResult] ) -> None: """Tokens with CHECK_FAILED status should not trigger alerts (they are reported differently in the summary).""" alerts_needed = [r for r in results if needs_alert(r)] for alert in alerts_needed: assert alert.status != PatStatus.CHECK_FAILED, ( f"Check-failed token should not trigger alert: " f"{alert.entry.service}/{alert.entry.token_name}" ) # --- Property 8 Tests --- # Feature: pat-renewal, Property 8: Alert message completeness class TestAlertMessageCompleteness: """ **Validates: Requirements 4.2, 4.3** Property 8: Alert message completeness - If the alert channel is OrgMyLife, the task title SHALL contain both the service name and token identifier. - The description SHALL contain service name, token identifier, and expiry information. """ @given(result=check_results_requiring_alert()) @settings(max_examples=100) def test_orgmylife_task_title_contains_service_and_token( self, result: CheckResult ) -> None: """OrgMyLife task title contains both service name and token identifier.""" sender = AlertSender( AlertChannel.ORGMYLIFE, {"orgmylife_url": "https://orgmylife.app", "orgmylife_api_key": "test-key"}, ) title = sender._build_task_title(result) # Title SHALL contain both the service name and token identifier assert result.entry.service in title, ( f"Title '{title}' does not contain service name '{result.entry.service}'" ) assert result.entry.token_name in title, ( f"Title '{title}' does not contain token name '{result.entry.token_name}'" ) @given(result=check_results_requiring_alert()) @settings(max_examples=100) def test_orgmylife_task_description_contains_required_fields( self, result: CheckResult ) -> None: """OrgMyLife task description contains service name, token name, and days remaining info.""" sender = AlertSender( AlertChannel.ORGMYLIFE, {"orgmylife_url": "https://orgmylife.app", "orgmylife_api_key": "test-key"}, ) description = sender._build_task_description(result) # Description SHALL contain service name assert result.entry.service in description, ( f"Description does not contain service name '{result.entry.service}'" ) # Description SHALL contain token identifier assert result.entry.token_name in description, ( f"Description does not contain token name '{result.entry.token_name}'" ) # Description SHALL contain days remaining information days = result.days_remaining if result.days_remaining is not None else 0 if days == 0: assert "0 days remaining" in description, ( f"Description does not contain '0 days remaining' for expired token" ) else: assert str(days) in description, ( f"Description does not contain days remaining '{days}'" ) # --- Additional imports for Property 10 --- import respx from httpx import Response # --- Property 10 Test --- # Feature: pat-renewal, Property 10: Alert idempotence class TestAlertIdempotence: """ **Validates: Requirements 4.5** Property 10: For any PAT that requires an alert, if an open OrgMyLife task already exists for the same service and token_name, THEN no new task SHALL be created. Running the alert flow twice for the same PAT state SHALL produce at most one task. """ @given( service=manual_services, token_name=token_names, expiry_date=expiry_dates, status=alert_statuses, days_remaining=days_remaining_values, ) @settings(max_examples=100, deadline=None) async def test_no_duplicate_task_when_open_task_exists( self, service: str, token_name: str, expiry_date: str, status: PatStatus, days_remaining: int, ) -> None: """When an open task already exists for a PAT, no new task is created.""" entry = PatEntry( service=service, token_name=token_name, expiry_date=expiry_date, renewal_method="manual", ) check_result = CheckResult( entry=entry, status=status, days_remaining=days_remaining, error_message=None, ) config = { "orgmylife_url": "https://orgmylife.app", "orgmylife_api_key": "test-api-key", } sender = AlertSender(AlertChannel.ORGMYLIFE, config) expected_title = f"[PAT Renewal] {service} - {token_name}" with respx.mock: # Mock GET: return a list containing a task with matching title respx.get("https://orgmylife.app/api/tasks").mock( return_value=Response( 200, json=[{"title": expected_title, "status": "open"}], ) ) # Mock POST: should NOT be called post_route = respx.post("https://orgmylife.app/api/tasks").mock( return_value=Response(201, json={"id": 1}) ) result = await sender.send_alert(check_result) # Alert was not sent (no new task created) assert result.sent is False # POST endpoint was never called (no duplicate task creation) assert not post_route.called