unittests passing on windows again

sanic/testing.py

@@ -1,12 +1,30 @@
+import sys
 import traceback
 from json import JSONDecodeError
 from sanic.log import logger
 from sanic.exceptions import MethodNotSupported
 from sanic.response import text
 
+try:
+    try:
+        # direct use
+        import packaging as _packaging
+        version = _packaging.version
+    except (ImportError, AttributeError):
+        # setuptools v39.0 and above.
+        try:
+            from setuptools.extern import packaging as _packaging
+        except ImportError:
+            # Before setuptools v39.0
+            from pkg_resources.extern import packaging as _packaging
+    version = _packaging.version
+except ImportError:
+    raise RuntimeError("The 'packaging' library is missing.")
+
 
 HOST = '127.0.0.1'
 PORT = 42101
+is_windows = sys.platform in ['win32', 'cygwin']
 
 
 class SanicTestClient:

sanic/worker.py

@@ -12,15 +12,18 @@ except ImportError:
 
 try:
     import uvloop
+    import gunicorn.workers.base as base
     asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
 except ImportError:
+    base = None
     pass
-import gunicorn.workers.base as base
 
 from sanic.server import trigger_events, serve, HttpProtocol, Signal
 from sanic.websocket import WebSocketProtocol
 
 
+# gunicorn is not available on windows
+if base is not None:
     class GunicornWorker(base.Worker):
 
         http_protocol = HttpProtocol

tests/test_config.py

@@ -1,10 +1,20 @@
 from os import environ
+from pathlib import Path
+from contextlib import contextmanager
+from tempfile import TemporaryDirectory
+from textwrap import dedent
 import pytest
-from tempfile import NamedTemporaryFile
 
 from sanic import Sanic
 
 
+@contextmanager
+def temp_path():
+    """ a simple cross platform replacement for NamedTemporaryFile """
+    with TemporaryDirectory() as td:
+        yield Path(td, 'file')
+
+
 def test_load_from_object(app):
     class Config:
         not_for_config = 'should not be used'
@@ -15,35 +25,38 @@ def test_load_from_object(app):
     assert app.config.CONFIG_VALUE == 'should be used'
     assert 'not_for_config' not in app.config
 
+
 def test_auto_load_env():
     environ["SANIC_TEST_ANSWER"] = "42"
     app = Sanic()
     assert app.config.TEST_ANSWER == 42
     del environ["SANIC_TEST_ANSWER"]
 
+
 def test_dont_load_env():
     environ["SANIC_TEST_ANSWER"] = "42"
     app = Sanic(load_env=False)
-    assert getattr(app.config, 'TEST_ANSWER', None) == None
+    assert getattr(app.config, 'TEST_ANSWER', None) is None
     del environ["SANIC_TEST_ANSWER"]
 
+
 def test_load_env_prefix():
     environ["MYAPP_TEST_ANSWER"] = "42"
     app = Sanic(load_env='MYAPP_')
     assert app.config.TEST_ANSWER == 42
     del environ["MYAPP_TEST_ANSWER"]
 
+
 def test_load_from_file(app):
-    config = b"""
+    config = dedent("""
     VALUE = 'some value'
     condition = 1 == 1
     if condition:
         CONDITIONAL = 'should be set'
-    """
-    with NamedTemporaryFile() as config_file:
-        config_file.write(config)
-        config_file.seek(0)
-        app.config.from_pyfile(config_file.name)
+    """)
+    with temp_path() as config_path:
+        config_path.write_text(config)
+        app.config.from_pyfile(str(config_path))
         assert 'VALUE' in app.config
         assert app.config.VALUE == 'some value'
         assert 'CONDITIONAL' in app.config
@@ -57,11 +70,10 @@ def test_load_from_missing_file(app):
 
 
 def test_load_from_envvar(app):
-    config = b"VALUE = 'some value'"
-    with NamedTemporaryFile() as config_file:
-        config_file.write(config)
-        config_file.seek(0)
-        environ['APP_CONFIG'] = config_file.name
+    config = "VALUE = 'some value'"
+    with temp_path() as config_path:
+        config_path.write_text(config)
+        environ['APP_CONFIG'] = str(config_path)
         app.config.from_envvar('APP_CONFIG')
         assert 'VALUE' in app.config
         assert app.config.VALUE == 'some value'

