cleveragents-core/features/steps/postgresql_config_steps.py

"""Step definitions for PostgreSQL connection utility scenarios in langgraph_platform_remote_graph.feature."""

from __future__ import annotations

from behave import then, use_step_matcher, when  # type: ignore[import-untyped]
from behave.runner import Context  # type: ignore[import-untyped]

from cleveragents.infrastructure.database.postgresql import (
    PostgreSQLConnectionConfig,
    build_postgresql_url,
    is_postgresql_url,
)

# ---------------------------------------------------------------------------
# PostgreSQL connection utilities steps
# ---------------------------------------------------------------------------


@when(
    'I create a PostgreSQLConnectionConfig with host "{host}" database "{database}" username "{username}" password "{password}"'
)
def step_create_pg_config(
    context: Context, host: str, database: str, username: str, password: str
) -> None:
    context.pg_config = PostgreSQLConnectionConfig(
        host=host,
        database=database,
        username=username,
        password=password,
    )


@when(
    'I create a PostgreSQLConnectionConfig with host "{host}" database "{database}" username "{username}" password "{password}" port {port:d}'
)
def step_create_pg_config_with_port(
    context: Context,
    host: str,
    database: str,
    username: str,
    password: str,
    port: int,
) -> None:
    context.pg_config = PostgreSQLConnectionConfig(
        host=host,
        database=database,
        username=username,
        password=password,
        port=port,
    )


@then('the pg config host should be "{expected}"')
def step_pg_config_host(context: Context, expected: str) -> None:
    assert context.pg_config.host == expected, (
        f"Expected host={expected!r}, got {context.pg_config.host!r}"
    )


@then('the pg config database should be "{expected}"')
def step_pg_config_database(context: Context, expected: str) -> None:
    assert context.pg_config.database == expected, (
        f"Expected database={expected!r}, got {context.pg_config.database!r}"
    )


@then('the pg config username should be "{expected}"')
def step_pg_config_username(context: Context, expected: str) -> None:
    assert context.pg_config.username == expected, (
        f"Expected username={expected!r}, got {context.pg_config.username!r}"
    )


@then("the pg config port should be {expected:d}")
def step_pg_config_port(context: Context, expected: int) -> None:
    assert context.pg_config.port == expected, (
        f"Expected port={expected}, got {context.pg_config.port}"
    )


@then('the pg config ssl_mode should be "{expected}"')
def step_pg_config_ssl_mode(context: Context, expected: str) -> None:
    assert context.pg_config.ssl_mode == expected, (
        f"Expected ssl_mode={expected!r}, got {context.pg_config.ssl_mode!r}"
    )


@then('the pg config async URL should start with "{prefix}"')
def step_pg_config_async_url(context: Context, prefix: str) -> None:
    url = context.pg_config.to_url(async_driver=True)
    assert url.startswith(prefix), f"Expected URL to start with {prefix!r}, got {url!r}"


@then('the pg config sync URL should start with "{prefix}"')
def step_pg_config_sync_url(context: Context, prefix: str) -> None:
    url = context.pg_config.to_url(async_driver=False)
    assert url.startswith(prefix), f"Expected URL to start with {prefix!r}, got {url!r}"


@when("I try to create a PostgreSQLConnectionConfig with empty host")
def step_try_create_pg_config_empty_host(context: Context) -> None:
    try:
        PostgreSQLConnectionConfig(
            host="",
            database="mydb",
            username="user",
            password="pass",
        )
        context.pg_config_error = None
    except Exception as exc:
        context.pg_config_error = exc


@when("I try to create a PostgreSQLConnectionConfig with port {port:d}")
def step_try_create_pg_config_bad_port(context: Context, port: int) -> None:
    try:
        PostgreSQLConnectionConfig(
            host="db.example.com",
            database="mydb",
            username="user",
            password="pass",
            port=port,
        )
        context.pg_config_error = None
    except Exception as exc:
        context.pg_config_error = exc


@when("I try to create a PostgreSQLConnectionConfig with pool_size {pool_size:d}")
def step_try_create_pg_config_bad_pool_size(context: Context, pool_size: int) -> None:
    try:
        PostgreSQLConnectionConfig(
            host="db.example.com",
            database="mydb",
            username="user",
            password="pass",
            pool_size=pool_size,
        )
        context.pg_config_error = None
    except Exception as exc:
        context.pg_config_error = exc


@then("a pg config validation error should be raised")
def step_pg_config_validation_error(context: Context) -> None:
    assert context.pg_config_error is not None, (
        "Expected a validation error but none was raised"
    )


@when(
    'I call build_postgresql_url with host "{host}" database "{database}" username "{username}" password "{password}"'
)
def step_build_postgresql_url(
    context: Context, host: str, database: str, username: str, password: str
) -> None:
    context.built_url = build_postgresql_url(
        host=host,
        database=database,
        username=username,
        password=password,
    )


@when("I call build_postgresql_url with async_driver False")
def step_build_postgresql_url_sync(context: Context) -> None:
    context.built_url = build_postgresql_url(
        host="db.example.com",
        database="mydb",
        username="user",
        password="pass",
        async_driver=False,
    )


@then('the built URL should start with "{prefix}"')
def step_built_url_starts_with(context: Context, prefix: str) -> None:
    assert context.built_url.startswith(prefix), (
        f"Expected URL to start with {prefix!r}, got {context.built_url!r}"
    )


@then('the built URL should contain "{substring}"')
def step_built_url_contains(context: Context, substring: str) -> None:
    assert substring in context.built_url, (
        f"Expected URL to contain {substring!r}, got {context.built_url!r}"
    )


use_step_matcher("re")


@when(r'I check if "(?P<url>.*)" is a PostgreSQL URL')
def step_check_is_postgresql_url(context: Context, url: str) -> None:
    context.is_pg_url_result = is_postgresql_url(url)


use_step_matcher("parse")


@then("the pg url check result should be True")
def step_pg_url_check_true(context: Context) -> None:
    assert context.is_pg_url_result is True, (
        f"Expected True, got {context.is_pg_url_result!r}"
    )


@then("the pg url check result should be False")
def step_pg_url_check_false(context: Context) -> None:
    assert context.is_pg_url_result is False, (
        f"Expected False, got {context.is_pg_url_result!r}"
    )