test_d1_client.py

import asyncio
import unittest
from contextlib import redirect_stdout
from io import StringIO
from unittest.mock import patch

import httpx
import d1_client


class _RaisingClient:
    async def __aenter__(self):
        return self

    async def __aexit__(self, exc_type, exc, tb):
        return False

    async def post(self, *args, **kwargs):
        raise httpx.ReadTimeout('')


class D1ClientDiagnosticsTests(unittest.TestCase):
    def test_timeout_log_includes_exception_type(self):
        output = StringIO()
        with patch.object(d1_client, 'ACCOUNT_ID', 'acct'), \
             patch.object(d1_client, 'DATABASE_ID', 'db'), \
             patch.object(d1_client, 'API_TOKEN', 'token'), \
             patch.object(d1_client.httpx, 'AsyncClient', return_value=_RaisingClient()), \
             redirect_stdout(output):
            result = asyncio.run(d1_client.execute_sql('SELECT 1'))

        self.assertEqual(result, [])
        self.assertIn('ReadTimeout', output.getvalue())


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