tests/test_keep_alive_timeout.py

@@ -7,24 +7,12 @@ from sanic.config import Config
 from sanic import server
 import aiohttp
 from aiohttp import TCPConnector
-from sanic.testing import SanicTestClient, HOST, PORT
+from sanic.testing import SanicTestClient, HOST, PORT, version
 
-try:
-    try:
-        import packaging # direct use
-    except ImportError:
-        # setuptools v39.0 and above.
-        try:
-            from setuptools.extern import packaging
-        except ImportError:
-            # Before setuptools v39.0
-            from pkg_resources.extern import packaging
-    version = packaging.version
-except ImportError:
-    raise RuntimeError("The 'packaging' library is missing.")
 
 aiohttp_version = version.parse(aiohttp.__version__)
 
+
 class ReuseableTCPConnector(TCPConnector):
     def __init__(self, *args, **kwargs):
         super(ReuseableTCPConnector, self).__init__(*args, **kwargs)

tests/test_multiprocessing.py

@@ -1,10 +1,16 @@
 import multiprocessing
 import random
 import signal
+import pytest
 
 from sanic.testing import HOST, PORT
 
 
+@pytest.mark.skipif(
+    not hasattr(signal, 'SIGALRM'),
+    reason='SIGALRM is not implemented for this platform, we have to come '
+    'up with another timeout strategy to test these'
+)
 def test_multiprocessing(app):
     """Tests that the number of children we produce is correct"""
     # Selects a number at random so we can spot check

tests/test_request_timeout.py

@@ -5,25 +5,13 @@ import asyncio
 from sanic.response import text
 from sanic.config import Config
 import aiohttp
-from aiohttp import TCPConnector, ClientResponse
-from sanic.testing import SanicTestClient, HOST, PORT
+from aiohttp import TCPConnector
+from sanic.testing import SanicTestClient, HOST, version
 
-try:
-    try:
-        import packaging # direct use
-    except ImportError:
-        # setuptools v39.0 and above.
-        try:
-            from setuptools.extern import packaging
-        except ImportError:
-            # Before setuptools v39.0
-            from pkg_resources.extern import packaging
-    version = packaging.version
-except ImportError:
-    raise RuntimeError("The 'packaging' library is missing.")
 
 aiohttp_version = version.parse(aiohttp.__version__)
 
+
 class DelayableTCPConnector(TCPConnector):
 
     class RequestContextManager(object):

tests/test_response.py

@@ -8,9 +8,11 @@ from urllib.parse import unquote
 import pytest
 from random import choice
 
-from sanic.response import HTTPResponse, stream, StreamingHTTPResponse, file, file_stream, json
+from sanic.response import (
+    HTTPResponse, stream, StreamingHTTPResponse, file, file_stream, json
+)
 from sanic.server import HttpProtocol
-from sanic.testing import HOST, PORT
+from sanic.testing import HOST, PORT, is_windows
 from unittest.mock import MagicMock
 
 JSON_DATA = {'ok': True}
@@ -75,8 +77,11 @@ def test_response_header(app):
     request, response = app.test_client.get('/')
     assert dict(response.headers) == {
         'Connection': 'keep-alive',
-        'Keep-Alive': '2',
-        'Content-Length': '11',
+        'Keep-Alive': str(app.config.KEEP_ALIVE_TIMEOUT),
+        # response body contains an extra \r at the end if its windows
+        # TODO: this is the only place this difference shows up in our tests
+        # we should figure out a way to unify testing on both platforms
+        'Content-Length': '12' if is_windows else '11',
         'Content-Type': 'application/json',
     }
 

tests/test_routes.py

@@ -165,10 +165,11 @@ def test_route_optional_slash(app):
     request, response = app.test_client.get('/get/')
     assert response.text == 'OK'
 
+
 def test_route_strict_slashes_set_to_false_and_host_is_a_list(app):
     # Part of regression test for issue #1120
 
