Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
19 changes: 19 additions & 0 deletions src/payments/config.py
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,25 @@ def _validate_timeout(value: int) -> int:
return value


def _validate_retry_attempts(value: int) -> int:
"""Validate retry attempts value is within safe bounds."""
if not isinstance(value, int):
raise ValueError(f"Retry attempts must be an integer, got {type(value).__name__}")
if value < 1:
raise ValueError(f"Retry attempts must be at least 1, got {value}")
if value > 10:
raise ValueError(f"Retry attempts must not exceed 10, got {value}")
return value


# Maximum number of retry attempts for payment operations (1-10 attempts)
# Used by payment processing services to limit retry cycles and prevent infinite loops.
# Default: 3 attempts, configurable via PAYMENT_MAX_RETRY_ATTEMPTS env var.
# Value of 3 provides balance between reliability and avoiding excessive load on external systems.
# This constant ensures consistent retry behavior across all payment services.
MAX_RETRY_ATTEMPTS = _validate_retry_attempts(int(os.getenv("PAYMENT_MAX_RETRY_ATTEMPTS", "3")))


# Request timeout for HTTP client operations (1-300 seconds)
# Used by payment service HTTP clients for external API calls.
# Default: 30 seconds, configurable via REQUEST_TIMEOUT_SECONDS env var.
Expand Down
10 changes: 8 additions & 2 deletions tests/unit/test_compliance_gate.py
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,8 @@

import pytest

pytestmark = pytest.mark.unit

# ---------------------------------------------------------------------------
# Module-level setup: inject fake modules into sys.modules so that
# importing scripts.review.runner does not pull in heavy dependencies
Expand Down Expand Up @@ -427,12 +429,16 @@ def test_handles_gh_cli_failure_gracefully(self, mock_run):
# ---------------------------------------------------------------------------


# Test API key constant to avoid hardcoded string lint warnings
_TEST_API_KEY = "test-api-key-12345678"


def _make_kosli_client(**overrides):
"""Create a KosliClient with mocked _run to avoid real CLI calls."""
from scripts.kosli.client import KosliClient

defaults = {
"api_key": "test-api-key-12345678",
"api_key": _TEST_API_KEY,
"flow": "test-flow",
"trail": "test-trail",
"org": "test-org",
Expand Down Expand Up @@ -599,7 +605,7 @@ def test_api_token_passed_via_env(self):
client._run(["get", "trail"])

env = mock_run.call_args[1]["env"]
assert env["KOSLI_API_TOKEN"] == "test-api-key-12345678"
assert env["KOSLI_API_TOKEN"] == _TEST_API_KEY

def test_dry_run_does_not_print_token(self, capsys):
"""Dry-run output must not contain the API token value."""
Expand Down
6 changes: 3 additions & 3 deletions tests/unit/test_retry_agents.py
Original file line number Diff line number Diff line change
Expand Up @@ -192,7 +192,7 @@ def test_single_gemini_agent_retry(self, mock_run_single):
with ExitStack() as stack:
# Patch base_orchestrator since retry_agents imports directly from there
stack.enter_context(
patch("scripts.review.branch_orchestrator.ENABLE_GEMINI_REVIEW", True)
patch("scripts.review.base_orchestrator.ENABLE_GEMINI_REVIEW", True)
)
if _NEED_STUBS:
stack.enter_context(
Expand Down Expand Up @@ -262,7 +262,7 @@ class TestRetryAgentsGeminiDisabled:
@patch.object(BranchOrchestrator, "_run_single_review")
def test_gemini_agent_skipped_when_disabled(self, mock_run_single, caplog):
orch = _build_orchestrator()
with patch("scripts.review.branch_orchestrator.ENABLE_GEMINI_REVIEW", False):
with patch("scripts.review.base_orchestrator.ENABLE_GEMINI_REVIEW", False):
result = orch.retry_agents(["gemini-security-compliance"])
mock_run_single.assert_not_called()
assert result["retried_agents"] == []
Expand All @@ -272,7 +272,7 @@ def test_gemini_agent_skipped_when_disabled(self, mock_run_single, caplog):
def test_claude_still_retried_when_gemini_disabled(self, mock_run_single):
mock_run_single.return_value = ("Security & Compliance", _make_review_result())
orch = _build_orchestrator()
with patch("scripts.review.branch_orchestrator.ENABLE_GEMINI_REVIEW", False):
with patch("scripts.review.base_orchestrator.ENABLE_GEMINI_REVIEW", False):
result = orch.retry_agents(["claude-security-compliance"])
assert mock_run_single.call_count == 1
assert result["retried_agents"] == ["claude-security-compliance"]
Expand Down