tests/test_blackd.py

   1 import re
   2 from typing import TYPE_CHECKING, Any, Callable, TypeVar
   3 from unittest.mock import patch
   4
   5 import pytest
   6 from click.testing import CliRunner
   7
   8 from tests.util import DETERMINISTIC_HEADER, read_data
   9
  10 try:
  11     from aiohttp import web
  12     from aiohttp.test_utils import AioHTTPTestCase
  13
  14     import blackd
  15 except ImportError as e:
  16     raise RuntimeError("Please install Black with the 'd' extra") from e
  17
  18 if TYPE_CHECKING:
  19     F = TypeVar("F", bound=Callable[..., Any])
  20
  21     unittest_run_loop: Callable[[F], F] = lambda x: x
  22 else:
  23     try:
  24         from aiohttp.test_utils import unittest_run_loop
  25     except ImportError:
  26         # unittest_run_loop is unnecessary and a no-op since aiohttp 3.8, and
  27         # aiohttp 4 removed it. To maintain compatibility we can make our own
  28         # no-op decorator.
  29         def unittest_run_loop(func, *args, **kwargs):
  30             return func
  31
  32
  33 @pytest.mark.blackd
  34 class BlackDTestCase(AioHTTPTestCase):  # type: ignore[misc]
  35     def test_blackd_main(self) -> None:
  36         with patch("blackd.web.run_app"):
  37             result = CliRunner().invoke(blackd.main, [])
  38             if result.exception is not None:
  39                 raise result.exception
  40             self.assertEqual(result.exit_code, 0)
  41
  42     async def get_application(self) -> web.Application:
  43         return blackd.make_app()
  44
  45     @unittest_run_loop
  46     async def test_blackd_request_needs_formatting(self) -> None:
  47         response = await self.client.post("/", data=b"print('hello world')")
  48         self.assertEqual(response.status, 200)
  49         self.assertEqual(response.charset, "utf8")
  50         self.assertEqual(await response.read(), b'print("hello world")\n')
  51
  52     @unittest_run_loop
  53     async def test_blackd_request_no_change(self) -> None:
  54         response = await self.client.post("/", data=b'print("hello world")\n')
  55         self.assertEqual(response.status, 204)
  56         self.assertEqual(await response.read(), b"")
  57
  58     @unittest_run_loop
  59     async def test_blackd_request_syntax_error(self) -> None:
  60         response = await self.client.post("/", data=b"what even ( is")
  61         self.assertEqual(response.status, 400)
  62         content = await response.text()
  63         self.assertTrue(
  64             content.startswith("Cannot parse"),
  65             msg=f"Expected error to start with 'Cannot parse', got {repr(content)}",
  66         )
  67
  68     @unittest_run_loop
  69     async def test_blackd_unsupported_version(self) -> None:
  70         response = await self.client.post(
  71             "/", data=b"what", headers={blackd.PROTOCOL_VERSION_HEADER: "2"}
  72         )
  73         self.assertEqual(response.status, 501)
  74
  75     @unittest_run_loop
  76     async def test_blackd_supported_version(self) -> None:
  77         response = await self.client.post(
  78             "/", data=b"what", headers={blackd.PROTOCOL_VERSION_HEADER: "1"}
  79         )
  80         self.assertEqual(response.status, 200)
  81
  82     @unittest_run_loop
  83     async def test_blackd_invalid_python_variant(self) -> None:
  84         async def check(header_value: str, expected_status: int = 400) -> None:
  85             response = await self.client.post(
  86                 "/",
  87                 data=b"what",
  88                 headers={blackd.PYTHON_VARIANT_HEADER: header_value},
  89             )
  90             self.assertEqual(response.status, expected_status)
  91
  92         await check("lol")
  93         await check("ruby3.5")
  94         await check("pyi3.6")
  95         await check("py1.5")
  96         await check("2")
  97         await check("2.7")
  98         await check("py2.7")
  99         await check("2.8")
 100         await check("py2.8")
 101         await check("3.0")
 102         await check("pypy3.0")
 103         await check("jython3.4")
 104
 105     @unittest_run_loop
 106     async def test_blackd_pyi(self) -> None:
 107         source, expected = read_data("miscellaneous", "stub.pyi")
 108         response = await self.client.post(
 109             "/", data=source, headers={blackd.PYTHON_VARIANT_HEADER: "pyi"}
 110         )
 111         self.assertEqual(response.status, 200)
 112         self.assertEqual(await response.text(), expected)
 113
 114     @unittest_run_loop
 115     async def test_blackd_diff(self) -> None:
 116         diff_header = re.compile(
 117             r"(In|Out)\t\d\d\d\d-\d\d-\d\d \d\d:\d\d:\d\d\.\d\d\d\d\d\d \+\d\d\d\d"
 118         )
 119
 120         source, _ = read_data("miscellaneous", "blackd_diff")
 121         expected, _ = read_data("miscellaneous", "blackd_diff.diff")
 122
 123         response = await self.client.post(
 124             "/", data=source, headers={blackd.DIFF_HEADER: "true"}
 125         )
 126         self.assertEqual(response.status, 200)
 127
 128         actual = await response.text()
 129         actual = diff_header.sub(DETERMINISTIC_HEADER, actual)
 130         self.assertEqual(actual, expected)
 131
 132     @unittest_run_loop
 133     async def test_blackd_python_variant(self) -> None:
 134         code = (
 135             "def f(\n"
 136             "    and_has_a_bunch_of,\n"
 137             "    very_long_arguments_too,\n"
 138             "    and_lots_of_them_as_well_lol,\n"
 139             "    **and_very_long_keyword_arguments\n"
 140             "):\n"
 141             "    pass\n"
 142         )
 143
 144         async def check(header_value: str, expected_status: int) -> None:
 145             response = await self.client.post(
 146                 "/", data=code, headers={blackd.PYTHON_VARIANT_HEADER: header_value}
 147             )
 148             self.assertEqual(
 149                 response.status, expected_status, msg=await response.text()
 150             )
 151
 152         await check("3.6", 200)
 153         await check("py3.6", 200)
 154         await check("3.6,3.7", 200)
 155         await check("3.6,py3.7", 200)
 156         await check("py36,py37", 200)
 157         await check("36", 200)
 158         await check("3.6.4", 200)
 159         await check("3.4", 204)
 160         await check("py3.4", 204)
 161         await check("py34,py36", 204)
 162         await check("34", 204)
 163
 164     @unittest_run_loop
 165     async def test_blackd_line_length(self) -> None:
 166         response = await self.client.post(
 167             "/", data=b'print("hello")\n', headers={blackd.LINE_LENGTH_HEADER: "7"}
 168         )
 169         self.assertEqual(response.status, 200)
 170
 171     @unittest_run_loop
 172     async def test_blackd_invalid_line_length(self) -> None:
 173         response = await self.client.post(
 174             "/",
 175             data=b'print("hello")\n',
 176             headers={blackd.LINE_LENGTH_HEADER: "NaN"},
 177         )
 178         self.assertEqual(response.status, 400)
 179
 180     @unittest_run_loop
 181     async def test_blackd_preview(self) -> None:
 182         response = await self.client.post(
 183             "/", data=b'print("hello")\n', headers={blackd.PREVIEW: "true"}
 184         )
 185         self.assertEqual(response.status, 204)
 186
 187     @unittest_run_loop
 188     async def test_blackd_response_black_version_header(self) -> None:
 189         response = await self.client.post("/")
 190         self.assertIsNotNone(response.headers.get(blackd.BLACK_VERSION_HEADER))
 191
 192     @unittest_run_loop
 193     async def test_cors_preflight(self) -> None:
 194         response = await self.client.options(
 195             "/",
 196             headers={
 197                 "Access-Control-Request-Method": "POST",
 198                 "Origin": "*",
 199                 "Access-Control-Request-Headers": "Content-Type",
 200             },
 201         )
 202         self.assertEqual(response.status, 200)
 203         self.assertIsNotNone(response.headers.get("Access-Control-Allow-Origin"))
 204         self.assertIsNotNone(response.headers.get("Access-Control-Allow-Headers"))
 205         self.assertIsNotNone(response.headers.get("Access-Control-Allow-Methods"))
 206
 207     @unittest_run_loop
 208     async def test_cors_headers_present(self) -> None:
 209         response = await self.client.post("/", headers={"Origin": "*"})
 210         self.assertIsNotNone(response.headers.get("Access-Control-Allow-Origin"))
 211         self.assertIsNotNone(response.headers.get("Access-Control-Expose-Headers"))