Skip to content

Testing

The test suite is split between deterministic unit tests and external-service integration tests.

The latest sanitized validation summary is maintained in Latest Test Report. Keep only that single report in git and overwrite it after new validation runs. Do not paste raw logs, server addresses, usernames, passwords, tokens, certificate material, wallet files, full connection strings, or sensitive payloads into test reports.

This rule matters in public repositories and it matters even more for mission-oriented or defence-adjacent deployments. Test evidence should prove behavior without disclosing network topology, credentials, classified labels, operational subjects, or real payload content.

Unit Tests

Unit tests must not make network calls and must not require NATS, Oracle, or any other external service.

pytest -m "not integration"

Unit tests cover:

  • envelope creation,
  • idempotency key generation,
  • JSON configuration loading,
  • strict JSON boundary handling for configuration, automation manifests, payloads, mission metadata, encryption envelopes, observability policy files, NATS monitoring responses, and public timing evidence,
  • secret redaction,
  • batch creation,
  • retry policy,
  • sink protocol contract,
  • safe registry behavior,
  • file sink path mapping,
  • file sink duplicate handling,
  • file sink filesystem error classification,
  • Oracle SQL generation,
  • Oracle identifier validation,
  • Oracle row mapping,
  • AES-256-GCM and AES-256-CCM payload encryption,
  • payload decryption verification,
  • commit-then-ack ordering,
  • fan-out route selection and ACK-gate certification,
  • no ACK when payload encryption fails before sink delivery,
  • DLQ-before-ACK ordering,
  • no ACK on sink failure,
  • no payload logging by default.

Routing And Fan-Out Certification

Routing and fan-out tests remain deterministic and local. They do not connect to NATS, Oracle Database, Oracle MySQL, or any other backend. The focused suite uses synthetic NatsEnvelope instances, route policies, in-memory fan-out operation plans, and the production FanoutSink to prove that route selection, child-sink dispatch, and ACK gating behave correctly.

Run the focused suite with:

pytest tests/unit/test_fanout_certification.py tests/unit/test_fanout_sink.py

Run the multi-sink routing end-to-end probe with:

python scripts/run-multi-sink-routing-e2e.py \
  --mode reduced \
  --output .local/multi-sink-routing-e2e/report.json

The suite covers:

  • one-to-one routing to a single logical sink target;
  • one-to-many routing to required and optional targets;
  • required target failure after partial success, proving no synthetic ACK is recorded;
  • optional target timeout, proving bounded wait behavior and payload-free logs;
  • no-route actions for reject, ignore, and default_route;
  • matching by subject, priority, classification, labels_all, labels_any, labels_none, approved non-secret headers, and combined match sets;
  • nats-sink validate coverage for the documented named multi-sink example and invalid route, sink, match, optional wait, fan-out, and redaction scenarios.
  • a deterministic multi-sink routing flow for Oracle Database, Oracle MySQL Database, File, and Oracle Coherence Community Edition logical targets, using local file-backed probe sinks so no credentials or network calls are needed.

The same suite is part of scripts/check-sinks.sh and therefore part of the deterministic release readiness path. Live NATS-to-destination fan-out tests remain opt-in because they need the same live destination credentials as the underlying child sinks.

See Multi-Sink Routing End-To-End Flow for the route matrix, example configuration, command output, and live-backend layering guidance.

Ordered-consumer inspection has focused unit coverage for fail-closed client capability detection, redacted default output, explicit payload opt-in, message and payload-byte limits, JSONL output-path validation, unsubscribe cleanup, and the CLI contract that no sink is built or written. Durable replay-to-sinks tests should stay on the normal commit-then-acknowledge contract and prove that replay never ACKs before durable sink success. See Ordered Consumer Evaluation and Durable Replay To Sinks.

The durable replay design is covered by a documentation guardrail test. That test keeps the page in the MkDocs navigation and checks that public guidance continues to mention durable pull consumers, no early ACK, start sequence or start time boundaries, maximum message limits, dry-run behavior, redacted reports, idempotency review, and future tests for duplicate replay and DLQ handling. When replay becomes executable tooling, those documentation checks must be complemented by runner-level tests that use fake messages and fake sinks to prove the same behavior in code.

Push-consumer support is opt-in and covered by unit tests for default-disabled configuration, manual ACK enforcement, bounded callback intake, queue overflow, shutdown intake stop, temporary failure without ACK, DLQ-before-ACK permanent failure handling, flow-control and idle-heartbeat option propagation, callback exception containment, and unchanged pull-mode behavior. Optional live NATS push-consumer certification is isolated behind NATS_SINKS_PUSH_CONSUMER_INTEGRATION=1 so ordinary unit and release checks do not depend on a local server. See Push Consumer Evaluation.

Bounded Property-Style Tests

The repository uses deterministic bounded generator tests for security-sensitive validators and normalizers. These tests provide many of the benefits of property-based testing without adding a new dependency or introducing non-determinism into CI. The generators are deliberately small, explicit, and reviewable so a failing case is easy to understand.

Run the focused generator suite with:

pytest tests/unit/test_property_generators.py

The current generator suite covers:

  • NATS subject pattern validation and wildcard matching invariants;
  • payload normalization into JSON-compatible storage values;
  • message metadata normalization for priority, classification, and labels;
  • mission metadata JSON validation, freezing, and duplicate-key rejection;
  • file sink path sanitization for traversal-like, control-character, Unicode, oversized, and hostile string-conversion values.

The file path generator found and now guards a real hardening case: a hostile object whose string conversion raises must not crash path sanitization. The sanitizer now falls back to a deterministic safe component derived from type metadata rather than using a raw object representation.

These tests are intentionally payload-safe. Error assertions check that invalid payload content is not echoed back in exception text. Generated values are fake and must not include credentials, private infrastructure details, real operational subjects, or sensitive payload fragments.

If a future maintainer adds Hypothesis or another property-based testing tool, it must be a development-only dependency, documented in CHANGELOG.md, and reflected in the generated dependency manifests.

Oracle Benchmark Tests

Oracle benchmark tooling is split into two layers:

  • unit tests for command validation, report redaction, phase rendering, and public-output safety;
  • a live benchmark script for non-production NATS and Oracle environments.

The unit tests run in normal CI and do not connect to NATS or Oracle:

pytest tests/unit/test_oracle_benchmark.py

The live benchmark is opt-in and must use ignored local environment files:

scripts/run-oracle-benchmark.sh \
  --message-count 256 \
  --batch-size 64 \
  --payload-shape mixed \
  --sink-mode merge \
  --format markdown \
  --report-file .local/oracle-benchmark/report.md

