Zenn
👌

DockerイメージをPythonでテストする

2022/10/17に公開
Docker
Python
Test
tech

はじめに

この記事ではDockerイメージをテストする方法を簡単に模索します。といってもtestinfraやory/dockettestのようなフレームワークを使うわけではなくPythonでテストコードを書いていきます。

Dockerイメージを用意する

テスト対象のDockerイメージを作成し、テストしたい動作を考え、テストコードの実装方針を決めます。

テスト対象のイメージ

私が作ったリポジトリhoratos/docker-imagesのcluttexイメージをテスト対象とします。cluttexイメージはtexlive/texliveをベースとしたイメージです。cluttex --watchを利用するためにfswatchを追加でインストールしてあります。

cluttexはLaTeX文書のコンパイルを自動で実行するコマンドラインツールです。LaTeX文書は参照の解決等のために複数回コンパイルする必要があります。cluttexは必要なコンパイル回数を自動で判定してコンパイルを実行します。

cluttexに--watchオプションを渡して起動することでコンパイル対象のファイルが更新された時に自動でコンパイルを実行します。cluttexはこの機能を実現するためにfswatchを利用しています。

テストしたい動作

cluttexイメージの動作のうちテストしたい動作を決めておきます。次の2つ機能のテストは外せません。

  1. cluttexを実行することでLaTeX文書がコンパイルされること。
  2. cluttex --watchを実行することでLaTeX文書のコンパイルが実行され、さらにLaTeX文書を変更することでもう一度コンパイルが実行されること。

後述のテストコードの実装では補助的なテストケースを作成しますが、主にテストしたい機能は上の2つです。
実行結果の確認方法としてcluttexが出力するログを確認する方法をとります。生成されるPDFファイルを確認することもできますが、今回はログを信じることにします。（なお生成されたPDFファイルを確認する場合、ファイルサイズの変化だけを確認するテストとPDFファイルの内容を確認するテストの2種類が考えられると思います。）

テストコードを実行する場所を検討する

テストコードを実行する場所を検討します。ざっと思いつく選択肢を書き出すと以下の3つが挙げられます。

  1. ホスト側
  2. テスト対象イメージのコンテナの中
  3. テスト対象イメージとは異なるイメージから作られたコンテナの中

どれを選ぶかを決めるためにpros/consを考えてみましょう。

pros

  1. ホスト側
    • 外から実行するためマウント位置を把握しやすい。
  2. テスト対象イメージのコンテナの中
    • docker execでテスト対象のプログラムと同じコンテナの中で動かせるため、操作がやりやすい。
  3. テスト対象イメージとは異なるイメージから作られたコンテナの中
    • テストに必要な処理系をテスト用イメージに分離できる。

cons

  1. ホスト側
    • テストコードの実行に必要な処理系をインストールする必要がある。
  2. テスト対象イメージのコンテナの中
    • テスト対象イメージにテスト用のプログラムが混ざってしまう。
  3. テスト対象イメージとは異なるイメージから作られたコンテナの中
    • Dockerの操作が複雑になる。

今回はテストコードを動かすための処理系をホスト側に入れたくないため選択肢1は取らないことにします。そしてテストコードとテスト対象を同じイメージに混在させたくないので選択肢2も取らないことにします。Dockerの操作が複雑になりますが選択肢3を取ることにします。

テストコード用のイメージを用意する

テストコードを動かすイメージを用意します。コンテナの中からホスト側のDockerを操作するためには以下の2つが必要になります。

  1. コンテナ内部で利用するDockerクライアント
  2. ホスト側のDockerデーモンと通信する手段

コンテナ内部で利用するDockerクライアントは、今回はPythonでテストコードを書くためDocker SDKを利用します。

テストの実行のためにdocker composeをタスクランナー的に使います。テストコードを実行するコンテナとテスト対象のコンテナをComposeファイルに記述します。cluttexイメージをcluttexサービスとして実行します。また、テスト実行用のイメージをtestsサービスとして実行します。

