Skip to content
Merged
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
3 changes: 3 additions & 0 deletions capiscio_sdk/connect.py
Original file line number Diff line number Diff line change
Expand Up @@ -815,12 +815,15 @@ def _setup_badge(self):
)

# Set up BadgeKeeper with correct parameters
# Wire on_renew so the guard's badge token stays in sync
# when the keeper renews it in the background.
keeper = BadgeKeeper(
api_url=self.server_url,
api_key=self.api_key,
agent_id=self.agent_id,
mode="dev" if self.dev_mode else "ca",
output_file=str(self.keys_dir / "badge.jwt"),
on_renew=lambda token: guard.set_badge_token(token),
)

# Start the keeper and get initial badge
Expand Down
36 changes: 36 additions & 0 deletions tests/unit/test_connect.py
Original file line number Diff line number Diff line change
Expand Up @@ -997,6 +997,42 @@ def test_setup_badge_success(self, tmp_path):
mock_keeper.start.assert_called_once()
mock_keeper.get_current_badge.assert_called_once()

def test_setup_badge_wires_on_renew(self, tmp_path):
"""Test _setup_badge wires on_renew callback to guard.set_badge_token."""
connector = _Connector(
api_key="sk_test",
name="Test",
agent_id="agent-123",
server_url="https://test.server.com",
keys_dir=tmp_path,
auto_badge=True,
dev_mode=False,
)
connector.did = "did:key:z6Mktest"

mock_keeper = MagicMock()
mock_keeper.get_current_badge.return_value = "badge-jwt"
mock_keeper.badge_expires_at = "2026-12-31T00:00:00Z"

mock_guard = MagicMock()

captured_on_renew = None

def capture_keeper_init(**kwargs):
nonlocal captured_on_renew
captured_on_renew = kwargs.get("on_renew")
return mock_keeper

with patch("capiscio_sdk.badge_keeper.BadgeKeeper", side_effect=capture_keeper_init):
with patch("capiscio_sdk.simple_guard.SimpleGuard", return_value=mock_guard):
connector._setup_badge()

# on_renew should have been passed to BadgeKeeper
assert captured_on_renew is not None
# Calling on_renew should forward to guard.set_badge_token
captured_on_renew("new-token-xyz")
mock_guard.set_badge_token.assert_called_once_with("new-token-xyz")

def test_setup_badge_failure_continues(self, tmp_path):
"""Test _setup_badge returns None on failure without raising."""
connector = _Connector(
Expand Down
Loading