The script reports publish, fetch, map, write, commit, ACK, retry, and shutdown phases separately. It uses the real OracleSink and JetStreamSinkRunner, so Oracle commit still happens before JetStream ACK. The script must not be used against production streams or production tables unless an operator has reviewed retention, duplicate handling, table cleanup flags, and the impact of synthetic messages.

Benchmark reports are sanitized by default. They are suitable for summaries in docs/test-report.md, but maintainers must still inspect them before copying them into public issues or release notes.

The benchmark report helpers reject non-finite phase timings before JSON or Markdown evidence is rendered. This keeps public reports standards-compliant and avoids accidentally publishing NaN or Infinity in issue comments, release notes, or retained test reports.

Local File Sink End-To-End Test

The file sink has a deterministic local end-to-end test because it only needs a temporary directory. This test exercises the core runner, FileSink, payload normalization, metadata persistence, and ACK-after-durable-file-success behavior without requiring a NATS server.

pytest tests/integration/test_file_sink_e2e.py

The test publishes fake JetStream message objects into JetStreamSinkRunner.process_raw_batch(...), writes local JSON files, verifies JSON, text, empty, and non-UTF-8 payload handling, and confirms that every fake message is ACKed only after the sink returns success.

The same module also covers gzip-compressed file output and verifies that one message still maps to one durable compressed file. It also covers core payload encryption and verifies that encrypted payloads written by the file sink can be decrypted back to the original JSON, text, empty, and binary message bodies. By default, any explicit file e2e output directory is deleted after the test. To retain generated files for inspection, set:

export NATS_SINKS_FILE_E2E_DIRECTORY='.local/file-sink-e2e'
export NATS_SINKS_FILE_E2E_DELETE_AFTER=false
pytest tests/integration/test_file_sink_e2e.py

HTTP Sink NGINX Container E2E Test

The HTTP sink has a local container-backed e2e test endpoint based on Oracle Linux 9 slim FIPS and NGINX. The test starts a short-lived local endpoint, sends fake events through the production HttpSink, copies bounded request evidence from the container, and verifies the HTTP envelope, subject, idempotency key, route header, and payload marker.

Run the focused test with:

python scripts/run-http-sink-nginx-e2e.py

Expected output:

HTTP sink NGINX container e2e test passed.

The runner binds NGINX to a random 127.0.0.1 host port, drops Linux capabilities, uses a read-only root filesystem, and removes the container plus copied evidence by default. See HTTP Sink NGINX FIPS Test Endpoint for the complete local test contract and troubleshooting guidance.

The focused static unit tests run without Docker:

pytest tests/unit/test_http_nginx_test_container.py

NATS JetStream FIPS Test Service

The local Docker smoke test can use a repository-owned NATS JetStream test service built from Oracle Linux 9 slim FIPS:

python scripts/run-docker-local-smoke.py --build-first-party-nats-image

The public nats:2.10 image remains the default developer path. Use the first-party image when release evidence or a controlled local environment requires the NATS test service to follow the same base-image policy as other repository-built test containers. See NATS JetStream FIPS Test Service for the image construction, checksum verification, architecture support, and limitations.

Palantir Foundry Mock Certification

The experimental Palantir Foundry sink is certified locally with a fake FoundryStreamClient. This test path does not connect to Foundry and does not require tenant URLs, credentials, client identifiers, resource identifiers, or private response bodies.

python -m pytest tests/unit/test_foundry_sink.py -q

The focused suite covers:

  • configuration validation for HTTPS endpoints, endpoint allow-listing, bearer-token environment variables, OAuth2 client-credentials fields, record field names, and size limits;
  • record mapping for payloads, payload metadata, priority, classification, labels, NATS metadata, mission metadata, security labels, and custody metadata;
  • deterministic idempotency key strategies and fail-closed behavior when a selected strategy lacks required metadata;
  • fake-client success, permanent rejection, temporary failure, ambiguous partial acceptance, batching, and duplicate redelivery;
  • runner-level ACK evidence proving Foundry acceptance happens before ACK and sink failure prevents ACK.

Mock certification is necessary before live certification, but it is not a substitute for testing against an approved Foundry environment.

Palantir Gotham Mock Certification

The experimental Palantir Gotham sink is certified locally with a fake GothamObjectClient. This test path does not connect to Gotham and does not require tenant URLs, credentials, client identifiers, object type names, property type names, primary keys, or private response bodies.

python -m pytest tests/unit/test_gotham_sink.py -q

The focused suite covers:

  • configuration validation for HTTPS base URLs, endpoint allow-listing, bearer-token environment variables, OAuth2 client-credentials fields, object type names, property type mappings, security portion markings, and size limits;
  • object-create mapping for payloads, payload metadata, selected NATS metadata, classification, labels, security markings, and deterministic external IDs;
  • deterministic idempotency strategies and fail-closed behavior when a selected strategy lacks required metadata;
  • fake-client success, duplicate redelivery, permanent rejection, temporary failure, ambiguous partial acceptance, and batching;
  • runner-level ACK evidence proving Gotham acceptance happens before ACK and sink failure prevents ACK.

Mock certification is necessary before live certification, but it is not a substitute for testing against an approved Gotham environment and object model.

The test creates a unique child directory under NATS_SINKS_FILE_E2E_DIRECTORY for each run. Keep that directory under .local/ or another ignored location when retaining files.

Oracle MySQL Sink And Test Container

The repository includes a local Oracle MySQL test database container and a container-backed e2e test for the first-party Oracle MySQL sink. The container is intentionally separate from the production Oracle Database sink. It provides a clean local database backend for transactions, schema validation, duplicate handling, routed writes, and throughput work without requiring a long-lived developer database.

Run the deterministic asset tests without Docker:

python -m pytest tests/unit/test_oracle_mysql_test_container.py -q

Run the optional Docker smoke test when a local Docker daemon is available:

python scripts/run-oracle-mysql-container-smoke.py

Expected sanitized output:

Oracle MySQL container smoke test passed with one verified test record.

Run the Oracle MySQL sink e2e test against a fresh short-lived container:

python scripts/run-mysql-sink-e2e.py

Expected sanitized output:

Oracle MySQL sink container e2e test passed.

That e2e runner builds the same Oracle Linux 9 slim FIPS based Oracle MySQL image, starts a fresh container with generated credentials, exposes it only on a random loopback port, runs tests/integration/test_mysql_sink.py, and removes the container, Docker volume, and generated secret files by default.