compose.yaml
services:
  cluttex:
    build:
      context: cluttex
  textlint-ja:
    build:
      context: textlint-ja
  tests:
    build:
      context: tests
    environment:
      HOST_PWD: $PWD
    volumes:
      - type: bind
        source: /var/run/docker.sock
        target: /var/run/docker.sock
      - type: bind
        source: tests/
        target: /home/tests/
      - type: bind # pylintにrcファイルを読み込ませるため
        source: tests/pylintrc
        target: /home/pylintrc
      - type: bind
        source: data/
        target: /home/data/
    working_dir: /home
    entrypoint: ["python3"]

ホスト側のDockerデーモンと通信する手段としてホスト側のDockerのソケットをテストコードを実行するコンテナでマウントする方法を用います。そのため、上記のComposeファイルで /var/run/docker.sock をバインドマウントしています。

テストコードを tests/ に配置しているため、このディレクトリを /home/tests にバインドマウントします。
テストデータは data/ に配置しています。このディレクトリは /home/data にバインドマウントします。

テストコードを動かすコンテナの内部からコンテナを実行する

テストコードを動かすコンテナの内部からテスト対象のイメージのコンテナを実行します。Docker SDKでコンテナを起動し、cluttexが実行できることとマウントポイントが正しいことを確認します。それから、cluttexでLaTeX文書をコンパイルできることを確認するテストコードを書きます。

Docker SDKを使ってコンテナを実行する

Docker SDK for Pythonを使ってコンテナを実行します。DockerClientオブジェクトを作成してからcontainers.runを呼び出すことで指定したイメージからコンテナを実行できます。

以下のコードはDockerClientオブジェクトを作成します。関数docker.from_envを呼び出すことで環境変数から必要な設定を読み込んだ上でDockerClientオブジェクトを作成します。

tests/test_cluttex.py
def client() -> docker.DockerClient:
    """Dockerのクライアントを作成するfixture"""
    return docker.from_env()

以下のコードの関数run_containerは指定されたイメージにコマンドを渡してコンテナを実行します。client.containers.runにdetach=Trueを渡すことでdocker run -dと同様にデタッチした状態でコンテナを実行できます。コンテナの実行の開始に成功すると、Containerオブジェクトを返します。

テストコードからの使いやすさを考えて、run_containerをコンテキストマネージャとして実装しています。with文から抜けた時にremoveを呼び出すことでコンテナを削除するようにしています。

tests/test_cluttex.py
@contextmanager
def run_container(client: docker.DockerClient,
                  image: str,
                  command: Union[list[str], None] = None,
                  **kwargs) -> docker.models.containers.Container:
    """
    Dockerイメージを実行してコンテナを返す。
    """
    container = client.containers.run(image, command, detach=True, **kwargs)
    try:
        yield container
    finally:
        container.remove()

cluttexが起動することを確認する

複雑なことをやる前に、まずは単純にコンテナを実行できることを確認しましょう。先ほど定義した関数run_containerにIMAGE_NAME=docker-images_cluttexを渡して呼び出します。Containerオブジェクトのwaitメソッドを呼び出して終了を待ちます。そしてlogsメソッドを呼び出してコンテナが出力したデータを取得します。このテストコードではlogsメソッドはバイナリデータを返すのでdecodeを呼び出すことでUTF-8文字列に変換しています。

cluttexは引数なしで呼び出すとヘルプメッセージを出力します。そのため、何らかの文字列を出力することをassert logs != ''でテストします。

また、cluttexは引数なしで呼び出された時には終了コードとして1を返します。そのことをassert result['StatusCode'] == 1で表明します。

tests/test_cluttex.py
def test_with_no_input(client: docker.DockerClient):
    """何も入力を渡さないときにステータス1で終了することを確認する"""
    with run_container(client, IMAGE_NAME) as container:
        result = container.wait()
        logs = container.logs().decode()

    assert logs != ''
    assert result['StatusCode'] == 1

マウントポイントが正しいことを確認する

テストデータのマウントポイントがずれている可能性を排除するためにマウントポイントが正しいことを確認するテストtest_lsを書きます。

