How to use test_static_root method in Airtest

Best Python code snippet using Airtest

tests.py

Source:tests.py

1import asyncio2import sys3import threading4from pathlib import Path5from unittest import skipIf6from asgiref.sync import SyncToAsync7from asgiref.testing import ApplicationCommunicator8from django.contrib.staticfiles.handlers import ASGIStaticFilesHandler9from django.core.asgi import get_asgi_application10from django.core.signals import request_finished, request_started11from django.db import close_old_connections12from django.test import (13    AsyncRequestFactory, SimpleTestCase, modify_settings, override_settings,14)15from django.utils.http import http_date16from .urls import test_filename17TEST_STATIC_ROOT = Path(__file__).parent / 'project' / 'static'18@skipIf(sys.platform == 'win32' and (3, 8, 0) < sys.version_info < (3, 8, 1), 'https://bugs.python.org/issue38563')19@override_settings(ROOT_URLCONF='asgi.urls')20class ASGITest(SimpleTestCase):21    async_request_factory = AsyncRequestFactory()22    def setUp(self):23        request_started.disconnect(close_old_connections)24    def tearDown(self):25        request_started.connect(close_old_connections)26    async def test_get_asgi_application(self):27        """28        get_asgi_application() returns a functioning ASGI callable.29        """30        application = get_asgi_application()31        # Construct HTTP request.32        scope = self.async_request_factory._base_scope(path='/')33        communicator = ApplicationCommunicator(application, scope)34        await communicator.send_input({'type': 'http.request'})35        # Read the response.36        response_start = await communicator.receive_output()37        self.assertEqual(response_start['type'], 'http.response.start')38        self.assertEqual(response_start['status'], 200)39        self.assertEqual(40            set(response_start['headers']),41            {42                (b'Content-Length', b'12'),43                (b'Content-Type', b'text/html; charset=utf-8'),44            },45        )46        response_body = await communicator.receive_output()47        self.assertEqual(response_body['type'], 'http.response.body')48        self.assertEqual(response_body['body'], b'Hello World!')49    async def test_file_response(self):50        """51        Makes sure that FileResponse works over ASGI.52        """53        application = get_asgi_application()54        # Construct HTTP request.55        scope = self.async_request_factory._base_scope(path='/file/')56        communicator = ApplicationCommunicator(application, scope)57        await communicator.send_input({'type': 'http.request'})58        # Get the file content.59        with open(test_filename, 'rb') as test_file:60            test_file_contents = test_file.read()61        # Read the response.62        response_start = await communicator.receive_output()63        self.assertEqual(response_start['type'], 'http.response.start')64        self.assertEqual(response_start['status'], 200)65        self.assertEqual(66            set(response_start['headers']),67            {68                (b'Content-Length', str(len(test_file_contents)).encode('ascii')),69                (b'Content-Type', b'text/plain' if sys.platform == 'win32' else b'text/x-python'),70                (b'Content-Disposition', b'inline; filename="urls.py"'),71            },72        )73        response_body = await communicator.receive_output()74        self.assertEqual(response_body['type'], 'http.response.body')75        self.assertEqual(response_body['body'], test_file_contents)76        # Allow response.close() to finish.77        await communicator.wait()78    @modify_settings(INSTALLED_APPS={'append': 'django.contrib.staticfiles'})79    @override_settings(80        STATIC_URL='/static/',81        STATIC_ROOT=TEST_STATIC_ROOT,82        STATICFILES_DIRS=[TEST_STATIC_ROOT],83        STATICFILES_FINDERS=[84            'django.contrib.staticfiles.finders.FileSystemFinder',85        ],86    )87    async def test_static_file_response(self):88        application = ASGIStaticFilesHandler(get_asgi_application())89        # Construct HTTP request.90        scope = self.async_request_factory._base_scope(path='/static/file.txt')91        communicator = ApplicationCommunicator(application, scope)92        await communicator.send_input({'type': 'http.request'})93        # Get the file content.94        file_path = TEST_STATIC_ROOT / 'file.txt'95        with open(file_path, 'rb') as test_file:96            test_file_contents = test_file.read()97        # Read the response.98        stat = file_path.stat()99        response_start = await communicator.receive_output()100        self.assertEqual(response_start['type'], 'http.response.start')101        self.assertEqual(response_start['status'], 200)102        self.assertEqual(103            set(response_start['headers']),104            {105                (b'Content-Length', str(len(test_file_contents)).encode('ascii')),106                (b'Content-Type', b'text/plain'),107                (b'Content-Disposition', b'inline; filename="file.txt"'),108                (b'Last-Modified', http_date(stat.st_mtime).encode('ascii')),109            },110        )111        response_body = await communicator.receive_output()112        self.assertEqual(response_body['type'], 'http.response.body')113        self.assertEqual(response_body['body'], test_file_contents)114        # Allow response.close() to finish.115        await communicator.wait()116    async def test_headers(self):117        application = get_asgi_application()118        communicator = ApplicationCommunicator(119            application,120            self.async_request_factory._base_scope(121                path='/meta/',122                headers=[123                    [b'content-type', b'text/plain; charset=utf-8'],124                    [b'content-length', b'77'],125                    [b'referer', b'Scotland'],126                    [b'referer', b'Wales'],127                ],128            ),129        )130        await communicator.send_input({'type': 'http.request'})131        response_start = await communicator.receive_output()132        self.assertEqual(response_start['type'], 'http.response.start')133        self.assertEqual(response_start['status'], 200)134        self.assertEqual(135            set(response_start['headers']),136            {137                (b'Content-Length', b'19'),138                (b'Content-Type', b'text/plain; charset=utf-8'),139            },140        )141        response_body = await communicator.receive_output()142        self.assertEqual(response_body['type'], 'http.response.body')143        self.assertEqual(response_body['body'], b'From Scotland,Wales')144    async def test_get_query_string(self):145        application = get_asgi_application()146        for query_string in (b'name=Andrew', 'name=Andrew'):147            with self.subTest(query_string=query_string):148                scope = self.async_request_factory._base_scope(149                    path='/',150                    query_string=query_string,151                )152                communicator = ApplicationCommunicator(application, scope)153                await communicator.send_input({'type': 'http.request'})154                response_start = await communicator.receive_output()155                self.assertEqual(response_start['type'], 'http.response.start')156                self.assertEqual(response_start['status'], 200)157                response_body = await communicator.receive_output()158                self.assertEqual(response_body['type'], 'http.response.body')159                self.assertEqual(response_body['body'], b'Hello Andrew!')160    async def test_disconnect(self):161        application = get_asgi_application()162        scope = self.async_request_factory._base_scope(path='/')163        communicator = ApplicationCommunicator(application, scope)164        await communicator.send_input({'type': 'http.disconnect'})165        with self.assertRaises(asyncio.TimeoutError):166            await communicator.receive_output()167    async def test_wrong_connection_type(self):168        application = get_asgi_application()169        scope = self.async_request_factory._base_scope(path='/', type='other')170        communicator = ApplicationCommunicator(application, scope)171        await communicator.send_input({'type': 'http.request'})172        msg = 'Django can only handle ASGI/HTTP connections, not other.'173        with self.assertRaisesMessage(ValueError, msg):174            await communicator.receive_output()175    async def test_non_unicode_query_string(self):176        application = get_asgi_application()177        scope = self.async_request_factory._base_scope(path='/', query_string=b'\xff')178        communicator = ApplicationCommunicator(application, scope)179        await communicator.send_input({'type': 'http.request'})180        response_start = await communicator.receive_output()181        self.assertEqual(response_start['type'], 'http.response.start')182        self.assertEqual(response_start['status'], 400)183        response_body = await communicator.receive_output()184        self.assertEqual(response_body['type'], 'http.response.body')185        self.assertEqual(response_body['body'], b'')186    async def test_request_lifecycle_signals_dispatched_with_thread_sensitive(self):187        class SignalHandler:188            """Track threads handler is dispatched on."""189            threads = []190            def __call__(self, **kwargs):191                self.threads.append(threading.current_thread())192        signal_handler = SignalHandler()193        request_started.connect(signal_handler)194        request_finished.connect(signal_handler)195        # Perform a basic request.196        application = get_asgi_application()197        scope = self.async_request_factory._base_scope(path='/')198        communicator = ApplicationCommunicator(application, scope)199        await communicator.send_input({'type': 'http.request'})200        response_start = await communicator.receive_output()201        self.assertEqual(response_start['type'], 'http.response.start')202        self.assertEqual(response_start['status'], 200)203        response_body = await communicator.receive_output()204        self.assertEqual(response_body['type'], 'http.response.body')205        self.assertEqual(response_body['body'], b'Hello World!')206        # Give response.close() time to finish.207        await communicator.wait()208        # At this point, AsyncToSync does not have a current executor. Thus209        # SyncToAsync falls-back to .single_thread_executor.210        target_thread = next(iter(SyncToAsync.single_thread_executor._threads))211        request_started_thread, request_finished_thread = signal_handler.threads212        self.assertEqual(request_started_thread, target_thread)213        self.assertEqual(request_finished_thread, target_thread)214        request_started.disconnect(signal_handler)...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.