tests: added popup tests (#39)

mxschmitt · web-flow · commit 7df1b61c639d · 2020-07-10T21:45:40.000-07:00
diff --git a/tests/assets/popup/popup.html b/tests/assets/popup/popup.html
@@ -0,0 +1,12 @@
+<!DOCTYPE html>
+<html>
+  <head>
+    <title>Popup</title>
+    <script>
+      window.initialUserAgent = navigator.userAgent;
+    </script>
+  </head>
+  <body>
+    I am a popup
+  </body>
+</html>
diff --git a/tests/assets/popup/window-open.html b/tests/assets/popup/window-open.html
@@ -0,0 +1,11 @@
+<!DOCTYPE html>
+<html>
+  <head>
+    <title>Popup test</title>
+  </head>
+  <body>
+    <script>
+      window.open('./popup.html');
+    </script>
+  </body>
+</html>
diff --git a/tests/assets/title.html b/tests/assets/title.html
@@ -0,0 +1 @@
+<title>Woof-Woof</title>
diff --git a/tests/conftest.py b/tests/conftest.py
@@ -13,15 +13,8 @@
 # limitations under the License.
 
 import asyncio
-import os
 import pytest
 import playwright
-import threading
-import time
-
-from twisted.internet import reactor
-from twisted.web.static import File
-from twisted.web import server as web_server, resource
 
 from .server import server as server_object
 from .utils import utils as utils_object
@@ -46,8 +39,10 @@ def event_loop():
 
 
 @pytest.fixture(scope="session")
-async def browser(browser_name):
-    browser = await playwright.browser_types[browser_name].launch()
+async def browser(browser_name, pytestconfig):
+    browser = await playwright.browser_types[browser_name].launch(
+        headless=not pytestconfig.getoption("--headful")
+    )
     yield browser
     await browser.close()
 
@@ -78,15 +73,15 @@ def utils():
 
 @pytest.fixture(autouse=True, scope="session")
 async def start_http_server():
-    static_path = os.path.join(os.path.dirname(__file__), "assets")
-    resource = File(static_path)
-    site = web_server.Site(resource)
-    reactor.listenTCP(server_object.PORT, site)
-    t = threading.Thread(target=lambda: reactor.run(installSignalHandlers=0))
-    t.start()
+    server_object.start()
     yield
-    reactor.stop()
-    t.join()
+    server_object.stop()
+
+
+@pytest.fixture(autouse=True)
+async def after_each_hook():
+    yield
+    server_object.reset()
 
 
 @pytest.fixture(scope="session")
@@ -136,3 +131,9 @@ def pytest_addoption(parser):
         default=[],
         help="Browsers which should be used. By default on all the browsers.",
     )
+    parser.addoption(
+        "--headful",
+        action="store_true",
+        default=False,
+        help="Run tests in headful mode.",
+    )
diff --git a/tests/server.py b/tests/server.py
@@ -12,10 +12,19 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import asyncio
 from contextlib import closing
 
 import os
 import socket
+import threading
+import os
+import binascii
+
+from twisted.internet import reactor
+from twisted.web.static import File
+from twisted.web import server as web_server, resource
+from twisted.web._responses import UNAUTHORIZED
 
 
 def find_free_port():
@@ -25,12 +34,76 @@ def find_free_port():
         return s.getsockname()[1]
 
 
+class UnauthorizedResource(resource.ErrorPage):
+    def __init__(self, message="Sorry, resource is unauthorized."):
+        resource.ErrorPage.__init__(
+            self, UNAUTHORIZED, "Unauthorized Resource", message
+        )
+
+
 class Server:
     def __init__(self):
         self.PORT = find_free_port()
         self.EMPTY_PAGE = f"http://localhost:{self.PORT}/empty.html"
         self.PREFIX = f"http://localhost:{self.PORT}"
         self.CROSS_PROCESS_PREFIX = f"http://127.0.0.1:{self.PORT}"
 
+    def start(self):
+        request_subscribers = {}
+        auth = {}
+        self.request_subscribers = request_subscribers
+        self.auth = auth
+
+        class CustomFileServer(File):
+            def getChild(self, path, request):
+                uri_path = request.uri.decode()
+                if request_subscribers.get(uri_path):
+                    request_subscribers[uri_path].set_result(request)
+                    request_subscribers.pop(uri_path)
+
+                if auth.get(uri_path):
+                    authorization_header = request.requestHeaders.getRawHeaders(
+                        "authorization"
+                    )
+                    creds_correct = False
+                    if authorization_header:
+                        creds = binascii.a2b_base64(
+                            authorization_header[0].split(" ")[1].encode() + b"==="
+                        ).decode()
+                        creds_correct = ":".join(auth.get(uri_path)) == creds
+                    if not (authorization_header or creds_correct):
+                        request.responseHeaders.addRawHeader(
+                            b"www-authenticate", b'Basic realm="Secure Area"'
+                        )
+                        return UnauthorizedResource()
+
+                return super().getChild(path, request)
+
+        static_path = os.path.join(os.path.dirname(__file__), "assets")
+        resource = CustomFileServer(static_path)
+        site = web_server.Site(resource)
+        reactor.listenTCP(self.PORT, site)
+        self.thread = threading.Thread(
+            target=lambda: reactor.run(installSignalHandlers=0)
+        )
+        self.thread.start()
+
+    def stop(self):
+        reactor.stop()
+        self.thread.join()
+
+    async def wait_for_request(self, path):
+        future = asyncio.Future()
+        self.request_subscribers[path] = future
+        future.add_done_callback(lambda f: self.request_subscribers.pop(path, None))
+        return await future
+
+    def set_auth(self, path: str, username: str, password: str):
+        self.auth[path] = (username, password)
+
+    def reset(self):
+        self.request_subscribers.clear()
+        self.auth.clear()
+
 
 server = Server()
diff --git a/tests/test_input.py b/tests/test_input.py
@@ -13,9 +13,11 @@
 # limitations under the License.
 
 import asyncio
+from asyncio.futures import Future
 import os
 
 from playwright.page import Page
+from playwright.file_chooser import FileChooser
 
 FILE_TO_UPLOAD = os.path.join(
     os.path.dirname(os.path.realpath(__file__)), "assets/file-to-upload.txt"
diff --git a/tests/test_popup.py b/tests/test_popup.py