マウントポイントはフィクスチャmount_data_tex_dirとして与えられます。これを関数run_containerに渡し、bashでls -aを実行してファイルの一覧を取得します。lsの出力は1行1ファイルのテキストで返ってくるのでsplitlinesで文字列のリストにします。

このテストでは hello.tex がマウントポイントに存在することを確かめます。このファイルは次に実装するテストケースでcluttexに入力として渡されることになるファイルです。

tests/test_cluttex.py
def test_ls(client: docker.DockerClient, mount_data_tex_dir: Mount):
    """lsコマンドを実行してマウント位置が正しいことを確認する"""
    with run_container(client,
                       IMAGE_NAME,
                       entrypoint="bash -c 'ls -a'",
                       mounts=[mount_data_tex_dir]) as container:
        result = container.wait()
        file_list = container.logs().decode().splitlines()

    assert 'hello.tex' in file_list
    assert result['StatusCode'] == 0

cluttexでLaTeX文書をコンパイルできることを確認する

いよいよテストしたい対象であったcluttexでLaTeX文書をコンパイルできること確認するテストを書きます。

一度だけコンパイルできることを確認するテストtest_compile_oneshotのソースコードを以下に示します。

関数run_containerに適切な引数を渡してcluttexを実行します。cluttexが出力したログをlogs = container.logs().decode()で文字列として取得します。

今回はログだけを見てうまく動いているかを確認します。cluttexはLaTeX文書をコンパイルしてPDFファイルを出力するとOutput written on <ファイル名>'という形式でログを出力するので、この文字列があるかどうかをassert`で表明します。

tests/test_cluttex.py
def test_compile_oneshot(client: docker.DockerClient,
                         mount_data_tex_dir: Mount):
    """一度だけコンパイルすることを確認する"""
    with run_container(client,
                       IMAGE_NAME, ["-e", "lualatex", "hello"],
                       mounts=[mount_data_tex_dir]) as container:
        result = container.wait()
        logs = container.logs().decode()

    assert 'Output written on hello.pdf' in logs
    assert result['StatusCode'] == 0

テスト対象のコンテナの実行と並行してログを確認する

ここまでで、単純にコンテナを実行して期待した結果を得られたことを確認するテストを書くことができました。次はコンテナを実行しながら並行してプログラムの出力を確認するテストを書いていきます。

このセクションでは非同期プログラミングが必要になるため、Docker SDKの代わりにaiodockerを使います。また、いくつかの補助的な関数を作成してテストコードを書きやすくします。

テストの流れを考える

テストしたい動作は「cluttex --watchを実行することでLaTeX文書のコンパイルが実行され、さらにLaTeX文書を変更することでもう一度コンパイルが実行されること」でした。まずはcluttex --watchを実行する流れを確認します。

  1. cluttex --watch -e lualatex helloを実行する。
  2. hello.texがコンパイルされPDFファイルhello.pdfが生成される。
  3. hello.texを別のLaTeXファイルで上書きする。
  4. hello.texがコンパイルされ新しいPDFファイルhello.pdfが生成される。

ここで、 2.と4.で生成されるhello.pdfの内容を区別できるように、前者では1ページのPDFファイルに、後者では2ページのPDFファイルになるようにhello.texの中身を調整しておきます。

テストコードのための補助的な関数を実装する

補助的な関数としてcluttex --watchを実行するコンテナを作成する関数とOutput writtenが含まれる行に出会ったらその行を返す非同期ジェネレータを書きます。

ログを確認しながらファイル操作を実行する必要があるため非同期プログラミングが必要になります。そのため、Dockerクライアントライブラリの非同期版としてaiodockerを利用します。aiodockerはDocker SDKに比べると抽象化がなされておらず、Docker Engine APIに近い使用感になっています。

関数run_cluttex_watchはcluttex --watch -e lualatex helloを実行する関数です。利便性のために非同期コンテクストマネージャとして実装します。コンテナを起動したらそのコンテナをyieldします。async withから抜けるときにコンテナの停止と削除を実行します。

