diff options
Diffstat (limited to 'tests/integration/test_real_browser.py')
| -rw-r--r-- | tests/integration/test_real_browser.py | 209 |
1 files changed, 90 insertions, 119 deletions
diff --git a/tests/integration/test_real_browser.py b/tests/integration/test_real_browser.py index 06a828b..a5c9030 100644 --- a/tests/integration/test_real_browser.py +++ b/tests/integration/test_real_browser.py @@ -19,12 +19,12 @@ import os import json import asyncio import socket -import unittest import pathlib import logging import webbrowser from aiohttp import web, hdrs +import pytest import selenium.common.exceptions from selenium import webdriver @@ -33,9 +33,7 @@ from selenium.webdriver.common.keys import Keys from selenium.webdriver.common.by import By from selenium.webdriver.support import expected_conditions as EC -from aiohttp_cors import setup, ResourceOptions - -from ..aio_test_base import create_server, AioTestBase, asynctest +from aiohttp_cors import setup as _setup, ResourceOptions, CorsViewMixin class _ServerDescr: @@ -52,7 +50,7 @@ class _ServerDescr: class IntegrationServers: """Integration servers starting/stopping manager""" - def __init__(self, use_resources, *, loop=None): + def __init__(self, use_resources, use_webview, *, loop=None): self.servers = {} self.loop = loop @@ -60,6 +58,7 @@ class IntegrationServers: self.loop = asyncio.get_event_loop() self.use_resources = use_resources + self.use_webview = use_webview self._logger = logging.getLogger("IntegrationServers") @@ -67,37 +66,37 @@ class IntegrationServers: def origin_server_url(self): return self.servers["origin"].url - @asyncio.coroutine - def start_servers(self): + async def start_servers(self): test_page_path = pathlib.Path(__file__).with_name("test_page.html") - @asyncio.coroutine - def handle_test_page(request: web.Request) -> web.StreamResponse: + async def handle_test_page(request: web.Request) -> web.StreamResponse: with test_page_path.open("r", encoding="utf-8") as f: return web.Response( text=f.read(), headers={hdrs.CONTENT_TYPE: "text/html"}) - @asyncio.coroutine - def handle_no_cors(request: web.Request) -> web.StreamResponse: + async def handle_no_cors(request: web.Request) -> web.StreamResponse: return web.Response( text="""{"type": "no_cors.json"}""", headers={hdrs.CONTENT_TYPE: "application/json"}) - @asyncio.coroutine - def handle_resource(request: web.Request) -> web.StreamResponse: + async def handle_resource(request: web.Request) -> web.StreamResponse: return web.Response( text="""{"type": "resource"}""", headers={hdrs.CONTENT_TYPE: "application/json"}) - @asyncio.coroutine - def handle_servers_addresses( + async def handle_servers_addresses( request: web.Request) -> web.StreamResponse: servers_addresses = \ {name: descr.url for name, descr in self.servers.items()} return web.Response( text=json.dumps(servers_addresses)) + class ResourceView(web.View, CorsViewMixin): + + async def get(self) -> web.StreamResponse: + return await handle_resource(self.request) + # For most resources: # "origin" server has no CORS configuration. # "allowing" server explicitly allows CORS requests to "origin" server. @@ -137,8 +136,12 @@ class IntegrationServers: for server_name in server_names: app = self.servers[server_name].app app.router.add_route("GET", "/no_cors.json", handle_no_cors) - app.router.add_route("GET", "/cors_resource", handle_resource, - name="cors_resource") + if self.use_webview: + app.router.add_route("*", "/cors_resource", ResourceView, + name="cors_resource") + else: + app.router.add_route("GET", "/cors_resource", handle_resource, + name="cors_resource") cors_default_configs = { "allowing": { @@ -167,7 +170,7 @@ class IntegrationServers: default_config = cors_default_configs.get(server_name) if default_config is None: continue - server_descr.cors = setup( + server_descr.cors = _setup( server_descr.app, defaults=default_config) # Add CORS routes. @@ -182,27 +185,30 @@ class IntegrationServers: server_descr.cors.add(resource) server_descr.cors.add(route) + elif self.use_webview: + server_descr.cors.add(route) + else: server_descr.cors.add(route) # Start servers. for server_name, server_descr in self.servers.items(): handler = server_descr.app.make_handler() - server = yield from create_server(handler, self.loop, - sock=server_sockets[server_name]) + server = await self.loop.create_server( + handler, + sock=server_sockets[server_name]) server_descr.handler = handler server_descr.server = server self._logger.info("Started server '%s' at '%s'", server_name, server_descr.url) - @asyncio.coroutine - def stop_servers(self): + async def stop_servers(self): for server_descr in self.servers.values(): server_descr.server.close() - yield from server_descr.handler.finish_connections() - yield from server_descr.server.wait_closed() - yield from server_descr.app.cleanup() + await server_descr.handler.shutdown() + await server_descr.server.wait_closed() + await server_descr.app.cleanup() self.servers = {} @@ -218,99 +224,64 @@ def _get_chrome_driver(): return driver -class TestInBrowser(AioTestBase): - @asyncio.coroutine - def _test_in_webdriver(self, driver, use_resources): - # TODO: Use pytest's fixtures to test use resources/not use resources. - servers = IntegrationServers(use_resources) - yield from servers.start_servers() - - def selenium_thread(): - driver.get(servers.origin_server_url) - assert "aiohttp_cors" in driver.title - - wait = WebDriverWait(driver, 10) - - run_button = wait.until(EC.element_to_be_clickable( - (By.ID, "runTestsButton"))) - - # Start tests. - run_button.send_keys(Keys.RETURN) - - # Wait while test will finish (until clear button is not - # activated). - wait.until(EC.element_to_be_clickable( - (By.ID, "clearResultsButton"))) - - # Get results json - results_area = driver.find_element_by_id("results") - - return json.loads(results_area.get_attribute("value")) - - try: - results = yield from self.loop.run_in_executor( - self.thread_pool_executor, selenium_thread) - - self.assertEqual(results["status"], "success") - for test_name, test_data in results["data"].items(): - with self.subTest(group_name=test_name): - self.assertEqual(test_data["status"], "success", - msg=(test_name, test_data)) - - finally: - yield from servers.stop_servers() - - @asynctest - @asyncio.coroutine - def test_firefox(self): - try: - driver = webdriver.Firefox() - except selenium.common.exceptions.WebDriverException: - raise unittest.SkipTest - - try: - yield from self._test_in_webdriver(driver, False) - finally: - driver.close() - - @asynctest - @asyncio.coroutine - def test_chromium(self): - try: - driver = _get_chrome_driver() - except selenium.common.exceptions.WebDriverException: - raise unittest.SkipTest - - try: - yield from self._test_in_webdriver(driver, False) - finally: - driver.close() - - @asynctest - @asyncio.coroutine - def test_firefox_resource(self): - try: - driver = webdriver.Firefox() - except selenium.common.exceptions.WebDriverException: - raise unittest.SkipTest - - try: - yield from self._test_in_webdriver(driver, True) - finally: - driver.close() - - @asynctest - @asyncio.coroutine - def test_chromium_resource(self): - try: - driver = _get_chrome_driver() - except selenium.common.exceptions.WebDriverException: - raise unittest.SkipTest - - try: - yield from self._test_in_webdriver(driver, True) - finally: - driver.close() +@pytest.fixture(params=[(False, False), + (True, False), + (False, True)]) +def server(request, loop): + async def inner(): + # to grab implicit loop + return IntegrationServers(*request.param) + return loop.run_until_complete(inner()) + + +@pytest.fixture(params=[webdriver.Firefox, + _get_chrome_driver]) +def driver(request): + try: + driver = request.param() + except selenium.common.exceptions.WebDriverException: + pytest.skip("Driver is not supported") + + yield driver + driver.close() + + +async def test_in_webdriver(driver, server): + loop = asyncio.get_event_loop() + await server.start_servers() + + def selenium_thread(): + driver.get(server.origin_server_url) + assert "aiohttp_cors" in driver.title + + wait = WebDriverWait(driver, 10) + + run_button = wait.until(EC.element_to_be_clickable( + (By.ID, "runTestsButton"))) + + # Start tests. + run_button.send_keys(Keys.RETURN) + + # Wait while test will finish (until clear button is not + # activated). + wait.until(EC.element_to_be_clickable( + (By.ID, "clearResultsButton"))) + + # Get results json + results_area = driver.find_element_by_id("results") + + return json.loads(results_area.get_attribute("value")) + + try: + results = await loop.run_in_executor( + None, selenium_thread) + + assert results["status"] == "success" + for test_name, test_data in results["data"].items(): + assert test_data["status"] == "success" + + finally: + await server.stop_servers() def _run_integration_server(): @@ -322,7 +293,7 @@ def _run_integration_server(): loop = asyncio.get_event_loop() - servers = IntegrationServers() + servers = IntegrationServers(False, True) logger.info("Starting integration servers...") loop.run_until_complete(servers.start_servers()) |