API Reference - Core Module

This section documents the core framework classes and functions.

BasePage

The BasePage class is the foundation for all API testing in socialseed-e2e.

Enhanced BasePage with logging, retries, and rate limiting.

This module provides an enhanced BasePage class for API testing with production-ready features including structured logging, automatic retries, rate limiting, and comprehensive request/response logging.

class socialseed_e2e.core.base_page.RetryConfig(max_retries: int = 3, backoff_factor: float = 1.0, max_backoff: float = 60.0, retry_on: List[int] | None = None, retry_exceptions: List[Type[Exception]] | None = None)[source]

Bases: object

Configuration for automatic retry mechanism.

max_retries

Maximum number of retry attempts (default: 3)

Type:: int

backoff_factor

Exponential backoff multiplier (default: 1.0)

Type:: float

max_backoff

Maximum backoff time in seconds (default: 60)

Type:: float

retry_on

List of HTTP status codes to retry on (default: [502, 503, 504, 429])

Type:: List[int] | None

retry_exceptions

List of exception types to retry on

Type:: List[Type[Exception]] | None

max_retries: int = 3

backoff_factor: float = 1.0

max_backoff: float = 60.0

retry_on: List[int] | None = None

retry_exceptions: List[Type[Exception]] | None = None

__post_init__()[source]: Initialize default retry configuration values.

__init__(max_retries: int = 3, backoff_factor: float = 1.0, max_backoff: float = 60.0, retry_on: List[int] | None = None, retry_exceptions: List[Type[Exception]] | None = None) → None

class socialseed_e2e.core.base_page.RateLimitConfig(enabled: bool = False, requests_per_second: float = 10.0, requests_per_minute: float = 600.0, burst_size: int = 5)[source]

Bases: object

Configuration for rate limiting.

enabled

Whether rate limiting is enabled (default: False)

Type:: bool

requests_per_second

Maximum requests per second (default: 10)

Type:: float

requests_per_minute

Maximum requests per minute (default: 600)

Type:: float

burst_size

Allow burst of requests (default: 5)

Type:: int

enabled: bool = False

requests_per_second: float = 10.0

requests_per_minute: float = 600.0

burst_size: int = 5

__init__(enabled: bool = False, requests_per_second: float = 10.0, requests_per_minute: float = 600.0, burst_size: int = 5) → None

class socialseed_e2e.core.base_page.RequestLog(method: str, url: str, headers: Dict[str, str], body: str | None, timestamp: float, duration_ms: float = 0.0, status: int | None = None, response_headers: Dict[str, str] | None = None, response_body: str | None = None, error: str | None = None)[source]

Bases: object

Log entry for a single request.

method

HTTP method

Type:: str

url

Full request URL

Type:: str

headers

Request headers (may be filtered)

Type:: Dict[str, str]

body

Request body (JSON string)

Type:: str | None

timestamp

When the request was made

Type:: float

duration_ms

Request duration in milliseconds

Type:: float

status

Response status code

Type:: int | None

response_headers

Response headers

Type:: Dict[str, str] | None

response_body

Response body (truncated if too large)

Type:: str | None

error

Error message if request failed

Type:: str | None

method: str

url: str

headers: Dict[str, str]

body: str | None

timestamp: float

duration_ms: float = 0.0

status: int | None = None

response_headers: Dict[str, str] | None = None

response_body: str | None = None

error: str | None = None

__init__(method: str, url: str, headers: Dict[str, str], body: str | None, timestamp: float, duration_ms: float = 0.0, status: int | None = None, response_headers: Dict[str, str] | None = None, response_body: str | None = None, error: str | None = None) → None

class socialseed_e2e.core.base_page.ServiceHealth(healthy: bool, status_code: int | None = None, response_time_ms: float = 0.0, message: str = '', timestamp: float = <factory>)[source]

Bases: object

Service health status.

healthy

Whether the service is healthy

Type:: bool

status_code

HTTP status code from health check

Type:: int | None

response_time_ms

Response time in milliseconds

Type:: float

message

Health check message

