diff --git a/tests/test_black.py b/tests/test_black.py index 015243f..a465d54 100644 --- a/tests/test_black.py +++ b/tests/test_black.py @@ -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,160 +1513,6 @@ def fail(*args: Any, **kwargs: Any) -> None: ): ff(THIS_FILE) - # 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 - 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') - - @skip_if_exception("ClientOSError") - @unittest.skipUnless(has_blackd_deps, "blackd's dependencies are not installed") - @async_test - 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"") - - @skip_if_exception("ClientOSError") - @unittest.skipUnless(has_blackd_deps, "blackd's dependencies are not installed") - @async_test - 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)}", - ) - - @skip_if_exception("ClientOSError") - @unittest.skipUnless(has_blackd_deps, "blackd's dependencies are not installed") - @async_test - 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) - - @skip_if_exception("ClientOSError") - @unittest.skipUnless(has_blackd_deps, "blackd's dependencies are not installed") - @async_test - 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) - - @skip_if_exception("ClientOSError") - @unittest.skipUnless(has_blackd_deps, "blackd's dependencies are not installed") - @async_test - 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") - - @skip_if_exception("ClientOSError") - @unittest.skipUnless(has_blackd_deps, "blackd's dependencies are not installed") - @async_test - 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) - - @skip_if_exception("ClientOSError") - @unittest.skipUnless(has_blackd_deps, "blackd's dependencies are not installed") - @async_test - async def test_blackd_python_variant(self) -> None: - app = blackd.make_app() - code = ( - "def f(\n" - " and_has_a_bunch_of,\n" - " very_long_arguments_too,\n" - " and_lots_of_them_as_well_lol,\n" - " **and_very_long_keyword_arguments\n" - "):\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) - - 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) - - @skip_if_exception("ClientOSError") - @unittest.skipUnless(has_blackd_deps, "blackd's dependencies are not installed") - @async_test - 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) - - @skip_if_exception("ClientOSError") - @unittest.skipUnless(has_blackd_deps, "blackd's dependencies are not installed") - @async_test - 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"): @@ -1694,5 +1522,139 @@ def test_blackd_main(self) -> None: 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") + @unittest_run_loop + async def test_blackd_request_needs_formatting(self) -> None: + 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") + @unittest_run_loop + async def test_blackd_request_no_change(self) -> None: + 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") + @unittest_run_loop + async def test_blackd_request_syntax_error(self) -> None: + 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") + @unittest_run_loop + async def test_blackd_unsupported_version(self) -> None: + 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") + @unittest_run_loop + async def test_blackd_supported_version(self) -> None: + 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") + @unittest_run_loop + async def test_blackd_invalid_python_variant(self) -> None: + 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") + @unittest_run_loop + async def test_blackd_pyi(self) -> None: + 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") + @unittest_run_loop + async def test_blackd_python_variant(self) -> None: + code = ( + "def f(\n" + " and_has_a_bunch_of,\n" + " very_long_arguments_too,\n" + " and_lots_of_them_as_well_lol,\n" + " **and_very_long_keyword_arguments\n" + "):\n" + " pass\n" + ) + + 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("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") + @unittest_run_loop + async def test_blackd_line_length(self) -> None: + 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") + @unittest_run_loop + async def test_blackd_invalid_line_length(self) -> None: + response = await self.client.post( + "/", data=b'print("hello")\n', headers={blackd.LINE_LENGTH_HEADER: "NaN"} + ) + self.assertEqual(response.status, 400) + + if __name__ == "__main__": unittest.main(module="test_black")