# python-test-fixtures

Shared pytest fixtures for Python integration test suites. Provides typed database connection helpers with credentials sourced from AWS Secrets Manager, intended for use alongside [`python-aws-testing-utils`](https://github.com/theorchard/python-aws-testing-utils).

## Installation

Install only the extras you need:

```bash
# MySQL only
pip install test_fixtures[mysql]

# Neo4j only
pip install test_fixtures[neo4j]

# Snowflake only
pip install test_fixtures[snowflake]

# Multiple
pip install test_fixtures[mysql,neo4j]

# The integration-test runner (parallel + HTML reporting)
pip install test_fixtures[runner]
```

## Usage

### Secrets

All connection helpers fetch credentials from AWS Secrets Manager. You can also call `get_secret` directly:

```python
from test_fixtures.secrets import get_secret

# Returns a dict if the secret value is JSON, otherwise a plain string
secret = get_secret('your/secret/name')
```

### MySQL

```python
from test_fixtures.mysql import MySQLConnection

# Secret must be a JSON dict with keys: host, user, password, database
# Optional key: port (defaults to 3306)
db = MySQLConnection('your/secret/name')

rows = db.fetchall('SELECT * FROM my_table WHERE status = %s', ('active',))
row  = db.fetchone('SELECT * FROM my_table WHERE id = %s', (1,))
affected = db.execute('DELETE FROM my_table WHERE id = %s', (1,))
db.close()
```

In pytest, wrap in a session-scoped fixture to share the connection across tests:

```python
@pytest.fixture(scope='session')
def my_db() -> Generator[MySQLConnection, None, None]:
    conn = MySQLConnection('your/secret/name')
    yield conn
    conn.close()
```

### Generic CRUD helpers

Schema-agnostic setup/teardown helpers that build parameterised SQL from plain
dicts. They take any connection exposing `execute`/`fetchone` and know nothing
about your schema — table names, seed builders and domain fixtures stay in your
repo. The emitted SQL is **MySQL/MariaDB-specific** (backtick-quoted
identifiers, `%s` placeholders and `LAST_INSERT_ID()`), so use them with
`MySQLConnection` — not the Neo4j or Snowflake connections.

```python
from test_fixtures.crud import (
    insert_entity, get_entity, update_entity, delete_entity, clone_row,
)

# INSERT and return the inserted row (auto-increment key fetched via LAST_INSERT_ID())
job = insert_entity(db, 'project_transfer_job',
                    {'project_id': 5, 'vendor_id': 7}, id_column='job_id')

row = get_entity(db, 'project_transfer_job', {'job_id': job['job_id']})

update_entity(db, 'project_transfer_job',
              {'job_id': job['job_id']}, {'status': 'complete'})

# Copy an existing row, overriding some columns; id_column is auto-generated.
clone = clone_row(db, 'project', {'project_id': 5},
                  overrides={'vendor_id': 8}, id_column='project_id')

# Returns the number of rows deleted. Conditions are required.
delete_entity(db, 'project_transfer_job', {'job_id': job['job_id']})
```

`delete_entity`/`update_entity` reject empty conditions to prevent accidental
full-table writes. For FK-safe teardown, delete child rows before parents.
Identifiers (table/column names) are backtick-quoted; a backtick in an
identifier is rejected.

### Neo4j

```python
from test_fixtures.neo4j import Neo4jConnection

# Each argument is a Secrets Manager secret name containing a plain string value.
# If the hostname secret does not include a URI scheme, bolt:// is prepended automatically.
conn = Neo4jConnection(
    username_secret='your/neo4j/username-secret',
    password_secret='your/neo4j/password-secret',
    hostname_secret='your/neo4j/hostname-secret',
)

records = conn.run('MATCH (n:Artist) WHERE n.id = $id RETURN n', id='123')
conn.close()
```

### Snowflake

```python
from test_fixtures.snowflake import SnowflakeConnection

# key_secret: Secrets Manager secret name containing the private key (PEM string).
# Bare base64 keys (without -----BEGIN/END----- headers) are wrapped automatically.
# All other kwargs are passed directly to snowflake.connector.connect().
conn = SnowflakeConnection(
    'your/snowflake/private-key-secret',
    account='your-account',
    user='YOUR_SERVICE_USER',
    warehouse='YOUR_WAREHOUSE',
    database='YOUR_DATABASE',
    schema='YOUR_SCHEMA',
    role='YOUR_ROLE',
)

rows = conn.fetchall('SELECT * FROM my_table WHERE id = %s', (1,))
row  = conn.fetchone('SELECT * FROM my_table WHERE id = %s', (1,))
affected = conn.execute('DELETE FROM my_table WHERE id = %s', (1,))
conn.close()
```

## Lambda / Step Function integration test runner

The `runner` extra ships the standard way to run a repo's **Lambda and Step
Function** integration suite (it is not a general-purpose test runner) — so no
repo has to re-declare parallelism / reporting flags in a Makefile or
Dockerfile. It has two parts, both installed with `test_fixtures[runner]`.

### `run-integration-tests` console command

Runs pytest with the org-standard options — parallel execution
(`-n auto --dist loadfile`), a self-contained HTML report under `report/`, and
Datadog tracing when `ddtrace` is installed:

```bash
run-integration-tests                          # whole suite, parallel + report
run-integration-tests --lambda finalize-job    # one lambda (sets LAMBDA_FUNCTION_NAMES)
run-integration-tests --serial -k test_x       # serial (debugger-friendly), single case
```

Unrecognised arguments pass straight through to pytest. Use it as the container
entry point: `CMD ["run-integration-tests"]`.

`loadfile` keeps each file's tests on one worker, so tests are serial *within* a
lambda (safe for shared per-lambda seed) and parallel *across* lambdas.

### Running against a local RIE

`--rie` points the Lambda client at `--lambda`'s already-running local RIE
container **without you having to know or pass its port** — it discovers it by
running `docker compose port <lambda> 8080` itself, the same way you'd look it
up by hand:

```bash
docker compose --profile finalize-job up -d --build --wait finalize-job
run-integration-tests --lambda finalize-job --rie --serial
```

If you already know the URL (or aren't using docker-compose profiles), pass it
directly instead — the two are mutually exclusive:

```bash
run-integration-tests --lambda finalize-job --endpoint-url http://localhost:9002
```

**Running several lambdas' RIEs in parallel** works the same way, once each has
its own docker-compose host port (a consuming repo's setup, not this package's
concern): start each container, then run each lambda's tests concurrently —
in separate terminals/processes, or backgrounded:

```bash
docker compose --profile finalize-job  up -d --build --wait finalize-job
docker compose --profile handle-error  up -d --build --wait handle-error

run-integration-tests --lambda finalize-job --rie --serial &
run-integration-tests --lambda handle-error --rie --serial &
wait
```

### Marker-based filtering (pytest plugin)

Installing `test_fixtures` registers a pytest plugin (via a `pytest11` entry
point) that provides the `lambda_name` marker and the changed-lambda filter —
tests whose lambda is not listed in the `LAMBDA_FUNCTION_NAMES` environment
variable are deselected. This replaces the hand-written
`pytest_collection_modifyitems` that each repo used to keep in its `conftest.py`.

Tag each test with its lambda:

```python
pytestmark = pytest.mark.lambda_name('finalize-job')
```

The one per-repo setting is which tags should only run when explicitly named
(e.g. a slow step-function suite):

```toml
[tool.pytest.ini_options]
always_deselect_lambdas = ['state-machine']
```

An empty `LAMBDA_FUNCTION_NAMES` runs everything (except `always_deselect_lambdas`
tags); when no `lambda_name` markers are present the plugin is a no-op.

## Design principles

All connection helpers fetch credentials exclusively from AWS Secrets Manager. There is no support for passing credentials directly as constructor arguments. This is intentional — the library enforces the org-standard pattern so integration test suites cannot accidentally hardcode or bypass secrets management.

If you need to run tests locally, AWS credentials must be present in the environment (via `AWS_ACCESS_KEY_ID`/`AWS_SESSION_TOKEN` or an active SSO session). To mock the secrets layer in unit tests, patch `get_secret` directly.

## Development

```bash
# Install all dependencies (including all optional extras)
poetry install

# Lint
make lint

# Format
make format

# Run unit tests
make run_unit_tests

# Lint and test together
make lint_and_test

# Run via Docker (mirrors CI)
make docker_run_lint_and_test
```

## Versioning and Publishing

Version bumps and publishing are automated via the Jenkins pipeline. Do not manually edit the version in `test_fixtures/__init__.py` or `pyproject.toml` — run the `publish-pypi-package-v2` pipeline instead.