ankn
a5f8fb49ab
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.
735 lines
28 KiB
Python
735 lines
28 KiB
Python
"""Unit tests for the PAT Manager expiry checker module."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import date, timedelta
|
|
|
|
import pytest
|
|
|
|
import httpx
|
|
|
|
from scripts.pat_manager.checker import ExpiryChecker, classify_by_date
|
|
from scripts.pat_manager.models import CheckResult, PatEntry, PatStatus
|
|
|
|
|
|
class TestClassifyByDate:
|
|
"""Tests for the classify_by_date pure function."""
|
|
|
|
def test_expired_token(self) -> None:
|
|
"""A token with expiry_date before reference_date is expired."""
|
|
status, days = classify_by_date("2025-01-01", date(2025, 1, 15), window=14)
|
|
assert status == PatStatus.EXPIRED
|
|
assert days == 0
|
|
|
|
def test_expiring_soon_on_boundary(self) -> None:
|
|
"""A token expiring exactly on reference_date is expiring soon."""
|
|
status, days = classify_by_date("2025-01-15", date(2025, 1, 15), window=14)
|
|
assert status == PatStatus.EXPIRING_SOON
|
|
assert days == 0
|
|
|
|
def test_expiring_soon_within_window(self) -> None:
|
|
"""A token expiring within the window is expiring soon."""
|
|
status, days = classify_by_date("2025-01-20", date(2025, 1, 15), window=14)
|
|
assert status == PatStatus.EXPIRING_SOON
|
|
assert days == 5
|
|
|
|
def test_expiring_soon_at_window_end(self) -> None:
|
|
"""A token expiring exactly at window end is expiring soon."""
|
|
status, days = classify_by_date("2025-01-29", date(2025, 1, 15), window=14)
|
|
assert status == PatStatus.EXPIRING_SOON
|
|
assert days == 14
|
|
|
|
def test_healthy_beyond_window(self) -> None:
|
|
"""A token expiring beyond the window is healthy."""
|
|
status, days = classify_by_date("2025-01-30", date(2025, 1, 15), window=14)
|
|
assert status == PatStatus.HEALTHY
|
|
assert days == 15
|
|
|
|
def test_healthy_far_future(self) -> None:
|
|
"""A token expiring far in the future is healthy."""
|
|
status, days = classify_by_date("2026-06-01", date(2025, 1, 15), window=14)
|
|
assert status == PatStatus.HEALTHY
|
|
assert days == 502
|
|
|
|
def test_custom_window(self) -> None:
|
|
"""Classification respects custom window size."""
|
|
# With window=7, 10 days out is healthy
|
|
status, days = classify_by_date("2025-01-25", date(2025, 1, 15), window=7)
|
|
assert status == PatStatus.HEALTHY
|
|
assert days == 10
|
|
|
|
# With window=30, 10 days out is expiring soon
|
|
status, days = classify_by_date("2025-01-25", date(2025, 1, 15), window=30)
|
|
assert status == PatStatus.EXPIRING_SOON
|
|
assert days == 10
|
|
|
|
|
|
class TestExpiryCheckerInit:
|
|
"""Tests for ExpiryChecker initialization."""
|
|
|
|
def test_default_values(self) -> None:
|
|
"""ExpiryChecker uses correct defaults."""
|
|
checker = ExpiryChecker()
|
|
assert checker.expiry_window_days == 14
|
|
assert checker.timeout == 30
|
|
assert checker.max_retries == 2
|
|
|
|
def test_custom_values(self) -> None:
|
|
"""ExpiryChecker accepts custom configuration."""
|
|
checker = ExpiryChecker(expiry_window_days=7, timeout=60, max_retries=3)
|
|
assert checker.expiry_window_days == 7
|
|
assert checker.timeout == 60
|
|
assert checker.max_retries == 3
|
|
|
|
|
|
class TestCheckAll:
|
|
"""Tests for ExpiryChecker.check_all method."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_missing_secret_returns_check_failed(self) -> None:
|
|
"""Entries with no matching secret get check_failed status."""
|
|
checker = ExpiryChecker()
|
|
entry = PatEntry(
|
|
service="GitLab",
|
|
token_name="test-token",
|
|
expiry_date="2025-06-01",
|
|
renewal_method="auto",
|
|
)
|
|
results = await checker.check_all([entry], secrets={})
|
|
assert len(results) == 1
|
|
assert results[0].status == PatStatus.CHECK_FAILED
|
|
assert results[0].days_remaining is None
|
|
assert "No secret found" in results[0].error_message
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_unsupported_service_returns_check_failed(self) -> None:
|
|
"""Entries with unsupported service get check_failed status."""
|
|
checker = ExpiryChecker()
|
|
entry = PatEntry(
|
|
service="UnknownService",
|
|
token_name="test-token",
|
|
expiry_date="2025-06-01",
|
|
renewal_method="manual",
|
|
)
|
|
# Provide a secret that would match via the fallback empty key
|
|
results = await checker.check_all([entry], secrets={"": "some-token"})
|
|
assert len(results) == 1
|
|
assert results[0].status == PatStatus.CHECK_FAILED
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_gitlab_check_with_mock_api(self) -> None:
|
|
"""GitLab check with a mocked API response classifies correctly."""
|
|
import respx
|
|
|
|
checker = ExpiryChecker()
|
|
entry = PatEntry(
|
|
service="GitLab",
|
|
token_name="test-token",
|
|
expiry_date="2025-06-01",
|
|
renewal_method="auto",
|
|
)
|
|
# Mock the GitLab API to return a 401 (expired token)
|
|
with respx.mock:
|
|
respx.get("https://gitlab.com/api/v4/personal_access_tokens/self").mock(
|
|
return_value=httpx.Response(401)
|
|
)
|
|
results = await checker.check_all(
|
|
[entry], secrets={"GITLAB_PAT": "glpat-test123"}
|
|
)
|
|
assert len(results) == 1
|
|
assert results[0].status == PatStatus.EXPIRED
|
|
assert results[0].days_remaining == 0
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_empty_entries_returns_empty_list(self) -> None:
|
|
"""No entries produces no results."""
|
|
checker = ExpiryChecker()
|
|
results = await checker.check_all([], secrets={"GITLAB_PAT": "token"})
|
|
assert results == []
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_multiple_entries_returns_one_result_per_entry(self) -> None:
|
|
"""Each entry produces exactly one result."""
|
|
checker = ExpiryChecker()
|
|
entries = [
|
|
PatEntry("GitLab", "token-1", "2025-06-01", "auto"),
|
|
PatEntry("Jira", "token-2", "2025-07-01", "manual"),
|
|
PatEntry("Confluence", "token-3", "2025-08-01", "manual"),
|
|
]
|
|
results = await checker.check_all(entries, secrets={})
|
|
assert len(results) == 3
|
|
# All should be check_failed since no secrets provided
|
|
for result in results:
|
|
assert result.status == PatStatus.CHECK_FAILED
|
|
|
|
|
|
class TestCheckGitLab:
|
|
"""Tests for ExpiryChecker.check_gitlab method."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_gitlab_healthy_token(self, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""GitLab token with future expiry is classified as healthy."""
|
|
import respx
|
|
from datetime import date, timedelta
|
|
|
|
future_date = (date.today() + timedelta(days=30)).isoformat()
|
|
monkeypatch.setenv("GITLAB_URL", "https://gitlab.example.com")
|
|
|
|
checker = ExpiryChecker()
|
|
entry = PatEntry("GitLab", "ci-token", future_date, "auto")
|
|
|
|
with respx.mock:
|
|
respx.get("https://gitlab.example.com/api/v4/personal_access_tokens/self").mock(
|
|
return_value=httpx.Response(200, json={"expires_at": future_date})
|
|
)
|
|
result = await checker.check_gitlab(entry, "glpat-test")
|
|
|
|
assert result.status == PatStatus.HEALTHY
|
|
assert result.days_remaining == 30
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_gitlab_expiring_soon_token(self, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""GitLab token expiring within window is classified as expiring soon."""
|
|
import respx
|
|
from datetime import date, timedelta
|
|
|
|
soon_date = (date.today() + timedelta(days=7)).isoformat()
|
|
monkeypatch.setenv("GITLAB_URL", "https://gitlab.example.com")
|
|
|
|
checker = ExpiryChecker()
|
|
entry = PatEntry("GitLab", "ci-token", soon_date, "auto")
|
|
|
|
with respx.mock:
|
|
respx.get("https://gitlab.example.com/api/v4/personal_access_tokens/self").mock(
|
|
return_value=httpx.Response(200, json={"expires_at": soon_date})
|
|
)
|
|
result = await checker.check_gitlab(entry, "glpat-test")
|
|
|
|
assert result.status == PatStatus.EXPIRING_SOON
|
|
assert result.days_remaining == 7
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_gitlab_expired_token_via_api(self, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""GitLab token with past expiry from API is classified as expired."""
|
|
import respx
|
|
from datetime import date, timedelta
|
|
|
|
past_date = (date.today() - timedelta(days=5)).isoformat()
|
|
monkeypatch.setenv("GITLAB_URL", "https://gitlab.example.com")
|
|
|
|
checker = ExpiryChecker()
|
|
entry = PatEntry("GitLab", "ci-token", past_date, "auto")
|
|
|
|
with respx.mock:
|
|
respx.get("https://gitlab.example.com/api/v4/personal_access_tokens/self").mock(
|
|
return_value=httpx.Response(200, json={"expires_at": past_date})
|
|
)
|
|
result = await checker.check_gitlab(entry, "glpat-test")
|
|
|
|
assert result.status == PatStatus.EXPIRED
|
|
assert result.days_remaining == 0
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_gitlab_401_means_expired(self, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""GitLab API returning 401 means token is expired."""
|
|
import respx
|
|
|
|
monkeypatch.setenv("GITLAB_URL", "https://gitlab.example.com")
|
|
|
|
checker = ExpiryChecker()
|
|
entry = PatEntry("GitLab", "ci-token", "2025-12-01", "auto")
|
|
|
|
with respx.mock:
|
|
respx.get("https://gitlab.example.com/api/v4/personal_access_tokens/self").mock(
|
|
return_value=httpx.Response(401)
|
|
)
|
|
result = await checker.check_gitlab(entry, "glpat-test")
|
|
|
|
assert result.status == PatStatus.EXPIRED
|
|
assert result.days_remaining == 0
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_gitlab_500_means_check_failed(self, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""GitLab API returning 500 means check failed."""
|
|
import respx
|
|
|
|
monkeypatch.setenv("GITLAB_URL", "https://gitlab.example.com")
|
|
|
|
checker = ExpiryChecker()
|
|
entry = PatEntry("GitLab", "ci-token", "2025-12-01", "auto")
|
|
|
|
with respx.mock:
|
|
respx.get("https://gitlab.example.com/api/v4/personal_access_tokens/self").mock(
|
|
return_value=httpx.Response(500)
|
|
)
|
|
result = await checker.check_gitlab(entry, "glpat-test")
|
|
|
|
assert result.status == PatStatus.CHECK_FAILED
|
|
assert "HTTP 500" in result.error_message
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_gitlab_null_expires_at_healthy(self, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""GitLab token with null expires_at and no stored expiry is healthy."""
|
|
import respx
|
|
|
|
monkeypatch.setenv("GITLAB_URL", "https://gitlab.example.com")
|
|
|
|
checker = ExpiryChecker()
|
|
entry = PatEntry("GitLab", "ci-token", None, "auto")
|
|
|
|
with respx.mock:
|
|
respx.get("https://gitlab.example.com/api/v4/personal_access_tokens/self").mock(
|
|
return_value=httpx.Response(200, json={"expires_at": None})
|
|
)
|
|
result = await checker.check_gitlab(entry, "glpat-test")
|
|
|
|
assert result.status == PatStatus.HEALTHY
|
|
assert result.days_remaining is None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_gitlab_stale_expiry_warning(self, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""GitLab API succeeds but stored expiry_date has passed → expired with stale warning."""
|
|
import respx
|
|
from datetime import date, timedelta
|
|
|
|
future_api_date = (date.today() + timedelta(days=30)).isoformat()
|
|
past_stored_date = (date.today() - timedelta(days=5)).isoformat()
|
|
monkeypatch.setenv("GITLAB_URL", "https://gitlab.example.com")
|
|
|
|
checker = ExpiryChecker()
|
|
entry = PatEntry("GitLab", "ci-token", past_stored_date, "auto")
|
|
|
|
with respx.mock:
|
|
respx.get("https://gitlab.example.com/api/v4/personal_access_tokens/self").mock(
|
|
return_value=httpx.Response(200, json={"expires_at": future_api_date})
|
|
)
|
|
result = await checker.check_gitlab(entry, "glpat-test")
|
|
|
|
assert result.status == PatStatus.EXPIRED
|
|
assert result.stale_warning is True
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_gitlab_timeout_retries_then_fails(self, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""GitLab API timing out after retries results in check_failed."""
|
|
import asyncio
|
|
import respx
|
|
|
|
monkeypatch.setenv("GITLAB_URL", "https://gitlab.example.com")
|
|
|
|
checker = ExpiryChecker(max_retries=1, timeout=1)
|
|
entry = PatEntry("GitLab", "ci-token", "2025-12-01", "auto")
|
|
|
|
async def noop_sleep(_: float) -> None:
|
|
pass
|
|
|
|
with respx.mock:
|
|
respx.get("https://gitlab.example.com/api/v4/personal_access_tokens/self").mock(
|
|
side_effect=httpx.ConnectError("Connection refused")
|
|
)
|
|
monkeypatch.setattr("asyncio.sleep", noop_sleep)
|
|
result = await checker.check_gitlab(entry, "glpat-test")
|
|
|
|
assert result.status == PatStatus.CHECK_FAILED
|
|
assert result.error_message is not None
|
|
|
|
|
|
class TestCheckJira:
|
|
"""Tests for ExpiryChecker.check_jira method."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_jira_healthy_token(self, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""Jira token with successful API call and future expiry is healthy."""
|
|
import respx
|
|
from datetime import date, timedelta
|
|
|
|
future_date = (date.today() + timedelta(days=30)).isoformat()
|
|
monkeypatch.setenv("JIRA_URL", "https://jira.example.com")
|
|
|
|
checker = ExpiryChecker()
|
|
entry = PatEntry("Jira", "jira-token", future_date, "manual")
|
|
|
|
with respx.mock:
|
|
respx.get("https://jira.example.com/rest/api/2/myself").mock(
|
|
return_value=httpx.Response(200, json={"displayName": "Test User"})
|
|
)
|
|
result = await checker.check_jira(entry, "jira-pat-123")
|
|
|
|
assert result.status == PatStatus.HEALTHY
|
|
assert result.days_remaining == 30
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_jira_401_means_expired(self, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""Jira API returning 401 means token is expired."""
|
|
import respx
|
|
|
|
monkeypatch.setenv("JIRA_URL", "https://jira.example.com")
|
|
|
|
checker = ExpiryChecker()
|
|
entry = PatEntry("Jira", "jira-token", "2025-12-01", "manual")
|
|
|
|
with respx.mock:
|
|
respx.get("https://jira.example.com/rest/api/2/myself").mock(
|
|
return_value=httpx.Response(401)
|
|
)
|
|
result = await checker.check_jira(entry, "jira-pat-123")
|
|
|
|
assert result.status == PatStatus.EXPIRED
|
|
assert result.days_remaining == 0
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_jira_no_url_env_var(self) -> None:
|
|
"""Jira check fails when JIRA_URL is not set."""
|
|
import os
|
|
|
|
# Ensure JIRA_URL is not set
|
|
os.environ.pop("JIRA_URL", None)
|
|
|
|
checker = ExpiryChecker()
|
|
entry = PatEntry("Jira", "jira-token", "2025-12-01", "manual")
|
|
result = await checker.check_jira(entry, "jira-pat-123")
|
|
|
|
assert result.status == PatStatus.CHECK_FAILED
|
|
assert "JIRA_URL" in result.error_message
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_jira_null_expiry_healthy(self, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""Jira token with no expiry_date and successful API call is healthy."""
|
|
import respx
|
|
|
|
monkeypatch.setenv("JIRA_URL", "https://jira.example.com")
|
|
|
|
checker = ExpiryChecker()
|
|
entry = PatEntry("Jira", "jira-token", None, "manual")
|
|
|
|
with respx.mock:
|
|
respx.get("https://jira.example.com/rest/api/2/myself").mock(
|
|
return_value=httpx.Response(200, json={"displayName": "Test User"})
|
|
)
|
|
result = await checker.check_jira(entry, "jira-pat-123")
|
|
|
|
assert result.status == PatStatus.HEALTHY
|
|
assert result.days_remaining is None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_jira_stale_expiry_warning(self, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""Jira API succeeds but stored expiry_date has passed → expired with stale warning."""
|
|
import respx
|
|
from datetime import date, timedelta
|
|
|
|
past_date = (date.today() - timedelta(days=5)).isoformat()
|
|
monkeypatch.setenv("JIRA_URL", "https://jira.example.com")
|
|
|
|
checker = ExpiryChecker()
|
|
entry = PatEntry("Jira", "jira-token", past_date, "manual")
|
|
|
|
with respx.mock:
|
|
respx.get("https://jira.example.com/rest/api/2/myself").mock(
|
|
return_value=httpx.Response(200, json={"displayName": "Test User"})
|
|
)
|
|
result = await checker.check_jira(entry, "jira-pat-123")
|
|
|
|
assert result.status == PatStatus.EXPIRED
|
|
assert result.stale_warning is True
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_jira_500_means_check_failed(self, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""Jira API returning 500 means check failed."""
|
|
import respx
|
|
|
|
monkeypatch.setenv("JIRA_URL", "https://jira.example.com")
|
|
|
|
checker = ExpiryChecker()
|
|
entry = PatEntry("Jira", "jira-token", "2025-12-01", "manual")
|
|
|
|
with respx.mock:
|
|
respx.get("https://jira.example.com/rest/api/2/myself").mock(
|
|
return_value=httpx.Response(500)
|
|
)
|
|
result = await checker.check_jira(entry, "jira-pat-123")
|
|
|
|
assert result.status == PatStatus.CHECK_FAILED
|
|
assert "HTTP 500" in result.error_message
|
|
|
|
|
|
class TestCheckConfluence:
|
|
"""Tests for ExpiryChecker.check_confluence method."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_confluence_healthy_token(self, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""Confluence token with successful API call and future expiry is healthy."""
|
|
import respx
|
|
from datetime import date, timedelta
|
|
|
|
future_date = (date.today() + timedelta(days=30)).isoformat()
|
|
monkeypatch.setenv("CONFLUENCE_URL", "https://confluence.example.com")
|
|
|
|
checker = ExpiryChecker()
|
|
entry = PatEntry("Confluence", "conf-token", future_date, "manual")
|
|
|
|
with respx.mock:
|
|
respx.get("https://confluence.example.com/rest/api/user/current").mock(
|
|
return_value=httpx.Response(200, json={"displayName": "Test User"})
|
|
)
|
|
result = await checker.check_confluence(entry, "conf-pat-123")
|
|
|
|
assert result.status == PatStatus.HEALTHY
|
|
assert result.days_remaining == 30
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_confluence_401_means_expired(self, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""Confluence API returning 401 means token is expired."""
|
|
import respx
|
|
|
|
monkeypatch.setenv("CONFLUENCE_URL", "https://confluence.example.com")
|
|
|
|
checker = ExpiryChecker()
|
|
entry = PatEntry("Confluence", "conf-token", "2025-12-01", "manual")
|
|
|
|
with respx.mock:
|
|
respx.get("https://confluence.example.com/rest/api/user/current").mock(
|
|
return_value=httpx.Response(401)
|
|
)
|
|
result = await checker.check_confluence(entry, "conf-pat-123")
|
|
|
|
assert result.status == PatStatus.EXPIRED
|
|
assert result.days_remaining == 0
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_confluence_no_url_env_var(self) -> None:
|
|
"""Confluence check fails when CONFLUENCE_URL is not set."""
|
|
import os
|
|
|
|
os.environ.pop("CONFLUENCE_URL", None)
|
|
|
|
checker = ExpiryChecker()
|
|
entry = PatEntry("Confluence", "conf-token", "2025-12-01", "manual")
|
|
result = await checker.check_confluence(entry, "conf-pat-123")
|
|
|
|
assert result.status == PatStatus.CHECK_FAILED
|
|
assert "CONFLUENCE_URL" in result.error_message
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_confluence_null_expiry_healthy(self, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""Confluence token with no expiry_date and successful API call is healthy."""
|
|
import respx
|
|
|
|
monkeypatch.setenv("CONFLUENCE_URL", "https://confluence.example.com")
|
|
|
|
checker = ExpiryChecker()
|
|
entry = PatEntry("Confluence", "conf-token", None, "manual")
|
|
|
|
with respx.mock:
|
|
respx.get("https://confluence.example.com/rest/api/user/current").mock(
|
|
return_value=httpx.Response(200, json={"displayName": "Test User"})
|
|
)
|
|
result = await checker.check_confluence(entry, "conf-pat-123")
|
|
|
|
assert result.status == PatStatus.HEALTHY
|
|
assert result.days_remaining is None
|
|
|
|
|
|
class TestCheckOrgMyLife:
|
|
"""Tests for ExpiryChecker.check_orgmylife method."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_orgmylife_healthy_token(self, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""OrgMyLife token with successful API call and future expiry is healthy."""
|
|
import respx
|
|
from datetime import date, timedelta
|
|
|
|
future_date = (date.today() + timedelta(days=30)).isoformat()
|
|
monkeypatch.setenv("ORGMYLIFE_URL", "https://orgmylife.example.com")
|
|
|
|
checker = ExpiryChecker()
|
|
entry = PatEntry("OrgMyLife", "oml-token", future_date, "manual")
|
|
|
|
with respx.mock:
|
|
respx.get("https://orgmylife.example.com/api/tasks").mock(
|
|
return_value=httpx.Response(200, json=[])
|
|
)
|
|
result = await checker.check_orgmylife(entry, "oml-key-123")
|
|
|
|
assert result.status == PatStatus.HEALTHY
|
|
assert result.days_remaining == 30
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_orgmylife_401_means_expired(self, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""OrgMyLife API returning 401 means token is expired."""
|
|
import respx
|
|
|
|
monkeypatch.setenv("ORGMYLIFE_URL", "https://orgmylife.example.com")
|
|
|
|
checker = ExpiryChecker()
|
|
entry = PatEntry("OrgMyLife", "oml-token", "2025-12-01", "manual")
|
|
|
|
with respx.mock:
|
|
respx.get("https://orgmylife.example.com/api/tasks").mock(
|
|
return_value=httpx.Response(401)
|
|
)
|
|
result = await checker.check_orgmylife(entry, "oml-key-123")
|
|
|
|
assert result.status == PatStatus.EXPIRED
|
|
assert result.days_remaining == 0
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_orgmylife_403_means_expired(self, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""OrgMyLife API returning 403 means token is expired."""
|
|
import respx
|
|
|
|
monkeypatch.setenv("ORGMYLIFE_URL", "https://orgmylife.example.com")
|
|
|
|
checker = ExpiryChecker()
|
|
entry = PatEntry("OrgMyLife", "oml-token", "2025-12-01", "manual")
|
|
|
|
with respx.mock:
|
|
respx.get("https://orgmylife.example.com/api/tasks").mock(
|
|
return_value=httpx.Response(403)
|
|
)
|
|
result = await checker.check_orgmylife(entry, "oml-key-123")
|
|
|
|
assert result.status == PatStatus.EXPIRED
|
|
assert result.days_remaining == 0
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_orgmylife_default_url(self) -> None:
|
|
"""OrgMyLife uses default URL when env var not set."""
|
|
import respx
|
|
import os
|
|
from datetime import date, timedelta
|
|
|
|
os.environ.pop("ORGMYLIFE_URL", None)
|
|
future_date = (date.today() + timedelta(days=30)).isoformat()
|
|
|
|
checker = ExpiryChecker()
|
|
entry = PatEntry("OrgMyLife", "oml-token", future_date, "manual")
|
|
|
|
with respx.mock:
|
|
respx.get("https://orgmylife.app/api/tasks").mock(
|
|
return_value=httpx.Response(200, json=[])
|
|
)
|
|
result = await checker.check_orgmylife(entry, "oml-key-123")
|
|
|
|
assert result.status == PatStatus.HEALTHY
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_orgmylife_null_expiry_healthy(self, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""OrgMyLife token with no expiry_date and successful API call is healthy."""
|
|
import respx
|
|
|
|
monkeypatch.setenv("ORGMYLIFE_URL", "https://orgmylife.example.com")
|
|
|
|
checker = ExpiryChecker()
|
|
entry = PatEntry("OrgMyLife", "oml-token", None, "manual")
|
|
|
|
with respx.mock:
|
|
respx.get("https://orgmylife.example.com/api/tasks").mock(
|
|
return_value=httpx.Response(200, json=[])
|
|
)
|
|
result = await checker.check_orgmylife(entry, "oml-key-123")
|
|
|
|
assert result.status == PatStatus.HEALTHY
|
|
assert result.days_remaining is None
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_orgmylife_null_expiry_401_expired(self, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""OrgMyLife token with no expiry_date and 401 response is expired."""
|
|
import respx
|
|
|
|
monkeypatch.setenv("ORGMYLIFE_URL", "https://orgmylife.example.com")
|
|
|
|
checker = ExpiryChecker()
|
|
entry = PatEntry("OrgMyLife", "oml-token", None, "manual")
|
|
|
|
with respx.mock:
|
|
respx.get("https://orgmylife.example.com/api/tasks").mock(
|
|
return_value=httpx.Response(401)
|
|
)
|
|
result = await checker.check_orgmylife(entry, "oml-key-123")
|
|
|
|
assert result.status == PatStatus.EXPIRED
|
|
assert result.days_remaining == 0
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_orgmylife_500_means_check_failed(self, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""OrgMyLife API returning 500 means check failed."""
|
|
import respx
|
|
|
|
monkeypatch.setenv("ORGMYLIFE_URL", "https://orgmylife.example.com")
|
|
|
|
checker = ExpiryChecker()
|
|
entry = PatEntry("OrgMyLife", "oml-token", "2025-12-01", "manual")
|
|
|
|
with respx.mock:
|
|
respx.get("https://orgmylife.example.com/api/tasks").mock(
|
|
return_value=httpx.Response(500)
|
|
)
|
|
result = await checker.check_orgmylife(entry, "oml-key-123")
|
|
|
|
assert result.status == PatStatus.CHECK_FAILED
|
|
assert "HTTP 500" in result.error_message
|
|
|
|
|
|
class TestRetryLogic:
|
|
"""Tests for the retry mechanism in service checks."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_retry_succeeds_on_second_attempt(self, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""Request succeeds after first timeout, second attempt works."""
|
|
import asyncio
|
|
import respx
|
|
from datetime import date, timedelta
|
|
|
|
future_date = (date.today() + timedelta(days=30)).isoformat()
|
|
monkeypatch.setenv("JIRA_URL", "https://jira.example.com")
|
|
|
|
async def noop_sleep(_: float) -> None:
|
|
pass
|
|
|
|
monkeypatch.setattr("asyncio.sleep", noop_sleep)
|
|
|
|
checker = ExpiryChecker(max_retries=2)
|
|
entry = PatEntry("Jira", "jira-token", future_date, "manual")
|
|
|
|
call_count = 0
|
|
|
|
def side_effect(request):
|
|
nonlocal call_count
|
|
call_count += 1
|
|
if call_count == 1:
|
|
raise httpx.ConnectError("Connection refused")
|
|
return httpx.Response(200, json={"displayName": "Test User"})
|
|
|
|
with respx.mock:
|
|
respx.get("https://jira.example.com/rest/api/2/myself").mock(
|
|
side_effect=side_effect
|
|
)
|
|
result = await checker.check_jira(entry, "jira-pat-123")
|
|
|
|
assert result.status == PatStatus.HEALTHY
|
|
assert call_count == 2
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_all_retries_exhausted(self, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""All retries exhausted results in check_failed."""
|
|
import asyncio
|
|
import respx
|
|
|
|
monkeypatch.setenv("JIRA_URL", "https://jira.example.com")
|
|
|
|
async def noop_sleep(_: float) -> None:
|
|
pass
|
|
|
|
monkeypatch.setattr("asyncio.sleep", noop_sleep)
|
|
|
|
checker = ExpiryChecker(max_retries=2)
|
|
entry = PatEntry("Jira", "jira-token", "2025-12-01", "manual")
|
|
|
|
with respx.mock:
|
|
respx.get("https://jira.example.com/rest/api/2/myself").mock(
|
|
side_effect=httpx.TimeoutException("Request timed out")
|
|
)
|
|
result = await checker.check_jira(entry, "jira-pat-123")
|
|
|
|
assert result.status == PatStatus.CHECK_FAILED
|
|
assert result.error_message is not None
|