actions/black
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Fix async blackd tests which won't fail currently (#966)

Browse Source
This commit is contained in:
Joe Antonakakis 2019-08-05 02:06:12 -07:00 committed by Zsolt Dollenstein
parent c7495b9aa0
commit 154b98579d
1 changed files with 140 additions and 178 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

120
tests/test_black.py
View File

@ -3,24 +3,14 @@
import logging
from concurrent.futures import ThreadPoolExecutor
from contextlib import contextmanager
from functools import partial, wraps
from functools import partial
from io import BytesIO, TextIOWrapper
import os
from pathlib import Path
import re
import sys
from tempfile import TemporaryDirectory
from typing import (
Any,
BinaryIO,
Callable,
Coroutine,
Generator,
List,
Tuple,
Iterator,
TypeVar,
)
from typing import Any, BinaryIO, Generator, List, Tuple, Iterator, TypeVar
import unittest
from unittest.mock import patch, MagicMock
@ -32,7 +22,8 @@
try:
import blackd
from aiohttp.test_utils import TestClient, TestServer
from aiohttp.test_utils import AioHTTPTestCase, unittest_run_loop
from aiohttp import web
except ImportError:
has_blackd_deps = False
else:
@ -90,27 +81,16 @@ def cache_dir(exists: bool = True) -> Iterator[Path]:
@contextmanager
def event_loop(close: bool) -> Iterator[None]:
policy = asyncio.get_event_loop_policy()
old_loop = policy.get_event_loop()
loop = policy.new_event_loop()
asyncio.set_event_loop(loop)
try:
yield
finally:
policy.set_event_loop(old_loop)
if close:
loop.close()
def async_test(f: Callable[..., Coroutine[Any, None, R]]) -> Callable[..., None]:
@event_loop(close=True)
@wraps(f)
def wrapper(*args: Any, **kwargs: Any) -> None:
asyncio.get_event_loop().run_until_complete(f(*args, **kwargs))
return wrapper
@contextmanager
def skip_if_exception(e: str) -> Iterator[None]:
try:
@ -118,6 +98,8 @@ def skip_if_exception(e: str) -> Iterator[None]:
except Exception as exc:
if exc.__class__.__name__ == e:
unittest.skip(f"Encountered expected exception {exc}, skipping")
else:
raise exc
class BlackRunner(CliRunner):
@ -1531,36 +1513,43 @@ def fail(*args: Any, **kwargs: Any) -> None:
):
ff(THIS_FILE)
@unittest.skipUnless(has_blackd_deps, "blackd's dependencies are not installed")
def test_blackd_main(self) -> None:
with patch("blackd.web.run_app"):
result = CliRunner().invoke(blackd.main, [])
if result.exception is not None:
raise result.exception
self.assertEqual(result.exit_code, 0)
class BlackDTestCase(AioHTTPTestCase):
async def get_application(self) -> web.Application:
return blackd.make_app()
# TODO: remove these decorators once the below is released
# https://github.com/aio-libs/aiohttp/pull/3727
@skip_if_exception("ClientOSError")
@unittest.skipUnless(has_blackd_deps, "blackd's dependencies are not installed")
@async_test
@unittest_run_loop
async def test_blackd_request_needs_formatting(self) -> None:
app = blackd.make_app()
async with TestClient(TestServer(app)) as client:
response = await client.post("/", data=b"print('hello world')")
response = await self.client.post("/", data=b"print('hello world')")
self.assertEqual(response.status, 200)
self.assertEqual(response.charset, "utf8")
self.assertEqual(await response.read(), b'print("hello world")\n')
@skip_if_exception("ClientOSError")
@unittest.skipUnless(has_blackd_deps, "blackd's dependencies are not installed")
@async_test
@unittest_run_loop
async def test_blackd_request_no_change(self) -> None:
app = blackd.make_app()
async with TestClient(TestServer(app)) as client:
response = await client.post("/", data=b'print("hello world")\n')
response = await self.client.post("/", data=b'print("hello world")\n')
self.assertEqual(response.status, 204)
self.assertEqual(await response.read(), b"")
@skip_if_exception("ClientOSError")
@unittest.skipUnless(has_blackd_deps, "blackd's dependencies are not installed")
@async_test
@unittest_run_loop
async def test_blackd_request_syntax_error(self) -> None:
app = blackd.make_app()
async with TestClient(TestServer(app)) as client:
response = await client.post("/", data=b"what even ( is")
response = await self.client.post("/", data=b"what even ( is")
self.assertEqual(response.status, 400)
content = await response.text()
self.assertTrue(
@ -1570,38 +1559,29 @@ async def test_blackd_request_syntax_error(self) -> None:
@skip_if_exception("ClientOSError")
@unittest.skipUnless(has_blackd_deps, "blackd's dependencies are not installed")
@async_test
@unittest_run_loop
async def test_blackd_unsupported_version(self) -> None:
app = blackd.make_app()
async with TestClient(TestServer(app)) as client:
response = await client.post(
response = await self.client.post(
"/", data=b"what", headers={blackd.VERSION_HEADER: "2"}
)
self.assertEqual(response.status, 501)
@skip_if_exception("ClientOSError")
@unittest.skipUnless(has_blackd_deps, "blackd's dependencies are not installed")
@async_test
@unittest_run_loop
async def test_blackd_supported_version(self) -> None:
app = blackd.make_app()
async with TestClient(TestServer(app)) as client:
response = await client.post(
response = await self.client.post(
"/", data=b"what", headers={blackd.VERSION_HEADER: "1"}
)
self.assertEqual(response.status, 200)
@skip_if_exception("ClientOSError")
@unittest.skipUnless(has_blackd_deps, "blackd's dependencies are not installed")
@async_test
@unittest_run_loop
async def test_blackd_invalid_python_variant(self) -> None:
app = blackd.make_app()
async with TestClient(TestServer(app)) as client:
async def check(header_value: str, expected_status: int = 400) -> None:
response = await client.post(
"/",
data=b"what",
headers={blackd.PYTHON_VARIANT_HEADER: header_value},
response = await self.client.post(
"/", data=b"what", headers={blackd.PYTHON_VARIANT_HEADER: header_value}
)
self.assertEqual(response.status, expected_status)
@ -1617,12 +1597,10 @@ async def check(header_value: str, expected_status: int = 400) -> None:
@skip_if_exception("ClientOSError")
@unittest.skipUnless(has_blackd_deps, "blackd's dependencies are not installed")
@async_test
@unittest_run_loop
async def test_blackd_pyi(self) -> None:
app = blackd.make_app()
async with TestClient(TestServer(app)) as client:
source, expected = read_data("stub.pyi")
response = await client.post(
response = await self.client.post(
"/", data=source, headers={blackd.PYTHON_VARIANT_HEADER: "pyi"}
)
self.assertEqual(response.status, 200)
@ -1630,9 +1608,8 @@ async def test_blackd_pyi(self) -> None:
@skip_if_exception("ClientOSError")
@unittest.skipUnless(has_blackd_deps, "blackd's dependencies are not installed")
@async_test
@unittest_run_loop
async def test_blackd_python_variant(self) -> None:
app = blackd.make_app()
code = (
"def f(\n"
" and_has_a_bunch_of,\n"
@ -1642,10 +1619,9 @@ async def test_blackd_python_variant(self) -> None:
"):\n"
" pass\n"
)
async with TestClient(TestServer(app)) as client:
async def check(header_value: str, expected_status: int) -> None:
response = await client.post(
response = await self.client.post(
"/", data=code, headers={blackd.PYTHON_VARIANT_HEADER: header_value}
)
self.assertEqual(response.status, expected_status)
@ -1663,36 +1639,22 @@ async def check(header_value: str, expected_status: int) -> None:
@skip_if_exception("ClientOSError")
@unittest.skipUnless(has_blackd_deps, "blackd's dependencies are not installed")
@async_test
@unittest_run_loop
async def test_blackd_line_length(self) -> None:
app = blackd.make_app()
async with TestClient(TestServer(app)) as client:
response = await client.post(
response = await self.client.post(
"/", data=b'print("hello")\n', headers={blackd.LINE_LENGTH_HEADER: "7"}
)
self.assertEqual(response.status, 200)
@skip_if_exception("ClientOSError")
@unittest.skipUnless(has_blackd_deps, "blackd's dependencies are not installed")
@async_test
@unittest_run_loop
async def test_blackd_invalid_line_length(self) -> None:
app = blackd.make_app()
async with TestClient(TestServer(app)) as client:
response = await client.post(
"/",
data=b'print("hello")\n',
headers={blackd.LINE_LENGTH_HEADER: "NaN"},
response = await self.client.post(
"/", data=b'print("hello")\n', headers={blackd.LINE_LENGTH_HEADER: "NaN"}
)
self.assertEqual(response.status, 400)
@unittest.skipUnless(has_blackd_deps, "blackd's dependencies are not installed")
def test_blackd_main(self) -> None:
with patch("blackd.web.run_app"):
result = CliRunner().invoke(blackd.main, [])
if result.exception is not None:
raise result.exception
self.assertEqual(result.exit_code, 0)
if __name__ == "__main__":
unittest.main(module="test_black")