mirror of
				https://github.com/ets-labs/python-dependency-injector.git
				synced 2025-11-04 09:57:37 +03:00 
			
		
		
		
	* Fix asyncio tests * Convert class-private attributes to just private * Upgrade to Cython 3 * Regenerate C files * Fix tox coverage report
		
			
				
	
	
		
			102 lines
		
	
	
		
			2.7 KiB
		
	
	
	
		
			Python
		
	
	
	
	
	
			
		
		
	
	
			102 lines
		
	
	
		
			2.7 KiB
		
	
	
	
		
			Python
		
	
	
	
	
	
"""Aiohttp extension tests."""
 | 
						|
 | 
						|
from aiohttp import web, test_utils
 | 
						|
from dependency_injector import containers, providers
 | 
						|
from dependency_injector.ext import aiohttp
 | 
						|
from pytest import fixture, mark
 | 
						|
from pytest_asyncio import fixture as aio_fixture
 | 
						|
 | 
						|
 | 
						|
async def index_view(_):
 | 
						|
    return web.Response(text="Hello World!")
 | 
						|
 | 
						|
 | 
						|
async def second_view(_):
 | 
						|
    return web.Response(text="Test!")
 | 
						|
 | 
						|
 | 
						|
class OtherClassBasedView(web.View):
 | 
						|
    async def get(self):
 | 
						|
        return web.Response(text="Test class-based!")
 | 
						|
 | 
						|
 | 
						|
@web.middleware
 | 
						|
async def middleware(request, handler):
 | 
						|
    resp = await handler(request)
 | 
						|
    resp.text = resp.text + " wink1"
 | 
						|
    return resp
 | 
						|
 | 
						|
 | 
						|
def middleware_factory(text):
 | 
						|
    @web.middleware
 | 
						|
    async def sample_middleware(request, handler):
 | 
						|
        resp = await handler(request)
 | 
						|
        resp.text = resp.text + text
 | 
						|
        return resp
 | 
						|
    return sample_middleware
 | 
						|
 | 
						|
 | 
						|
class ApplicationContainer(containers.DeclarativeContainer):
 | 
						|
 | 
						|
    app = aiohttp.Application(
 | 
						|
        web.Application,
 | 
						|
        middlewares=providers.List(
 | 
						|
            aiohttp.Middleware(middleware),
 | 
						|
            aiohttp.MiddlewareFactory(middleware_factory, text=" wink2"),
 | 
						|
        ),
 | 
						|
    )
 | 
						|
 | 
						|
    index_view = aiohttp.View(index_view)
 | 
						|
    second_view = aiohttp.View(second_view)
 | 
						|
    other_class_based_view = aiohttp.ClassBasedView(OtherClassBasedView)
 | 
						|
 | 
						|
 | 
						|
@fixture
 | 
						|
def app():
 | 
						|
    container = ApplicationContainer()
 | 
						|
    app = container.app()
 | 
						|
    app.container = container
 | 
						|
    app.add_routes([
 | 
						|
        web.get("/", container.index_view.as_view()),
 | 
						|
        web.get("/second", container.second_view.as_view(), name="second"),
 | 
						|
        web.get("/class-based", container.other_class_based_view.as_view()),
 | 
						|
    ])
 | 
						|
    return app
 | 
						|
 | 
						|
 | 
						|
@aio_fixture
 | 
						|
async def client(app):
 | 
						|
    async with test_utils.TestClient(test_utils.TestServer(app)) as client:
 | 
						|
        yield client
 | 
						|
 | 
						|
 | 
						|
@mark.asyncio
 | 
						|
@mark.filterwarnings("ignore:The loop argument is deprecated:DeprecationWarning")
 | 
						|
async def test_index(client):
 | 
						|
    response = await client.get("/")
 | 
						|
 | 
						|
    assert response.status == 200
 | 
						|
    assert await response.text() == "Hello World! wink2 wink1"
 | 
						|
 | 
						|
 | 
						|
@mark.asyncio
 | 
						|
@mark.filterwarnings("ignore:The loop argument is deprecated:DeprecationWarning")
 | 
						|
async def test_second(client):
 | 
						|
    response = await client.get("/second")
 | 
						|
 | 
						|
    assert response.status == 200
 | 
						|
    assert await response.text() == "Test! wink2 wink1"
 | 
						|
 | 
						|
 | 
						|
@mark.asyncio
 | 
						|
@mark.filterwarnings("ignore:The loop argument is deprecated:DeprecationWarning")
 | 
						|
async def test_class_based(client):
 | 
						|
    response = await client.get("/class-based")
 | 
						|
 | 
						|
    assert response.status == 200
 | 
						|
    assert await response.text() == "Test class-based! wink2 wink1"
 | 
						|
 | 
						|
 | 
						|
def test_endpoints(app):
 | 
						|
    assert str(app.router["second"].url_for()) == "/second"
 |