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
29 changes: 27 additions & 2 deletions tests/integration/conftest.py
Original file line number Diff line number Diff line change
Expand Up @@ -37,6 +37,9 @@ def test_config_fixture() -> Generator:

This fixture loads the actual configuration file used in testing,
demonstrating integration with the configuration system.

Yields:
The `configuration` module with the loaded settings.
"""
config_path = (
Path(__file__).parent.parent / "configuration" / "lightspeed-stack.yaml"
Expand All @@ -56,6 +59,9 @@ def current_config_fixture() -> Generator:

This fixture loads the actual configuration file from project root (current configuration),
demonstrating integration with the configuration system.

Yields:
configuration: The loaded configuration object.
"""
config_path = Path(__file__).parent.parent.parent / "lightspeed-stack.yaml"
assert config_path.exists(), f"Config file not found: {config_path}"
Expand All @@ -73,6 +79,9 @@ def test_db_engine_fixture() -> Generator:

This provides a real database (not mocked) for integration tests.
Each test gets a fresh database.

Yields:
engine (Engine): A SQLAlchemy Engine connected to a new in-memory SQLite database.
"""
# Create in-memory SQLite database
engine = create_engine(
Expand All @@ -96,6 +105,10 @@ def test_db_session_fixture(test_db_engine: Engine) -> Generator[Session, None,
"""Create a database session for testing.

Provides a real database session connected to the in-memory test database.

Yields:
session (Session): A database session bound to the test engine; the
fixture closes the session after the test.
"""
session_local = sessionmaker(autocommit=False, autoflush=False, bind=test_db_engine)
session = session_local()
Expand All @@ -107,7 +120,12 @@ def test_db_session_fixture(test_db_engine: Engine) -> Generator[Session, None,

@pytest.fixture(name="test_request")
def test_request_fixture() -> Request:
"""Create a test FastAPI Request object with proper scope."""
"""Create a test FastAPI Request object with proper scope.

Returns:
request (fastapi.Request): A Request object whose scope has `"type":
"http"`, an empty `query_string`, and no headers.
"""
return Request(
scope={
"type": "http",
Expand All @@ -119,7 +137,11 @@ def test_request_fixture() -> Request:

@pytest.fixture(name="test_response")
def test_response_fixture() -> Response:
"""Create a test FastAPI Response object with proper scope."""
"""Create a test FastAPI Response object with proper scope.

Returns:
Response: Response with empty content, status 200, and media_type "application/json".
"""
return Response(content="", status_code=200, media_type="application/json")


Expand All @@ -129,6 +151,9 @@ async def test_auth_fixture(test_request: Request) -> AuthTuple:

This uses the actual NoopAuthDependency instead of mocking,
making this a true integration test.

Returns:
AuthTuple: Authentication information produced by NoopAuthDependency.
"""
noop_auth = NoopAuthDependency()
return await noop_auth(test_request)
14 changes: 13 additions & 1 deletion tests/integration/test_configuration.py
Original file line number Diff line number Diff line change
Expand Up @@ -21,7 +21,19 @@ def test_default_configuration() -> None:


def test_loading_proper_configuration(configuration_filename: str) -> None:
"""Test the configuration loading."""
"""Test the configuration loading.

Validate that loading the given configuration YAML populates all expected sections and values.

Loads configuration from the provided file and asserts presence and
correctness of top-level sections (configuration, service, llama_stack,
user_data_collection, mcp_servers) and selected field values including
service host and flags, CORS settings, llama stack URL and API key secret,
user data collection settings, and three MCP server entries.

Parameters:
configuration_filename (str): Path to the YAML configuration file used for the test.
"""
cfg = configuration
cfg.load_configuration(configuration_filename)

Expand Down
125 changes: 108 additions & 17 deletions tests/integration/test_openapi_json.py
Original file line number Diff line number Diff line change
Expand Up @@ -22,7 +22,20 @@


def _load_openapi_spec_from_file() -> dict[str, Any]:
"""Load OpenAPI specification from configured path."""
"""Load OpenAPI specification from configured path.

Load and return the OpenAPI JSON document from the configured file path.

If the configured file is present, its contents are parsed as JSON and
returned as a dictionary.
If the file is missing, the running test is failed via pytest.fail.

Returns:
spec (dict[str, Any]): Parsed OpenAPI specification.

Raises:
AssertionError: Causes the test to fail through pytest.fail when the file is not found.
"""
path = Path(OPENAPI_FILE)
if path.is_file():
with path.open("r", encoding="utf-8") as f:
Expand All @@ -33,7 +46,13 @@ def _load_openapi_spec_from_file() -> dict[str, Any]:


def _load_openapi_spec_from_url() -> dict[str, Any]:
"""Load OpenAPI specification from URL."""
"""Load OpenAPI specification from URL.

Retrieve the OpenAPI specification by requesting the application's /openapi.json endpoint.

Returns:
dict[str, Any]: The parsed OpenAPI specification as a dictionary.
"""
configuration_filename = "tests/configuration/lightspeed-stack-proper-name.yaml"
cfg = configuration
cfg.load_configuration(configuration_filename)
Expand All @@ -52,18 +71,42 @@ def _load_openapi_spec_from_url() -> dict[str, Any]:

@pytest.fixture(scope="module", name="spec_from_file")
def open_api_spec_from_file() -> dict[str, Any]:
"""Fixture containing OpenAPI specification represented as a dictionary."""
"""Fixture containing OpenAPI specification represented as a dictionary.

Provides the parsed OpenAPI specification as a dictionary for tests.

Returns:
openapi_spec (dict[str, Any]): The OpenAPI document parsed from docs/openapi.json.
"""
return _load_openapi_spec_from_file()


@pytest.fixture(scope="module", name="spec_from_url")
def open_api_spec_from_url() -> dict[str, Any]:
"""Fixture containing OpenAPI specification represented as a dictionary."""
"""Fixture containing OpenAPI specification represented as a dictionary.

Provides the OpenAPI specification loaded from the running application's
/openapi.json endpoint.

Returns:
dict: The OpenAPI document parsed into a dictionary.
"""
return _load_openapi_spec_from_url()


def _check_openapi_top_level_info(spec: dict[str, Any]) -> None:
"""Check all top level informations stored in OpenAPI specification."""
"""Check all top level informations stored in OpenAPI specification.

Checks that the OpenAPI version, info section (title and version), contact, and license
(name and URL) match the expected values used by the project.

Parameters:
spec (dict): Parsed OpenAPI specification document.

Raises:
AssertionError: If any required top-level field is missing or does not
match the expected value.
"""
assert spec.get("openapi") == "3.1.0"

info = spec.get("info") or {}
Expand All @@ -79,7 +122,14 @@ def _check_openapi_top_level_info(spec: dict[str, Any]) -> None:


def _check_server_section_present(spec: dict[str, Any]) -> None:
"""Check if the servers section stored in OpenAPI specification."""
"""Check if the servers section stored in OpenAPI specification.

Parameters:
spec (dict[str, Any]): Parsed OpenAPI specification.

Raises:
AssertionError: If the 'servers' field is missing, not a list, or empty.
"""
servers = spec.get("servers")
assert isinstance(servers, list) and servers, "servers must be a non-empty list"

