This article describes an approach for creating comprehensive end-to-end system tests. We’ll automate the setup of the entire testing environment, including data preparation.

By leveraging features offered by pytest and docker compose, we can maintain a clean design for the tests, promote maintainability and reduce boilerplate code.

System testing

#

If you’re new to testing, I always recommend starting with the test pyramid and Martin Fowler’s article on the practical test pyramid.

In the section titled Acceptance Tests — Do Your Features Work Correctly? you’ll learn that:

The higher you move up in your test pyramid the more likely you enter the realms of testing whether the features you’re building work correctly from a user’s perspective. You can treat your application as a black box and shift the focus in your tests from:

• when I enter the values x and y, the return value should be z

towards:

• given there’s a logged in user
• and there’s an article “bicycle”
• when the user navigates to the “bicycle” article’s detail page
• and clicks the “add to basket” button
• then the article “bicycle” should be in their shopping basket

This is exactly what you want: a black-box style automated system-wide test that verifies the system behaves as expected given certain user activity.

First automated system-test for Popyka

#

For the first automated system test of Popyka, we will test the following scenario:

• Given: A user performs an INSERT, UPDATE, or DELETE operation in the database.
• Then: The changes made by the INSERT, UPDATE, or DELETE operation should be captured by Popyka.
  • Additionally, messages describing these changes should be published to Kafka.

However, in real-world scenarios, our user inserting a row will likely be another system (not a person). To make the system test more realistic, I decided to create a simple Django-based application that simulates this user behavior.

In this case, the application leverages the built-in Django Admin interface. The test will involve a programmatic login to the Django Admin, mimicking a user’s action.

The updated scenario can be described more precisely:

• Given: A user logs in to the Django Admin.
• When: Django executes its expected behavior of performing one insert and two update operations in the database.
• Then: Popyka should capture all three changes (one insert and two updates).
  • Additionally, three messages describing these changes should be published to Kafka.

The system

#

The whole system (testing environment) is defined in a single docker compose file.

services:
  demo-db:
    build:
      context: postgres16-wal2json
    ports:
      - 54091:5432

  demo-django-admin:
    build:
      context: .
    environment:
      DATABASE_URL: "postgresql://postgres:pass@demo-db:5432/postgres"
    ports:
      - 8081:8080

  demo-popyka:
    build:
      context: ../../
    environment:
      POPYKA_CONFIG: "${POPYKA_CONFIG:-}"
      POPYKA_DB_DSN: "postgresql://postgres:pass@demo-db:5432/postgres"
      POPYKA_KAFKA_BOOTSTRAP_SERVERS: 'demo-kafka:9092'
      POPYKA_DB_SLOT_NAME: "popyka"
      POPYKA_KAFKA_TOPIC: "popyka"
      LAZYTOSTR_COMPACT: "${LAZYTOSTR_COMPACT:-0}"
    volumes:
      - ./popyka-config:/popyka-config

  demo-kafka:
    image: 'bitnami/kafka:3.5'
    ports:
      - 54092:54092

Our testing environment consists of four services:

• demo-popyka: This is the Popyka instance we intend to test.
• demo-db: A PostgreSql instances.
• demo-django-admin: The demo Django Admin application.
• demo-kafka: The Kafka instance where Popyka will publish messages.

The system test

#

We want to bring up the entire testing environment (including Popyka and its dependencies) and then simulate a user logging in to the Django Admin. This login will trigger INSERT and UPDATE operations on the database. We will then verify that Popyka captures these changes and publishes them to Kafka:

The test is implemented using pytest ( code).

We’ll explain each of the code blocks in the next sections. Here’s an overview of the system test:

@system_test
def test_django_admin_login_with_default_config(
    setup_env,  # (1) Environment setup
    dc_popyka_default_config,
    consumer
):
    # (2) Django login
    django_admin_login()

    # (3) Waiting for database changes
    dc_popyka_default_config.wait_for_change(timeout=5).assert_insert().assert_table("django_session")
    dc_popyka_default_config.wait_for_change(timeout=5).assert_update().assert_table("auth_user")
    dc_popyka_default_config.wait_for_change(timeout=5).assert_update().assert_table("django_session")

    # (4) Verifying Kafka messages
    expected_summaries = [("I", "django_session"), ("U", "auth_user"), ("U", "django_session")]
    actual_summaries = consumer.wait_for_count_summarized(3, timeout=10)
    assert sorted(actual_summaries) == sorted(expected_summaries)

