Testing a Click app with streaming input#

For sqlite-utils#364 I needed to write a test for a Click app which dealt with input streamed to standard input. I needed to run some assertions during that process, which ruled out the usual CliRunner.invoke() testing tool since that works by running the command until completion.

I decided to use subprocess to run the application. Here’s the pattern I came up with for the test:

1
def test_insert_streaming_batch_size_1(db_path):
2
    # https://github.com/simonw/sqlite-utils/issues/364
3
    # Streaming with --batch-size 1 should commit on each record
4
    # Can't use CliRunner().invoke() here bacuse we need to
5
    # run assertions in between writing to process stdin
6
    proc = subprocess.Popen(
7
        [
8
            sys.executable,
9
            "-m",
10
            "sqlite_utils",
11
            "insert",
12
            db_path,
13
            "rows",
14
            "-",
15
            "--nl",
16
            "--batch-size",
17
            "1",
18
        ],
19
        stdin=subprocess.PIPE
20
    )
21
    proc.stdin.write(b'{"name": "Azi"}\n')
22
    proc.stdin.flush()
23
    # Without this delay the data wasn't yet visible
24
    time.sleep(0.2)
25
    assert list(Database(db_path)["rows"].rows) == [{"name": "Azi"}]
26
    proc.stdin.write(b'{"name": "Suna"}\n')
27
    proc.stdin.flush()
28
    time.sleep(0.2)
29
    assert list(Database(db_path)["rows"].rows) == [{"name": "Azi"}, {"name": "Suna"}]
30
    proc.stdin.close()
31
    proc.wait()
32
    assert proc.returncode == 0

The first trick I’m using here is running sys.executable to start a Python process. This ensures I run the Python that is available in the current virtual environment.

I modified my sqlite-utils command such that it could also be run using python -m sqlite_utils - see sqlite-utils#368 for details.

Setting stdin=subprocess.PIPE allows me to write data to the process’s standard input using proc.stdin.write().

I realized I needed to call proc.stdin.flush() after each write to ensure the write was pushed to the process in a predictable manner.

At the end of the test, running proc.stdin.close() is equivalent to sending an end-of-file, then proc.wait() ensures the process has finished and terminated.

Newsletter

Testing a Click app with streaming input#

Newsletter