The integration test validates:

  • MySqlSink.start, healthcheck, ensure_schema, write_batch, and stop;
  • idempotent upsert with no-op duplicate redelivery;
  • subject-to-table routing;
  • metadata persistence for priority, classification, and labels;
  • non-JSON payload wrapping;
  • empty payload handling;
  • Oracle MySQL duplicate/upsert metrics through the shared metrics recorder;
  • disconnected backend recovery by writing 1001 records directly, spooling 1001 records during a simulated outage, replaying them, and writing a final 1001 records before verifying 3003 unique backend records.

The smoke test builds the Oracle Linux 9 slim FIPS based Oracle MySQL image, starts a fresh short-lived container with generated random credentials, waits for readiness, creates a test table, inserts and reads one row, and removes the container, Docker volume, and generated secret files by default. Use --preserve-artifacts only for local debugging and clean up preserved files under .local/oracle-mysql-test/ afterward.

See Oracle MySQL Sink for sink configuration and Oracle MySQL Test Container for the complete container security model, runtime sequence, capability exception, and troubleshooting guidance.

Oracle NoSQL Database Sink And Test Backend

The repository includes a local Oracle NoSQL Database KVLite backend and a container-backed e2e test for the experimental first-party Oracle NoSQL Database sink. The backend wraps Oracle's documented Community Edition KVLite image from GitHub Container Registry and is local test infrastructure only.

Run the deterministic asset tests without Docker:

python -m pytest tests/unit/test_oracle_nosql_test_container.py -q

Run the optional Docker smoke test when a local Docker daemon and the optional Oracle NoSQL Python SDK are available:

python -m pip install -e ".[oracle-nosql]"
python scripts/run-oracle-nosql-container-smoke.py

Expected sanitized output:

Oracle NoSQL Database container smoke test passed with one verified JSON key/value entry.

Run the Oracle NoSQL sink e2e test against a fresh short-lived KVLite container:

python scripts/run-oracle-nosql-sink-e2e.py

Expected sanitized output:

Oracle NoSQL sink container e2e test passed.

That e2e runner starts the same short-lived KVLite backend, binds the HTTP proxy to a random loopback port, enables the live-gated tests/integration/test_oracle_nosql_sink_e2e.py test, creates the narrow key/value table through generated safe DDL, writes the same normalized envelope twice, runs the disconnected spool replay certification, and removes the container by default.

See Oracle NoSQL Database Sink for sink configuration and Oracle NoSQL Database Test Backend for the image strategy, local-only security posture, JSON verification, cleanup behavior, and troubleshooting guidance.

Oracle Coherence Community Edition Test Backend

The repository includes a local Oracle Coherence Community Edition test backend for future Oracle Coherence sink and multi-sink routing work. The backend is a test-only Oracle Linux 9 slim FIPS image that resolves the Oracle Coherence Community Edition runtime, gRPC proxy, and JSON modules during build. It starts a short-lived container, exposes the client endpoint only on a random loopback port, writes one complete fake event JSON object as a key/value entry, reads it back, and removes the container by default.

Run the deterministic asset tests without Docker:

python -m pytest tests/unit/test_oracle_coherence_test_container.py -q

Run the optional Docker smoke test when a local Docker daemon and the optional Coherence Python client are available. Use an isolated local virtual environment for the optional client:

python -m venv .local/coherence-smoke-venv
. .local/coherence-smoke-venv/bin/activate
python -m pip install coherence-client
python scripts/run-oracle-coherence-container-smoke.py

Expected sanitized output:

Oracle Coherence CE container smoke test passed with one verified JSON key/value entry.

This backend does not implement the Oracle Coherence Community Edition sink and does not prove production Coherence durability. It is local test infrastructure for future sink certification, routing, and fan-out validation.

The Oracle Coherence sink e2e runner also runs the disconnected spool replay certification. It writes 1001 records directly, proves an unreachable Coherence endpoint fails closed, writes 1001 records into encrypted local spool custody, replays them to the reachable cache, writes another 1001 records directly, and verifies the expected keys through the Coherence client.

See Oracle Coherence Community Edition Test Backend for Oracle Linux 9 slim FIPS base-image selection, runtime module selection, runtime sequence, optional client setup, smoke-test commands, cleanup behavior, and security boundaries.

Disconnected Backend Spool Replay

The disconnected spool replay certification is part of the standard deterministic sink checks and the explicit local container gate. The deterministic unit test proves the full 1001 + 1001 + 1001 message flow without network calls:

python -m pytest \
  tests/unit/test_disconnected_spool_replay.py \
  tests/unit/test_oracle_disconnected_replay_verification.py \
  -q

Expected output:

3 passed

Backend-specific coverage is enabled by the maintained e2e runners for Oracle MySQL Database, Oracle NoSQL Database, and Oracle Coherence Community Edition. Oracle Database coverage remains live-environment gated:

export NATS_SINKS_ORACLE_DISCONNECTED_REPLAY=1
python -m pytest -m integration tests/integration/test_oracle_sink.py -q

See Disconnected Spool Replay Testing for the phase model, backend support matrix, commands, expected output, cleanup behavior, and troubleshooting.

Oracle Database Version Matrix

Oracle Database version compatibility is validated through an explicit local matrix runner rather than routine unit tests or routine GitHub Actions. The runner is intended for release preparation and local certification evidence when a maintainer has Docker resources, Oracle registry or license setup, and approved image references available.

List the default inventory:

python scripts/run-oracle-db-version-matrix.py --list

Dry-run a selected version plan without starting Docker:

python scripts/run-oracle-db-version-matrix.py \
  --include oracle-free-26ai \
  --image-ref oracle-free-26ai=registry.example.invalid/oracle/database-free:26.1.0 \
  --dry-run

The live runner starts a single selected Oracle Database container, waits for readiness, runs the existing Oracle sink integration certification, optionally includes disconnected replay, writes sanitized evidence under .local/, and removes the container by default. See Oracle Database Version Matrix for the inventory, status model, prerequisites, exclusions, and evidence format.

Subject-Aware Observability Certification

Subject-aware observability has its own focused certification suite because subject-family labels can expose operational routing structure or create high-cardinality metric sets. The tests use synthetic subjects only and prove disabled-by-default behavior, allow and deny handling, malformed policy rejection, cardinality caps, overflow behavior, sanitized connector output, and delivery non-interference.

Run the focused suite with:

python -m pytest tests/unit/test_subject_observability_certification.py -q

The reusable helper is available for connector authors:

from nats_sinks.testing import run_subject_observability_certification


