tests/test_blackd.py

   1 import re
   2 import unittest
   3 from unittest.mock import patch
   4
   5 from click.testing import CliRunner
   6
   7 from tests.util import read_data, DETERMINISTIC_HEADER, skip_if_exception
   8
   9 try:
  10     import blackd
  11     from aiohttp.test_utils import AioHTTPTestCase, unittest_run_loop
  12     from aiohttp import web
  13 except ImportError:
  14     has_blackd_deps = False
  15 else:
  16     has_blackd_deps = True
  17
  18
  19 class BlackDTestCase(AioHTTPTestCase):
  20     @unittest.skipUnless(has_blackd_deps, "blackd's dependencies are not installed")
  21     def test_blackd_main(self) -> None:
  22         with patch("blackd.web.run_app"):
  23             result = CliRunner().invoke(blackd.main, [])
  24             if result.exception is not None:
  25                 raise result.exception
  26             self.assertEqual(result.exit_code, 0)
  27
  28     async def get_application(self) -> web.Application:
  29         return blackd.make_app()
  30
  31     # TODO: remove these decorators once the below is released
  32     # https://github.com/aio-libs/aiohttp/pull/3727
  33     @skip_if_exception("ClientOSError")
  34     @unittest.skipUnless(has_blackd_deps, "blackd's dependencies are not installed")
  35     @unittest_run_loop
  36     async def test_blackd_request_needs_formatting(self) -> None:
  37         response = await self.client.post("/", data=b"print('hello world')")
  38         self.assertEqual(response.status, 200)
  39         self.assertEqual(response.charset, "utf8")
  40         self.assertEqual(await response.read(), b'print("hello world")\n')
  41
  42     @skip_if_exception("ClientOSError")
  43     @unittest.skipUnless(has_blackd_deps, "blackd's dependencies are not installed")
  44     @unittest_run_loop
  45     async def test_blackd_request_no_change(self) -> None:
  46         response = await self.client.post("/", data=b'print("hello world")\n')
  47         self.assertEqual(response.status, 204)
  48         self.assertEqual(await response.read(), b"")
  49
  50     @skip_if_exception("ClientOSError")
  51     @unittest.skipUnless(has_blackd_deps, "blackd's dependencies are not installed")
  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     @skip_if_exception("ClientOSError")
  63     @unittest.skipUnless(has_blackd_deps, "blackd's dependencies are not installed")
  64     @unittest_run_loop
  65     async def test_blackd_unsupported_version(self) -> None:
  66         response = await self.client.post(
  67             "/", data=b"what", headers={blackd.PROTOCOL_VERSION_HEADER: "2"}
  68         )
  69         self.assertEqual(response.status, 501)
  70
  71     @skip_if_exception("ClientOSError")
  72     @unittest.skipUnless(has_blackd_deps, "blackd's dependencies are not installed")
  73     @unittest_run_loop
  74     async def test_blackd_supported_version(self) -> None:
  75         response = await self.client.post(
  76             "/", data=b"what", headers={blackd.PROTOCOL_VERSION_HEADER: "1"}
  77         )
  78         self.assertEqual(response.status, 200)
  79
  80     @skip_if_exception("ClientOSError")
  81     @unittest.skipUnless(has_blackd_deps, "blackd's dependencies are not installed")
  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                 "/", data=b"what", headers={blackd.PYTHON_VARIANT_HEADER: header_value}
  87             )
  88             self.assertEqual(response.status, expected_status)
  89
  90         await check("lol")
  91         await check("ruby3.5")
  92         await check("pyi3.6")
  93         await check("py1.5")
  94         await check("2.8")
  95         await check("py2.8")
  96         await check("3.0")
  97         await check("pypy3.0")
  98         await check("jython3.4")
  99
 100     @skip_if_exception("ClientOSError")
 101     @unittest.skipUnless(has_blackd_deps, "blackd's dependencies are not installed")
 102     @unittest_run_loop
 103     async def test_blackd_pyi(self) -> None:
 104         source, expected = read_data("stub.pyi")
 105         response = await self.client.post(
 106             "/", data=source, headers={blackd.PYTHON_VARIANT_HEADER: "pyi"}
 107         )
 108         self.assertEqual(response.status, 200)
 109         self.assertEqual(await response.text(), expected)
 110
 111     @skip_if_exception("ClientOSError")
 112     @unittest.skipUnless(has_blackd_deps, "blackd's dependencies are not installed")
 113     @unittest_run_loop
 114     async def test_blackd_diff(self) -> None:
 115         diff_header = re.compile(
 116             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"
 117         )
 118
 119         source, _ = read_data("blackd_diff.py")
 120         expected, _ = read_data("blackd_diff.diff")
 121
 122         response = await self.client.post(
 123             "/", data=source, headers={blackd.DIFF_HEADER: "true"}
 124         )
 125         self.assertEqual(response.status, 200)
 126
 127         actual = await response.text()
 128         actual = diff_header.sub(DETERMINISTIC_HEADER, actual)
 129         self.assertEqual(actual, expected)
 130
 131     @skip_if_exception("ClientOSError")
 132     @unittest.skipUnless(has_blackd_deps, "blackd's dependencies are not installed")
 133     @unittest_run_loop
 134     async def test_blackd_python_variant(self) -> None:
 135         code = (
 136             "def f(\n"
 137             "    and_has_a_bunch_of,\n"
 138             "    very_long_arguments_too,\n"
 139             "    and_lots_of_them_as_well_lol,\n"
 140             "    **and_very_long_keyword_arguments\n"
 141             "):\n"
 142             "    pass\n"
 143         )
 144
 145         async def check(header_value: str, expected_status: int) -> None:
 146             response = await self.client.post(
 147                 "/", data=code, headers={blackd.PYTHON_VARIANT_HEADER: header_value}
 148             )
 149             self.assertEqual(
 150                 response.status, expected_status, msg=await response.text()
 151             )
 152
 153         await check("3.6", 200)
 154         await check("py3.6", 200)
 155         await check("3.6,3.7", 200)
 156         await check("3.6,py3.7", 200)
 157         await check("py36,py37", 200)
 158         await check("36", 200)
 159         await check("3.6.4", 200)
 160
 161         await check("2", 204)
 162         await check("2.7", 204)
 163         await check("py2.7", 204)
 164         await check("3.4", 204)
 165         await check("py3.4", 204)
 166         await check("py34,py36", 204)
 167         await check("34", 204)
 168
 169     @skip_if_exception("ClientOSError")
 170     @unittest.skipUnless(has_blackd_deps, "blackd's dependencies are not installed")
 171     @unittest_run_loop
 172     async def test_blackd_line_length(self) -> None:
 173         response = await self.client.post(
 174             "/", data=b'print("hello")\n', headers={blackd.LINE_LENGTH_HEADER: "7"}
 175         )
 176         self.assertEqual(response.status, 200)
 177
 178     @skip_if_exception("ClientOSError")
 179     @unittest.skipUnless(has_blackd_deps, "blackd's dependencies are not installed")
 180     @unittest_run_loop
 181     async def test_blackd_invalid_line_length(self) -> None:
 182         response = await self.client.post(
 183             "/", data=b'print("hello")\n', headers={blackd.LINE_LENGTH_HEADER: "NaN"}
 184         )
 185         self.assertEqual(response.status, 400)
 186
 187     @skip_if_exception("ClientOSError")
 188     @unittest.skipUnless(has_blackd_deps, "blackd's dependencies are not installed")
 189     @unittest_run_loop
 190     async def test_blackd_response_black_version_header(self) -> None:
 191         response = await self.client.post("/")
 192         self.assertIsNotNone(response.headers.get(blackd.BLACK_VERSION_HEADER))