Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

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

Merged
merged 1 commit into from
Aug 5, 2019
Merged

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

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
226 changes: 94 additions & 132 deletions tests/test_black.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand All @@ -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:
Expand Down Expand Up @@ -90,34 +81,25 @@ 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:
yield
except Exception as exc:
if exc.__class__.__name__ == e:
unittest.skip(f"Encountered expected exception {exc}, skipping")
else:
raise exc


class BlackRunner(CliRunner):
Expand Down Expand Up @@ -1531,108 +1513,103 @@ 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')")
self.assertEqual(response.status, 200)
self.assertEqual(response.charset, "utf8")
self.assertEqual(await response.read(), b'print("hello world")\n')
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')
self.assertEqual(response.status, 204)
self.assertEqual(await response.read(), b"")
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")
self.assertEqual(response.status, 400)
content = await response.text()
self.assertTrue(
content.startswith("Cannot parse"),
msg=f"Expected error to start with 'Cannot parse', got {repr(content)}",
)
response = await self.client.post("/", data=b"what even ( is")
self.assertEqual(response.status, 400)
content = await response.text()
self.assertTrue(
content.startswith("Cannot parse"),
msg=f"Expected error to start with 'Cannot parse', got {repr(content)}",
)

@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(
"/", data=b"what", headers={blackd.VERSION_HEADER: "2"}
)
self.assertEqual(response.status, 501)
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(
"/", data=b"what", headers={blackd.VERSION_HEADER: "1"}
)
self.assertEqual(response.status, 200)
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},
)
self.assertEqual(response.status, expected_status)

await check("lol")
await check("ruby3.5")
await check("pyi3.6")
await check("py1.5")
await check("2.8")
await check("py2.8")
await check("3.0")
await check("pypy3.0")
await check("jython3.4")
async def check(header_value: str, expected_status: int = 400) -> None:
response = await self.client.post(
"/", data=b"what", headers={blackd.PYTHON_VARIANT_HEADER: header_value}
)
self.assertEqual(response.status, expected_status)

await check("lol")
await check("ruby3.5")
await check("pyi3.6")
await check("py1.5")
await check("2.8")
await check("py2.8")
await check("3.0")
await check("pypy3.0")
await check("jython3.4")

@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(
"/", data=source, headers={blackd.PYTHON_VARIANT_HEADER: "pyi"}
)
self.assertEqual(response.status, 200)
self.assertEqual(await response.text(), expected)
source, expected = read_data("stub.pyi")
response = await self.client.post(
"/", data=source, headers={blackd.PYTHON_VARIANT_HEADER: "pyi"}
)
self.assertEqual(response.status, 200)
self.assertEqual(await response.text(), expected)

@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"
Expand All @@ -1642,56 +1619,41 @@ 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(
"/", data=code, headers={blackd.PYTHON_VARIANT_HEADER: header_value}
)
self.assertEqual(response.status, expected_status)
async def check(header_value: str, expected_status: int) -> None:
response = await self.client.post(
"/", data=code, headers={blackd.PYTHON_VARIANT_HEADER: header_value}
)
self.assertEqual(response.status, expected_status)

await check("3.6", 200)
await check("py3.6", 200)
await check("3.6,3.7", 200)
await check("3.6,py3.7", 200)
await check("3.6", 200)
await check("py3.6", 200)
await check("3.6,3.7", 200)
await check("3.6,py3.7", 200)

await check("2", 204)
await check("2.7", 204)
await check("py2.7", 204)
await check("3.4", 204)
await check("py3.4", 204)
await check("2", 204)
await check("2.7", 204)
await check("py2.7", 204)
await check("3.4", 204)
await check("py3.4", 204)

@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(
"/", data=b'print("hello")\n', headers={blackd.LINE_LENGTH_HEADER: "7"}
)
self.assertEqual(response.status, 200)
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"},
)
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)
response = await self.client.post(
"/", data=b'print("hello")\n', headers={blackd.LINE_LENGTH_HEADER: "NaN"}
)
self.assertEqual(response.status, 400)


if __name__ == "__main__":
Expand Down