Skip to main content
You set up your etcd cluster and got it running in our environment in part 1. Now, you’ll continue working in the same project directory to test it. Here’s the source code for this tutorial. A distributed datastore needs to be consistent. You put data into it, get a successful response, fetch it in the future, and it matches your expectation. While it’s easy enough to write a script that inserts and reads key-value pairs, making it into a meaningful test that simulates production conditions — with parallel requests, a varied cadence, and a faulty environment — is more complicated. Antithesis simplifies this greatly by providing a Test Composer — a framework that takes care of the parallelism, variation in command order, and more. All you need to provide are the basic functions that exercise the system.

​
Create some helper functions

In this section, you’ll create helper functions that make requests to the etcd cluster. Add a new client directory to your etcd-antithesis directory: 
$ mkdir client && cd client
You’ll use this directory for all client code that uses Test Composer to exercise your system. Inside the client directory, add python-generate-traffic/resources directories: 
$ mkdir python-generate-traffic/resources && cd python-generate-traffic/resources
Inside resources, add a helper.py file. You should now have the following file structure in your etcd-antithesis directory: 
etcd-antithesis/
├── client/
│   └── python-generate-traffic/
│       └── resources/
│           └── helper.py
├── config/
│   ├── docker-compose.yaml
│   └── Dockerfile.config
└── health-checker/
    ├── Dockerfile.health-checker
    └── entrypoint.py

​
Make requests to the cluster

Inside helper.py, add functions to connect to a randomly chosen etcd node, and to get and put data: 
import etcd3, string

# Antithesis SDK
from antithesis.random import (
    random_choice,
)

def connect_to_host():
    host = random_choice(["etcd0", "etcd1", "etcd2"])
    try:
        client = etcd3.client(host=host, port=2379)
        print(f"Client: connected to {host}")
        return client
    except Exception as e:
        print(f"Client: failed to connect to {host}. exiting")
        sys.exit(1)

def get_request(c, key):
    try:
        response = c.get(key)
        database_value = response[0].decode('utf-8')
        return True, None, database_value
    except Exception as e:
        return False, e, None

def put_request(c, key, value):
    try:
        c.put(key, value)
        return True, None
    except Exception as e:
        return False, e
Note that we’re selecting a random node with the built-in random_choice function from Antithesis’s Python SDK.

​
Generate random strings

Next, add a helper function that uses random_choice to generate some random strings to insert: 
def generate_random_string():
    random_str = []\
    for _ in range(16):
        random_str.append(random_choice(list(string.ascii_letters + string.digits)))
    return "".join(random_str)
Add another helper function that uses the built-in get-random function from the Antithesis Python SDK to generate numbers between 1-100: 
# Antithesis SDK
from antithesis.random import (
    random_choice,
    get_random,
)

# ...

def generate_num_requests():
    return (get_random() % 100) + 1
You’ll use this later to generate the number of requests inserting data into the cluster.

​
Add a Test Composer command

Next, we’ll add a Test Composer command that uses these helper functions to add key-value pairs to the etcd cluster and then check for mismatches. The Test Composer relies on an opinionated framework that identifies executable scripts as test commands using a naming convention. There are a few types of test commands, but we’ll only use the parallel driver command in this tutorial. Parallel driver commands make up the core of a typical Antithesis test, and can run in parallel with other parallel driver commands, including other copies of the same one. To make a script a parallel driver command , all we do is name the file parallel_driver_<name>. In this case, we’ll name it parallel_driver_generate_traffic.py: 
etcd-antithesis/
└── client/
    └── python-generate-traffic/
        ├── parallel_driver_generate_traffic.py
        └── resources/
            └── helper.py