def test_subject_observability_contract() -> None:
    report = run_subject_observability_certification()

    assert report.raw_subject_leaks == ()
    assert report.delivery_probe_before == report.delivery_probe_after

See the Subject-Aware Observability Runbook before enabling subject-family metrics in an operator policy.

Local PyPI Artifact Container Validation

After a release has been published to PyPI, maintainers can run a local post-release validation in a clean Oracle Linux 9 slim FIPS based container:

python scripts/run-pypi-release-container-validation.py --version 0.4.1

This check is intentionally not part of GitHub Actions. It validates public registry state after publication by installing nats-sinks from PyPI inside a short-lived container and confirming that the package does not import from the local checkout. It then checks CLI help, version output, Python imports, configuration validation, FileSink smoke behavior, metrics snapshot behavior, and observability CLI startup.

Use latest to check the newest public package:

python scripts/run-pypi-release-container-validation.py --version latest

Use optional extras only when they can be validated without private services:

python scripts/run-pypi-release-container-validation.py \
  --version 0.4.1 \
  --extras crypto,mysql,oci

The script removes the validation container, temporary image, and generated validator files by default. It writes sanitized reports under .local/pypi-release-validation/reports/, which is ignored by Git. If the check finds a failure, create a sanitized GitHub bug report before starting the fix, attach the relevant local report summary, and follow the normal test-driven bug workflow.

Local WebSocket End-To-End Test

WebSocket transport has two layers of testing:

  • unit tests for configuration guardrails, optional header validation, redaction, TLS context construction, and collision-safe harness helpers;
  • an optional live local script that starts a temporary NATS server with JetStream and WebSocket enabled.

The unit tests never bind sockets and never start nats-server:

pytest tests/unit/test_nats_connection_options.py tests/unit/test_websocket_harness.py

The live script requires the nats-server executable on PATH:

scripts/run-websocket-e2e.sh --message-count 16 --batch-size 8

The script is designed to avoid collisions with developer machines where NATS is already running. It checks the default local NATS, monitoring, and WebSocket ports and chooses free loopback alternatives when needed. It only terminates the temporary nats-server process that it started.

Example sanitized output:

{"monitoring_port": 8222, "nats_port": 4222, "transport": "websocket", "websocket_port": 8080}
{"commit_then_ack": "verified by JetStreamSinkRunner.process_raw_batch", "files_written": 16, "messages_processed": 16, "messages_published": 16}
{"status": "passed"}

Use --message-count to change the number of synthetic messages and --batch-size to exercise partial or multiple fetch batches:

scripts/run-websocket-e2e.sh --message-count 23 --batch-size 8

Use --preserve-work-dir only for local debugging:

scripts/run-websocket-e2e.sh --preserve-work-dir

Generated files live under .local/websocket-e2e by default and must not be committed. The generated NATS config and output files are synthetic and loopback-only, but maintainers should still review preserved files before copying excerpts into public GitHub issues or release evidence.

Synthetic Mission Scenario Harness

The repository includes a synthetic scenario harness for repeatable mission-style testing without real operational content. The harness generates immutable NatsEnvelope objects with fake subjects, payloads, metadata, and JetStream sequence values. It is safe for unit tests, smoke tests, release evidence, and public issue comments because generated reports exclude payload contents, service locators, usernames, passwords, certificates, keys, wallet material, local paths, and private infrastructure details.

The harness currently covers these scenario cases:

  • valid_json, a normal JSON payload that should pass through every JSON-capable sink.
  • malformed_json_text, text that looks like JSON but is intentionally incomplete. Default payload handling wraps it as text instead of crashing.
  • duplicate, a redelivery-style message that reuses the previous message's stream sequence, subject, message ID, and payload so idempotent sinks can treat the duplicate as already processed.
  • stale, a message with an older event timestamp so future freshness checks and operators can test delayed-event behavior.
  • encrypted_marker, a fake encrypted-payload envelope shape that proves sinks can persist encryption-envelope-like JSON without exposing real key material.
  • classified, priority, and labeled, which exercise NATO-style classification values, urgency values, and semicolon-compatible labels.
  • empty, an empty message body that must be persisted safely rather than causing a hard crash.

Generate a destination-neutral sanitized JSON report:

python scripts/run-synthetic-harness.py --message-count 18

Example output:

{
  "classification_values": {
    "NATO RESTRICTED": 16,
    "NATO SECRET": 2
  },
  "duplicate_messages": 2,
  "encrypted_marker_messages": 2,
  "file_count": null,
  "generated_messages": 18,
  "malformed_json_text_messages": 2,
  "profile": "mission-smoke",
  "sink": "core",
  "stale_messages": 2,
  "unique_idempotency_keys": 16
}

Run the same scenario through the file sink without external services:

python scripts/run-synthetic-harness.py \
  --sink file \
  --message-count 18 \
  --output-dir .local/synthetic-file-smoke \
  --format markdown \
  --report-file .local/synthetic-file-smoke-report.md

By default the script deletes file-sink output after the run. Add --preserve-files when you intentionally want to inspect generated records under an ignored local directory:

python scripts/run-synthetic-harness.py \
  --sink file \
  --message-count 18 \
  --output-dir .local/synthetic-file-smoke \
  --preserve-files

The file-sink adapter also supports gzip output through Python's standard-library gzip module:

python scripts/run-synthetic-harness.py \
  --sink file \
  --compression gzip \
  --message-count 18

Python projects can import the harness directly:

from nats_sinks.testing import SyntheticScenarioProfile, generate_synthetic_scenario

profile = SyntheticScenarioProfile(message_count=18, seed=7)
messages = generate_synthetic_scenario(profile)

for synthetic in messages:
    envelope = synthetic.envelope
    assert envelope.subject.startswith("mission.synthetic.")

Future sink adapters should reuse the same generated NatsEnvelope objects and produce sanitized reports with the same shape. Live NATS and Oracle execution must remain gated behind ignored .local/ configuration files and explicit environment flags; the synthetic harness should never connect to live systems unless a separate, opt-in integration wrapper is added for that purpose.

Synthetic Load Profile Tests

Load profiles are local, deterministic pressure rehearsals for runtime behavior. They are separate from the live Oracle benchmark and from the local file-sink e2e test. Their job is to make normal, retry, DLQ, and shutdown behavior repeatable without contacting private infrastructure.

Run the unit coverage for the profile API and CLI argument handling:

pytest tests/unit/test_load_profiles.py

Run a normal profile and write a sanitized Markdown report under an ignored local path:

mkdir -p .local/load-profile
python scripts/run-load-profile.py \
  --profile normal \
  --message-count 256 \
  --batch-size 64 \
  --with-encryption \
  --format markdown \
  --report-file .local/load-profile/normal.md

Run the failure-oriented profiles:

python scripts/run-load-profile.py --profile retry --message-count 256 --batch-size 64
python scripts/run-load-profile.py --profile dlq --message-count 256 --batch-size 64
python scripts/run-load-profile.py --profile shutdown --message-count 250 --batch-size 64

The shutdown example intentionally uses a message count that is not a clean multiple of the batch size. That keeps partial-batch behavior visible while the profile models stop-fetch behavior and leaves unfetched messages outside the ACK boundary.

Load-profile phase rates use phase-specific completed-work counters. For example, fetch timing uses messages_fetched, backend-write timing uses messages_written, DLQ timing uses messages_dlq, and ACK timing uses messages_acked. This prevents a shutdown or DLQ rehearsal from reporting inflated throughput by dividing every phase by the total generated message count.

The profile output is public-safe by design. It contains only generated counts, aggregate timings, selected retry and DLQ counters, and optional metrics snapshot data. It must not be edited to include real subjects, server addresses, usernames, passwords, wallet locations, certificate material, table names, or payload bodies.

For operational interpretation guidance, see Performance.

Payload Encryption Test Matrix

Payload encryption is tested as a generic core feature and through each production sink. Use the tracked helper script:

scripts/check-encryption.sh

The script generates temporary AES-256 key material, exports it to the test process as NATS_SINKS_TEST_ENCRYPTION_KEY_B64, and runs encryption-focused unit and local end-to-end tests. By default, the generated key file is deleted when the script exits.

To preserve the generated key file for local debugging:

scripts/check-encryption.sh --preserve-key-material

Do not commit preserved key files. Keep them under the temporary path reported by the script or move them only to ignored local directories.

The encryption matrix covers:

  • AES-256-GCM encrypt/decrypt,
  • AES-256-CCM encrypt/decrypt,
  • empty payload encryption,
  • wrong key identifier rejection,
  • redacted direct key configuration,
  • subject-specific encryption rules, including matching subjects, unmatched subjects, disabled-rule exemptions, and first-match-wins behavior,
  • runner encryption before sink writes,
  • no ACK when encryption fails,
  • file sink storage with and without gzip,
  • file sink end-to-end decrypt verification,
  • Oracle sink encrypted payload storage through mocked Oracle objects.

Sink Release Test Matrix

Every production release should validate each production sink at the strongest available level:

Sink Unit tests Smoke tests End-to-end tests
Oracle SQL, mapping, routing, payload, idempotency, encrypted payload storage, and contract tests. nats-sink validate examples/oracle-jetstream/config.json; live test-sink when Oracle env is available. Live NATS-to-Oracle e2e when .local integration env is available.
Oracle MySQL SQL, mapping, routing, payload, idempotency, TLS configuration, metrics, disconnected replay, and contract tests. nats-sink validate examples/oracle-mysql-basic/config.json; python scripts/run-oracle-mysql-container-smoke.py. Local short-lived Oracle MySQL container e2e with python scripts/run-mysql-sink-e2e.py.
Oracle NoSQL Database Config validation, SDK authorization-provider construction, namespace and compartment handle defaults, cloud table-limit setup, key strategy, JSON value mapping, generated table DDL, duplicate policy, timeout handling, optional dependency failure, disconnected replay, fan-out defaults, test-container asset checks, and contract tests with fake SDK clients. nats-sink validate examples/oracle-nosql-basic/config.json; python scripts/run-oracle-nosql-container-smoke.py. Local short-lived Oracle NoSQL Database KVLite container e2e with python scripts/run-oracle-nosql-sink-e2e.py; Cloud Simulator and Oracle NoSQL Database Cloud Service certification remains explicitly environment-gated.
Oracle Coherence Community Edition Config validation, key strategy, JSON value mapping, duplicate policy, timeout handling, optional dependency failure, disconnected replay, fan-out defaults, and contract tests. nats-sink validate examples/oracle-coherence-basic/config.json; python scripts/run-oracle-coherence-container-smoke.py. Local short-lived Oracle Coherence Community Edition container e2e with python scripts/run-coherence-sink-e2e.py.
File Path mapping, payload, duplicate policy, compression, encryption, healthcheck, filesystem errors, and fuzz-style path safety tests. nats-sink validate examples/file-basic/config.json; nats-sink test-sink examples/file-basic/config.json. Local deterministic runner-to-file e2e in tests/integration/test_file_sink_e2e.py, with uncompressed, gzip, and encrypted output.
S3-compatible object storage Config validation, bucket and endpoint validation, key strategy, object value mapping, duplicate policy, metadata sidecar behavior, timeout handling, optional dependency failure, fan-out defaults, and contract tests with fake SDK clients. nats-sink validate examples/s3-basic/config.json. Live S3-compatible object-storage certification remains explicitly environment-gated and should use scoped temporary prefixes, fake payloads, and sanitized reports.

If a live external-service test is not run, the latest test report must say so explicitly. Do not imply that Oracle, NATS, or any other external service was validated when only deterministic local tests were executed.

For mission systems, keep that distinction visible. A deterministic local test, a mocked integration test, a lab e2e test, and a production-like acceptance run are different levels of evidence and should not be described as if they prove the same operational readiness.

The deterministic sink release matrix can be run with:

scripts/check-sinks.sh

scripts/check-sinks.sh includes the reusable sink certification helpers from Sink Certification. Those helpers are not a replacement for destination-specific tests; they are the shared baseline that proves a sink respects the framework boundary, receives only NatsEnvelope objects, does not own JetStream ACK behavior, and returns success only after the sink-specific durable assertion has passed.

To include the local container-backed Oracle-family sink e2e suite, install the optional backend clients, make sure Docker is running, and enable the explicit container gate:

python -m pip install -e ".[mysql,coherence,oracle-nosql]"
NATS_SINKS_RUN_CONTAINER_E2E=1 scripts/check-sinks.sh

That gate runs:

  • python scripts/run-mysql-sink-e2e.py, which builds and starts a fresh Oracle MySQL Database test container and verifies the Oracle MySQL sink;
  • python scripts/run-oracle-nosql-sink-e2e.py, which starts a fresh Oracle NoSQL Database KVLite container and verifies the Oracle NoSQL Database sink;
  • python scripts/run-coherence-sink-e2e.py, which builds and starts the Oracle Coherence Community Edition test container and verifies the Oracle Coherence Community Edition sink.

