【助けてください】Pythonでmainクラスの標準出力に関する単体テスト

Question

Python初心者です。回答得られず困ってます。

Pythonにて、単体テストを記載しておりまして、以下のようなmainクラスの単体テストを書いております。
Pythonのバージョンは3.8となります。

テストの期待する結果としては、タイムアウトが発生した時に正しくタイムアウトエラーが
発生することを目的としてます。

メソッド自体の働きとしてmainメソッドでhoge.fuga()を実施します。
その際にタイムアウトがfugaで発生したらexcept asyncio.TimeoutError:に飛ばすようにしております。
ただ、タイムアウトが発生したあともタスクは実行するという形にしております。
こちらのソースが実際に動くことは検証済です。

• 実処理を行うソースコード

import logging
import traceback
import asyncio
import os
from hoge import fuga
from get_file import get_file
from datetime import datetime
from dotenv import load_dotenv

async def main():
    logging.basicConfig(level=logging.INFO)
    logger = logging.getLogger(__name__)
    load_dotenv(override=True)
    logger.info(f"start {datetime.now()}")
    task = asyncio.create_task(fuga(get_file()))
    try:
        await asyncio.wait_for(asyncio.shield(task), timeout=60)

    except asyncio.TimeoutError:
        logger.warning(
            f"WARNING\nTIMEOUT")
    except Exception as e:
        logger.critical(f"ERROR\n{traceback.format_exc()}")
    # タイムアウト後も実行されているので、終了を待機する
    await task
    logger.info(f"end {datetime.now()}")

if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    loop.run_until_complete(main())

現状のテストコードは以下のように記載してますが、全然うまく書けてない状態です。
元コードのmain()で呼ばれるhoge.fuga()での処理をmockとしてスリープを実施する_mock_fuga()を用意しております。

• テストコード

from test.support import captured_stdout
from testfixtures import LogCapture
import logging
import os
import sys
import time


testdir = os.path.dirname(__file__)
srcdir = '../src'
sys.path.insert(0, os.path.abspath(os.path.join(testdir, srcdir)))

import unittest
from src.main import main
from unittest import mock

class TestMain(unittest.TestCase):
    def _mock_fuga(self):
        """fugaの処理のモック"""
        time.sleep(15)
        pass

    @mock.patch("src.hoge.fuga", new=_mock_fuga)
    def test_delayed_time(self):
        with LogCapture() as l:
            main()
            # ログ出力の文字列として`WARNING`と`TIMEOUT`という文字列があるかのチェック
            l.check_present(
                ("root", "WARNING", "WARNING")
            )
            l.check_present(
                ("root", "WARNING", "TIMEOUT")
            )

以上、ちょっと言葉足らずのところがあるかもしれませんが、心優しいご回答お待ちしております。
宜しくお願い致します。

Answer 1

非同期処理のテストを同期処理のコードでテストしようとしてますが、
python 3.8 以降であれば、unittest.IsolatedAsyncioTestCase が使えます。

import asyncio
import unittest


async def fuga(num):
    await asyncio.sleep(num)


async def main(num=10):
    loop = asyncio.get_running_loop()
    task = loop.create_task(fuga(num))

    await asyncio.wait_for(
        asyncio.shield(task),
        timeout=20)

    return 1


class AsyncTestCase(unittest.IsolatedAsyncioTestCase):

    async def test_main(self):
        self.assertEqual(1, await main(10))

    async def test_main_timeout(self):
        with self.assertRaises(asyncio.TimeoutError):
            await main(21)


if __name__ == '__main__':
    unittest.main()

---

現状のコードをテストしようとする前に、
対象のコードをテストしやすい設計にしましょう。

• time.sleep はイベントループを止めるブロッキング処理

　全体の動作が止まるので、asyncio版のスリープを使う。

• 単体テスト内でlogging.basicConfig のような

　グローバルに影響のある操作は行わない

• ログ出力をテスト対象とするのはお勧めしません。

　ログ出力は柔軟にカスタマイズできるのですが、
テストを通す為に、ログ出力を気軽に弄れなくなってしまいます。
(例えば、ログ出力先を変更しただけでテストが失敗してしまう）