Test commands should be marked as executables. From the python-generate-traffic directory, run: 
$ chmod 777 parallel_driver_generate_traffic.py
They also require an appropriate shebang in the first line (in this case #!/usr/bin/env -S python3 -u). Add a simulate_traffic function to parallel_driver_generate_traffic.py that uses your helper functions to connect to an etcd host, then generate a random number of put requests that each add a randomly generated key-value pair: 
#!/usr/bin/env -S python3 -u

import sys
sys.path.append("/opt/antithesis/resources")
import helper

def simulate_traffic(prefix):
    """
        This function will first connect to an etcd host, then execute a certain number of put requests. 
        The key and value for each put request are generated using Antithesis randomness (check within the helper.py file). 
        We return the key/value pairs from successful requests.
    """
    client = helper.connect_to_host()
    num_requests = helper.generate_num_requests()
    kvs = []

    for _ in range(num_requests):

        # generating random str for the key and value
        key = prefix+helper.generate_random_string()
        value = helper.generate_random_string()

        # response of the put request
        success, error = helper.put_request(client, key, value)

        if success:
            kvs.append((key, value))
            print(f"Client: successful put with key '{key}' and value '{value}'")
        else:
            print(f"Client: unsuccessful put with key '{key}', value '{value}', and error '{error}'")

    print(f"Client: traffic simulated!")
    return kvs
It’s okay for put_request to be unsuccessful during faults, so we print an “unsuccessful” message rather than stopping execution. Next, add a validate_puts function that checks for key-value mismatches. This function takes an array of expected key-value pairs. For each key, it makes a get request to a random etcd host, and checks whether the value matches the expected one: 
def validate_puts(kvs):
    """
        This function will first connect to an etcd host, then perform a get request on each key in the key/value array. 
        For each successful response, we check that the get request value == value from the key/value array. 
        If we ever find a mismatch, we return it. 
    """
    client = helper.connect_to_host()

    for kv in kvs:
        key, value = kv[0], kv[1]
        success, error, database_value = helper.get_request(client, key)

        if not success:
            print(f"Client: unsuccessful get with key '{key}', and error '{error}'")
        elif value != database_value:
            print(f"Client: a key value mismatch! This shouldn't happen.")
            return False, (value, database_value)

    print(f"Client: validation ok!")
    return True, None
Now bring it all together. Call the simulate_traffic function to generate random key-value pairs and add them to the datastore, then call validate_puts to get the values back from the datastore and check that they match: 
if __name__ == "__main__":
    prefix = helper.generate_random_string()
    kvs = simulate_traffic(prefix)
    values_stay_consistent, mismatch = validate_puts(kvs)
If the values match, values_stay_consistent will be true and mismatch will be None.

​
Add some assertions to validate

Assertions express properties your system should have, and Antithesis relies on assertions to understand what you’re testing for. Assertions in Antithesis describes the mechanics in a lot more detail. Antithesis’s SDKs provide many types of assertions, but we’ll only use two here.

​
Add an always assertion

The first is an Always assertion — these assertions are similar to the programming assertions you’re familiar with, but they don’t crash your program. They create a property that Antithesis will test, and list in the triage report as passing or failing. You always want the datastore to be consistent. So, in your parallel driver command, values_stay_consistent must always be true. Import always assertions from the Python SDK, and add one to your __main__ block to test that values stay consistent: 
# ...

from antithesis.assertions import (
    always,
)

# ...

if __name__ == "__main__":
    prefix = helper.generate_random_string()
    kvs = simulate_traffic(prefix)
    values_stay_consistent, mismatch = validate_puts(kvs)

    # Always assertion: we expect that the values we put in the database stay consistent
    always(values_stay_consistent, "Database key values stay consistent", {"mismatch":mismatch})

​
Add a sometimes assertion

The second assertion we’ll use is a Sometimes assertion (these are so valuable and unusual they get a whole section of documentation to themselves). When inserting key-value pairs into a distributed datastore in the face of network and environmental faults, it’s okay for some requests to fail. But if none of them succeed then your system is never able to insert keys into etcd and that’s either a bug or a test misconfiguration that needs attention. Here’s what a sometimes assertion looks like. 
sometimes(success, "Client can make successful put requests", {"error":error})
sometimes(error!=None, "Client put requests can fail", None)
The first parameter is the something that should happen sometimes. The second describes the property we’re asserting. Import sometimes assertions from the Python SDK and add them to your simulate_traffic function to assert that put requests are sometimes successful, and sometimes fail: 
from antithesis.assertions import (
    always,
    sometimes,
)


def simulate_traffic(prefix):
    """
        This function will first connect to an etcd host, then execute a certain number of put requests. 
        The key and value for each put request are generated using Antithesis randomness (check within the helper.py file). 
        We return the key/value pairs from successful requests.
    """
    client = helper.connect_to_host()
    num_requests = helper.generate_requests()
    kvs = []

    for _ in range(num_requests):

        # generating random str for the key and value
        key = prefix+helper.generate_random_string()
        value = helper.generate_random_string()

        # response of the put request
        success, error = helper.put_request(client, key, value)

        # Sometimes assertion: sometimes put requests are successful. A failed request is OK since we expect them to happen.
        sometimes(success, "Client can make successful put requests", {"error":str(error)})
        sometimes(error!=None, "Client put requests can fail", None)

        if success:
            kvs.append((key, value))
            print(f"Client: successful put with key '{key}' and value '{value}'")
        else:
            print(f"Client: unsuccessful put with key '{key}', value '{value}', and error '{error}'")

    print(f"Client: traffic simulated!")
    return kvs
Similarly, add sometimes assertions to your validate_puts function to assert that get requests sometimes succeed and sometimes fail: 
def validate_puts(kvs):
    """
        This function will first connect to an etcd host, then perform a get request on each key in the key/value array. 
        For each successful response, we check that the get request value == value from the key/value array. 
        If we ever find a mismatch, we return it. 
    """
    client = helper.connect_to_host()

    for kv in kvs:
        key, value = kv[0], kv[1]
        success, error, database_value = helper.get_request(client, key)

        # Sometimes assertion: sometimes get requests are successful. A failed request is OK since we expect them to happen.
        sometimes(success, "Client can make successful get requests", {"error":str(error)})
        sometimes(error!=None, "Client get requests can fail", None)

        if not success:
            print(f"Client: unsuccessful get with key '{key}', and error '{error}'")
        elif value != database_value:
            print(f"Client: a key value mismatch! This shouldn't happen.")
            return False, (value, database_value)

    print(f"Client: validation ok!")
    return True, None
The assertions you’ve added will show up in the triage report as properties, and the report will show if they passed or failed in testing.

​
Build your client

Now you have a test template with one test command to exercise the etcd cluster. To package it, add a Dockerfile.client file in the client directory: 
etcd-antithesis/
├── client
│   ├── Dockerfile.client
│   └── python-generate-traffic
│       ├── parallel_driver_generate_traffic.py
│       └── resources
│           └── helper.py
├── config
│   ├── docker-compose.yaml  
│   └── Dockerfile.config
└── health-checker
    ├── Dockerfile.health-checker
    └── entrypoint.py
Add the following to your Dockerfile.client:
.enumerated}
FROM python:3.12-slim

