当社TORICOのサービスにおける、自動テストの構成・取り組みを紹介します。
自動テストの実行環境
当社では、社内サーバーに構築した Jenkins での実行が多くを占めています。
Jenkins 以外では、Github actions, Magic Pod, Travis CI なども少量ありますが、多くは Jenkins です。 社内サーバーで起動しているため、コストやリソースの制限を気にせずにテストメソッドを起動できます。
Jenkins には、ユニットテストやE2E テストが50個ほど登録されており、コードのプッシュ時、もしくはタイマーで自動的に、テストを実行しています。
実行プログラムのビルドを行う場合はあまりなく、「テストコードを実行し、失敗した場合に通知する」という使い方が主となります。
自動テストのコードは、作っても自動実行させる環境が無いと有効に機能しません。 アプリ全体の自動テストの実行が手動での実行しか方法が無い場合、失敗した時の実行者のストレスが多くどこで失敗したのかの追跡も難しいため、常に自動的に実行し続ける環境の整備が必要です。
Jenkins内でのテストの起動方法
各サービスのソースコードを Github からプルし、同時に各サービスの検証環境のDockerイメージを AWS ECR からプルし、ソースコードを Docker イメージにマウントさせて自動テストを行っています。すべて Docker で実行します。
Jenkins 自体も Kubernetes 上で Pod として起動していますが、ホストノードの Docker のドメインソケットをマウントしており、ホストノードの docker 実行環境をそのまま操作する形となっています。
Jenkins の Kubernetes マニフェスト
Jenkins は Kubernetes 上で起動しています。参考までに、文末に Dockerfile と Kubernetes のマニフェストファイルを参考までに書きます。
作ったのは少し古く、今はもう公開されていないベースイメージを使っていたため、イメージのタグ等は伏せたものとなります。
Dockerfile, Kubernetes マニフェスト
ユニットテスト
ユニットテストには、書いたコードが仕様通り動いているかの動作保証をする他に、依存関係や実行環境のアップデートがあった場合、変わらずに動作し続けるかを保証することができます。
現在は10年前と比べて、苛烈とも言える勢いで依存環境がアップデートされていきます。乗り遅れずに、事故を少なくアップデートしていくにあたり、ユニットテストは必須です。
Django のユニットテスト
当社のサービスはウェブフレームーワークとして Django を多く使っています。 Django にはテスト環境がビルトインされていますので、それを使います。
Writing and running tests | Django documentation | Django
使用ライブラリ
ほぼすべてのサービスの開発環境に含まれるライブラリを紹介します。
freezegun
時間に関わるテストの際に使います。
https://github.com/spulec/freezegun
factory-boy + Faker
https://factoryboy.readthedocs.io/en/stable/
検証用のモデルデータの作成が非常に便利になります。
Django のモデルを作った際は、対応する DjangoModelFactory を作り、テストの実行を容易にします。
tenacity
https://github.com/jd/tenacity
リトライに使います。
ユニットテストという範疇からは超えるかもしれませんが、テストコード内で外部との通信を行う際はリトライを考慮する必要があるため、 tenacity を使ってリトライを行います。
テストコードで行うか、内部のロジックで行う必要があるのかは場合によると思いますが、まれに発生する通信障害で自動テストが失敗しアラートを出さないためにリトライを行うようにします。
カバレッジ計測
現在、多くのプロダクトでカバレッジの計測はしていません。
データベースエンジンにオンメモリの SQLite3 を使う
Django のユニットテストは通常、テスト関数の実行時にデータベースのスキーマを作成し、DDLを実行(スキーママイグレーション)し、終了時に消します。 そのまま使うと、マイグレーションファイルの量だけ時間がかかってしまい、非効率ですので、順次のマイグレーションを行わず、かつデータベースエンジンはオンメモリの SQLite を使うようにしています。
Setting では下記のようになります。
myapp/settings/unittest.py
from .base import * # NOQA
class DisableMigrations(object):
def __contains__(self, item):
return True
def __getitem__(self, item):
return None
DATABASES = {
'default': {
'ENGINE': 'django.db.backends.sqlite3',
'NAME': ':memory:',
},
}
MIGRATION_MODULES = DisableMigrations()
DATABASE_ROUTERS = []
生SQL の SQLite3 でのユニットテスト
サービス内では、まれに生SQL を実行する箇所があります。
本番サービスのデータベースエンジンは SQLite3 ではなく MySQL のため、生SQL を SQLite3 で実行する時にエラーが発生することがあります。
簡易的な文字列置換で MySQL から SQLite3 への変換が行える場合は、変換をして SQLite3 でユニットテストを実行させます。 変換コードを付録に記載します。
SQL を MySQL から SQLite3 に変換するコード
Python の doctest
Python にはコード内のドキュメンテーションにテストコードを書く機能「doctest」があります。
doctest --- 対話的な実行例をテストする — Python 3.12.1 ドキュメント
シンプルな割に非常に便利で強力で、Python のキラー機能の一つといえます。私は大好きです。
関数やクラスのドキュメントに doctest が書いてあると、コードの使い方もすぐわかりますし、 その doctest が常に Jenkins で実行され、結果にエラーが無いことが保証されるため、安心感もあります。
そして、エディターによっては右クリックから一発でテストを実行できるようになっていますので、開発時も速度・精度の向上に繋がります。
ただし、DBを使うようなステートフルな処理に doctest を使うのは不向きです。
コンパクトでステートレスで、かつ読みにくい処理に使うと最適です。
例えば、正規表現で文字列をマッチさせるような関数や、文字コードやマルチバイト文字に関する関数、暗号化・復号化処理には doctest が必須といえます。
def calc_transition_timing_count(
transition_duration_ms: int,
transition_fps: int
) -> int:
"""
transition_duration_ms, transition_fps を元に、
transition_timing_count を求める
>>> calc_transition_timing_count(2000, 10)
20
>>> calc_transition_timing_count(500, 10)
5
>>> calc_transition_timing_count(1000, 20)
20
"""
return int(transition_fps * (transition_duration_ms / 1000.0))
re_jp_phonenumbers = [
re.compile(r'^(0\d{2})[- ](\d{4})[- ](\d{4})$'),
re.compile(r'^(0\d{1,5}?)[- ](\d{1,4})[- ](\d{4})$'),
re.compile(r'^(0\d{1,2}?)(\d{4})(\d{4})$'),
]
def format_jp_phonenumber(value: str) -> str | None:
"""
>>> v = format_jp_phonenumber
>>> v('080-0000-1111')
'080-0000-1111'
>>> v('180-0000-1111')
>>> v('080-0000')
>>> v('08000001111')
'080-0000-1111'
>>> v('080 0000 1111')
'080-0000-1111'
>>> v('08O 0000 1111')
>>> v('0300001111')
'03-0000-1111'
>>> v('0250001111')
'02-5000-1111'
"""
for rpn in re_jp_phonenumbers:
if r := rpn.match(value)
return '{}-{}-{}'.format(r.group(1), r.group(2), r.group(3))
Python の Unittest
Django に依存しないライブラリを開発する際は、Python の Unittest フレームワークを使ったユニットテストを書きます。
unittest --- ユニットテストフレームワーク — Python 3.12.1 ドキュメント
当社では、API通信を行うクライアントライブラリが多くあります。API通信はモックを使ってテストを書く場合もありますが、モックを使わず実際に対向環境と通信をするユニットテストも多いです。 実際に通信を行うユニットテストが整備されていれば、通信先の破壊的アップデートによるトラブル時の原因の究明が素早く行えますので便利です。
E2E テスト
E2E (エンドツーエンド) テストは、ブラウザを操作してアプリの一連の動作を検証するテストです。
当社では、フロントンドフレームワークのユニットテストはあまり充実していませんが、システムの一連の動作を検証する E2E テストは整備しており、プログラムの反映時もしくは定期的に実行しています。
テストクライアント
RobobrowserDash
JavaScript を考慮しなくてよい環境であれば、ウェブブラウザは起動せずに単純な HTTP クライアントを使います。
Pythonでよく使われる HTTP クライアントである requests をラップし、HTMLパーサーの BeautifulSoup と組み合わせて使うソリューションに 「Robobrowser 」というものがあり、簡単なテストであれば十分に機能します。
ただし、Robobrowser は2015年のアップデート以降更新が止まっており、最近の依存関係セットの中では動かすことはできません。 そのため、Robobrowser をフォークして Roborowserdash というパッケージ名として PyPI に登録し、使っています。
JS を動かさないため、安定したテストが行なえます。
ウェブブラウザを起動するものよりリソース消費が少ないため、より攻撃的な検証に使うこともあります。
例えば、残り1つの商品を100人で同時に購入を行い、排他制御とエラー処理が適切に行われているかのテストをする場合は、Robobrowserdash を使って検証を行います。
Selenium
Selenium と Python のテストフレームワークを用いた E2E テストはいくつかあります。
ただ、Selenium は、E2Eテストとして使うことは今後はあまり無いと思っています。
まれに、 Selenium の Python ラッパーである Selene を使っているものもあります。
Puppeteer, Pyppeteer
比較的最近は、Chrome を操作する node.js のライブラリの Puppetter やその Python 版の Pyppeteer を使ってE2Eテストが書かれています。
ブラウザからの E2E テストを安定して行うにあたり、HTML のエレメントに data-annotate="xxx" という E2E テスト専用の属性を設定し、ブラウザはそのエレメントに対して QuerySelector を行って自動操作するような形式としています。
Pyautogui
印刷時のダイアログ等、ブラウザウインドウ範囲外のものを自動操作する場合は、Puppeteer では行えないため、 pyautogui を使います。
関連記事: pyautogui を使ってみた #Python - Qiita
セキュリティーテスト
製品に対してのセキュリティー面のテストは、開発エンジニアではなく専門の品質管理部門が行います。
AppScan
アプリケーションに対してのセキュリティーテストは、主に HCL Software の AppScan を使います。
基本的に自動的に脆弱性をスキャンするソフトウェアですが、APIのエンドポイントやパラメーターを指定して重点的な検証も行えます。
HCL AppScan Dynamic Application Security Testing (DAST) - HCLSoftware
OWASP ZAP
AppScan と併用し、OWASP ZAP も使います。こちらも AppScan と同様に、自動操作型の脆弱性スキャナーです。
AppScan では使用可能ライセンス数に限りがあるため、ライセンスを持っていない領域では OWASP ZAP を使っています。
手動テスト
そのほか、セキュリティーのテストに関しては Burp suite や Postman を使った手動の検証が多くを占めますが、今回の記事の範囲外ではないため割愛します。
付録
Jenkins の Dockerfile
昔作ったイメージを今でも使っており、今はもう存在しないベースイメージなどありますのでイメージタグは伏せています。参考程度にご覧ください。
FROM alpine:** as builder
RUN apk --no-cache add \
gcc \
make \
python3-dev \
musl-dev \
libxml2-dev \
libxslt-dev \
libffi-dev \
libressl-dev
RUN pip3 install docker-compose
FROM jenkins/jenkins:lts-alpine
USER root
COPY --from=builder /usr/lib/python**/site-packages/ /usr/lib/python**/site-packages/
COPY --from=builder /usr/bin/docker-compose /usr/bin/docker-compose
RUN apk --no-cache add \
python3
ENV DOCKER_VERSION **
RUN curl -fL -o docker.tgz "https://download.docker.com/linux/static/test/x86_64/docker-$DOCKER_VERSION.tgz" && \
tar --strip-components=1 -xvzf docker.tgz -C /usr/bin
RUN apk --update add tzdata && \
cp /usr/share/zoneinfo/Asia/Tokyo /etc/localtime && \
apk del tzdata && \
rm -rf /var/cache/apk/*
ENV JENKINS_REF /usr/share/jenkins/ref
# install jenkins plugins
COPY ./plugins.txt $JENKINS_REF/
RUN /usr/local/bin/install-plugins.sh < $JENKINS_REF/plugins.txt
# add docker group
RUN addgroup -S docker && adduser jenkins docker
VOLUME /var/jenkins
Kubernetes のマニフェスト
deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
namespace: torico
name: jenkins
spec:
selector:
matchLabels:
app: jenkins
strategy:
type: Recreate
template:
metadata:
labels:
app: jenkins
spec:
containers:
- image: **/torico/jenkins:latest
name: jenkins
env:
- name: JAVA_OPTS
value: "-Duser.timezone=Asia/Tokyo"
ports:
- containerPort: 8080
name: http-port
- containerPort: 50000
name: api-port
volumeMounts:
- name: home
mountPath: /var/jenkins_home
- name: ssh-keys
mountPath: /root/.ssh
- name: docker-socket
mountPath: /var/run/docker.sock
- name: docker-credentials
mountPath: /root/.docker
imagePullSecrets:
- name: ecr-credeintial
volumes:
- name: home
hostPath:
path: /data/jenkins
- name: ssh-keys
hostPath:
path: /data/jenkins/ssh
- name: docker-socket
hostPath:
path: /var/run/docker.sock
- name: docker-credentials
hostPath:
path: /home/ubuntu/.docker
service.yaml
apiVersion: v1
kind: Service
metadata:
name: jenkins-service
namespace: torico
spec:
type: NodePort
ports:
- port: 8080
protocol: TCP
targetPort: 8080
name: jenkins-http
selector:
app: jenkins
ingress.yaml
apiVersion: networking.k8s.io/v1
kind: Ingress
metadata:
name: jenkins-ingress
namespace: torico
spec:
tls:
- hosts:
- jenkins.torico-tokyo.com
secretName: tls-certificate
rules:
- host: jenkins.torico-tokyo.com
http:
paths:
- path: /
pathType: Prefix
backend:
service:
name: jenkins-service
port:
number: 8080
ユニットテスト時、MySQL の生 SQL を SQLite3 へ変換するコード
def db_engine_is_mysql():
"""
DBエンジンが MySQLか
通常、エンジンは MySQL だが、ユニットテストで SQLite3 を使うことがある。
SQLite3 の場合、特定の raw_query は動作しないので、それを回避するためのコード。
SQLite3を使う時、この分岐の中は充分なユニットテストはできない
"""
return settings.DATABASES['default']['ENGINE'].endswith('.mysql')
def db_engine_is_sqlite3():
"""
DBエンジンが SQLite3 か
テストの時用
"""
return settings.DATABASES['default']['ENGINE'].endswith('.sqlite3')
def _replace_sql_named_placeholder(sql: str) -> str:
"""
名前付きプレースホルダーを MySQL形式 -> SQLite形式に置換する
- MySQL形式: `SELECT * FROM user WHERE email = %(email)s`
- SQLite形式: `SELECT * FROM user WHERE email = :email`
>>> _replace_sql_named_placeholder("customer_id = %(customer_id)s")
"customer_id = :customer_id"
>>> _replace_sql_named_placeholder(
... "AND (email = %(email_1)s OR email = %(email_2)s)")
"AND (email = :email_1 OR email = :email_2)"
"""
pattern = re.compile(r'%\(([a-zA-Z0-9_]+)\)s')
return pattern.sub(r":\1", sql)
GROUP_CONCAT = 'GROUP_CONCAT' # GROUP_CONCATは置換対象にしない
def _replace_sql_concat(sql: str) -> str:
"""
CONCAT を || に、簡易的に置換する
>>> _replace_sql_concat("A = CONCAT(%s, 'hoge') AS b")
"A = %s || 'hoge' AS b"
>>> _replace_sql_concat(
... "CONCAT('ebook-continuation-', t1.asp_title_code) AS ()")
"'ebook-continuation-' || t1.asp_title_code AS ()"
>>> _replace_sql_concat(
... "GROUP_CONCAT('ebook-continuation', 'a') as ()")
"GROUP_CONCAT('ebook-continuation', 'a') as ()"
"""
splited_texts = []
re_concat = re.compile(r'CONCAT\(([^)]+),([^)]+)\)')
for s in sql.split(GROUP_CONCAT):
splited_texts.append(re_concat.sub(r'\1 ||\2', s))
return GROUP_CONCAT.join(splited_texts)
def _replace_sql_concat_separator(sql: str) -> str:
"""
GROUP_CONCAT(category_id SEPARATOR ',')
↓
GROUP_CONCAT(category_id, ',')
>>> _replace_sql_concat_separator("(category_id SEPARATOR ',')")
"(category_id, ',')"
>>> _replace_sql_concat_separator(
... "GROUP_CONCAT(category_id order by sort_key)")
'GROUP_CONCAT(category_id)'
"""
sql = sql.replace('group_concat', 'GROUP_CONCAT')
sql = re.sub(
'GROUP_CONCAT\((.*) order by [^)]+\)', # NOQA
r'GROUP_CONCAT(\1)', sql)
sql = sql.replace(' SEPARATOR', ',').replace(' separator', ',')
return sql
def _replace_sql_now(sql: str) -> str:
"""
NOW() を DATE('now') に変換する
>>> _replace_sql_now("AND NOW() <= c.coupon_end_date")
"AND DATE('now') <= c.coupon_end_date"
"""
return sql.replace('NOW()', "DATE('now')")
def _replace_sql_subdate(sql: str) -> str:
"""
SUBDATE(NOW() を DATE('now' +第2引数に変換
subdate で INTERVAL が正の値のみサポート。必要に応じて拡張する
必ず _replace_sql_now より先に行う
>>> _replace_sql_subdate(
... "d < SUBDATE(NOW(), INTERVAL 8 DAY) /* コメ */")
"d < DATE('now', '-8 DAY') /* コメ */"
>>> _replace_sql_subdate(
... "d < SUBDATE('2021-05-01', INTERVAL 8 DAY) /* コメ */")
"d < DATE('2021-05-01', '-8 DAY') /* コメ */"
>>> _replace_sql_subdate(
... "d < SUBDATE(%s, INTERVAL 8 DAY) /* コメ */")
"d < DATE(%s, '-8 DAY') /* コメ */"
>>> _replace_sql_subdate(
... "d < SUBDATE(:placeholder, INTERVAL 8 DAY) /* コメ */")
"d < DATE(:placeholder, '-8 DAY') /* コメ */"
"""
re_subdate = re.compile(r'SUBDATE\(NOW\(\), INTERVAL (\d+) (\w+)\)')
sql = re_subdate.sub(r"DATE('now', '-\1 \2')", sql)
re_subdate = re.compile(r'SUBDATE\(([^,]+), INTERVAL (\d+) (\w+)\)')
sql = re_subdate.sub(r"DATE(\1, '-\2 \3')", sql)
return sql
def _replace_sql_force_index(sql: str) -> str:
"""
FORCE INDEX (index_name_01) をINDEXED BY index_name_01 に変換する
>>> _replace_sql_force_index('FORCE INDEX (index_name_01) ')
'INDEXED BY index_name_01 '
"""
re_index_name = re.compile(r'FORCE INDEX \(([^)]+)\)')
sql = re_index_name.sub(r'INDEXED BY \1', sql)
return sql
def _replace_sql_word_boundary(sql: str) -> str:
"""
MySQLの正規表現のワード境界をSQLite3の正規表現に変換する
>>> _replace_sql_word_boundary("[[:<:]]123[[:>:]]")
"\\b123\\b"
>>> _replace_sql_word_boundary("[[:<:]]123あいうえお[[:>:]]")
"\\b123あいうえお\\b"
>>> _replace_sql_word_boundary("[[:<:]]123あいうえおABC[[:>:]]")
"\\b123あいうえおABC\\b"
"""
re_str = re.compile(r'\[\[:<:]](.*)\[\[:>:]]')
return re_str.sub(r'\\b\1\\b', sql)
def replace_sql_mysql_to_sqlite(sql: str) -> str:
"""
SQL文を MySQL から SQLite にできるだけ置換する。
ユニットテスト用。完全な置換はできない。
"""
sql = _replace_sql_named_placeholder(sql)
sql = _replace_sql_concat(sql)
sql = _replace_sql_concat_separator(sql)
sql = _replace_sql_subdate(sql)
sql = _replace_sql_now(sql)
sql = _replace_sql_force_index(sql)
sql = _replace_sql_word_boundary(sql)
return sql
def replace_sql_mysql_to_sqlite_if_sqlite_test(sql: str) -> str:
"""
もし、DB エンジンが SQLite3 なら、SQL を SQLite3 用に置換する。
ユニットテスト用。非常に簡易的なものなので、
置換できないような複雑なSQLをユニットテストで使う場合は
@skipIf(not db_engine_is_mysql(), 'MySQL only.') でスキップする
"""
if db_engine_is_mysql():
return sql
return replace_sql_mysql_to_sqlite(sql)