Expected successful tail output:

Oracle MySQL sink container e2e test passed.
Oracle NoSQL sink container e2e test passed.
Oracle Coherence sink e2e test passed.
Full container-backed sink e2e suite passed.

The gate is local and opt-in. It is not a GitHub Actions default and it does not run unless NATS_SINKS_RUN_CONTAINER_E2E=1 is set. The backend helpers bind to loopback, use fake data, bound readiness waits, and remove containers by default.

To include live Oracle checks, source the ignored local integration environment files first and set:

export NATS_SINKS_RUN_LIVE_ORACLE=1
export NATS_SINKS_RUN_LIVE_E2E=1
scripts/check-sinks.sh

Integration Tests

Integration tests are marked:

pytest -m integration

They require isolated local services. Do not point integration tests at production NATS or Oracle instances.

Oracle Integration Tests

Oracle integration tests are disabled unless explicitly enabled. This keeps the default test suite deterministic and prevents accidental network calls.

The Oracle tests use OracleSink(auto_create=True). When enabled, the sink attempts to create the configured test table before writing. If the table already exists, Oracle raises ORA-00955 and the sink treats that as success. This is a create-if-missing flow rather than a destructive schema migration.

Required environment:

export NATS_SINKS_ORACLE_INTEGRATION=1
export NATS_SINKS_ORACLE_DSN='localhost:1521/FREEPDB1'
export NATS_SINKS_ORACLE_USER='NATS_SINK_TEST'
export ORACLE_PASSWORD='replace-with-test-password'

Optional environment:

export NATS_SINKS_ORACLE_TABLE='NATS_SINKS_ORACLE_TEST_EVENTS_V2'
export NATS_SINKS_ORACLE_PASSWORD_ENV='ORACLE_PASSWORD'
export NATS_SINKS_ORACLE_CONFIG_DIR='.local/oracle-adb/wallet'
export NATS_SINKS_ORACLE_WALLET_LOCATION='.local/oracle-adb/wallet'
export NATS_SINKS_ORACLE_WALLET_PASSWORD_ENV='ORACLE_WALLET_PASSWORD'
export NATS_SINKS_ORACLE_SSL_SERVER_DN_MATCH='true'
export NATS_SINKS_ORACLE_RETRY_COUNT='20'
export NATS_SINKS_ORACLE_RETRY_DELAY='3'
export NATS_SINKS_ORACLE_DROP_TABLE_BEFORE='false'
export NATS_SINKS_ORACLE_DROP_TABLE_AFTER='false'

For Autonomous Database wallet/mTLS tests, unzip the downloaded wallet into an ignored directory and keep both the database password and wallet password in environment variables:

mkdir -p .local/oracle-adb/wallet
unzip Wallet_MYDB.zip -d .local/oracle-adb/wallet
export NATS_SINKS_ORACLE_DSN='mydb_low'
export NATS_SINKS_ORACLE_CONFIG_DIR='.local/oracle-adb/wallet'
export NATS_SINKS_ORACLE_WALLET_LOCATION='.local/oracle-adb/wallet'
export NATS_SINKS_ORACLE_WALLET_PASSWORD_ENV='ORACLE_WALLET_PASSWORD'
export ORACLE_WALLET_PASSWORD='replace-with-wallet-password'

Run only the Oracle integration test module:

pytest -m integration tests/integration/test_oracle_sink.py

The Oracle integration tests use a specific retained test table. The default is NATS_SINKS_ORACLE_TEST_EVENTS_V2; override it with NATS_SINKS_ORACLE_TABLE. The table is not dropped before or after tests unless NATS_SINKS_ORACLE_DROP_TABLE_BEFORE=true or NATS_SINKS_ORACLE_DROP_TABLE_AFTER=true is set. Keeping the table by default lets operators inspect rows after a test run.

Before writing, the integration test verifies that the retained table contains the current required columns. If it finds an older table layout, it fails with a clear message instead of surfacing a lower-level Oracle invalid-identifier error. Set NATS_SINKS_ORACLE_DROP_TABLE_BEFORE=true for the test table, or choose a fresh table name, when you intentionally want the test to recreate the schema.

The database user must have enough privilege to create the configured test table when it is missing, insert or merge rows into it, optionally drop the test table when cleanup flags are enabled, and select row counts for assertions. Use a dedicated non-production schema.

sequenceDiagram
    participant T as Pytest
    participant S as OracleSink
    participant DB as Oracle or Autonomous Database

    T->>S: start(auto_create=True)
    S->>DB: create table NATS_SINKS_ORACLE_TEST_EVENTS_V2
    DB-->>S: created or ORA-00955 already exists
    T->>S: write_batch(test envelope)
    S->>DB: merge row and commit
    T->>DB: select count(*) for unique test stream
    DB-->>T: one row

Manual Live NATS Probe

For a real NATS server, use the tracked manual probe script:

scripts/nats-live-probe.py

The script is not part of the automated unit test suite because it makes a network connection. It is useful for validating:

  • TLS connection setup,
  • local CA certificate trust,
  • token or username/password authentication,
  • credentials-file, NKEY seed-file, and TLS client certificate workflows when the target NATS deployment is configured for them,
  • subscribing to a subject,
  • optionally publishing and receiving a test message.

Keep live material out of git:

mkdir -p .local/nats-live
chmod 700 .local/nats-live
$EDITOR .local/nats-live/ca.crt
cat > .local/nats-live/nats-sink.env <<'EOF'
NATS_PASSWORD=replace-with-test-password
EOF
chmod 600 .local/nats-live/ca.crt .local/nats-live/nats-sink.env

Subscribe-only probe:

python scripts/nats-live-probe.py \
  --server tls://nats.example.com:4222 \
  --user example_user \
  --password-env NATS_PASSWORD \
  --env-file .local/nats-live/nats-sink.env \
  --ca-file .local/nats-live/ca.crt \
  --subject example.test.subject

Publish-and-receive probe:

python scripts/nats-live-probe.py \
  --server tls://nats.example.com:4222 \
  --user example_user \
  --password-env NATS_PASSWORD \
  --env-file .local/nats-live/nats-sink.env \
  --ca-file .local/nats-live/ca.crt \
  --subject example.test.subject \
  --publish \
  --message '{"probe":"nats-sinks","kind":"live-test"}'

Only use --publish with an explicitly approved test subject. The probe does not print payload contents by default.

For client-side authentication workflow certification, use the gated pytest module instead of adding live credentials to unit tests:

export NATS_SINKS_NATS_AUTH_INTEGRATION=1
export NATS_SINKS_AUTH_MODE=nkey_seed_file
export NATS_SINKS_AUTH_URL=tls://nats.example.com:4222
export NATS_SINKS_AUTH_NKEY_SEED_FILE=/run/secrets/nats/orders-sink.nk
export NATS_SINKS_AUTH_TLS_CA_FILE=/etc/nats/certs/root-ca.crt
python -m pytest tests/integration/test_nats_auth_workflows.py -q

Supported modes are none, username_password, token, credentials_file, nkey_seed_file, and tls_client_certificate. Keep all real endpoints and identity files in ignored local configuration or runtime secret mounts.

Ordering Tests

sequenceDiagram
    participant T as Test
    participant R as Runner
    participant S as Mock Sink
    participant M as Mock Message

    T->>R: process_raw_batch([message])
    R->>S: write_batch
    S-->>R: success
    R->>M: ack
    T->>T: assert write before ack

This style keeps the most important delivery invariant executable in CI.

Failure Scenarios Covered

The unit and integration tests intentionally exercise common non-happy paths:

  • a message without Nats-Msg-Id is persisted when stream-sequence idempotency is active,
  • a message with Nats-Expected-Stream persists that reserved NATS header in metadata_json,
  • an empty message body is wrapped and stored instead of crashing,
  • malformed JSON-looking text is preserved as text unless payload_mode is explicitly json_only,
  • non-UTF-8 bytes are base64-wrapped so opaque payloads remain durable,
  • invalid mission metadata is treated as a permanent validation failure and follows DLQ-before-ACK behavior when DLQ is configured.

Live NATS To Oracle End-To-End Test

The end-to-end integration test is the repeatable Oracle-backend acceptance test. It publishes real messages to NATS JetStream, uses JetStreamSinkRunner to fetch the messages, writes them through OracleSink, and verifies that every row exists in Oracle. It is disabled by default and only runs when explicitly enabled.

The default message count is 256. Override it with NATS_SINKS_E2E_MESSAGE_COUNT when you want a shorter smoke test or a larger throughput-style run. The default sink batch size is 64 and can be overridden with NATS_SINKS_E2E_BATCH_SIZE.

Use a message count that is not an exact multiple of the batch size when you want to prove final partial-batch behavior. For example, with NATS_SINKS_E2E_MESSAGE_COUNT=250 and NATS_SINKS_E2E_BATCH_SIZE=64, the test expects four backend writes and verifies that the final batch size is 58. This confirms that the runner does not require a full batch before writing to the backend.

Local Secret Layout

Place live test material under .local/. The repository ignores .local/, so these files stay out of git:

.local/
  nats-live/
    ca.crt
    nats-sink.env
  oracle-adb/
    integration.env
    wallet/
      tnsnames.ora
      sqlnet.ora
      ewallet.pem
      cwallet.sso
      ewallet.p12
      ...
  nats-oracle-e2e/
    integration.env

The NATS env file should contain only the NATS client secret:

mkdir -p .local/nats-live
chmod 700 .local/nats-live
$EDITOR .local/nats-live/ca.crt
cat > .local/nats-live/nats-sink.env <<'EOF'
NATS_PASSWORD=replace-with-test-password
EOF
chmod 600 .local/nats-live/ca.crt .local/nats-live/nats-sink.env

The Oracle ADB env file should contain the Oracle connection settings and database or wallet secrets. For wallet/mTLS, unzip the wallet directly into .local/oracle-adb/wallet.

mkdir -p .local/oracle-adb/wallet
chmod 700 .local/oracle-adb .local/oracle-adb/wallet
unzip Wallet_MYDB.zip -d .local/oracle-adb/wallet

cat > .local/oracle-adb/integration.env <<'EOF'
NATS_SINKS_ORACLE_INTEGRATION=1
NATS_SINKS_ORACLE_DSN=natstest_high
NATS_SINKS_ORACLE_USER=NATS_SINK_TEST
NATS_SINKS_ORACLE_PASSWORD_ENV=ORACLE_PASSWORD
ORACLE_PASSWORD=replace-with-database-password
NATS_SINKS_ORACLE_CONFIG_DIR=.local/oracle-adb/wallet
NATS_SINKS_ORACLE_WALLET_LOCATION=.local/oracle-adb/wallet
NATS_SINKS_ORACLE_WALLET_PASSWORD_ENV=ORACLE_WALLET_PASSWORD
ORACLE_WALLET_PASSWORD=replace-with-wallet-password
NATS_SINKS_ORACLE_TABLE=NATS_SINKS_ORACLE_TEST_EVENTS_V2
NATS_SINKS_ORACLE_SSL_SERVER_DN_MATCH=true
NATS_SINKS_ORACLE_RETRY_COUNT=20
NATS_SINKS_ORACLE_RETRY_DELAY=3
NATS_SINKS_ORACLE_DROP_TABLE_BEFORE=false
NATS_SINKS_ORACLE_DROP_TABLE_AFTER=false
EOF
chmod 600 .local/oracle-adb/integration.env

For walletless TLS, set NATS_SINKS_ORACLE_DSN to the full tcps connect descriptor and omit the wallet fields.

The e2e env file should contain the NATS endpoint, the test stream/subject, and the optional message-count controls:

mkdir -p .local/nats-oracle-e2e
chmod 700 .local/nats-oracle-e2e
cat > .local/nats-oracle-e2e/integration.env <<'EOF'
NATS_SINKS_E2E_INTEGRATION=1
NATS_SINKS_E2E_NATS_URL=tls://nats.example.com:4222
NATS_SINKS_E2E_NATS_USER=example_nats_user
NATS_SINKS_E2E_NATS_PASSWORD_ENV=NATS_PASSWORD
NATS_SINKS_E2E_NATS_TLS_CA_FILE=.local/nats-live/ca.crt
NATS_SINKS_E2E_STREAM=NATS_SINKS_E2E
NATS_SINKS_E2E_SUBJECT=example.test.*
NATS_SINKS_E2E_PUBLISH_SUBJECT=example.test.subject
NATS_SINKS_E2E_ORACLE_TABLE=NATS_SINKS_E2E_EVENTS_V2
NATS_SINKS_E2E_MESSAGE_COUNT=256
NATS_SINKS_E2E_BATCH_SIZE=64
NATS_SINKS_E2E_TEXT_PAYLOAD_INTERVAL=17
NATS_SINKS_E2E_EMPTY_PAYLOAD_INTERVAL=31
NATS_SINKS_E2E_MISSING_MESSAGE_ID_INTERVAL=23
NATS_SINKS_E2E_EXPECTED_STREAM_HEADER_INTERVAL=29
NATS_SINKS_E2E_DROP_TABLE_BEFORE=false
NATS_SINKS_E2E_DROP_TABLE_AFTER=false
NATS_SINKS_E2E_PRINT_TIMINGS=true
# Optional encrypted Oracle e2e mode:
NATS_SINKS_E2E_ENCRYPTION_ENABLED=false
NATS_SINKS_E2E_ENCRYPTION_ALGORITHM=aes-256-gcm
NATS_SINKS_E2E_ENCRYPTION_KEY_ID=nats-sinks-e2e-key
NATS_SINKS_E2E_ENCRYPTION_KEY_B64_ENV=NATS_SINKS_E2E_ENCRYPTION_KEY_B64
EOF
chmod 600 .local/nats-oracle-e2e/integration.env