# Install dependencies and Antithesis Python SDK
RUN pip install --no-cache-dir etcd3 numpy protobuf filelock antithesis cffi 

# Fixes some compability issues by forcing pure-Python protobuf.
ENV PROTOCOL_BUFFERS_PYTHON_IMPLEMENTATION=python

# Copy the test driver into the Test Composer directory
COPY ./python-generate-traffic/parallel_driver_generate_traffic.py /opt/antithesis/test/v1/main/parallel_driver_generate_traffic.py

# Copy additional resources into a resources folder
COPY ./python-generate-traffic/resources/helper.py /opt/antithesis/resources/helper.py
This adds the Antithesis Python SDK and other dependencies to a Python base image, copies the test driver into the test directory, and the helper function into a resources directory. Build your client container image: 
$ docker build . -f Dockerfile.client -t us-central1-docker.pkg.dev/molten-verve-216720/$TENANT_NAME-repository/etcd-client:v1
Replace $TENANT_NAME with your tenant’s name. To make the simulation more realistic, you should always run multiple client containers using the same container image. Update the docker-compose.yaml to configure two client containers: 
  client1:
    image: 'etcd-client:v1'
    container_name: client1
    entrypoint: "sleep infinity"

  client2:
    image: 'etcd-client:v1'
    container_name: client2
    entrypoint: "sleep infinity"
The client containers must be kept running for Antithesis to keep testing. To do this we’ve added a “sleep infinity” command as the entrypoint attribute. Rebuild your config image and tag as v2: 
$ docker build . -f Dockerfile.config -t us-central1-docker.pkg.dev/molten-verve-216720/$TENANT_NAME-repository/etcd-config:v2

​
Push your images

Now push your updated images to the Antithesis registry. As before, authenticate to your container registry with: 
$ cat $TENANT_NAME.key.json | docker login -u _json_key https://us-central1-docker.pkg.dev --password-stdin
Then push your updated container images with the following commands: 
$ docker push us-central1-docker.pkg.dev/molten-verve-216720/$TENANT_NAME-repository/etcd-health-checker:v1
$ docker push us-central1-docker.pkg.dev/molten-verve-216720/$TENANT_NAME-repository/etcd-config:v2
$ docker push us-central1-docker.pkg.dev/molten-verve-216720/$TENANT_NAME-repository/etcd-client:v1
You might want to check your test commands and set up locally before running it in Antithesis.

​
Run your test

Call the basic_test webhook again endpoint with this updated curl command: 
curl --fail -u '$USER:$PASSWORD' \
-X POST https://<$TENANT_NAME>.antithesis.com/api/v1/launch/basic_test \
-d '{"params": { "antithesis.description":"basic_test on main",
    "antithesis.duration":"30",
    "antithesis.config_image":"etcd-config:v2",
    "antithesis.report.recipients":"foo@email.com;bar@email.com"
    } }'
Note that we’re now using v2 of the config image. Also, remember to replace $USER, $PASSWORD, $TENANT_NAME and the antithesis.report.recipients email addresses. The test is set up to run for 30 minutes this time. To view the progress and results of your test, go to your Runs page at https://$TENANT_NAME.antithesis.com/runs and click the Triage results button to see your report when it finishes. You’ll also receive an email when the run completes.

​
Summary

To recap, your first test run validated that Antithesis could run your etcd cluster. You checked that the cluster was healthy and then sent a signal to Antithesis that it was ready to test. In this second test run, you added a test template in the client container to add data to your cluster and check for mismatches. There’s a lot of depth to test templates, and iterating on your test template is a great way to improve your testing. Check out the Test Composer docs for more ideas. Or if you’re keen to get your own system under test in Antithesis, try our setup guide.