Unit test example

TestLogs.py

import unittest
from unittest.mock import patch, MagicMock, AsyncMock
import aiohttp

from logs import logs

class TestLogs(unittest.IsolatedAsyncioTestCase):
    @patch('aiohttp.UnixConnector')
    async def test_connector(self, mock_connector):
        """
        Test docker socket connection
        """
        mock_session = MagicMock()
        mock_connector.return_value = mock_session
        await logs('test_container', 'test_name')
        mock_connector.assert_called_once_with(path="/var/run/docker.sock")
        
    @patch('aiohttp.ClientSession')
    async def test_url(self, mock_session):
        """
        URL validation test with logs
        """
        mock_response = MagicMock()
        mock_session.get.return_value = mock_response
        await logs('test_container', 'test_name')
        mock_session.get.assert_called_once_with(f"http://xx/containers/test_container/logs?follow=1&stdout=1")
        
    @patch('aiohttp.ClientSession')
    async def test_log_stream(self, mock_session):
        """
        Test correct log streaming
        """
        mock_response = MagicMock()
        mock_iterator = AsyncMock()
        mock_iterator.__aiter__.return_value = mock_iterator
        mock_iterator.__anext__.side_effect = ["log line 1", "log line 2", StopAsyncIteration]
        mock_response.content = mock_iterator
        mock_session.get.return_value = mock_response        
        await logs('test_container', 'test_name')
        mock_iterator.__aiter__.assert_called_once()
        self.assertEqual(mock_iterator.__anext__.call_count, 3)
       
    @patch('builtins.print')
    @patch('aiohttp.ClientSession')
    async def test_log_prefix(self, mock_session, mock_print):
        """
        Test correct log formatting and printing
        """        
        mock_response = MagicMock()        
        mock_iterator = AsyncMock()
        mock_iterator.__aiter__.return_value = mock_iterator
        mock_iterator.__anext__.side_effect = ["log line 1", "log line 2", StopAsyncIteration]
        mock_response.content = mock_iterator
        mock_session.get.return_value = mock_response
        await logs('test_container', 'test_name')        
        mock_print.assert_has_calls([
            unittest.mock.call('test_name', 'log line 1'),
            unittest.mock.call('test_name', 'log line 2')
        ])

def test_suite():
    suite = unittest.TestSuite()
    suite.addTest(TestLogs('test_connector'))
    suite.addTest(TestLogs('test_url'))
    suite.addTest(TestLogs('test_log_stream'))
    suite.addTest(TestLogs('test_log_prefix'))
    return suite

def run_tests():
    runner = unittest.TextTestRunner()
    runner.run(test_suite())

if __name__ == "__main__":
    run_tests()