test: add cors test

test/12.cors.py

@@ -0,0 +1,72 @@
+import signal
+import subprocess
+
+import time
+from selenium.webdriver import Chrome
+
+import template
+import util
+from pywebio.input import *
+from pywebio.utils import run_as_function
+
+
+def target():
+    template.basic_output()
+    template.background_output()
+
+    run_as_function(template.basic_input())
+    actions(buttons=['Continue'])
+    template.background_input()
+
+
+def test_once(browser: Chrome, output_file):
+    template.test_output(browser)
+    template.test_input(browser)
+    time.sleep(1)
+    template.save_output(browser, output_file)
+
+
+def test(server_proc: subprocess.Popen, browser: Chrome):
+    test_once(browser, '12.tornado_cors.html')
+    server_proc.send_signal(signal.SIGINT)
+
+    time.sleep(4)
+    browser.get('http://localhost:5000/?pywebio_api=http://localhost:8081/io')
+    test_once(browser, '12.aiohttp_cors.html')
+    server_proc.send_signal(signal.SIGINT)
+
+    time.sleep(4)
+    browser.get('http://localhost:5000/?pywebio_api=http://localhost:8082/io')
+    test_once(browser, '12.flask_cors.html')
+
+
+def start_test_server():
+    from pywebio import start_server as tornado_server
+    from pywebio.platform.flask import start_server as flask_server
+    from pywebio.platform.aiohttp import start_server as aiohttp_server
+    import asyncio
+
+    util.start_static_server()
+
+    try:
+        tornado_server(target, port=8080, allowed_origins=['http://localhost:5000'])
+    except:
+        print('tornado_server exit')
+
+    loop = asyncio.new_event_loop()
+    asyncio.set_event_loop(loop)
+
+    try:
+        aiohttp_server(target, port=8081, allowed_origins=['http://localhost:5000'])
+    except:
+        print('aiohttp_server exit')
+
+    try:
+        flask_server(target, port=8082, allowed_origins=['http://localhost:5000'])
+    except:
+        print('flask_server exit')
+
+
+if __name__ == '__main__':
+    util.run_test(start_test_server, test,
+                  address='http://localhost:5000/?pywebio_api=http://localhost:8080/io')

test/util.py

@@ -2,9 +2,13 @@ import asyncio
 import signal
 import subprocess
 import sys
+import threading
+from functools import partial
+from urllib.parse import urlparse
 
 from selenium import webdriver
 
+from pywebio import STATIC_PATH
 from pywebio.utils import wait_host_port
 
 default_chrome_options = webdriver.ChromeOptions()
@@ -25,7 +29,7 @@ python {name} debug
 """
 
 
-def run_test(server_func, test_func, port=8080, chrome_options=None):
+def run_test(server_func, test_func, address='http://localhost:8080?_pywebio_debug=1', chrome_options=None):
     """
     :param server_func: 启动PyWebIO服务器的函数
     :param test_func: 测试的函数。人工测试时不会被运行 (server_proc, browser)
@@ -56,8 +60,9 @@ def run_test(server_func, test_func, port=8080, chrome_options=None):
     try:
         browser = webdriver.Chrome(chrome_options=chrome_options)
         browser.set_window_size(1000, 900)
-        asyncio.run(wait_host_port('localhost', port))
-        browser.get('http://localhost:%s?_pywebio_debug=1' % port)
+        port_str = urlparse(address).netloc.split(':', 1)[-1] or '80'
+        asyncio.run(wait_host_port('localhost', int(port_str)))
+        browser.get(address)
         browser.implicitly_wait(10)
         test_func(proc, browser)
     finally:
@@ -70,3 +75,10 @@ def run_test(server_func, test_func, port=8080, chrome_options=None):
         # 不要使用 proc.terminate() ，因为coverage会无法保存分析数据
         proc.send_signal(signal.SIGINT)
         print("Closed browser and PyWebIO server")
+
+
+def start_static_server(port=5000):
+    from http.server import SimpleHTTPRequestHandler, test
+
+    handler_class = partial(SimpleHTTPRequestHandler, directory=STATIC_PATH)
+    threading.Thread(target=test, kwargs=dict(HandlerClass=handler_class, port=port, bind='localhost'), daemon=True).start()