Expand All @@ -89,14 +139,14 @@ def _check_paths_and_responses_exist(
) -> None:
"""Checks if the specified paths and responses exist in the API specification.

Args:
spec (dict): The API specification.
path (str): The API endpoint path to check.
method (str): The HTTP method to check.
expected_codes (set[str]): The set of expected HTTP status codes.
Parameters:
spec (dict): The API specification.
path (str): The API endpoint path to check.
method (str): The HTTP method to check.
expected_codes (set[str]): The set of expected HTTP status codes.

Raises:
AssertionError: If the path, method, or any of the expected response codes are missing.
Raises:
AssertionError: If the path, method, or any of the expected response codes are missing.
"""
paths = spec.get("paths") or {}
assert path in paths, f"Missing path: {path}"
Expand All @@ -111,12 +161,29 @@ def _check_paths_and_responses_exist(


def test_openapi_top_level_info_from_file(spec_from_file: dict[str, Any]) -> None:
"""Test all top level informations stored in OpenAPI specification."""
"""Test all top level informations stored in OpenAPI specification.

Asserts that the OpenAPI version, info (title and version), contact, and
license fields meet the repository's expected values.

Parameters:
spec_from_file (dict[str, Any]): OpenAPI specification dictionary
loaded from docs/openapi.json.
"""
_check_openapi_top_level_info(spec_from_file)


def test_openapi_top_level_info_from_url(spec_from_url: dict[str, Any]) -> None:
"""Test all top level informations stored in OpenAPI specification."""
"""Test all top level informations stored in OpenAPI specification.

Asserts that the OpenAPI version, info section (including title, version,
and contact), and license (name and URL) meet the project's expectations
used by the tests.

Parameters:
spec_from_url (dict[str, Any]): OpenAPI document parsed from the
application's /openapi.json endpoint.
"""
_check_openapi_top_level_info(spec_from_url)


Expand Down Expand Up @@ -196,7 +263,19 @@ def test_servers_section_present_from_url(spec_from_url: dict[str, Any]) -> None
def test_paths_and_responses_exist_from_file(
spec_from_file: dict, path: str, method: str, expected_codes: set[str]
) -> None:
"""Tests all paths defined in OpenAPI specification."""
"""Tests all paths defined in OpenAPI specification.

Verify that the given path and HTTP method are defined in the provided
OpenAPI specification and that the operation contains all expected response
status codes.

Parameters:
spec_from_file (dict): OpenAPI specification document loaded from the local file.
path (str): API path to check (e.g., "/items/{id}").
method (str): HTTP method to check for the path (e.g., "get", "post").
expected_codes (set[str]): Set of expected HTTP response status codes
as strings (e.g., {"200", "404"}).
"""
_check_paths_and_responses_exist(spec_from_file, path, method, expected_codes)


Expand Down Expand Up @@ -266,5 +345,17 @@ def test_paths_and_responses_exist_from_file(
def test_paths_and_responses_exist_from_url(
spec_from_url: dict, path: str, method: str, expected_codes: set[str]
) -> None:
"""Tests all paths defined in OpenAPI specification."""
"""Tests all paths defined in OpenAPI specification.

Verify that the OpenAPI spec served at /openapi.json contains the given
path and HTTP method and that the operation declares the specified response
status codes.

Parameters:
path (str): OpenAPI path string to check (for example, "/items/{id}").
method (str): HTTP method name for the operation to check (e.g., "get",
"post"); case-insensitive.
expected_codes (set[str]): Set of response status code strings expected
to be present for the operation (for example, {"200", "404"}).
"""
_check_paths_and_responses_exist(spec_from_url, path, method, expected_codes)
39 changes: 35 additions & 4 deletions tests/integration/test_rh_identity_integration.py
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,14 @@

@pytest.fixture
def client() -> Generator[TestClient, None, None]:
"""Create test client for FastAPI app with RH Identity config."""
"""Create test client for FastAPI app with RH Identity config.

Provides a TestClient for the FastAPI application configured with the RH
Identity test configuration.

Yields:
TestClient: A test client instance for the FastAPI app.
"""
# Save original env var if it exists
original_config = os.environ.get("LIGHTSPEED_STACK_CONFIG_PATH")

Expand Down Expand Up @@ -63,7 +70,16 @@ def user_identity_json() -> dict:

@pytest.fixture
def system_identity_json() -> dict:
"""Fixture providing valid System identity JSON."""
"""Fixture providing valid System identity JSON.

Provide a valid RH Identity "System" payload suitable for tests.

Returns:
dict: JSON with keys:
- "identity": contains "account_number", "org_id", "type" set to
"System", and "system" with "cn".
- "entitlements": contains "rhel" with "is_entitled" and "is_trial" boolean flags.
"""
return {
"identity": {
"account_number": "456",
Expand All @@ -78,7 +94,14 @@ def system_identity_json() -> dict:


def encode_identity(identity_json: dict) -> str:
"""Encode identity JSON to base64."""
"""Encode identity JSON to base64.

Parameters:
identity_json (dict): JSON-serializable identity payload to encode.

Returns:
str: Base64-encoded UTF-8 string representation of the JSON payload.
"""
json_str = json.dumps(identity_json)
return base64.b64encode(json_str.encode("utf-8")).decode("utf-8")

Expand All @@ -89,7 +112,15 @@ class TestRHIdentityIntegration:
def test_valid_user_identity(
self, client: TestClient, user_identity_json: dict
) -> None:
"""Test successful request with valid User identity."""
"""Test successful request with valid User identity.

Verify that a GET to /api/v1/conversations with a valid User RH
Identity is accepted.

Sends the provided User identity encoded in the `x-rh-identity` header
and asserts the response status code is `200` (success) or `404` (no
conversations).
"""
headers = {"x-rh-identity": encode_identity(user_identity_json)}

response = client.get("/api/v1/conversations", headers=headers)
Expand Down
23 changes: 21 additions & 2 deletions tests/integration/test_version.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,19 @@


def read_version_from_pyproject() -> str:
"""Read version from pyproject.toml file."""
"""Read version from pyproject.toml file.

Retrieve the project's version as reported by the PDM tool.

Invokes the `pdm show --version` command and returns the resulting version string
decoded as UTF-8 with surrounding whitespace removed.

Returns:
version (str): The project version reported by PDM.

Raises:
subprocess.CalledProcessError: If the `pdm` command exits with a non-zero status.
"""
# it is not safe to just try to read version from pyproject.toml file directly
# the PDM tool itself is able to retrieve the version, even if the version
# is generated dynamically
Expand All @@ -19,7 +31,14 @@ def read_version_from_pyproject() -> str:


def test_version_handling() -> None:
"""Test how version is handled by the project."""
"""Test how version is handled by the project.

Verify that the package's source __version__ matches the version reported by the project tool.

Raises:
AssertionError: If the source version and the project-reported version
differ; the message includes both versions.
"""
source_version = __version__
project_version = read_version_from_pyproject()
assert (
Expand Down
Loading