nat.test.plugin#

Functions#

`pytest_addoption`(parser)	Adds command line options for running specfic tests that are disabled by default
`pytest_runtest_setup`(item)
`register_components_fixture`()
`module_registry_fixture`()	Resets and returns the global type registry for testing
`function_registry_fixture`()	Resets and returns the global type registry for testing
`fail_missing_fixture`(→ bool)	Returns the value of the `fail_missing` flag, when false tests requiring unmet dependencies will be skipped, when
`require_env_variables`(→ dict[str, str])	Checks if the given environment variable is set, and returns its value if it is. If the variable is not set, and
`openai_api_key_fixture`(fail_missing)	Use for integration tests that require an Openai API key.
`nvidia_api_key_fixture`(fail_missing)	Use for integration tests that require an Nvidia API key.
`serp_api_key_fixture`(fail_missing)	Use for integration tests that require a SERP API (serpapi.com) key.
`serperdev_api_key_fixture`(fail_missing)	Use for integration tests that require a Serper Dev API (https://serper.dev) key.
`tavily_api_key_fixture`(fail_missing)	Use for integration tests that require a Tavily API key.
`mem0_api_key_fixture`(fail_missing)	Use for integration tests that require a Mem0 API key.
`aws_keys_fixture`(fail_missing)	Use for integration tests that require AWS credentials.
`azure_openai_keys_fixture`(fail_missing)	Use for integration tests that require Azure OpenAI credentials.
`langfuse_keys_fixture`(fail_missing)	Use for integration tests that require Langfuse credentials.
`wandb_api_key_fixture`(fail_missing)	Use for integration tests that require a Weights & Biases API key.
`require_weave_fixture`(→ types.ModuleType)	Use for integration tests that require Weave to be running.
`langsmith_api_key_fixture`(fail_missing)	Use for integration tests that require a LangSmith API key.
`langsmith_client_fixture`(→ langsmith.client.Client)
`project_name_fixture`(→ str)
`langsmith_project_name_fixture`(...)
`galileo_api_key_fixture`(fail_missing)	Use for integration tests that require a Galileo API key.
`galileo_project_fixture`(...)	Creates a unique Galileo project and deletes it after the test run.
`galileo_log_stream_fixture`(→ galileo.log_streams.LogStream)	Creates a Galileo log stream for integration tests.
`catalyst_keys_fixture`(fail_missing)	Use for integration tests that require RagaAI Catalyst credentials.
`catalyst_project_name_fixture`(→ str)
`catalyst_dataset_name_fixture`(→ str)	We can't create and delete projects, but we can create and delete datasets, so use a unique dataset name
`require_docker_fixture`(→ docker.client.DockerClient)	Use for integration tests that require Docker to be running.
`restore_environ_fixture`()
`root_repo_dir_fixture`(→ pathlib.Path)
`examples_dir_fixture`(→ pathlib.Path)
`env_without_nat_log_level_fixture`(→ dict[str, str])
`etcd_url_fixture`(→ str)	To run these tests, an etcd server must be running
`milvus_uri_fixture`(→ str)	To run these tests, a Milvus server must be running
`populate_milvus_fixture`(milvus_uri, root_repo_dir)	Populate Milvus with some test data.
`require_nest_asyncio_fixture`()	Some tests require the nest_asyncio2 patch to be applied to allow nested event loops, calling
`phoenix_url_fixture`(→ str)	To run these tests, a phoenix server must be running.
`phoenix_trace_url_fixture`(→ str)	Some of our tools expect the base url provided by the phoenix_url fixture, however the
`fixture_redis_server`(...)	Fixture to safely skip redis based tests if redis is not running
`fixture_mysql_server`(...)	Fixture to safely skip MySQL based tests if MySQL is not running
`minio_server_fixture`(...)	Fixture to safely skip MinIO based tests if MinIO is not running
`langfuse_bucket_fixture`(→ collections.abc.Generator[str])
`langfuse_url_fixture`(→ str)	To run these tests, a langfuse server must be running.
`langfuse_trace_url_fixture`(→ str)	The langfuse_url fixture provides the base url, however the general.telemetry.tracing["langfuse"].endpoint expects
`oauth2_server_url_fixture`(→ str)	To run these tests, an oauth2 server must be running.
`oauth2_client_credentials_fixture`(→ dict[str, Any])	Fixture to provide OAuth2 client credentials for testing
`local_sandbox_url_fixture`(→ str)	Check if sandbox server is running before running tests.
`sandbox_config_fixture`(→ dict[str, Any])	Configuration for sandbox testing.
`piston_url_fixture`(→ str)	Verify that a Piston server is running and has the required python version installed.
`import_adk_early`()	Import ADK early to work-around slow import issue (google/adk-python#2433),

Module Contents#

pytest_addoption(parser: pytest.Parser)#: Adds command line options for running specfic tests that are disabled by default

pytest_runtest_setup(item)#

register_components_fixture()#

module_registry_fixture()#

Resets and returns the global type registry for testing

This gets automatically used at the module level to ensure no state is leaked between modules

function_registry_fixture()#

Resets and returns the global type registry for testing

This gets automatically used at the function level to ensure no state is leaked between functions

fail_missing_fixture(pytestconfig: pytest.Config) → bool#: Returns the value of the fail_missing flag, when false tests requiring unmet dependencies will be skipped, when True they will fail.

require_env_variables( varnames: list[str], reason: str, fail_missing: bool = False, ) → dict[str, str]#: Checks if the given environment variable is set, and returns its value if it is. If the variable is not set, and fail_missing is False the test will ve skipped, otherwise a RuntimeError will be raised.

openai_api_key_fixture(fail_missing: bool)#: Use for integration tests that require an Openai API key.

nvidia_api_key_fixture(fail_missing: bool)#: Use for integration tests that require an Nvidia API key.

serp_api_key_fixture(fail_missing: bool)#: Use for integration tests that require a SERP API (serpapi.com) key.

serperdev_api_key_fixture(fail_missing: bool)#: Use for integration tests that require a Serper Dev API (https://serper.dev) key.

tavily_api_key_fixture(fail_missing: bool)#: Use for integration tests that require a Tavily API key.

mem0_api_key_fixture(fail_missing: bool)#: Use for integration tests that require a Mem0 API key.

aws_keys_fixture(fail_missing: bool)#: Use for integration tests that require AWS credentials.

azure_openai_keys_fixture(fail_missing: bool)#: Use for integration tests that require Azure OpenAI credentials.

langfuse_keys_fixture(fail_missing: bool)#: Use for integration tests that require Langfuse credentials.

wandb_api_key_fixture(fail_missing: bool)#: Use for integration tests that require a Weights & Biases API key.

require_weave_fixture(fail_missing: bool) → types.ModuleType#: Use for integration tests that require Weave to be running.

langsmith_api_key_fixture(fail_missing: bool)#: Use for integration tests that require a LangSmith API key.

langsmith_client_fixture( langsmith_api_key: str, fail_missing: bool, ) → langsmith.client.Client#

project_name_fixture() → str#

langsmith_project_name_fixture( langsmith_client: langsmith.client.Client, project_name: str, ) → collections.abc.Generator[str]#

galileo_api_key_fixture(fail_missing: bool)#: Use for integration tests that require a Galileo API key.

galileo_project_fixture( galileo_api_key: str, fail_missing: bool, project_name: str, ) → collections.abc.Generator[galileo.projects.Project]#: Creates a unique Galileo project and deletes it after the test run.

galileo_log_stream_fixture( galileo_project: galileo.projects.Project, ) → galileo.log_streams.LogStream#

Creates a Galileo log stream for integration tests.

The log stream is automatically deleted when the associated project is deleted.

catalyst_keys_fixture(fail_missing: bool)#: Use for integration tests that require RagaAI Catalyst credentials.

catalyst_project_name_fixture(catalyst_keys) → str#

catalyst_dataset_name_fixture( catalyst_project_name: str, project_name: str, ) → str#: We can’t create and delete projects, but we can create and delete datasets, so use a unique dataset name

require_docker_fixture( fail_missing: bool, ) → docker.client.DockerClient#: Use for integration tests that require Docker to be running.

restore_environ_fixture()#

root_repo_dir_fixture() → pathlib.Path#

examples_dir_fixture(root_repo_dir: pathlib.Path) → pathlib.Path#

env_without_nat_log_level_fixture() → dict[str, str]#

etcd_url_fixture(fail_missing: bool = False) → str#: To run these tests, an etcd server must be running

milvus_uri_fixture(etcd_url: str, fail_missing: bool = False) → str#: To run these tests, a Milvus server must be running

populate_milvus_fixture(milvus_uri: str, root_repo_dir: pathlib.Path)#: Populate Milvus with some test data.

require_nest_asyncio_fixture()#: Some tests require the nest_asyncio2 patch to be applied to allow nested event loops, calling nest_asyncio2.apply() more than once is a no-op. However we need to ensure that the nest_asyncio2 patch is applied prior to the older nest_asyncio patch is applied. Requiring us to ensure that any library which will apply the patch on import is lazily imported.

phoenix_url_fixture(fail_missing: bool) → str#: To run these tests, a phoenix server must be running. The phoenix server can be started by running the following command: docker run -p 6006:6006 -p 4317:4317 arizephoenix/phoenix:latest

phoenix_trace_url_fixture(phoenix_url: str) → str#: Some of our tools expect the base url provided by the phoenix_url fixture, however the general.telemetry.tracing[“phoenix”].endpoint expects the trace url which is what this fixture provides.

fixture_redis_server( fail_missing: bool, ) → collections.abc.Generator[dict[str, str | int]]#: Fixture to safely skip redis based tests if redis is not running

async fixture_mysql_server( fail_missing: bool, ) → collections.abc.AsyncGenerator[dict[str, str | int]]#: Fixture to safely skip MySQL based tests if MySQL is not running

minio_server_fixture( fail_missing: bool, ) → collections.abc.Generator[dict[str, str | int]]#: Fixture to safely skip MinIO based tests if MinIO is not running

langfuse_bucket_fixture( fail_missing: bool, minio_server: dict[str, str | int], ) → collections.abc.Generator[str]#

langfuse_url_fixture(fail_missing: bool, langfuse_bucket: str) → str#: To run these tests, a langfuse server must be running.

langfuse_trace_url_fixture(langfuse_url: str) → str#: The langfuse_url fixture provides the base url, however the general.telemetry.tracing[“langfuse”].endpoint expects the trace url which is what this fixture provides.

oauth2_server_url_fixture(fail_missing: bool) → str#: To run these tests, an oauth2 server must be running.

oauth2_client_credentials_fixture( oauth2_server_url: str, fail_missing: bool, ) → dict[str, Any]#

Fixture to provide OAuth2 client credentials for testing

Simulates the steps a user would take in a web browser to create a new OAuth2 client as documented in: examples/front_ends/simple_auth/README.md

local_sandbox_url_fixture(fail_missing: bool) → str#: Check if sandbox server is running before running tests.

sandbox_config_fixture(local_sandbox_url: str) → dict[str, Any]#: Configuration for sandbox testing.

piston_url_fixture(fail_missing: bool) → str#: Verify that a Piston server is running and has the required python version installed.

import_adk_early()#

Import ADK early to work-around slow import issue (google/adk-python#2433), when ADK is imported early it takes about 8 seconds, however if we wait until the packages/nvidia_nat_adk/tests run the same import will take about 70 seconds.

Since ADK is an optional dependency, we will ignore any import errors.