-    site1 = 'localhost:{}'.format(app.test_client.port)
+    site1 = '127.0.0.1:{}'.format(app.test_client.port)
 
     # before fix, this raises a RouteExists error
     @app.get('/get', host=[site1, 'site2.com'], strict_slashes=False)
@@ -178,21 +179,21 @@ def test_route_strict_slashes_set_to_false_and_host_is_a_list(app):
     request, response = app.test_client.get('http://' + site1 + '/get')
     assert response.text == 'OK'
 
-    @app.post('/post', host=[site1, 'site2.com'], strict_slashes=False)
+    @app.post('/post', host=[site1, 'site2.com'], strict_slashes=False)  # noqa
     def handler(request):
         return text('OK')
 
     request, response = app.test_client.post('http://' + site1 + '/post')
     assert response.text == 'OK'
 
-    @app.put('/put', host=[site1, 'site2.com'], strict_slashes=False)
+    @app.put('/put', host=[site1, 'site2.com'], strict_slashes=False)  # noqa
     def handler(request):
         return text('OK')
 
     request, response = app.test_client.put('http://' + site1 + '/put')
     assert response.text == 'OK'
 
-    @app.delete('/delete', host=[site1, 'site2.com'], strict_slashes=False)
+    @app.delete('/delete', host=[site1, 'site2.com'], strict_slashes=False)  # noqa
     def handler(request):
         return text('OK')
 

tests/test_server_events.py

@@ -11,6 +11,12 @@ AVAILABLE_LISTENERS = [
     'after_server_stop'
 ]
 
+skipif_no_alarm = pytest.mark.skipif(
+    not hasattr(signal, 'SIGALRM'),
+    reason='SIGALRM is not implemented for this platform, we have to come '
+    'up with another timeout strategy to test these'
+)
+
 
 def create_listener(listener_name, in_list):
     async def _listener(app, loop):
@@ -32,6 +38,7 @@ def start_stop_app(random_name_app, **run_kwargs):
         pass
 
 
+@skipif_no_alarm
 @pytest.mark.parametrize('listener_name', AVAILABLE_LISTENERS)
 def test_single_listener(app, listener_name):
     """Test that listeners on their own work"""
@@ -43,6 +50,7 @@ def test_single_listener(app, listener_name):
     assert app.name + listener_name == output.pop()
 
 
+@skipif_no_alarm
 @pytest.mark.parametrize('listener_name', AVAILABLE_LISTENERS)
 def test_register_listener(app, listener_name):
     """
@@ -52,12 +60,12 @@ def test_register_listener(app, listener_name):
     output = []
     # Register listener
     listener = create_listener(listener_name, output)
-    app.register_listener(listener,
-                                      event=listener_name)
+    app.register_listener(listener, event=listener_name)
     start_stop_app(app)
     assert app.name + listener_name == output.pop()
 
 
+@skipif_no_alarm
 def test_all_listeners(app):
     output = []
     for listener_name in AVAILABLE_LISTENERS:

tests/test_worker.py

@@ -4,16 +4,27 @@ import shlex
 import subprocess
 import urllib.request
 from unittest import mock
-from sanic.worker import GunicornWorker
-from sanic.app import Sanic
 import asyncio
-import logging
 import pytest
+from sanic.app import Sanic
+try:
+    from sanic.worker import GunicornWorker
+except ImportError:
+    pytestmark = pytest.mark.skip(
+        reason="GunicornWorker Not supported on this platform"
+    )
+    # this has to be defined or pytest will err on import
+    GunicornWorker = object
 
 
 @pytest.fixture(scope='module')
 def gunicorn_worker():
-    command = 'gunicorn --bind 127.0.0.1:1337 --worker-class sanic.worker.GunicornWorker examples.simple_server:app'
+    command = (
+        'gunicorn '
+        '--bind 127.0.0.1:1337 '
+        '--worker-class sanic.worker.GunicornWorker '
+        'examples.simple_server:app'
+    )
     worker = subprocess.Popen(shlex.split(command))
     time.sleep(3)
     yield