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
22 changes: 9 additions & 13 deletions infisical_sdk/api_types.py
Original file line number Diff line number Diff line change
Expand Up @@ -529,9 +529,8 @@ class AgentVaultSessionActor(BaseModel):

@dataclass
class AgentVaultSessionAccessBundle(BaseModel):
"""An access bundle carried by an Agent Vault session"""
"""The access bundle an Agent Vault session carries. id is None once the bundle is deleted."""
name: str
position: int
id: Optional[str] = None

@dataclass
Expand All @@ -544,18 +543,16 @@ class AgentVaultSession(BaseModel):
actor: Optional[AgentVaultSessionActor] = None
expiresAt: Optional[str] = None
revokedAt: Optional[str] = None
accessBundles: Optional[List[AgentVaultSessionAccessBundle]] = None
accessBundle: Optional[AgentVaultSessionAccessBundle] = None

@classmethod
def from_dict(cls, data: Dict) -> 'AgentVaultSession':
"""Create model from dictionary with nested actor and access bundles. accessBundles is None if the response leaves it out."""
"""Create model from dictionary with nested actor and access bundle. accessBundle is None if the response leaves it out."""
session_data = data.copy()
if data.get('actor') is not None:
session_data['actor'] = AgentVaultSessionActor.from_dict(data['actor'])
if data.get('accessBundles') is not None:
session_data['accessBundles'] = [
AgentVaultSessionAccessBundle.from_dict(bundle) for bundle in data['accessBundles']
]
if data.get('accessBundle') is not None:
session_data['accessBundle'] = AgentVaultSessionAccessBundle.from_dict(data['accessBundle'])
Comment thread
saifsmailbox98 marked this conversation as resolved.
Resolved

return super().from_dict(session_data)

Expand All @@ -567,15 +564,14 @@ class CreatedAgentVaultSession(BaseModel):
createdAt: str
label: Optional[str] = None
expiresAt: Optional[str] = None
accessBundles: List[AgentVaultSessionAccessBundle] = field(default_factory=list)
accessBundle: Optional[AgentVaultSessionAccessBundle] = None

@classmethod
def from_dict(cls, data: Dict) -> 'CreatedAgentVaultSession':
"""Create model from dictionary with nested access bundles"""
"""Create model from dictionary with nested access bundle"""
session_data = data.copy()
session_data['accessBundles'] = [
AgentVaultSessionAccessBundle.from_dict(bundle) for bundle in data.get('accessBundles', [])
]
if data.get('accessBundle') is not None:
session_data['accessBundle'] = AgentVaultSessionAccessBundle.from_dict(data['accessBundle'])

return super().from_dict(session_data)

Expand Down
4 changes: 2 additions & 2 deletions infisical_sdk/resources/agent_vault.py
Original file line number Diff line number Diff line change
Expand Up @@ -268,7 +268,7 @@ def create(
The created session, including its token. The token is only returned once.
"""
request_body = {
"accessBundles": [access_bundle],
"accessBundle": access_bundle,
"ttl": ttl,
"label": label,
}
Expand Down Expand Up @@ -311,7 +311,7 @@ def revoke(self, session_id: str) -> AgentVaultSession:
session_id: The ID of the session to revoke.

Returns:
The revoked session. Its accessBundles is None, because the response leaves the bundles out; call get for them.
The revoked session. Its accessBundle is None, because the response leaves the bundle out; call get for it.
"""
result = self.requests.post(
path=f"/api/v1/agent-vault/sessions/{session_id}/revoke",
Expand Down
27 changes: 21 additions & 6 deletions tests/test_agent_vault.py
Original file line number Diff line number Diff line change
Expand Up @@ -9,7 +9,7 @@
"label": None,
"status": "active",
"createdAt": "2026-10-27T00:00:00.000Z",
"accessBundles": [],
"accessBundle": {"id": "0d3f8a51-7c2e-4b9a-8e16-5a4c3b2d1f07", "name": "release"},
}


Expand Down Expand Up @@ -73,12 +73,27 @@ def test_proxies_rotate_token_is_not_retried_once_the_request_may_have_reached_t
assert api.session.post.call_count == 1


def test_sessions_revoke_leaves_access_bundles_unset_when_the_response_omits_them(api, agent_vault, json_response):
revoked = {key: value for key, value in SESSION.items() if key != "accessBundles"}
def test_sessions_revoke_leaves_the_access_bundle_unset_when_the_response_omits_it(api, agent_vault, json_response):
revoked = {key: value for key, value in SESSION.items() if key != "accessBundle"}
api.session.post.return_value = json_response({"session": dict(revoked, status="revoked")})
assert agent_vault.sessions.revoke(SESSION["id"]).accessBundles is None
assert agent_vault.sessions.revoke(SESSION["id"]).accessBundle is None


def test_sessions_get_keeps_an_empty_access_bundles_list(api, agent_vault, json_response):
def test_sessions_get_reads_the_access_bundle(api, agent_vault, json_response):
api.session.get.return_value = json_response({"session": SESSION})
assert agent_vault.sessions.get(SESSION["id"]).accessBundles == []
bundle = agent_vault.sessions.get(SESSION["id"]).accessBundle
assert (bundle.id, bundle.name) == (SESSION["accessBundle"]["id"], "release")


def test_sessions_get_keeps_the_name_of_a_deleted_access_bundle(api, agent_vault, json_response):
api.session.get.return_value = json_response({"session": dict(SESSION, accessBundle={"id": None, "name": "release"})})
bundle = agent_vault.sessions.get(SESSION["id"]).accessBundle
assert (bundle.id, bundle.name) == (None, "release")


def test_sessions_create_sends_one_access_bundle(api, agent_vault, json_response):
created = {"id": SESSION["id"], "token": "agv_token", "createdAt": SESSION["createdAt"], "accessBundle": SESSION["accessBundle"]}
api.session.post.return_value = json_response({"session": created})
session = agent_vault.sessions.create(access_bundle="release", ttl="1h")
assert api.session.post.call_args.kwargs["json"] == {"accessBundle": "release", "ttl": "1h"}
assert session.accessBundle.name == "release"
Loading