Skip to content
Closed
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
4 changes: 2 additions & 2 deletions src/pytfe/resources/policy_check.py
Original file line number Diff line number Diff line change
Expand Up @@ -43,7 +43,7 @@ def list(
for d in jd.get("data", []):
attrs = d.get("attributes", {})
attrs["id"] = d.get("id")
attrs["run"] = d.get("relationships", {}).get("run", {})
attrs["run"] = d.get("relationships", {}).get("run", {}).get("data", {})
items.append(PolicyCheck.model_validate(attrs))
return PolicyCheckList(
items=items,
Expand All @@ -66,7 +66,7 @@ def read(self, policy_check_id: str) -> PolicyCheck:
d = jd.get("data", {})
attrs = d.get("attributes", {})
attrs["id"] = d.get("id")
attrs["run"] = d.get("relationships", {}).get("run", {})
attrs["run"] = d.get("relationships", {}).get("run", {}).get("data", {})
return PolicyCheck.model_validate(attrs)

def override(self, policy_check_id: str) -> PolicyCheck:
Expand Down
121 changes: 121 additions & 0 deletions tests/units/test_policy_check.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,121 @@
"""Unit tests for the policy evaluation module."""

from unittest.mock import Mock

import pytest

from pytfe._http import HTTPTransport
from pytfe.errors import InvalidRunIDError
from pytfe.models.policy_check import (
PolicyCheck,
PolicyCheckListOptions,
PolicyStatus,
)
from pytfe.resources.policy_check import PolicyChecks


class TestPolicyChecks:
"""Test the PolicyChecks service class."""

@pytest.fixture
def mock_transport(self):
"""Create a mock HTTPTransport."""
return Mock(spec=HTTPTransport)

@pytest.fixture
def policy_checks_service(self, mock_transport):
"""Create a PolicyChecks service with mocked transport."""
return PolicyChecks(mock_transport)

def test_list_validations(self, policy_checks_service):
"""Test list method with invalid task stage ID."""

# Test empty run ID
with pytest.raises(InvalidRunIDError):
list(policy_checks_service.list(""))

# Test None run ID
with pytest.raises(InvalidRunIDError):
list(policy_checks_service.list(None))

def test_list_success_with_options(self, policy_checks_service, mock_transport):
"""Test successful iteration with custom pagination options."""

mock_response_data = {
"data": [
{
"id": "polchk-9VYRc9bpfJEsnwum",
"type": "policy-checks",
"attributes": {
"result": {
"result": False,
"passed": 0,
"total-failed": 1,
"hard-failed": 0,
"soft-failed": 1,
"advisory-failed": 0,
"duration-ms": 0,
"sentinel": {},
},
"scope": "organization",
"status": "soft_failed",
"status-timestamps": {
"queued-at": "2017-11-29T20:02:17+00:00",
"soft-failed-at": "2017-11-29T20:02:20+00:00",
},
"actions": {"is-overridable": True},
"permissions": {"can-override": True},
},
"relationships": {
"run": {"data": {"id": "run-veDoQbv6xh6TbnJD", "type": "runs"}}
},
"links": {
"output": "/api/v2/policy-checks/polchk-9VYRc9bpfJEsnwum/output"
},
}
]
}

mock_response = Mock()
mock_response.json.return_value = mock_response_data
mock_transport.request.return_value = mock_response

options = PolicyCheckListOptions(page_size=5) # type: ignore
result = policy_checks_service.list("run-123", options=options)

# Verify the request was made with correct parameters
assert mock_transport.request.call_count == 1
call_args = mock_transport.request.call_args
assert call_args[0][0] == "GET"
assert call_args[0][1] == "/api/v2/runs/run-123/policy-checks"

# Verify custom options were passed and merged with _list defaults
params = call_args[1]["params"]
assert params["page[size]"] == 5 # Custom value from options

# Verify the result
print(type(result))
assert len(result.items) == 1
assert isinstance(result.items[0], PolicyCheck)
assert result.items[0].id == "polchk-9VYRc9bpfJEsnwum"
assert result.items[0].status == PolicyStatus.POLICY_SOFT_FAILED
assert result.items[0].result.advisory_failed == 0
assert result.items[0].result.total_failed == 1

def test_list_empty_result(self, policy_checks_service, mock_transport):
"""Test iteration with no results."""

mock_response_data = {"data": []}

mock_response = Mock()
mock_response.json.return_value = mock_response_data
mock_transport.request.return_value = mock_response

result = policy_checks_service.list("run-empty")

# Verify the request was made
assert mock_transport.request.call_count == 1

# Verify iterator yields no items
assert len(result.items) == 0
assert result.items == []
Loading