Migrate all repos into monorepo context folders
Bahn: aisupport, Analyse-O2C-C2S, awesome-bahn-mcp-servers, beam-mcp,
Confluence_Bot, db-planet-mcp-server, O2C-Harness, project-audit,
Projekt-KIQ-HP, teamlandkarte-mcp
Dhive: Jury-Voting
Privat: CV, NoteGraph (NOTE: NoteGraph needs complete redo after consolidation)
Shared: AI-Orchestrator, OrgMyLife, power_skills_and_more
Shared/references: symphony (read-only)
Bahn repos remain available as independent remotes - this monorepo
pulls them in via subtree, the originals are untouched.
This commit is contained in:
@@ -0,0 +1,314 @@
|
||||
"""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
|
||||
Reference in New Issue
Block a user