tests/test_blackd.py

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