1. Environment setup: This step is handled by pytest fixtures, which will be explained in more detail later.
2. Django login: The django_admin_login() function is in charge of logging in to the Django Admin.
3. Waiting for database changes: The dc_popyka_default_config.wait_for_change() function waits until the expected database changes (caused by the login) are processed by Popyka.
4. Verifying Kafka messages: The last code block waits until Popyka publishes the 3 messages to Kafka.

Short intro to pytest fixtures

#

For the full explanation, refer to the pytest documentation about fixtures.

We utilize 3 specific pytest fixtures to set up the testing environment:

@system_test
def test_django_admin_login_with_default_config(
    setup_env,
    dc_popyka_default_config,
    consumer
):
    ...

• setup_env: This fixture sets up the testing environment by bringing up the required services and preparing any necessary test data. Code.
• dc_popyka_default_config: This fixture brings up a Popyka instance configured specifically for the scenario you want to test. Code.
• consumer: This fixture returns a Kafka consumer that is ready to be used within your test. This consumer can be used to consume the messages published by Popyka to Kafka during the test. Code.

Step 1: Environment setup

#

The test starts by referencing the setup_env fixture, which is responsible for setting up the test environment.

@system_test
def test_django_admin_login_with_default_config(
    setup_env,
    dc_popyka_default_config,
    consumer
):
    ...

The implementation of this fixture is straightforward:

@pytest.fixture
def setup_env(kill_popyka, clean_data, docker_compose_deps):
    yield

It relies on 3 other fixtures to handle specific tasks within the environment setup process:

• kill_popyka: This fixture ensures a clean slate by terminating any lingering instances of demo-popyka from prior test runs. Code.
• clean_data: This fixture cleans up the database and Kafka topic, guaranteeing a consistent starting point for your tests in terms of data. Code.
• docker_compose_deps: This fixture brings up the required dependencies (Django, PostgreSQL, and Kafka) using docker compose. Code.

A predictable testing environment is crucial for verifying system behavior. By utilizing these fixtures, we achieve a high degree of encapsulation and reusability within the test suite.

Step 2: Django login

#

Code:

django_admin_login()

This is not a fixture, this is a simple python function.

Our requirements here are simple: loading the Django Admin URL, filling in the login form, and submitting it. Libraries like mechanize, requests, or even the built-in http client can achieve this. However, in this case, mechanize is used because it often leads to clearer and more maintainable code for this specific kind of task.

Step 3: Waiting for database changes

#

Code:

dc_popyka_default_config.wait_for_change(timeout=5).assert_insert().assert_table("django_session")
dc_popyka_default_config.wait_for_change(timeout=5).assert_update().assert_table("auth_user")
dc_popyka_default_config.wait_for_change(timeout=5).assert_update().assert_table("django_session")

In this step, the test waits (blocks) until it can confirm that Popyka has received the changes from the PostgreSQL database. This might seem to contradict the black-box testing approach we’re aiming for, and that’s a valid concern.

There are practical reasons for using a busy wait here. By blocking the test until Popyka’s expected behavior is verified, we gain valuable visibility into potential failures:

if the test fails at this point, it indicates an issue before Popyka interacts with Kafka. This narrows down the root cause investigation by eliminating any Kafka-related issues and focusing on the communication between PostgreSql and Popyka, or more probably, in Popyka code itself.

Step 4: Verifying Kafka messages

#

Code:

expected_summaries = [("I", "django_session"), ("U", "auth_user"), ("U", "django_session")]
actual_summaries = consumer.wait_for_count_summarized(3, timeout=10)
assert sorted(actual_summaries) == sorted(expected_summaries)

Having confirmed that Popyka captured the changes from PostgreSQL (Step 3), we now verify that it published 3 messages to Kafka, reflecting the 3 database operations:

• INSERT into django_session.
• UPDATE of auth_user.
• UPDATE of django_session.

The test utilizes consumer.wait_for_count_summarized(3, timeout=10) to wait for a maximum of 10 seconds until 3 messages are received from Kafka. To enhance test readability, this function retrieves a summarized version of the messages rather than the full JSON content (which can be large).

The final assertion clearly demonstrates that Popyka’s functionality functioned as expected within the context of this specific test scenario 😀.

Conclusion

#

By following these steps and leveraging:

• pytest fixtures for reusability, isolation, readability and decoupling of test from setup,
• docker compose to have a controlled testing environment,
• mechanize to automate interaction with Django Admin,

this approach establishes a solid foundation for testing Popyka’s core functionality, end to end.

While this scenario focuses on a specific case, it demonstrates a clear testing pattern that can be adapted for various use cases.