Orchestrator/shared/AI-Orchestrator/tests/test_registry_properties.py

"""Property-based tests for the PAT Registry module using Hypothesis."""

from __future__ import annotations

import json
from pathlib import Path

import pytest
from hypothesis import given, settings, assume
from hypothesis import strategies as st

from scripts.pat_manager.errors import RegistryValidationError
from scripts.pat_manager.models import PatEntry
from scripts.pat_manager.registry import PatRegistry, VALID_SERVICES


# --- Strategies ---

# Valid service names and their required renewal methods
_SERVICES_AND_METHODS = [
    ("GitLab", "auto"),
    ("Jira", "manual"),
    ("Confluence", "manual"),
    ("OrgMyLife", "manual"),
]


@st.composite
def valid_iso_dates(draw: st.DrawFn) -> str:
    """Generate valid ISO 8601 date strings (YYYY-MM-DD)."""
    year = draw(st.integers(min_value=2020, max_value=2099))
    month = draw(st.integers(min_value=1, max_value=12))
    # Limit days based on month to avoid invalid dates
    if month in (1, 3, 5, 7, 8, 10, 12):
        max_day = 31
    elif month in (4, 6, 9, 11):
        max_day = 30
    elif month == 2:
        # Simplified leap year check
        if year % 4 == 0 and (year % 100 != 0 or year % 400 == 0):
            max_day = 29
        else:
            max_day = 28
    else:
        max_day = 28
    day = draw(st.integers(min_value=1, max_value=max_day))
    return f"{year:04d}-{month:02d}-{day:02d}"


@st.composite
def valid_token_names(draw: st.DrawFn) -> str:
    """Generate valid token names (1-128 chars, printable ASCII)."""
    name = draw(
        st.text(
            alphabet=st.characters(
                whitelist_categories=("L", "N", "P", "S"),
                whitelist_characters="-_",
            ),
            min_size=1,
            max_size=128,
        )
    )
    # Ensure no control characters or empty after strip
    assume(len(name.strip()) > 0)
    return name


@st.composite
def valid_pat_entries(draw: st.DrawFn) -> PatEntry:
    """Generate a valid PatEntry with consistent service/renewal_method."""
    service, renewal_method = draw(st.sampled_from(_SERVICES_AND_METHODS))
    token_name = draw(valid_token_names())
    expiry_date = draw(st.one_of(st.none(), valid_iso_dates()))
    return PatEntry(
        service=service,
        token_name=token_name,
        expiry_date=expiry_date,
        renewal_method=renewal_method,
    )


@st.composite
def unique_pat_entry_lists(draw: st.DrawFn) -> list[PatEntry]:
    """Generate a list of valid PatEntry objects with no duplicate (service, token_name) pairs."""
    entries = draw(st.lists(valid_pat_entries(), min_size=1, max_size=10))
    # Deduplicate by (service, token_name)
    seen: set[tuple[str, str]] = set()
    unique_entries: list[PatEntry] = []
    for entry in entries:
        key = (entry.service, entry.token_name)
        if key not in seen:
            seen.add(key)
            unique_entries.append(entry)
    assume(len(unique_entries) >= 1)
    return unique_entries


@st.composite
def invalid_entry_dicts(draw: st.DrawFn) -> dict:
    """Generate entry dicts that are invalid (missing fields, bad service, bad token_name, etc.)."""
    strategy_choice = draw(st.integers(min_value=0, max_value=4))

    if strategy_choice == 0:
        # Missing required field: service
        return {
            "token_name": draw(valid_token_names()),
            "renewal_method": draw(st.sampled_from(["auto", "manual"])),
        }
    elif strategy_choice == 1:
        # Missing required field: token_name
        service, method = draw(st.sampled_from(_SERVICES_AND_METHODS))
        return {
            "service": service,
            "renewal_method": method,
        }
    elif strategy_choice == 2:
        # Missing required field: renewal_method
        service, _ = draw(st.sampled_from(_SERVICES_AND_METHODS))
        return {
            "service": service,
            "token_name": draw(valid_token_names()),
        }
    elif strategy_choice == 3:
        # Invalid service name
        bad_service = draw(
            st.text(min_size=1, max_size=20).filter(
                lambda s: s not in VALID_SERVICES
            )
        )
        return {
            "service": bad_service,
            "token_name": draw(valid_token_names()),
            "renewal_method": draw(st.sampled_from(["auto", "manual"])),
        }
    else:
        # Empty token_name or too long
        bad_name = draw(
            st.one_of(
                st.just(""),
                st.text(min_size=129, max_size=200),
            )
        )
        service, method = draw(st.sampled_from(_SERVICES_AND_METHODS))
        return {
            "service": service,
            "token_name": bad_name,
            "renewal_method": method,
        }


@st.composite
def duplicate_entry_from_existing(draw: st.DrawFn, entries: list[PatEntry]) -> dict:
    """Generate a duplicate entry dict that has the same (service, token_name) as an existing entry."""
    assume(len(entries) > 0)
    existing = draw(st.sampled_from(entries))
    # Create a dict with same service+token_name but possibly different expiry
    expiry = draw(st.one_of(st.none(), valid_iso_dates()))
    return {
        "service": existing.service,
        "token_name": existing.token_name,
        "expiry_date": expiry,
        "renewal_method": existing.renewal_method,
    }