tests/test_cluttex.py
@asynccontextmanager
async def run_cluttex_watch(
) -> AsyncIterator[aiodocker.docker.DockerContainer]:
    """
    `cluttex --watch -e lualatex hello`を実行するコンテナを起動する
    この場合は並行してログを見たいので非同期にしてある。
    あとでコンテナを削除するためにコンテクストマネージャとして実装している。
    """
    docker = aiodocker.Docker()
    config = {
        'Image': IMAGE_NAME,
        'Cmd': ['--watch', '-e', 'lualatex', 'hello'],
        'User': 'root',
        'HostConfig': {
            'Mounts': [{
                "Type": "bind",
                "Source": f'{HOST_PWD}/data/test_compile_watch',
                "Target": "/home/cluttex",
                "Mode": "",
                "RW": True,
                "Propagation": "rprivate",
            }]
        }
    }
    container = await docker.containers.run(config)
    try:
        yield container
    finally:
        await container.kill()
        await container.delete()
        await docker.close()

関数output_log_waiterはコンテナを受け取ってログにOutput written on hello.pdfで始まる行を見つけたときにその行をyieldする非同期ジェネレータです。[ERROR]を含む行を見つけたときには例外を送出してジェネレータを終了します。

tests/test_cluttex.py
async def output_log_waiter(
        container: aiodocker.docker.DockerContainer) -> AsyncIterator[str]:
    """
    Output writtenのログ行を見つけたらその行をgenerateするasync generator
    [ERROR]を見つけた場合はその場でストップする
    """
    log = container.log(stdout=True, stderr=True, follow=True)
    async for line in log:
        print(line.encode())
        if line.startswith('Output written on hello.pdf'):
            yield line
        if '[ERROR]' in line:
            raise StopAsyncIteration(line)

テストケースを実装する

関数test_compile_watchにテスト手順を実装していきます。

まずはテストを実行するためのディレクトリ /home/data/test_compile_watch を作成し、その中に hello.tex をコピーします。ディレクトリ /home/data/test_compile_watch は関数run_cluttex_watchでコンテナを実行する際にマウントされます。

tests/test_cluttex.py
    data_root = Path("/home/data")
    tex_dir = data_root / 'tex'
    tmp_dir = data_root / 'test_compile_watch'

    hello_tex = tex_dir / 'hello.tex'

    assert tex_dir.exists()
    assert hello_tex.exists()

    tmp_dir.mkdir(exist_ok=True)

    shutil.copyfile(hello_tex, tmp_dir / 'hello.tex')

次に、コンテナを起動してテスト手順を実行していきます。まずは hello.tex のコンパイルが完了されるまで待ちます。このとき、なんらかの原因でコンパイルが進まなくなってしまったときにテストが終了しないことを避けるため、 asyncio.wait_for でタイムアウトするようにしておきます。

最初のコンパイルが終わった後に、shutil.copyfileを呼び出して hello.tex を更新します。その後、先ほどと同様にcluttexがコンパイルを終了するまで待ちます。

最後にログに出力された生成されたPDFファイルのページ数をassertで確認します。

tests/test_cluttex.py
    async with run_cluttex_watch() as container:
        waiter = output_log_waiter(container)

        # hello.texは1ページからなるPDFを出力する。
        line1 = await asyncio.wait_for(waiter.__anext__(), timeout=60)

        shutil.copyfile(tex_dir / 'hello2.tex', tmp_dir / 'hello.tex')

        # hello2.texは2ページからなるPDFを出力する。
        line2 = await asyncio.wait_for(waiter.__anext__(), timeout=60)

    assert '1 page' in line1
    assert '2 pages' in line2

おわりに

今回はログを確認するだけのテストケースでしたがそれなりに骨の折れるテストコードを書くことになったと感じています。テストフレームワークを使わない中での非同期なテストコードの実装は手間がかかります。

この記事では最初は同期的なプログラムを書きましたが、さまざまな状況に対応するためには非同期プログラミングが必須になると思います。Dockerコンテナに限らず、テスト対象のプログラムが状態を持っている場合や、複数のプログラム間の相互作用がある場合には、非同期プログラミングによる自動テストを書く必要があるように思います。

Discussion