<s> import unittest <EOL> import pytest <EOL> aiohttp = pytest . importorskip ( "<STR_LIT>" ) <EOL> from aiohttp . wsgi import WSGIServerHttpProtocol <EOL> import asyncio <EOL> from gunicorn . workers import gaiohttp <EOL> from gunicorn . workers . _gaiohttp import _wrp <EOL> from gunicorn . config import Config <EOL> from unittest import mock <EOL> class WorkerTests ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> self . loop = asyncio . new_event_loop ( ) <EOL> asyncio . set_event_loop ( None ) <EOL> self . worker = gaiohttp . AiohttpWorker ( '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> Config ( ) , <EOL> '<STR_LIT>' ) <EOL> def tearDown ( self ) : <EOL> self . loop . close ( ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> def test_init_process ( self , m_asyncio ) : <EOL> try : <EOL> self . worker . init_process ( ) <EOL> except TypeError : <EOL> pass <EOL> self . assertTrue ( m_asyncio . get_event_loop . return_value . close . called ) <EOL> self . assertTrue ( m_asyncio . new_event_loop . called ) <EOL> self . assertTrue ( m_asyncio . set_event_loop . called ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> def test_run ( self , m_asyncio ) : <EOL> self . worker . loop = mock . Mock ( ) <EOL> self . worker . run ( ) <EOL> self . assertTrue ( m_asyncio . async . called ) <EOL> self . assertTrue ( self . worker . loop . run_until_complete . called ) <EOL> self . assertTrue ( self . worker . loop . close . called ) <EOL> def test_factory ( self ) : <EOL> self . worker . wsgi = mock . Mock ( ) <EOL> self . worker . loop = mock . Mock ( ) <EOL> self . worker . log = mock . Mock ( ) <EOL> self . worker . cfg = Config ( ) <EOL> f = self . worker . factory ( <EOL> self . worker . wsgi , ( '<STR_LIT:localhost>' , <NUM_LIT> ) ) <EOL> self . assertIsInstance ( f , WSGIServerHttpProtocol ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> def test__run ( self , m_asyncio ) : <EOL> self . worker . ppid = <NUM_LIT:1> <EOL> self . worker . alive = True <EOL> self . worker . servers = [ ] <EOL> sock = mock . Mock ( ) <EOL> sock . cfg_addr = ( '<STR_LIT:localhost>' , <NUM_LIT> ) <EOL> self . worker . sockets = [ sock ] <EOL> self . worker . wsgi = mock . Mock ( ) <EOL> self . worker . log = mock . Mock ( ) <EOL> self . worker . notify = mock . Mock ( ) <EOL> loop = self . worker . loop = mock . Mock ( ) <EOL> loop . create_server . return_value = asyncio . Future ( loop = self . loop ) <EOL> loop . create_server . return_value . set_result ( sock ) <EOL> self . loop . run_until_complete ( self . worker . _run ( ) ) <EOL> self . assertTrue ( self . worker . log . info . called ) <EOL> self . assertTrue ( self . worker . notify . called ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> def test__run_unix_socket ( self , m_asyncio ) : <EOL> self . worker . ppid = <NUM_LIT:1> <EOL> self . worker . alive = True <EOL> self . worker . servers = [ ] <EOL> sock = mock . Mock ( ) <EOL> sock . cfg_addr = '<STR_LIT>' <EOL> self . worker . sockets = [ sock ] <EOL> self . worker . wsgi = mock . Mock ( ) <EOL> self . worker . log = mock . Mock ( ) <EOL> self . worker . notify = mock . Mock ( ) <EOL> loop = self . worker . loop = mock . Mock ( ) <EOL> loop . create_server . return_value = asyncio . Future ( loop = self . loop ) <EOL> loop . create_server . return_value . set_result ( sock ) <EOL> self . loop . run_until_complete ( self . worker . _run ( ) ) <EOL> self . assertTrue ( self . worker . log . info . called ) <EOL> self . assertTrue ( self . worker . notify . called ) <EOL> def test__run_connections ( self ) : <EOL> conn = mock . Mock ( ) <EOL> self . worker . ppid = <NUM_LIT:1> <EOL> self . worker . alive = False <EOL> self . worker . servers = [ mock . Mock ( ) ] <EOL> self . worker . connections = { <NUM_LIT:1> : conn } <EOL> self . worker . sockets = [ ] <EOL> self . worker . wsgi = mock . Mock ( ) <EOL> self . worker . log = mock . Mock ( ) <EOL> self . worker . loop = self . loop <EOL> self . worker . loop . create_server = mock . Mock ( ) <EOL> self . worker . notify = mock . Mock ( ) <EOL> def _close_conns ( ) : <EOL> self . worker . connections = { } <EOL> self . loop . call_later ( <NUM_LIT:0.1> , _close_conns ) <EOL> self . loop . run_until_complete ( self . worker . _run ( ) ) <EOL> self . assertTrue ( self . worker . log . info . called ) <EOL> self . assertTrue ( self . worker . notify . called ) <EOL> self . assertFalse ( self . worker . servers ) <EOL> self . assertTrue ( conn . closing . called ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> def test__run_exc ( self , m_sleep , m_os ) : <EOL> m_os . getpid . return_value = <NUM_LIT:1> <EOL> m_os . getppid . return_value = <NUM_LIT:1> <EOL> self . worker . servers = [ mock . Mock ( ) ] <EOL> self . worker . ppid = <NUM_LIT:1> <EOL> self . worker . alive = True <EOL> self . worker . sockets = [ ] <EOL> self . worker . log = mock . Mock ( ) <EOL> self . worker . loop = mock . Mock ( ) <EOL> self . worker . notify = mock . Mock ( ) <EOL> slp = asyncio . Future ( loop = self . loop ) <EOL> slp . set_exception ( KeyboardInterrupt ) <EOL> m_sleep . return_value = slp <EOL> self . loop . run_until_complete ( self . worker . _run ( ) ) <EOL> self . assertTrue ( m_sleep . called ) <EOL> self . assertTrue ( self . worker . servers [ <NUM_LIT:0> ] . close . called ) <EOL> def test_close_wsgi_app ( self ) : <EOL> self . worker . ppid = <NUM_LIT:1> <EOL> self . worker . alive = False <EOL> self . worker . servers = [ mock . Mock ( ) ] <EOL> self . worker . connections = { } <EOL> self . worker . sockets = [ ] <EOL> self . worker . log = mock . Mock ( ) <EOL> self . worker . loop = self . loop <EOL> self . worker . loop . create_server = mock . Mock ( ) <EOL> self . worker . notify = mock . Mock ( ) <EOL> self . worker . wsgi = mock . Mock ( ) <EOL> self . worker . wsgi . close . return_value = asyncio . Future ( loop = self . loop ) <EOL> self . worker . wsgi . close . return_value . set_result ( <NUM_LIT:1> ) <EOL> self . loop . run_until_complete ( self . worker . _run ( ) ) <EOL> self . assertTrue ( self . worker . wsgi . close . called ) <EOL> self . worker . wsgi = mock . Mock ( ) <EOL> self . worker . wsgi . close . return_value = asyncio . Future ( loop = self . loop ) <EOL> self . worker . wsgi . close . return_value . set_exception ( ValueError ( ) ) <EOL> self . loop . run_until_complete ( self . worker . _run ( ) ) <EOL> self . assertTrue ( self . worker . wsgi . close . called ) <EOL> def test_wrp ( self ) : <EOL> conn = object ( ) <EOL> tracking = { } <EOL> meth = mock . Mock ( ) <EOL> wrp = _wrp ( conn , meth , tracking ) <EOL> wrp ( ) <EOL> self . assertIn ( id ( conn ) , tracking ) <EOL> self . assertTrue ( meth . called ) <EOL> meth = mock . Mock ( ) <EOL> wrp = _wrp ( conn , meth , tracking , False ) <EOL> wrp ( ) <EOL> self . assertNotIn ( <NUM_LIT:1> , tracking ) <EOL> self . assertTrue ( meth . called ) </s>
