Skip to content
Merged
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
97 changes: 57 additions & 40 deletions tests/unit/authentication/test_rh_identity.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,7 @@

import base64
import json
from typing import Optional
from typing import Any, Optional

import pytest
from fastapi import HTTPException, Request
Expand All @@ -16,7 +16,7 @@


@pytest.fixture
def user_identity_data() -> dict:
def user_identity_data() -> dict[str, Any]:
"""Fixture providing valid User identity data.

Provide a valid Red Hat identity payload for a User, suitable for unit tests.
Expand Down Expand Up @@ -55,7 +55,7 @@ def user_identity_data() -> dict:


@pytest.fixture
def system_identity_data() -> dict:
def system_identity_data() -> dict[str, Any]:
"""Fixture providing valid System identity data.

Provide a sample System identity payload used by tests.
Expand Down Expand Up @@ -85,7 +85,7 @@ def system_identity_data() -> dict:


@pytest.fixture
def service_account_identity_data() -> dict:
def service_account_identity_data() -> dict[str, Any]:
"""Fixture providing valid ServiceAccount identity data.

Mirrors the canonical ServiceAccount payload from the identity-schemas repo.
Expand Down Expand Up @@ -113,7 +113,7 @@ def service_account_identity_data() -> dict:
}


def create_auth_header(identity_data: dict) -> str:
def create_auth_header(identity_data: dict[str, Any]) -> str:
"""Helper to create base64-encoded x-rh-identity header value.