The test creates the JetStream stream when it is missing. If the stream already exists, it verifies that the configured subject is included in the stream subjects. For each run, the test creates a unique durable consumer with DeliverPolicy.NEW, AckPolicy.EXPLICIT, and MaxAckPending large enough for the requested test batch. It then publishes the requested number of messages to NATS_SINKS_E2E_PUBLISH_SUBJECT. Most messages are JSON objects, and every NATS_SINKS_E2E_TEXT_PAYLOAD_INTERVAL message is encrypted-text-style non-JSON text. The test runs the sink runner subscribed to NATS_SINKS_E2E_SUBJECT, verifies the Oracle row count and distinct message-ID count, verifies that messages without Nats-Msg-Id do not crash, verifies that messages with Nats-Expected-Stream are represented in metadata_json, verifies that empty message bodies do not crash and are stored in the nats-sinks JSON payload envelope, verifies that the expected number of text payloads were stored, confirms backend write timing metrics were captured, confirms there are no pending ACKs on the test consumer, and deletes the test consumer.

When NATS_SINKS_E2E_ENCRYPTION_ENABLED=true, the same e2e test encrypts payloads before Oracle writes, verifies that every stored payload contains the standard _nats_sinks_encryption envelope, decrypts the stored values with the configured key, and compares them with the original JSON, text, empty, and binary test payloads.

The e2e test uses a specific retained Oracle table. The default is NATS_SINKS_E2E_EVENTS_V2; override it with NATS_SINKS_E2E_ORACLE_TABLE. The table is not dropped before or after the test unless NATS_SINKS_E2E_DROP_TABLE_BEFORE=true or NATS_SINKS_E2E_DROP_TABLE_AFTER=true is set. This is intentional: keeping the table lets operators inspect stored payloads, metadata, and timing columns after the test. If the table has an older layout and you want the test to recreate it, set NATS_SINKS_E2E_DROP_TABLE_BEFORE=true for that run.

Retained Oracle Table Schema Drift

Retained test tables are useful because they let you inspect rows after a live run, but they can outlive schema changes. When the project adds columns such as PRIORITY, CLASSIFICATION, LABELS, timestamp fields, METADATA_JSON, or MISSION_METADATA_JSON, an older retained table may no longer match the current e2e contract.

This is handled deliberately. The e2e test checks the table shape before it publishes messages or starts the runner. If required columns are missing, the test fails fast with a clear schema message and does not process messages against that table. Treat that result as a safety guard, not as a delivery failure.

You have three safe options:

  1. Choose a fresh table name for the run, letting OracleSink.ensure_schema() create the current test layout.
  2. Set NATS_SINKS_E2E_DROP_TABLE_BEFORE=true only for an isolated test table that may be destroyed and recreated.
  3. Migrate the retained table manually through your normal database change process, then rerun the e2e test.

For example, this command keeps the table after the run while avoiding any older retained table:

scripts/run-oracle-e2e.sh --table NATS_SINKS_E2E_METRICS --message-count 256 --batch-size 64

Do not use --drop-table-before or --drop-table-after on a table that holds evidence, audit records, production data, or data that another test still needs.

Set NATS_SINKS_E2E_SUBJECT to a wildcard such as example.test.* and NATS_SINKS_E2E_PUBLISH_SUBJECT to a concrete matching subject such as example.test.subject when you want the e2e test to prove wildcard subscription behavior.

When NATS_SINKS_E2E_PRINT_TIMINGS=true, the test prints a backend write timing summary based on the runner's sink_batch_write_seconds observations. This measures the duration of sink.write_batch(...), including the Oracle write and commit. The older batch_write_seconds alias remains available for compatibility with earlier local reports.

Run it with:

set -a
source .local/nats-live/nats-sink.env
source .local/oracle-adb/integration.env
source .local/nats-oracle-e2e/integration.env
set +a
pytest -m integration tests/integration/test_nats_oracle_e2e.py

The tracked helper script performs the same run and accepts explicit cleanup flags. Both cleanup flags default to false:

scripts/run-oracle-e2e.sh
scripts/run-oracle-e2e.sh --drop-table-before
scripts/run-oracle-e2e.sh --drop-table-after
scripts/run-oracle-e2e.sh --table NATS_SINKS_E2E_EVENTS_V2 --message-count 256
scripts/run-oracle-e2e.sh --table NATS_SINKS_E2E_EVENTS_V2 --message-count 250 --batch-size 64
scripts/run-oracle-e2e.sh --with-encryption --message-count 64 --batch-size 16
scripts/run-oracle-e2e.sh --with-encryption --encryption-algorithm aes-256-ccm --preserve-key-material

--with-encryption generates temporary AES-256 key material when NATS_SINKS_E2E_ENCRYPTION_KEY_B64 is not already set. The generated key file is deleted after the run by default. Use --preserve-key-material only for local debugging, and keep preserved files out of git.

sequenceDiagram
    participant T as Pytest
    participant N as NATS JetStream
    participant R as JetStreamSinkRunner
    participant E as PayloadEncryptor
    participant O as OracleSink
    participant DB as Oracle ADB

    T->>N: ensure stream and create durable consumer
    T->>N: publish N mixed JSON/text/empty/header-variation test messages
    loop bounded batches
        R->>N: pull batch
        opt encrypted e2e mode
            R->>E: encrypt payload bytes
            E-->>R: encrypted payload envelopes
        end
        R->>O: write_batch(envelopes)
        O->>DB: merge rows and commit
        O-->>R: durable success
        R->>N: ACK messages
    end
    T->>DB: verify N rows by message_id
    T->>N: verify no pending ACKs