How to use the gcloud Pub/Sub emulator with pytest - Part 1

The original pubsub emulator documentation is pretty cool. To get the emulator up and running requires some manual shell work even if it's running on your machine and for the purpose of automated testing it might not necessarily tick all the boxes. After all it's still in beta and pretty barebone as far as gcloud commands go.

Continue reading, if you want to learn about how to turn this little beautiful beta beast into a repeatable testing pattern (until it changes and you have to do it all again).

First write the below in your test/pytest.ini

[pytest]
env =
    PROJECT_ID=fake-project-id
    PUBSUB_EMULATOR_HOST=localhost:8085

For the above to work, install pytest-env. If you can't be bothered, simply export PROJECT_ID and PUBSUB_EMULATOR_HOST as environment variables and carry on.

export PROJECT_ID=fake-project-id
export PUBSUB_EMULATOR_HOST=localhost:8085

Import your environment variables with os.getenv.

from os import getenv

PROJECT_ID = getenv("PROJECT_ID")
PUBSUB_EMULATOR_HOST = getenv("PUBSUB_EMULATOR_HOST")

Now for the pièce de résistance. Let's define our pytest fixture function. If this is an unfamiliar term, check out the official pytest documentation for the juicy details.

from subprocess import PIPE, Popen

import pytest

@pytest.fixture(scope="module")
def emulator() -> None:
    start = [
        "gcloud",
        "beta",
        "emulators",
        "pubsub",
        "start",
        f"--project={PROJECT_ID}",
        f"--host-port={PUBSUB_EMULATOR_HOST}",
    ]
    Popen(start, stdout=PIPE)

    yield

    exec_ps = Popen(["ps", "-ef"], stdout=PIPE)
    exec_pgrep = Popen(
        ["pgrep", "-f", "cloud-pubsub-emulator"], stdin=exec_ps.stdout, stdout=PIPE
    )
    Popen(["xargs", "kill"], stdin=exec_pgrep.stdout, stdout=PIPE)

There are a lot of pieces to unpack here. Let's get started.

1. Scope

@pytest.fixture(scope="module")

If you don't want the emulator to be starting and stopping with every individual test it's perfectly ok to set it's scope to "module". In the interest of keeping this piece short read about different fixture scopes.

2. Function Annotation

def emulator() -> None:

that() -> arrow is called a function annotation. In this case it's indicating that the function does not output anything at all.

3. The subprocess module

This python module is generally used to execute shell commands. The recommended approach is to use subprocess.run. However, (a) using subprocess.run will cause our fixture function to wait until the emulator is terminated and (b) using pipes is not possible. Therefore, we will choose to use subprocess.Popen instead (which incidentally is what subprocess.run uses in the background).

The gcloud command in the pubsub emulator documentation is now transformed to its subprocess equivalent. More specifically:

gcloud beta emulators pubsub start

becomes:

start = [
        "gcloud",
        "beta",
        "emulators",
        "pubsub",
        "start",
        f"--project={PROJECT_ID}",
        f"--host-port={PUBSUB_EMULATOR_HOST}",
    ]
    Popen(start, stdout=PIPE)

Note that, we are not using Popen.communicate() which blocks any further execution until feedback from the process has been received. It's a long running process therefore not a good candidate for synchronous feedback. We could have done more magic with e.g using nohup but we choose to omit code rather than to add code for the purpose of simplicity.

4. The yield keyword

Oh yes! The mighty fixture function yield keyword. Now, this is some voodoo magic right here. In a nutshell, everything before the yield keyword is the "setup" and everything after the yield keyword is the "teardown". So there it is. You now know how to make tests clean after themselves using a fixture function.

5. Teardown

At the time of going to press gcloud beta emulators pubsub doesn't have a stop command:

Available commands for gcloud beta emulators pubsub:

      env-init                *(BETA)*  Print the commands required to export
                              pubsub emulator's env variables.
      start                   *(BETA)*  Start a local pubsub emulator.

We will have to kill that process ourselves. One way to do that is with:

ps -ef | pgrep -f cloud-pubsub-emulator | xargs kill

Read about ps, pgrep, xargs and kill if you fancy. If you are looking for the windows equivalent you are totally on your own. Sorry.

See what the above bash command looks like in python. It's definitely not nice but it works.

exec_ps = Popen(["ps", "-ef"], stdout=PIPE)
exec_pgrep = Popen(
        ["pgrep", "-f", "cloud-pubsub-emulator"], stdin=exec_ps.stdout, stdout=PIPE
    )
Popen(["xargs", "kill"], stdin=exec_pgrep.stdout, stdout=PIPE)

You have no doubt worked out that stdout=PIPE is the equivalent of | and the way to chain the next command is by passing its predecessor's output as an input with stdin=previous_command.stdout.

That's it! Join me next time to see how to use the above fixture in a real Pub/Sub test.