Create a base64-encoded string suitable for use as an x-rh-identity header from identity data.
Expand Down Expand Up @@ -160,22 +160,22 @@ def create_request_with_header(
class TestRHIdentityData:
"""Test suite for RHIdentityData class."""

def test_user_type_extraction(self, user_identity_data: dict) -> None:
def test_user_type_extraction(self, user_identity_data: dict[str, Any]) -> None:
"""Test extraction of User identity fields."""
rh_identity = RHIdentityData(user_identity_data)

assert rh_identity.get_user_id() == "abc123"
assert rh_identity.get_username() == "user@redhat.com"

def test_system_type_extraction(self, system_identity_data: dict) -> None:
def test_system_type_extraction(self, system_identity_data: dict[str, Any]) -> None:
"""Test extraction of System identity fields."""
rh_identity = RHIdentityData(system_identity_data)

assert rh_identity.get_user_id() == "c87dcb4c-8af1-40dd-878e-60c744edddd0"
assert rh_identity.get_username() == "123"

def test_service_account_type_extraction(
self, service_account_identity_data: dict
self, service_account_identity_data: dict[str, Any]
) -> None:
"""Test extraction of ServiceAccount identity fields."""
rh_identity = RHIdentityData(service_account_identity_data)
Expand All @@ -187,7 +187,7 @@ def test_service_account_type_extraction(
)

def test_service_account_system_id_empty(
self, service_account_identity_data: dict
self, service_account_identity_data: dict[str, Any]
) -> None:
"""Test that get_system_id returns empty string for ServiceAccount type."""
rh_identity = RHIdentityData(service_account_identity_data)
Expand All @@ -205,7 +205,7 @@ def test_get_org_id(
rh_identity = RHIdentityData(identity_data)
assert rh_identity.get_org_id() == "321"

def test_get_org_id_missing(self, user_identity_data: dict) -> None:
def test_get_org_id_missing(self, user_identity_data: dict[str, Any]) -> None:
"""Test org_id returns empty string when not present."""
identity_data = {
**user_identity_data,
Expand All @@ -225,7 +225,7 @@ def test_get_org_id_missing(self, user_identity_data: dict) -> None:
],
)
def test_has_entitlement(
self, user_identity_data: dict, service: str, expected: bool
self, user_identity_data: dict[str, Any], service: str, expected: bool
) -> None:
"""Test has_entitlement returns correct result for various services."""
rh_identity = RHIdentityData(user_identity_data)
Expand All @@ -240,7 +240,7 @@ def test_has_entitlement(
],
)
def test_has_entitlements(
self, user_identity_data: dict, services: list[str], expected: bool
self, user_identity_data: dict[str, Any], services: list[str], expected: bool
) -> None:
"""Test has_entitlements returns correct result for service lists."""
rh_identity = RHIdentityData(user_identity_data)
Expand All @@ -267,7 +267,7 @@ def test_has_entitlements(
)
def test_validate_entitlements(
self,
user_identity_data: dict,
user_identity_data: dict[str, Any],
required_entitlements: Optional[list[str]],
should_raise: bool,
expected_error: Optional[str],
Expand Down Expand Up @@ -388,7 +388,7 @@ def test_validate_entitlements(
)
def test_validation_failures(
self,
missing_field: dict,
missing_field: dict[str, Any],
expected_error: str,
mocker: MockerFixture,
) -> None:
Expand Down Expand Up @@ -422,7 +422,7 @@ class TestRHIdentityAuthDependency:

@pytest.mark.asyncio
async def test_user_authentication_success(
self, mocker: MockerFixture, user_identity_data: dict
self, mocker: MockerFixture, user_identity_data: dict[str, Any]
) -> None:
"""Test successful User authentication."""
auth_dep = RHIdentityAuthDependency()
Expand All @@ -436,7 +436,7 @@ async def test_user_authentication_success(

@pytest.mark.asyncio
async def test_system_authentication_success(
self, mocker: MockerFixture, system_identity_data: dict
self, mocker: MockerFixture, system_identity_data: dict[str, Any]
) -> None:
"""Test successful System authentication."""
auth_dep = RHIdentityAuthDependency()
Expand All @@ -452,7 +452,7 @@ async def test_system_authentication_success(

@pytest.mark.asyncio
async def test_service_account_authentication_success(
self, mocker: MockerFixture, service_account_identity_data: dict
self, mocker: MockerFixture, service_account_identity_data: dict[str, Any]
) -> None:
"""Test successful ServiceAccount authentication."""
auth_dep = RHIdentityAuthDependency()
Expand Down Expand Up @@ -558,7 +558,7 @@ async def test_invalid_json(self, mocker: MockerFixture) -> None:
async def test_entitlement_validation( # pylint: disable=too-many-arguments,too-many-positional-arguments
self,
mocker: MockerFixture,
user_identity_data: dict,
user_identity_data: dict[str, Any],
required_entitlements: Optional[list[str]],
should_raise: bool,
expected_error: Optional[str],
Expand All @@ -581,7 +581,7 @@ async def test_entitlement_validation( # pylint: disable=too-many-arguments,too

@pytest.mark.asyncio
async def test_no_entitlement_data(
self, mocker: MockerFixture, user_identity_data: dict
self, mocker: MockerFixture, user_identity_data: dict[str, Any]
) -> None:
"""Test authentication succeeds when no entitlements are present in identity data."""
user_identity_data_no_entitlements = user_identity_data.copy()
Expand Down Expand Up @@ -756,7 +756,7 @@ class TestRHIdentityHeaderSizeLimit:

@pytest.mark.asyncio
async def test_header_at_exact_limit_accepted(
self, mocker: MockerFixture, user_identity_data: dict
self, mocker: MockerFixture, user_identity_data: dict[str, Any]
) -> None:
"""Test that a header at exactly the size limit is accepted."""
header_value = create_auth_header(user_identity_data)
Expand Down Expand Up @@ -815,7 +815,10 @@ class TestRHIdentityFieldValidation: # pylint: disable=too-many-public-methods
],
)
def test_user_non_string_types_rejected(
self, user_identity_data: dict, field_path: tuple[str, str], bad_value: object
self,
user_identity_data: dict[str, Any],
field_path: tuple[str, str],
bad_value: object,
) -> None:
"""Reject non-string values in User identity string fields."""
user_identity_data["identity"][field_path[0]][field_path[1]] = bad_value
Expand All @@ -840,7 +843,7 @@ def test_user_non_string_types_rejected(
)
def test_system_non_string_types_rejected(
self,
system_identity_data: dict,
system_identity_data: dict[str, Any],
field_path: tuple[str, ...],
bad_value: object,
) -> None:
Expand Down Expand Up @@ -868,7 +871,7 @@ def test_system_non_string_types_rejected(
)
def test_service_account_non_string_types_rejected(
self,
service_account_identity_data: dict,
service_account_identity_data: dict[str, Any],
field_path: tuple[str, str],
bad_value: object,
) -> None:
Expand All @@ -882,7 +885,7 @@ def test_service_account_non_string_types_rejected(

@pytest.mark.parametrize("bad_value", ["", " ", "\t", "\n"])
def test_empty_whitespace_rejected(
self, user_identity_data: dict, bad_value: str
self, user_identity_data: dict[str, Any], bad_value: str
) -> None:
"""Reject empty and whitespace-only strings."""
user_identity_data["identity"]["user"]["user_id"] = bad_value
Expand All @@ -895,49 +898,55 @@ def test_empty_whitespace_rejected(
["user\x00id", "user\nid", "user\rid", "a\x1fb", "a\x7fb"],
)
def test_control_characters_rejected(
self, user_identity_data: dict, bad_value: str
self, user_identity_data: dict[str, Any], bad_value: str
) -> None:
"""Reject strings containing control characters."""
user_identity_data["identity"]["user"]["user_id"] = bad_value
with pytest.raises(HTTPException) as exc_info:
RHIdentityData(user_identity_data)
assert exc_info.value.status_code == 400

def test_oversized_value_rejected(self, user_identity_data: dict) -> None:
def test_oversized_value_rejected(self, user_identity_data: dict[str, Any]) -> None:
"""Reject values longer than 256 characters."""
user_identity_data["identity"]["user"]["user_id"] = "a" * 257
with pytest.raises(HTTPException) as exc_info:
RHIdentityData(user_identity_data)
assert exc_info.value.status_code == 400

def test_max_length_boundary_accepted(self, user_identity_data: dict) -> None:
def test_max_length_boundary_accepted(
self, user_identity_data: dict[str, Any]
) -> None:
"""Accept values exactly 256 characters long."""
user_identity_data["identity"]["user"]["user_id"] = "a" * 256
RHIdentityData(user_identity_data)

def test_org_id_missing_accepted(self, user_identity_data: dict) -> None:
def test_org_id_missing_accepted(self, user_identity_data: dict[str, Any]) -> None:
"""Allow missing org_id."""
user_identity_data["identity"].pop("org_id", None)
RHIdentityData(user_identity_data)

def test_org_id_empty_accepted(self, user_identity_data: dict) -> None:
def test_org_id_empty_accepted(self, user_identity_data: dict[str, Any]) -> None:
"""Allow empty org_id."""
user_identity_data["identity"]["org_id"] = ""
RHIdentityData(user_identity_data)

def test_org_id_non_string_rejected(self, user_identity_data: dict) -> None:
def test_org_id_non_string_rejected(
self, user_identity_data: dict[str, Any]
) -> None:
"""Reject non-string org_id when provided and non-empty."""
user_identity_data["identity"]["org_id"] = 12345
with pytest.raises(HTTPException) as exc_info:
RHIdentityData(user_identity_data)
assert exc_info.value.status_code == 400

def test_org_id_valid_accepted(self, user_identity_data: dict) -> None:
def test_org_id_valid_accepted(self, user_identity_data: dict[str, Any]) -> None:
"""Accept valid string org_id."""
user_identity_data["identity"]["org_id"] = "valid-org-id"
RHIdentityData(user_identity_data)

def test_org_id_oversized_rejected(self, user_identity_data: dict) -> None:
def test_org_id_oversized_rejected(
self, user_identity_data: dict[str, Any]
) -> None:
"""Reject oversized org_id.

Verifies that an identity with an org_id longer than 256 characters is rejected.
Expand All @@ -950,30 +959,36 @@ def test_org_id_oversized_rejected(self, user_identity_data: dict) -> None:
RHIdentityData(user_identity_data)
assert exc_info.value.status_code == 400

def test_org_id_control_chars_rejected(self, user_identity_data: dict) -> None:
def test_org_id_control_chars_rejected(
self, user_identity_data: dict[str, Any]
) -> None:
"""Reject org_id containing control characters."""
user_identity_data["identity"]["org_id"] = "org\x00id"
with pytest.raises(HTTPException) as exc_info:
RHIdentityData(user_identity_data)
assert exc_info.value.status_code == 400

def test_valid_user_data_still_passes(self, user_identity_data: dict) -> None:
def test_valid_user_data_still_passes(
self, user_identity_data: dict[str, Any]
) -> None:
"""Regression: valid User identity data passes validation."""
RHIdentityData(user_identity_data)

def test_valid_system_data_still_passes(self, system_identity_data: dict) -> None:
def test_valid_system_data_still_passes(
self, system_identity_data: dict[str, Any]
) -> None:
"""Regression: valid System identity data passes validation."""
RHIdentityData(system_identity_data)

def test_valid_service_account_data_still_passes(
self, service_account_identity_data: dict
self, service_account_identity_data: dict[str, Any]
) -> None:
"""Regression: valid ServiceAccount identity data passes validation."""
RHIdentityData(service_account_identity_data)

@pytest.mark.parametrize("account_number", ["", None])
def test_system_empty_or_absent_account_number_accepted(
self, system_identity_data: dict, account_number: Optional[str]
self, system_identity_data: dict[str, Any], account_number: Optional[str]
) -> None:
"""Accept System identity with empty or absent account_number.

Expand Down Expand Up @@ -1022,7 +1037,7 @@ def test_system_developer_subscription_identity_accepted(self) -> None:
assert rh_identity.get_org_id() == "18939564"

def test_system_present_account_number_unchanged(
self, system_identity_data: dict
self, system_identity_data: dict[str, Any]
) -> None:
"""Regression: System with a present, non-empty account_number is unchanged.

Expand All @@ -1034,7 +1049,7 @@ def test_system_present_account_number_unchanged(

@pytest.mark.parametrize("org_id", ["", None])
def test_system_missing_org_id_rejected(
self, system_identity_data: dict, org_id: Optional[str]
self, system_identity_data: dict[str, Any], org_id: Optional[str]
) -> None:
"""Reject System identity when org_id is empty or absent.

Expand All @@ -1050,7 +1065,9 @@ def test_system_missing_org_id_rejected(
RHIdentityData(system_identity_data)
assert exc_info.value.status_code == 400

def test_system_empty_cn_rejected(self, system_identity_data: dict) -> None:
def test_system_empty_cn_rejected(
self, system_identity_data: dict[str, Any]
) -> None:
"""Reject System identity when system cn is empty."""
system_identity_data["identity"]["system"]["cn"] = ""
with pytest.raises(HTTPException) as exc_info:
Expand Down
Loading