こんにちは。 Nintendo Switch 2 の抽選予選に無事コマを進め続けている、SREチームの高 (@nogtk) です。早く当選してこの大会からドロップアウトしたいものです。
今回は、構造化ログのユニットテストを書いたときのプチハマりと、それを解決した方法についてご紹介したいと思います。
なぜログのテストコードを書いたか
前回のブログで、Django アプリケーションが出力するログを構造化したお話をご紹介致しました。
ログを構造化するにあたって、出力されるログのエントリを定義しそれに沿ってログが出力されるような実装を追加しています。その中には Datadog APM への連携するためのトレース・スパン情報の付加したり、structlog の contextvars というモジュールを使ってコンテキストをログエントリに含めたりなど、徐々に込み入った実装が増えてきました。 基本的には関連するライブラリが提供している方法に準拠して実装を行っているだけではありますが、とはいえ出力結果が期待値通りであることは確認したいところです。
加えて、ログ出力がデグレを起こしていることを人間が発見するのは非常に難しいです。人間の目はJSONを良い感じにパースしてくれませんし、各エントリの差分やメッセージフォーマットの変化に気づくのは容易ではありません。
そこで、構造化ログの出力をテストするユニットテストを記述してみることにしました。
前提
今回のテストコードは Python の unittest を使って記述しています。 また構造化ロギングするにあたってのロガーの設定は Python の DictConfig 形式を利用していて、formatter / handler を以下のように設定しています。
{ 'formatters': { 'json': { '()': 'structlog.stdlib.ProcessorFormatter', "processor": structlog.processors.JSONRenderer(ensure_ascii=False), }, }, 'handlers': { 'json': { 'class': 'logging.StreamHandler', 'formatter': 'json', }, }, 'loggers': { 'struct_log_logger': { 'handlers': ['json'], 'level': 'INFO', 'propagate': True, }, }, }
assertLogs を使ったテストコード
今回は実際のログ出力をテスト対象にしたいので、例えば以下のようにモックを使ったログ呼び出しに対する引数のチェックでは不十分です。 呼び出されたことではなく、呼び出された結果、意図するフォーマットでログが出力されていることを担保したいためです。
def test_logger_info_called_with_hoge(self): # Arrange _logger = mock.Mock() # Act _logger.info("hoge") # Assert _logger.info.assert_called_once_with("hoge")
ログ出力まで含めてテストコードを書くには、unittest の assertLogs を使うのが王道でしょう。 以下のように、with 文を使用してコンテキストマネージャーとして assertLogs を実行し、ログの出力を捕捉します。assertLogs は指定されたロガーのログメッセージをキャプチャし、cm (context manager) オブジェクトを通じてアクセス可能にします。
def test_non_struct_log_info(self): # Act logger = logging.getLogger('non_struct_log_logger') with self.assertLogs('non_struct_log_logger', level='INFO') as cm: logger.info('hoge') # Assert self.assertIn('hoge', cm.output[0])
これに倣って、構造化ログについても assertLogs を使って検査してみます。 以下のようなテストコードですね。パッと見よさげです。
def test_struct_logger_json_output(self): # Act logger = logging.getLogger('struct_log_logger') with self.assertLogs('struct_log_logger', level='INFO') as cm: logger.info('hoge') # Arrange log_data = json.loads(cm.output[0]) self.assertEqual(log_data['message'], 'hoge') self.assertEqual(log_data['level'], 'info') self.assertEqual(log_data['logger'], 'struct_log_logger')
しかし、このテストコードを実行すると json.loads の部分で JSON パースに失敗します。
補足している cm.output[0] を出力させると、
INFO:struct_log_logger:hoge
という出力が得られ、構造化ログが出力されていません。
当然、同じロガーを使って python shell からログを発行すると構造化されたログが出力されます。
>>> logger = logging.getLogger('struct_log_logger') >>> logger.info("hoge") {"level": "info", "logger": "struct_log_logger", "timestamp": "2025-06-14T09:47:03.021630+09:00", "message": "hoge"}
assertLogs はログをどう捕捉しているのか
assertLogs の実装を確認しに行きます。
この _AssertLogsContext クラスで補足をしています。
コンテキストマネージャーのエントリで呼ばれる __enter__ メソッドの実装を一部抜粋します。
def __enter__(self): formatter = logging.Formatter(self.LOGGING_FORMAT) handler = _CapturingHandler() handler.setLevel(self.level) handler.setFormatter(formatter) ... logger.handlers = [handler] ... return handler.watcher
ログの formatter および handler を上書きしていますね。 おさらいですが、構造化ロギングの設定は、formatter / handler を JSON 出力するように変更することで実現しています。 それらの設定が assertLogs によって上書きされてしまったことで、結果構造化されたログを出力することができず、テストに失敗していました。
解決策: ログストリームを StringIO に差し替える
assertLogs が使えないことがわかったので、別のアプローチとして今回はログストリームを StringIO に差し替えてテストを行うようにしました。
setUp で構造化ロギング用の handler の stream を StringIO に差し替えます。
def setUp(self): # ログ出力をテストできるように出力先を StringIO に変更 logging_config = settings.LOGGING.copy() self.log_output = StringIO() logging_config['handlers']['json']['stream'] = self.log_output # ログ設定を上書き logging.config.dictConfig(logging_config)
このように差し替えることで当初の想定通り、JSON として出力されたログエントリに対する検査が記述できるようになりました。
def test_basic_log_format(self): # Act logging.getLogger('struct_log_logger').info("test message") # JSON パース log_output = self.log_output.getvalue().strip() log_data = [json.loads(line) for line in log_output.split('\n')] # Assert self.assertEqual(log_data.get('level'), 'info') self.assertEqual(log_data.get('logger'), 'struct_log_logger') self.assertEqual(log_data.get('message'), 'test message')
他の例として、冒頭で紹介した structlog の contextvars を使ったログエントリの追加についても同様にテストを記述することが出来ます。いい感じです。
def test_contextvars(self): # Arrange structlog.contextvars.bind_contextvars( additional_key='additional_value', ) # Act self.logger.info("test message") # JSON パース log_output = self.log_output.getvalue().strip() log_data = [json.loads(line) for line in log_output.split('\n')] # Assert self.assertEqual(log_data.get('additional_key'), 'additional_value')
終わりに
今回は構造化ログのテストで遭遇した問題と解決策について紹介しました。 assertLogs の動作を理解するために Python の標準ライブラリのソースコードを読み、単にAPIを使うだけでは気づけない制約を理解することが出来ました。
かなりニッチな内容だったかと思いますが、未来の誰かの一助になれば幸いです！