Type:: str

timestamp

When the check was performed

Type:: float

healthy: bool

status_code: int | None = None

response_time_ms: float = 0.0

message: str = ''

timestamp: float

__init__(healthy: bool, status_code: int | None = None, response_time_ms: float = 0.0, message: str = '', timestamp: float = <factory>) → None

Bases: Exception

Enhanced exception with request context.

__init__(message: str, url: str | None = None, method: str | None = None, status: int | None = None, response_text: str | None = None, request_log: RequestLog | None = None)[source]: Initialize exception with request context.

__str__() → str[source]: Return formatted error message with context.

class socialseed_e2e.core.base_page.BasePage(base_url: str, playwright: Playwright | None = None, default_headers: Dict[str, str] | None = None, retry_config: RetryConfig | None = None, rate_limit_config: RateLimitConfig | None = None, enable_request_logging: bool = True, max_log_body_size: int = 10000, health_endpoint: str = '/actuator/health')[source]

Bases: object

Enhanced base class for API testing with logging, retries, and rate limiting.

This class extends the basic API testing capabilities with production-ready features including:

Structured logging of all requests and responses
Automatic retry mechanism with exponential backoff
Rate limiting to avoid overwhelming APIs
Request timing and performance metrics
Enhanced error messages with full context
Helper methods for common assertions
Service health checking

Example

>>> page = BasePage("https://api.example.com")
>>> page.setup()
>>>
>>> # Enable retries for transient failures
>>> page.retry_config = RetryConfig(max_retries=3)
>>>
>>> # Enable rate limiting
>>> page.rate_limit_config = RateLimitConfig(
...     enabled=True,
...     requests_per_second=5
... )
>>>
>>> # Make request with automatic retry and logging
>>> response = page.get("/users/123")
>>>
>>> # Use helper methods for assertions
>>> page.assert_status(response, 200)
>>> user = page.assert_json(response)
>>>
>>> page.teardown()

base_url: The base URL for the API

default_headers: Headers applied to all requests

retry_config: Configuration for automatic retries

rate_limit_config: Configuration for rate limiting

enable_request_logging: Whether to log all requests

max_log_body_size: Maximum size for logged request/response bodies

request_history: List of RequestLog entries for recent requests

__init__(base_url: str, playwright: Playwright | None = None, default_headers: Dict[str, str] | None = None, retry_config: RetryConfig | None = None, rate_limit_config: RateLimitConfig | None = None, enable_request_logging: bool = True, max_log_body_size: int = 10000, health_endpoint: str = '/actuator/health') → None[source]

Initialize the BasePage.

Parameters:

base_url – The base URL for the API (e.g., “https://api.example.com”)
playwright – Optional Playwright instance (created if not provided)
default_headers – Headers to include in all requests
retry_config – Configuration for automatic retries (default: no retries)
rate_limit_config – Configuration for rate limiting (default: disabled)
enable_request_logging – Whether to log requests and responses
max_log_body_size – Maximum size for logged bodies (truncated if larger)
health_endpoint – Endpoint for health checks (default: /actuator/health)

base_url: str

playwright_manager: Any | None

playwright: Playwright | None

api_context: APIRequestContext | None

request_history: List[RequestLog]

metadata: Dict[str, Any]

set_metadata(key: str, value: Any) → None[source]: Set a metadata value for sharing state between tests.

get_metadata(key: str, default: Any = None) → Any[source]: Get a metadata value.

expect(value: Any, name: str = 'value') → AssertionBuilder[source]: Start a fluent assertion.

Example

page.expect(user_id).exists().equals(123)

classmethod from_config(config: ServiceConfig, playwright: Playwright | None = None, **kwargs) → BasePage[source]

Create a BasePage from a ServiceConfig object.

Parameters:

config – Service configuration object
playwright – Optional Playwright instance
**kwargs – Additional arguments passed to BasePage constructor

Returns:

Configured BasePage instance

setup() → None[source]

Initialize the API context.

This method creates the Playwright APIRequestContext. It is called automatically before making requests if not already set up.

teardown() → None[source]

Clean up the API context and resources.

Always call this method when done to release resources properly.

get(endpoint: str, headers: Dict[str, str] | None = None, params: Dict[str, Any] | None = None) → APIResponse[source]

Perform a GET request.

Parameters:

endpoint – API endpoint (e.g., “/users/123”)
headers – Optional request-specific headers
params – Optional query parameters

Returns:

APIResponse object

post(endpoint: str, data: Dict[str, Any] | str | None = None, json: Dict[str, Any] | None = None, headers: Dict[str, str] | None = None) → APIResponse[source]

Perform a POST request.

Parameters:

endpoint – API endpoint (e.g., “/users”)
data – Form data or dict (use either data or json, not both)
json – JSON payload (use either data or json, not both)
headers – Optional request-specific headers

Returns:

APIResponse object

put(endpoint: str, data: Dict[str, Any] | str | None = None, json: Dict[str, Any] | None = None, headers: Dict[str, str] | None = None) → APIResponse[source]

Perform a PUT request.

Parameters:

endpoint – API endpoint (e.g., “/users/123”)
data – Form data or dict (use either data or json, not both)
json – JSON payload (use either data or json, not both)
headers – Optional request-specific headers

Returns:

APIResponse object

delete(endpoint: str, data: Dict[str, Any] | str | None = None, json: Dict[str, Any] | None = None, headers: Dict[str, str] | None = None) → APIResponse[source]

Perform a DELETE request.

Parameters:

endpoint – API endpoint (e.g., “/users/123”)
data – Request body as dict or string (for non-standard REST APIs)
json – JSON payload (alternative to data)
headers – Optional request-specific headers

Returns:

APIResponse object

patch(endpoint: str, data: Dict[str, Any] | str | None = None, json: Dict[str, Any] | None = None, headers: Dict[str, str] | None = None) → APIResponse[source]

Perform a PATCH request.

Parameters:

endpoint – API endpoint (e.g., “/users/123”)
data – Form data or dict (use either data or json, not both)
json – JSON payload (use either data or json, not both)
headers – Optional request-specific headers

Returns:

APIResponse object

assert_status(response: APIResponse, expected_status: int | List[int], message: str | None = None) → APIResponse[source]

Assert that response status matches expected.

Parameters:

response – The API response to check
expected_status – Expected status code or list of acceptable codes
message – Optional custom error message

Returns:

The response (for chaining)

Raises:

BasePageError – If status doesn’t match

assert_ok(response: APIResponse) → APIResponse[source]

Assert that response status is 2xx (success).

Parameters:: response – The API response to check
Returns:: The response (for chaining)
Raises:: BasePageError – If status is not 2xx

assert_json(response: APIResponse, key: str | None = None) → Any[source]

Parse response as JSON with optional key extraction.

Parameters:

response – The API response to parse
key – Optional key to extract from JSON (e.g., “data.user.name”)

Returns:

Parsed JSON data, or value at key if specified

Raises:

BasePageError – If JSON parsing fails or key not found

assert_header(response: APIResponse, header_name: str, expected_value: str | None = None) → str[source]

Assert that response contains a specific header.

Parameters:

response – The API response to check
header_name – Name of the header to check (case-insensitive)
expected_value – Optional expected value (if None, just checks existence)

Returns:

The header value

Raises:

BasePageError – If header not found or value doesn’t match

assert_schema(response: APIResponse, schema: Dict[str, Any] | type, message: str | None = None) → Dict[str, Any][source]

Assert that response matches a JSON schema or Pydantic model.

Parameters:

response – The API response to validate
schema – JSON schema dict or Pydantic model class
message – Optional custom error message

Returns:

The parsed JSON data

Raises:

BasePageError – If validation fails

get_response_text(response: APIResponse) → str[source]

Get response text from Playwright APIResponse.

Parameters:: response – The API response
Returns:: Response body as string

get_last_request() → RequestLog | None[source]

Get the most recent request log entry.

Returns:: The last RequestLog or None if no requests made

get_request_stats() → Dict[str, Any][source]

Get statistics about requests made.

Returns:

total_requests: Total number of requests
successful_requests: Number of 2xx responses
failed_requests: Number of non-2xx responses
total_duration_ms: Total time spent in requests
average_duration_ms: Average request duration
status_distribution: Count of each status code

Return type:

Dictionary with request statistics

add_response_interceptor(interceptor: Callable[[APIResponse], None]) → None[source]

Add a response interceptor.

Interceptors are called after each successful response. They receive the APIResponse object and can be used for: - Logging - Token extraction - Response transformation - Metrics collection

Parameters:: interceptor – Callable that receives APIResponse

clear_response_interceptors() → None[source]: Remove all response interceptors.

check_health(timeout: int | None = None) → ServiceHealth[source]

Check if the service is healthy.

Makes a request to the health endpoint and returns status.

Parameters:: timeout – Request timeout in milliseconds (uses default if not specified)
Returns:: ServiceHealth object with health status

wait_for_healthy(timeout: int = 30, interval: float = 1.0, raise_on_timeout: bool = True) → bool[source]

Wait for the service to become healthy.

Polls the health endpoint until it returns healthy or timeout is reached.

Parameters:

timeout – Maximum time to wait in seconds
interval – Polling interval in seconds
raise_on_timeout – Whether to raise exception on timeout

Returns:

True if service became healthy, False if timed out

Raises:

BasePageError – If raise_on_timeout is True and service doesn’t become healthy

classmethod wait_for_service(url: str, health_endpoint: str = '/actuator/health', timeout: int = 30, interval: float = 1.0) → bool[source]

Wait for a service to become healthy without creating a persistent instance.

This is a class method that creates a temporary instance just for health checking.

Parameters:

url – Service base URL
health_endpoint – Health check endpoint
timeout – Maximum wait time in seconds
interval – Polling interval

Returns:

True if service became healthy

Example

>>> BasePage.wait_for_service("http://localhost:8081", timeout=60)

Configuration

ApiConfigLoader - Centralized Configuration Management

This module provides a centralized way to load and manage API configuration from the api.conf file. It supports environment variable substitution and provides easy access to all service configurations.

socialseed_e2e.core.config_loader.normalize_service_name(name: str) → str[source]

Normalize service name by converting hyphens to underscores.

This ensures that service names like ‘auth-service’ and ‘auth_service’ are treated as identical for matching purposes.

Parameters:: name – Service name (may contain hyphens or underscores)
Returns:: Normalized service name with underscores
Return type:: str

Examples

>>> normalize_service_name("auth-service")
'auth_service'
>>> normalize_service_name("auth_service")
'auth_service'
>>> normalize_service_name("user-api-v2")
'user_api_v2'

class socialseed_e2e.core.config_loader.ServiceEndpoint(name: str, path: str, method: str = 'POST', requires_auth: bool = False)[source]

Bases: object

Represents a single API endpoint configuration.

name: str

path: str

method: str = 'POST'

requires_auth: bool = False

__init__(name: str, path: str, method: str = 'POST', requires_auth: bool = False) → None

class socialseed_e2e.core.config_loader.ServiceConfig(name: str, base_url: str, health_endpoint: str = '/actuator/health', port: int = 8080, maven_module: str = '', timeout: int = 30000, headers: Dict[str, str]=<factory>, auto_start: bool = True, required: bool = True, endpoints: Dict[str, str]=<factory>)[source]

Bases: object

Complete configuration for a microservice.

name: str

base_url: str

health_endpoint: str = '/actuator/health'

port: int = 8080

maven_module: str = ''

timeout: int = 30000

headers: Dict[str, str]

auto_start: bool = True

required: bool = True

endpoints: Dict[str, str]

__init__(name: str, base_url: str, health_endpoint: str = '/actuator/health', port: int = 8080, maven_module: str = '', timeout: int = 30000, headers: Dict[str, str]=<factory>, auto_start: bool = True, required: bool = True, endpoints: Dict[str, str]=<factory>) → None

class socialseed_e2e.core.config_loader.ApiGatewayConfig(enabled: bool = False, url: str = '', prefix: str = '', auth_type: str = 'none', auth_token: str | None = None, api_key_header: str | None = None, api_key_value: str | None = None)[source]

Bases: object

API Gateway configuration.

enabled: bool = False

url: str = ''

prefix: str = ''

auth_type: str = 'none'

auth_token: str | None = None

api_key_header: str | None = None

api_key_value: str | None = None

__init__(enabled: bool = False, url: str = '', prefix: str = '', auth_type: str = 'none', auth_token: str | None = None, api_key_header: str | None = None, api_key_value: str | None = None) → None

class socialseed_e2e.core.config_loader.DatabaseConfig(host: str = 'localhost', port: int = 5432, database: str = '', username: str = '', password: str = '', enabled: bool = False)[source]

Bases: object

Database connection configuration.

host: str = 'localhost'

port: int = 5432

database: str = ''

username: str = ''

password: str = ''

enabled: bool = False

__init__(host: str = 'localhost', port: int = 5432, database: str = '', username: str = '', password: str = '', enabled: bool = False) → None

class socialseed_e2e.core.config_loader.TestDataConfig(email_domain: str = 'test.socialseed.com', password: str = 'StrongPass123!', username_prefix: str = 'testuser', step_delay: int = 100, async_timeout: int = 10000, max_retries: int = 3, retry_backoff_ms: int = 1000)[source]

Bases: object

Test data generation configuration.

email_domain: str = 'test.socialseed.com'

password: str = 'StrongPass123!'

username_prefix: str = 'testuser'

step_delay: int = 100

async_timeout: int = 10000

max_retries: int = 3

retry_backoff_ms: int = 1000

__init__(email_domain: str = 'test.socialseed.com', password: str = 'StrongPass123!', username_prefix: str = 'testuser', step_delay: int = 100, async_timeout: int = 10000, max_retries: int = 3, retry_backoff_ms: int = 1000) → None

class socialseed_e2e.core.config_loader.SecurityConfig(verify_ssl: bool = True, ssl_cert: str | None = None, ssl_key: str | None = None, ssl_ca: str | None = None, test_tokens: Dict[str, str]=<factory>)[source]

Bases: object

Security and SSL configuration.

verify_ssl: bool = True

ssl_cert: str | None = None

ssl_key: str | None = None

ssl_ca: str | None = None

test_tokens: Dict[str, str]

__init__(verify_ssl: bool = True, ssl_cert: str | None = None, ssl_key: str | None = None, ssl_ca: str | None = None, test_tokens: Dict[str, str]=<factory>) → None

class socialseed_e2e.core.config_loader.ReportingConfig(format: str = 'console', save_logs: bool = True, log_dir: str = './logs', include_payloads: bool = False, screenshot_on_failure: bool = False)[source]

Bases: object

Test reporting configuration.

format: str = 'console'

save_logs: bool = True

log_dir: str = './logs'

include_payloads: bool = False

screenshot_on_failure: bool = False

__init__(format: str = 'console', save_logs: bool = True, log_dir: str = './logs', include_payloads: bool = False, screenshot_on_failure: bool = False) → None

class socialseed_e2e.core.config_loader.ParallelConfig(enabled: bool = False, max_workers: int | None = None, mode: str = 'service', isolation_level: str = 'process')[source]

Bases: object

Parallel test execution configuration.

enabled

Whether parallel execution is enabled (default: False)

Type:: bool

max_workers

Maximum number of parallel workers (None = auto)

Type:: int | None

mode

Execution mode (‘service’ or ‘test’)

Type:: str

isolation_level

State isolation level (‘process’, ‘service’, ‘none’)

Type:: str

enabled: bool = False

max_workers: int | None = None

mode: str = 'service'

isolation_level: str = 'process'

__init__(enabled: bool = False, max_workers: int | None = None, mode: str = 'service', isolation_level: str = 'process') → None

class socialseed_e2e.core.config_loader.AppConfig(environment: str = 'dev', timeout: int = 30000, user_agent: str = 'SocialSeed-E2E-Agent/2.0', verification_level: str = 'strict', verbose: bool = True, project_name: str = 'SocialSeed', project_version: str = '0.0.0', api_gateway: ApiGatewayConfig = <factory>, services: Dict[str, ~socialseed_e2e.core.config_loader.ServiceConfig]=<factory>, databases: Dict[str, ~socialseed_e2e.core.config_loader.DatabaseConfig]=<factory>, test_data: TestDataConfig = <factory>, security: SecurityConfig = <factory>, reporting: ReportingConfig = <factory>, parallel: ParallelConfig = <factory>)[source]

Bases: object

Main application configuration container.

environment: str = 'dev'

timeout: int = 30000

user_agent: str = 'SocialSeed-E2E-Agent/2.0'

verification_level: str = 'strict'

verbose: bool = True

project_name: str = 'SocialSeed'

project_version: str = '0.0.0'

api_gateway: ApiGatewayConfig

services: Dict[str, ServiceConfig]

databases: Dict[str, DatabaseConfig]

test_data: TestDataConfig

security: SecurityConfig

reporting: ReportingConfig

parallel: ParallelConfig

__init__(environment: str = 'dev', timeout: int = 30000, user_agent: str = 'SocialSeed-E2E-Agent/2.0', verification_level: str = 'strict', verbose: bool = True, project_name: str = 'SocialSeed', project_version: str = '0.0.0', api_gateway: ApiGatewayConfig = <factory>, services: Dict[str, ~socialseed_e2e.core.config_loader.ServiceConfig]=<factory>, databases: Dict[str, ~socialseed_e2e.core.config_loader.DatabaseConfig]=<factory>, test_data: TestDataConfig = <factory>, security: SecurityConfig = <factory>, reporting: ReportingConfig = <factory>, parallel: ParallelConfig = <factory>) → None

exception socialseed_e2e.core.config_loader.ConfigError[source]

Bases: Exception

Custom exception for configuration errors.

class socialseed_e2e.core.config_loader.ApiConfigLoader[source]

Bases: object

Loads and manages API configuration from api.conf file.

Supports: - Environment variable substitution: ${VAR_NAME} or ${VAR_NAME:-default} - YAML or JSON format - Hot-reloading (reload config without restarting) - Singleton pattern for global access

Usage:

# Load default configuration config = ApiConfigLoader.load()

# Get specific service configuration auth_config = config.services.get(“auth”)

# Check if using API Gateway if config.api_gateway.enabled:

base_url = config.api_gateway.url

classmethod load(config_path: str | None = None) → AppConfig[source]

Load configuration from api.conf file.

Parameters:

config_path – Path to configuration file. If None, searches in: 1. Environment variable E2E_CONFIG_PATH 2. verify_services/api.conf (relative to project root) 3. Current working directory ./api.conf

Returns:

Parsed configuration object

Return type:

AppConfig

Raises:

FileNotFoundError – If configuration file not found
ValueError – If configuration file is invalid

classmethod reload() → AppConfig[source]: Reload configuration from file (useful for hot-reloading).

classmethod get_config_path() → Path | None[source]: Get the path of the currently loaded configuration file.

classmethod create_default_config(path: str | Path, overwrite: bool = False) → str[source]

Create a default e2e.conf configuration file.

Creates a default configuration file at the specified path using the e2e.conf.template template. Parent directories are created automatically if they don’t exist.

Parameters:

path – Path where to create the configuration file
overwrite – If True, overwrite existing file

Returns:

Path to the created configuration file

Return type:

str

Raises:

FileExistsError – If file exists and overwrite=False
FileNotFoundError – If template file doesn’t exist

classmethod validate_config(data: Dict[str, Any], strict: bool = False) → None[source]

Validate minimum configuration requirements.

Parameters:

data – Raw configuration dictionary
strict – If True, raises error for missing optional fields

Raises:

ConfigError – If configuration is invalid or missing required fields

classmethod get_service_url(service_name: str, use_gateway: bool | None = None) → str[source]

Get the effective URL for a service.

If API Gateway is enabled, returns gateway URL + service path. Otherwise, returns the service’s base_url.

Parameters:

service_name – Name of the service
use_gateway – Force gateway mode (None = use config setting)

Returns:

Full URL for the service

Return type:

str

classmethod get_all_required_services() → List[str][source]: Get list of services marked as required.

classmethod get_auto_start_services() → List[str][source]: Get list of services configured for auto-start.

classmethod get_service_by_maven_module(maven_module: str) → ServiceConfig | None[source]: Find service configuration by Maven module name.

socialseed_e2e.core.config_loader.get_config() → AppConfig[source]: Get the global configuration instance.

socialseed_e2e.core.config_loader.get_service_config(service_name: str) → ServiceConfig | None[source]

Get configuration for a specific service.

Parameters:: service_name – Name of the service (hyphens or underscores allowed)
Returns:: Service configuration if found, None otherwise
Return type:: Optional[ServiceConfig]

Example

>>> get_service_config("auth-service")  # Matches auth_service in config
>>> get_service_config("auth_service")  # Matches auth_service in config

socialseed_e2e.core.config_loader.get_service_url(service_name: str) → str[source]: Get effective URL for a service (considers API Gateway).

Models

class socialseed_e2e.core.models.ServiceConfig(*, name: str, base_url: str, default_headers: Dict[str, str]=<factory>, timeout: int = 30000, extra: Dict[str, ~typing.Any]=<factory>)[source]

Bases: BaseModel

Configuration for a specific service.

name: str

base_url: str

default_headers: Dict[str, str]

timeout: int

extra: Dict[str, Any]

model_config = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class socialseed_e2e.core.models.TestContext(*, env: str = 'dev', services: Dict[str, ~socialseed_e2e.core.models.ServiceConfig]=<factory>, metadata: Dict[str, ~typing.Any]=<factory>)[source]

Bases: BaseModel

Generic context for test execution.

env: str

services: Dict[str, ServiceConfig]

metadata: Dict[str, Any]

get_service(name: str) → ServiceConfig | None[source]

model_config = {}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

Test Orchestrator

class socialseed_e2e.core.test_orchestrator.TestOrchestrator(root_dir: str | None = None, services_path: str = 'verify_services/e2e/services')[source]

Bases: object

Orchestrates dynamic loading and execution of test modules. Auto-discovers modules in services/*/modules/ and runs them.

__init__(root_dir: str | None = None, services_path: str = 'verify_services/e2e/services')[source]

discover_modules()[source]: Discover all test modules in services directories using dynamic file loading.

run_service_tests(service_name: str, context: Any)[source]: Run all modules for a specific service.

run_all_tests(context_factory: Callable[[], Any])[source]: Run all discovered tests with a context factory.

socialseed_e2e.core.test_orchestrator.pytest_configure(config)[source]: Hook to set up orchestrator for pytest.

socialseed_e2e.core.test_orchestrator.pytest_collection_modifyitems(config, items)[source]: Modify test collection to include dynamic modules.

Module Loader

Module loaders for dynamic test discovery.

class socialseed_e2e.core.loaders.ModuleLoader[source]

Bases: object

Handles discovery and dynamic loading of Python modules from file paths.

static load_runnable_from_file(file_path: Path, function_name: str = 'run') → Callable | None[source]: Load a specific function from a python file.

discover_runnables(root_path: Path, pattern: str = '*.py') → List[Callable][source]: Discovers all runnable modules in a directory matching a pattern.

Interfaces

class socialseed_e2e.core.interfaces.IServicePage(*args, **kwargs)[source]

Bases: Protocol

Protocol that every service-specific page object should implement.

base_url: str

setup() → None[source]

teardown() → None[source]

__init__(*args, **kwargs)

class socialseed_e2e.core.interfaces.ITestModule(*args, **kwargs)[source]

Bases: Protocol

Protocol for test modules.

run(context: Any) → None[source]

__init__(*args, **kwargs)