# --- Property 1 Test ---

# Feature: pat-renewal, Property 1: Registry serialization round-trip
class TestRegistrySerializationRoundTrip:
    """
    **Validates: Requirements 1.1**

    Property 1: For any valid list of PAT entries, serializing the registry
    to JSON and deserializing it back SHALL produce an equivalent list of
    entries with identical field values.
    """

    @given(
        entries=st.lists(valid_pat_entries(), min_size=0, max_size=15).map(
            lambda entries: list({
                (e.service, e.token_name): e for e in entries
            }.values())
        )
    )
    @settings(max_examples=100)
    def test_save_load_round_trip(
        self, entries: list[PatEntry], tmp_path_factory: pytest.TempPathFactory
    ) -> None:
        """save(path, entries) followed by load(path) produces identical entries."""
        tmp_path = tmp_path_factory.mktemp("prop1_roundtrip")
        registry = PatRegistry()
        file_path = tmp_path / "registry.json"

        # Save entries to file
        registry.save(file_path, entries)

        # Load entries back
        loaded = registry.load(file_path)

        # Verify same count
        assert len(loaded) == len(entries)

        # Verify each entry has identical field values
        for original, restored in zip(entries, loaded):
            assert restored.service == original.service
            assert restored.token_name == original.token_name
            assert restored.expiry_date == original.expiry_date
            assert restored.renewal_method == original.renewal_method


# --- Property 3 Test ---

# Feature: pat-renewal, Property 3: Invalid entries and duplicates preserve registry state
class TestInvalidEntriesAndDuplicatesPreserveState:
    """
    **Validates: Requirements 1.3, 1.5**

    Property 3: For any PAT registry and any entry that is either invalid
    (fails validation) or a duplicate (same service + token_name already exists),
    attempting to add that entry SHALL leave the registry unchanged — the entry
    count and all existing entries remain identical.
    """

    @given(
        entries=unique_pat_entry_lists(),
        bad_entry=invalid_entry_dicts(),
    )
    @settings(max_examples=100)
    def test_invalid_entry_preserves_registry_state(
        self, entries: list[PatEntry], bad_entry: dict, tmp_path_factory: pytest.TempPathFactory
    ) -> None:
        """Loading a registry file with an appended invalid entry raises
        RegistryValidationError and the original saved file remains unchanged.
        """
        tmp_path = tmp_path_factory.mktemp("prop3_invalid")
        registry = PatRegistry()
        file_path = tmp_path / "registry.json"

        # Save the valid registry
        registry.save(file_path, entries)

        # Read the original file content for comparison
        original_content = file_path.read_text(encoding="utf-8")

        # Create a corrupted registry file with the bad entry appended
        corrupted_path = tmp_path / "corrupted.json"
        data = json.loads(original_content)
        data["tokens"].append(bad_entry)
        corrupted_path.write_text(json.dumps(data, indent=2), encoding="utf-8")

        # Attempting to load the corrupted file should raise RegistryValidationError
        with pytest.raises(RegistryValidationError):
            registry.load(corrupted_path)

        # The original file must remain unchanged
        assert file_path.read_text(encoding="utf-8") == original_content

        # The original file is still loadable and produces the same entries
        reloaded = registry.load(file_path)
        assert len(reloaded) == len(entries)
        for original, loaded in zip(entries, reloaded):
            assert loaded.service == original.service
            assert loaded.token_name == original.token_name
            assert loaded.expiry_date == original.expiry_date
            assert loaded.renewal_method == original.renewal_method

    @given(entries=unique_pat_entry_lists())
    @settings(max_examples=100)
    def test_duplicate_entry_preserves_registry_state(
        self, entries: list[PatEntry], tmp_path_factory: pytest.TempPathFactory
    ) -> None:
        """Loading a registry file with a duplicate (service, token_name) entry
        raises RegistryValidationError and the original saved file remains unchanged.
        """
        tmp_path = tmp_path_factory.mktemp("prop3_dup")
        registry = PatRegistry()
        file_path = tmp_path / "registry.json"

        # Save the valid registry
        registry.save(file_path, entries)

        # Read the original file content for comparison
        original_content = file_path.read_text(encoding="utf-8")

        # Pick an existing entry to duplicate
        existing = entries[0]
        duplicate_dict = {
            "service": existing.service,
            "token_name": existing.token_name,
            "expiry_date": "2099-12-31",  # Different expiry to prove it's a new entry
            "renewal_method": existing.renewal_method,
        }

        # Create a corrupted registry file with the duplicate appended
        corrupted_path = tmp_path / "corrupted_dup.json"
        data = json.loads(original_content)
        data["tokens"].append(duplicate_dict)
        corrupted_path.write_text(json.dumps(data, indent=2), encoding="utf-8")

        # Attempting to load the corrupted file should raise RegistryValidationError
        with pytest.raises(RegistryValidationError):
            registry.load(corrupted_path)

        # The original file must remain unchanged
        assert file_path.read_text(encoding="utf-8") == original_content

        # The original file is still loadable and produces the same entries
        reloaded = registry.load(file_path)
        assert len(reloaded) == len(entries)
        for original, loaded in zip(entries, reloaded):
            assert loaded.service == original.service
            assert loaded.token_name == original.token_name
            assert loaded.expiry_date == original.expiry_date
            assert loaded.renewal_method == original.renewal_method