Replace black and flake8 with ruff for linting and formatting

- Remove .flake8; add [tool.ruff] and [tool.ruff.format] config in pyproject.toml
  (line-length 120, excludes, ignore list, per-file-ignores, mccabe complexity)
- Update GitHub workflow lint job to run `ruff check .` and `ruff format --check .`
- Swap black and flake8 for ruff in requirements.txt and Pipfile
- Change capnp/__init__.py to ruff-style noqa comment
- Move max-complexity into [tool.ruff.lint.mccabe], lint options into [tool.ruff.lint]
- Add per-file-ignores for capnp/__init__.py (F401, F403, F405), remove inline noqa
- Run ruff format across codebase (24 files) for consistent style

.flake8

@@ -1,6 +0,0 @@
-[flake8]
-max-line-length = 120
-extend-ignore = E203,E211,E225,E226,E227,E231,E251,E261,E262,E265,E402,E999
-max-complexity = 10
-per-file-ignores =
-    test/test_examples.py: C901

.github/workflows/wheels.yml

@@ -79,17 +79,16 @@ jobs:
           path: dist/*.tar.gz
 
   lint:
-    name: Lint with flake8 and check black
+    name: Lint and format with ruff
     runs-on: ubuntu-latest
     steps:
       - uses: actions/checkout@v5
 
-      - name: Lint with flake8 and check black
+      - name: Lint and format with ruff
         run: |
-          pip install black flake8
+          pip install ruff
-          flake8 . --filename '*.py,*.pyx,*.pxd' --count --show-source --statistics --exclude benchmark,build,capnp/templates/module.pyx
+          ruff check .
-          flake8 . --count --show-source --statistics --exclude benchmark,build
+          ruff format --check .
-          black . --check --diff --color
 
   # upload_pypi:
   #   needs: [build_wheels, build_sdist]

Pipfile

@@ -6,8 +6,7 @@ verify_ssl = true
 [packages]
 Cython = "<3"
 Jinja2 = "*"
-black = "*"
+ruff = "*"
-flake8 = "*"
 pkgconfig = "*"
 pytest = "*"
 sphinx = "*"

_custom_build/backend.py

@@ -21,9 +21,7 @@ class _CustomBuildMetaBackend(backend_class):
                 sys.argv = sys.argv[:1] + ["build_ext"] + flags + sys.argv[1:]
         return super().run_setup(setup_script)
 
-    def build_wheel(
+    def build_wheel(self, wheel_directory, config_settings=None, metadata_directory=None):
-        self, wheel_directory, config_settings=None, metadata_directory=None
-    ):
         self.config_settings = config_settings
         return super().build_wheel(wheel_directory, config_settings, metadata_directory)
 

buildutils/bundle.py

@@ -10,7 +10,6 @@
 # Adapted for use in pycapnp from pyzmq. See https://github.com/zeromq/pyzmq
 # for original project.
 
-
 import fileinput  # noqa
 import os
 import shutil

capnp/__init__.py

@@ -32,7 +32,6 @@ Example Usage::
             print(phone.type, ':', phone.number)
 """
 
-# flake8: noqa F401 F403 F405
 from .version import version as __version__
 from .lib.capnp import *
 from .lib.capnp import (

capnp/_gen.py

@@ -21,21 +21,13 @@ def main():
 
     code = schema_capnp.CodeGeneratorRequest.read(sys.stdin)
     code = code.to_dict()
-    code["nodes"] = [
+    code["nodes"] = [node for node in code["nodes"] if "struct" in node and node["scopeId"] != 0]
-        node for node in code["nodes"] if "struct" in node and node["scopeId"] != 0
-    ]
     for node in code["nodes"]:
         displayName = node["displayName"]
         parent, path = displayName.split(":")
-        node["module_path"] = (
+        node["module_path"] = parent.replace(".", "_") + "." + ".".join([x[0].upper() + x[1:] for x in path.split(".")])
-            parent.replace(".", "_")
-            + "."
-            + ".".join([x[0].upper() + x[1:] for x in path.split(".")])
-        )
         node["module_name"] = path.replace(".", "_")
-        node["c_module_path"] = "::".join(
+        node["c_module_path"] = "::".join([x[0].upper() + x[1:] for x in path.split(".")])
-            [x[0].upper() + x[1:] for x in path.split(".")]
-        )
         node["schema"] = "_{}_Schema".format(node["module_name"])
         is_union = False
         for field in node["struct"]["fields"]:
@@ -63,18 +55,12 @@ def main():
         filename = f["filename"].replace(".", "_") + "_cython.pyx"
 
         file_code = dict(code)
-        file_code["nodes"] = [
+        file_code["nodes"] = [node for node in file_code["nodes"] if node["displayName"].startswith(f["filename"])]
-            node
-            for node in file_code["nodes"]
-            if node["displayName"].startswith(f["filename"])
-        ]
         with open(filename, "w") as out:
             out.write(module.render(code=file_code, file=f, include_dir=include_dir))
 
     setup = env.get_template("setup.py.tmpl")
     with open("setup_capnp.py", "w") as out:
         out.write(setup.render(code=code))
-    print(
+    print("You now need to build the cython module by running `python setup_capnp.py build_ext --inplace`.")
-        "You now need to build the cython module by running `python setup_capnp.py build_ext --inplace`."
-    )
     print()

examples/async_calculator_client.py

@@ -23,9 +23,7 @@ class PowerFunction(calculator_capnp.Calculator.Function.Server):
 
 
 def parse_args():
-    parser = argparse.ArgumentParser(
+    parser = argparse.ArgumentParser(usage="Connects to the Calculator server at the given address and does some RPCs")
-        usage="Connects to the Calculator server at the given address and does some RPCs"
-    )
     parser.add_argument("host", help="HOST:PORT")
 
     return parser.parse_args()

examples/async_calculator_server.py

@@ -111,9 +111,7 @@ async def new_connection(stream):
 
 
 def parse_args():
-    parser = argparse.ArgumentParser(
+    parser = argparse.ArgumentParser(usage="""Runs the server bound to the given address/port ADDRESS. """)
-        usage="""Runs the server bound to the given address/port ADDRESS. """
-    )
 
     parser.add_argument("address", help="ADDRESS:PORT")
 

examples/async_reconnecting_ssl_client.py

@@ -45,21 +45,15 @@ async def main(host):
     addr, port = host.split(":")
 
     # Setup SSL context
-    ctx = ssl.create_default_context(
+    ctx = ssl.create_default_context(ssl.Purpose.SERVER_AUTH, cafile=os.path.join(this_dir, "selfsigned.cert"))
-        ssl.Purpose.SERVER_AUTH, cafile=os.path.join(this_dir, "selfsigned.cert")
-    )
 
     # Handle both IPv4 and IPv6 cases
     try:
         print("Try IPv4")
-        stream = await capnp.AsyncIoStream.create_connection(
+        stream = await capnp.AsyncIoStream.create_connection(addr, port, ssl=ctx, family=socket.AF_INET)
-            addr, port, ssl=ctx, family=socket.AF_INET
-        )
     except Exception:
         print("Try IPv6")
-        stream = await capnp.AsyncIoStream.create_connection(
+        stream = await capnp.AsyncIoStream.create_connection(addr, port, ssl=ctx, family=socket.AF_INET6)
-            addr, port, ssl=ctx, family=socket.AF_INET6
-        )
 
     client = capnp.TwoPartyClient(stream)
     cap = client.bootstrap().cast_as(thread_capnp.Example)

examples/async_server.py

@@ -28,9 +28,7 @@ async def new_connection(stream):
 
 
 def parse_args():
-    parser = argparse.ArgumentParser(
+    parser = argparse.ArgumentParser(usage="""Runs the server bound to the given address/port ADDRESS. """)
-        usage="""Runs the server bound to the given address/port ADDRESS. """
-    )
 
     parser.add_argument("address", help="ADDRESS:PORT")
 

examples/async_socket_message_server.py

@@ -42,9 +42,7 @@ async def new_connection(stream):
 
 
 def parse_args():
-    parser = argparse.ArgumentParser(
+    parser = argparse.ArgumentParser(usage="""Runs the server bound to the given address/port ADDRESS. """)
-        usage="""Runs the server bound to the given address/port ADDRESS. """
-    )
 
     parser.add_argument("address", help="ADDRESS:PORT")
 

examples/async_ssl_calculator_client.py

@@ -28,9 +28,7 @@ class PowerFunction(calculator_capnp.Calculator.Function.Server):
 
 
 def parse_args():
-    parser = argparse.ArgumentParser(
+    parser = argparse.ArgumentParser(usage="Connects to the Calculator server at the given address and does some RPCs")
-        usage="Connects to the Calculator server at the given address and does some RPCs"
-    )
     parser.add_argument("host", help="HOST:PORT")
 
     return parser.parse_args()
@@ -40,21 +38,15 @@ async def main(host):
     addr, port = host.split(":")
 
     # Setup SSL context
-    ctx = ssl.create_default_context(
+    ctx = ssl.create_default_context(ssl.Purpose.SERVER_AUTH, cafile=os.path.join(this_dir, "selfsigned.cert"))
-        ssl.Purpose.SERVER_AUTH, cafile=os.path.join(this_dir, "selfsigned.cert")
-    )
 
     # Handle both IPv4 and IPv6 cases
     try:
         print("Try IPv4")
-        stream = await capnp.AsyncIoStream.create_connection(
+        stream = await capnp.AsyncIoStream.create_connection(addr, port, ssl=ctx, family=socket.AF_INET)
-            addr, port, ssl=ctx, family=socket.AF_INET
-        )
     except Exception:
         print("Try IPv6")
-        stream = await capnp.AsyncIoStream.create_connection(
+        stream = await capnp.AsyncIoStream.create_connection(addr, port, ssl=ctx, family=socket.AF_INET6)
-            addr, port, ssl=ctx, family=socket.AF_INET6
-        )
 
     client = capnp.TwoPartyClient(stream)
 

examples/async_ssl_calculator_server.py

@@ -112,9 +112,7 @@ class CalculatorImpl(calculator_capnp.Calculator.Server):
 
 
 def parse_args():
-    parser = argparse.ArgumentParser(
+    parser = argparse.ArgumentParser(usage="""Runs the server bound to the given address/port ADDRESS. """)
-        usage="""Runs the server bound to the given address/port ADDRESS. """
-    )
 
     parser.add_argument("address", help="ADDRESS:PORT")
 
@@ -138,14 +136,10 @@ async def main():
     # Handle both IPv4 and IPv6 cases
     try:
         print("Try IPv4")
-        server = await capnp.AsyncIoStream.create_server(
+        server = await capnp.AsyncIoStream.create_server(new_connection, host, port, ssl=ctx, family=socket.AF_INET)
-            new_connection, host, port, ssl=ctx, family=socket.AF_INET
-        )
     except Exception:
         print("Try IPv6")
-        server = await capnp.AsyncIoStream.create_server(
+        server = await capnp.AsyncIoStream.create_server(new_connection, host, port, ssl=ctx, family=socket.AF_INET6)
-            new_connection, host, port, ssl=ctx, family=socket.AF_INET6
-        )
 
     async with server:
         await server.serve_forever()

examples/async_ssl_client.py

@@ -33,21 +33,15 @@ async def main(host):
     addr, port = host.split(":")
 
     # Setup SSL context
-    ctx = ssl.create_default_context(
+    ctx = ssl.create_default_context(ssl.Purpose.SERVER_AUTH, cafile=os.path.join(this_dir, "selfsigned.cert"))
-        ssl.Purpose.SERVER_AUTH, cafile=os.path.join(this_dir, "selfsigned.cert")
-    )
 
     # Handle both IPv4 and IPv6 cases
     try:
         print("Try IPv4")
-        stream = await capnp.AsyncIoStream.create_connection(
+        stream = await capnp.AsyncIoStream.create_connection(addr, port, ssl=ctx, family=socket.AF_INET)
-            addr, port, ssl=ctx, family=socket.AF_INET
-        )
     except Exception:
         print("Try IPv6")
-        stream = await capnp.AsyncIoStream.create_connection(
+        stream = await capnp.AsyncIoStream.create_connection(addr, port, ssl=ctx, family=socket.AF_INET6)
-            addr, port, ssl=ctx, family=socket.AF_INET6
-        )
 
     client = capnp.TwoPartyClient(stream)
     cap = client.bootstrap().cast_as(thread_capnp.Example)

examples/async_ssl_server.py

@@ -36,9 +36,7 @@ async def new_connection(stream):
 
 
 def parse_args():
-    parser = argparse.ArgumentParser(
+    parser = argparse.ArgumentParser(usage="""Runs the server bound to the given address/port ADDRESS. """)
-        usage="""Runs the server bound to the given address/port ADDRESS. """
-    )
     parser.add_argument("address", help="ADDRESS:PORT")
     return parser.parse_args()
 
@@ -56,14 +54,10 @@ async def main():
     # Handle both IPv4 and IPv6 cases
     try:
         print("Try IPv4")
-        server = await capnp.AsyncIoStream.create_server(
+        server = await capnp.AsyncIoStream.create_server(new_connection, host, port, ssl=ctx, family=socket.AF_INET)
-            new_connection, host, port, ssl=ctx, family=socket.AF_INET
-        )
     except Exception:
         print("Try IPv6")
-        server = await capnp.AsyncIoStream.create_server(
+        server = await capnp.AsyncIoStream.create_server(new_connection, host, port, ssl=ctx, family=socket.AF_INET6)
-            new_connection, host, port, ssl=ctx, family=socket.AF_INET6
-        )
 
     async with server:
         await server.serve_forever()

examples/py_custom_message_builder.py

@@ -51,9 +51,7 @@ print(person.extraData)
 print(type(person.extraData))
 print()
 
-person = addressbook_capnp.Person.new_message(
+person = addressbook_capnp.Person.new_message(allocate_seg_callable=MemoryViewAllocator())
-    allocate_seg_callable=MemoryViewAllocator()
-)
 
 person.init("extraData", 5)
 print(person.extraData)

pyproject.toml

@@ -5,3 +5,24 @@ backend-path = ["_custom_build"]
 
 [tool.pytest.ini_options]
 asyncio_mode = "auto"
+
+[tool.ruff]
+target-version = "py37"
+line-length = 120
+exclude = ["benchmark", "build", "capnp/templates/module.pyx"]
+
+[tool.ruff.lint]
+ignore = [
+    "E203", "E211", "E225", "E226", "E227", "E231", "E251",
+    "E261", "E262", "E265", "E402",
+]
+
+[tool.ruff.lint.per-file-ignores]
+"test/test_examples.py" = ["C901"]
+"capnp/__init__.py" = ["F401", "F403", "F405"]
+
+[tool.ruff.lint.mccabe]
+max-complexity = 10
+
+[tool.ruff.format]
+quote-style = "double"

requirements.txt

@@ -1,7 +1,6 @@
 jinja2
-black
 cython>=3
-flake8
+ruff
 setuptools
 pkgconfig
 pytest

scripts/capnp-json.py

@@ -48,9 +48,7 @@ def main():
     kwargs = vars(args)
     del kwargs["command"]
 
-    globals()[command](
+    globals()[command](**kwargs)  # hacky way to get defined functions, and call function with name=command
-        **kwargs
-    )  # hacky way to get defined functions, and call function with name=command
 
 
 main()

scripts/capnp_test_pycapnp.py

@@ -4,9 +4,7 @@ import sys
 
 import capnp
 
-capnp.add_import_hook(
+capnp.add_import_hook([os.getcwd(), "/usr/local/include/"])  # change this to be auto-detected?
-    [os.getcwd(), "/usr/local/include/"]
-)  # change this to be auto-detected?
 
 import test_capnp  # noqa: E402
 

setup.py

@@ -124,11 +124,7 @@ class build_libcapnp_ext(build_ext_c):
             if capnp_executable:
                 capnp_dir = os.path.dirname(capnp_executable)
                 self.include_dirs += [os.path.join(capnp_dir, "..", "include")]
-                self.library_dirs += [
+                self.library_dirs += [os.path.join(capnp_dir, "..", "lib{}".format(8 * struct.calcsize("P")))]
-                    os.path.join(
-                        capnp_dir, "..", "lib{}".format(8 * struct.calcsize("P"))
-                    )
-                ]
                 self.library_dirs += [os.path.join(capnp_dir, "..", "lib")]
 
             # Look for capnproto using pkg-config (and minimum version)
@@ -151,9 +147,7 @@ class build_libcapnp_ext(build_ext_c):
             bundle_dir = os.path.join(_this_dir, "bundled")
             if not os.path.exists(bundle_dir):
                 os.mkdir(bundle_dir)
-            build_dir = os.path.join(
+            build_dir = os.path.join(_this_dir, "build{}".format(8 * struct.calcsize("P")))
-                _this_dir, "build{}".format(8 * struct.calcsize("P"))
-            )
             if not os.path.exists(build_dir):
                 os.mkdir(build_dir)
 

test/test_async_write_large_payload.py

@@ -63,9 +63,7 @@ async def test_large_response_sequential():
         response = await cap.foo(i=size, j=False)
 
         # Verify the response has the correct length
-        assert (
+        assert len(response.x) == size, f"Size mismatch for {size}: expected {size}, got {len(response.x)}"
-            len(response.x) == size
-        ), f"Size mismatch for {size}: expected {size}, got {len(response.x)}"
 
         # Verify the pattern is correct (not corrupted)
         expected = "".join(chr(65 + (k % 26)) for k in range(size))
@@ -107,6 +105,4 @@ async def test_large_response_pipelined():
         assert len(response.x) == size, f"Size mismatch for {size}"
 
         expected = "".join(chr(65 + (k % 26)) for k in range(size))
-        assert (
+        assert response.x == expected, f"Data corruption detected for {size} bytes payload!"
-            response.x == expected
-        ), f"Data corruption detected for {size} bytes payload!"

test/test_capability_context.py

@@ -212,9 +212,7 @@ class TailCaller(capability.TestTailCaller.Server):
     async def foo_context(self, context):
         self.count += 1
 
-        tail = context.params.callee.foo_request(
+        tail = context.params.callee.foo_request(i=context.params.i, t="from TailCaller")
-            i=context.params.i, t="from TailCaller"
-        )
         await context.tail_call(tail)
 
 

test/test_examples.py

@@ -19,9 +19,7 @@ def cleanup():
         p.kill()
 
 
-def run_subprocesses(
+def run_subprocesses(address, server, client, wildcard_server=False, ipv4_force=True):  # noqa
-    address, server, client, wildcard_server=False, ipv4_force=True
-):  # noqa
     server_attempt = 0
     server_attempts = 2
     done = False
@@ -29,9 +27,7 @@ def run_subprocesses(
     c_address = address
     s_address = address
     while not done:
-        assert server_attempt < server_attempts, "Failed {} server attempts".format(
+        assert server_attempt < server_attempts, "Failed {} server attempts".format(server_attempts)
-            server_attempts
-        )
         server_attempt += 1
 
         # Force ipv4 for tests (known issues on GitHub Actions with IPv6 for some targets)
@@ -125,9 +121,7 @@ def test_async_calculator_example(unused_tcp_port, cleanup):
 
 
 def test_addressbook_example(cleanup):
-    proc = subprocess.Popen(
+    proc = subprocess.Popen([sys.executable, os.path.join(examples_dir, "addressbook.py")])
-        [sys.executable, os.path.join(examples_dir, "addressbook.py")]
-    )
     ret = proc.wait()
     assert ret == 0
 

test/test_get_data_view.py

@@ -197,12 +197,10 @@ def test_view_keeps_message_alive(all_types):
     view = msg.get_data_as_view("dataField")
     new_ref_count = sys.getrefcount(msg)
 
-    assert (
+    assert new_ref_count > initial_ref_count, (
-        new_ref_count > initial_ref_count
+        f"View failed to hold reference to Message! (Old: {initial_ref_count}, New: {new_ref_count})"
-    ), f"View failed to hold reference to Message! (Old: {initial_ref_count}, New: {new_ref_count})"
-    print(
-        f"\n[Ref Check] Success: Ref count increased from {initial_ref_count} to {new_ref_count}"
     )
+    print(f"\n[Ref Check] Success: Ref count increased from {initial_ref_count} to {new_ref_count}")
 
     del msg
     gc.collect()

test/test_regression.py

@@ -216,9 +216,7 @@ def test_addressbook_explicit_fields(addressbook):
         alicePhones[0]._set_by_field(phone_fields["number"], "555-1212")
         alicePhones[0]._set_by_field(phone_fields["type"], "mobile")
         employment = alice._get_by_field(person_fields["employment"])
-        employment._set_by_field(
+        employment._set_by_field(addressbook.Person.Employment.schema.fields["school"], "MIT")
-            addressbook.Person.Employment.schema.fields["school"], "MIT"
-        )
 
         bob = people[1]
         bob._set_by_field(person_fields["id"], 456)
@@ -230,9 +228,7 @@ def test_addressbook_explicit_fields(addressbook):
         bobPhones[1]._set_by_field(phone_fields["number"], "555-7654")
         bobPhones[1]._set_by_field(phone_fields["type"], "work")
         employment = bob._get_by_field(person_fields["employment"])
-        employment._set_by_field(
+        employment._set_by_field(addressbook.Person.Employment.schema.fields["unemployed"], None)
-            addressbook.Person.Employment.schema.fields["unemployed"], None
-        )
 
         addresses.write(file)
 
@@ -252,9 +248,7 @@ def test_addressbook_explicit_fields(addressbook):
         assert alicePhones[0]._get_by_field(phone_fields["number"]) == "555-1212"
         assert alicePhones[0]._get_by_field(phone_fields["type"]) == "mobile"
         employment = alice._get_by_field(person_fields["employment"])
-        employment._get_by_field(
+        employment._get_by_field(addressbook.Person.Employment.schema.fields["school"]) == "MIT"
-            addressbook.Person.Employment.schema.fields["school"]
-        ) == "MIT"
 
         bob = people[1]
         assert bob._get_by_field(person_fields["id"]) == 456
@@ -266,9 +260,7 @@ def test_addressbook_explicit_fields(addressbook):
         assert bobPhones[1]._get_by_field(phone_fields["number"]) == "555-7654"
         assert bobPhones[1]._get_by_field(phone_fields["type"]) == "work"
         employment = bob._get_by_field(person_fields["employment"])
-        employment._get_by_field(
+        employment._get_by_field(addressbook.Person.Employment.schema.fields["unemployed"]) is None
-            addressbook.Person.Employment.schema.fields["unemployed"]
-        ) is None
 
     f = open("example", "w")
     writeAddressBook(f)
@@ -450,13 +442,9 @@ def check_all_types(reader):
     check_list(subReader.uInt8List, [12, 34, 0, 0xFF])
     check_list(subReader.uInt16List, [1234, 5678, 0, 0xFFFF])
     check_list(subReader.uInt32List, [12345678, 90123456, 0, 0xFFFFFFFF])
-    check_list(
+    check_list(subReader.uInt64List, [123456789012345, 678901234567890, 0, 0xFFFFFFFFFFFFFFFF])
-        subReader.uInt64List, [123456789012345, 678901234567890, 0, 0xFFFFFFFFFFFFFFFF]
-    )
     check_list(subReader.float32List, [0.0, 1234567.0, 1e37, -1e37, 1e-37, -1e-37])
-    check_list(
+    check_list(subReader.float64List, [0.0, 123456789012345.0, 1e306, -1e306, 1e-306, -1e-306])
-        subReader.float64List, [0.0, 123456789012345.0, 1e306, -1e306, 1e-306, -1e-306]
-    )
     check_list(subReader.textList, ["quux", "corge", "grault"])
     check_list(subReader.dataList, [b"garply", b"waldo", b"fred"])
 
@@ -510,25 +498,19 @@ def check_all_types(reader):
 def test_build(all_types):
     root = all_types.TestAllTypes.new_message()
     init_all_types(root)
-    expectedText = open(
+    expectedText = open(os.path.join(this_dir, "all-types.txt"), "r", encoding="utf8").read()
-        os.path.join(this_dir, "all-types.txt"), "r", encoding="utf8"
-    ).read()
     assert str(root) + "\n" == expectedText
 
 
 def test_build_first_segment_size(all_types):
     root = all_types.TestAllTypes.new_message(1)
     init_all_types(root)
-    expectedText = open(
+    expectedText = open(os.path.join(this_dir, "all-types.txt"), "r", encoding="utf8").read()
-        os.path.join(this_dir, "all-types.txt"), "r", encoding="utf8"
-    ).read()
     assert str(root) + "\n" == expectedText
 
     root = all_types.TestAllTypes.new_message(1024 * 1024)
     init_all_types(root)
-    expectedText = open(
+    expectedText = open(os.path.join(this_dir, "all-types.txt"), "r", encoding="utf8").read()
-        os.path.join(this_dir, "all-types.txt"), "r", encoding="utf8"
-    ).read()
     assert str(root) + "\n" == expectedText
 
 
@@ -537,9 +519,7 @@ def test_binary_read(all_types):
     root = all_types.TestAllTypes.read(f)
     check_all_types(root)
 
-    expectedText = open(
+    expectedText = open(os.path.join(this_dir, "all-types.txt"), "r", encoding="utf8").read()
-        os.path.join(this_dir, "all-types.txt"), "r", encoding="utf8"
-    ).read()
     assert str(root) + "\n" == expectedText
 
     # Test set_root().
@@ -557,9 +537,7 @@ def test_packed_read(all_types):
     root = all_types.TestAllTypes.read_packed(f)
     check_all_types(root)
 
-    expectedText = open(
+    expectedText = open(os.path.join(this_dir, "all-types.txt"), "r", encoding="utf8").read()
-        os.path.join(this_dir, "all-types.txt"), "r", encoding="utf8"
-    ).read()
     assert str(root) + "\n" == expectedText
 
 

test/test_schema.py

@@ -40,15 +40,11 @@ def test_annotations(annotations):
     assert annotation.value.struct.as_struct(annotations.AnnotationStruct).test == 100
 
     annotation = annotations.TestAnnotationThree.schema.node.annotations[0]
-    annotation_list = annotation.value.list.as_list(
+    annotation_list = annotation.value.list.as_list(capnp._ListSchema(annotations.AnnotationStruct))
-        capnp._ListSchema(annotations.AnnotationStruct)
-    )
     assert annotation_list[0].test == 100
     assert annotation_list[1].test == 101
 
     annotation = annotations.TestAnnotationFour.schema.node.annotations[0]
-    annotation_list = annotation.value.list.as_list(
+    annotation_list = annotation.value.list.as_list(capnp._ListSchema(capnp.types.UInt16))
-        capnp._ListSchema(capnp.types.UInt16)
-    )
     assert annotation_list[0] == 200
     assert annotation_list[1] == 201

test/test_serialization.py

@@ -114,9 +114,7 @@ def test_roundtrip_bytes_packed(all_types):
 
 
 @contextmanager
-def _warnings(
+def _warnings(expected_count=2, expected_text="This message has already been written once."):
-    expected_count=2, expected_text="This message has already been written once."
-):
     with warnings.catch_warnings(record=True) as w:
         yield
 

test/test_struct.py

@@ -168,9 +168,7 @@ def test_new_message(all_types):
 
     assert msg.structField.int32Field == 100
 
-    msg = all_types.TestAllTypes.new_message(
+    msg = all_types.TestAllTypes.new_message(structList=[{"int32Field": 100}, {"int32Field": 101}])
-        structList=[{"int32Field": 100}, {"int32Field": 101}]
-    )
 
     assert msg.structList[0].int32Field == 100
     assert msg.structList[1].int32Field == 101
@@ -199,9 +197,7 @@ def test_set_dict(all_types):
 
 
 def test_set_dict_union(addressbook):
-    person = addressbook.Person.new_message(
+    person = addressbook.Person.new_message(**{"employment": {"employer": {"name": "foo"}}})
-        **{"employment": {"employer": {"name": "foo"}}}
-    )
 
     assert person.employment.which == addressbook.Person.Employment.employer
 
@@ -212,16 +208,12 @@ def test_union_enum(all_types):
     assert all_types.UnionAllTypes.Union.UnionStructField1 == 0
     assert all_types.UnionAllTypes.Union.UnionStructField2 == 1
 
-    msg = all_types.UnionAllTypes.new_message(
+    msg = all_types.UnionAllTypes.new_message(**{"unionStructField1": {"textField": "foo"}})
-        **{"unionStructField1": {"textField": "foo"}}
-    )
     assert msg.which == all_types.UnionAllTypes.Union.UnionStructField1
     assert msg.which == "unionStructField1"
     assert msg.which == 0
 
-    msg = all_types.UnionAllTypes.new_message(
+    msg = all_types.UnionAllTypes.new_message(**{"unionStructField2": {"textField": "foo"}})
-        **{"unionStructField2": {"textField": "foo"}}
-    )
     assert msg.which == all_types.UnionAllTypes.Union.UnionStructField2
     assert msg.which == "unionStructField2"
     assert msg.which == 1
@@ -229,14 +221,10 @@ def test_union_enum(all_types):
     assert all_types.GroupedUnionAllTypes.Union.G1 == 0
     assert all_types.GroupedUnionAllTypes.Union.G2 == 1
 
-    msg = all_types.GroupedUnionAllTypes.new_message(
+    msg = all_types.GroupedUnionAllTypes.new_message(**{"g1": {"unionStructField1": {"textField": "foo"}}})
-        **{"g1": {"unionStructField1": {"textField": "foo"}}}
-    )
     assert msg.which == all_types.GroupedUnionAllTypes.Union.G1
 
-    msg = all_types.GroupedUnionAllTypes.new_message(
+    msg = all_types.GroupedUnionAllTypes.new_message(**{"g2": {"unionStructField2": {"textField": "foo"}}})
-        **{"g2": {"unionStructField2": {"textField": "foo"}}}
-    )
     assert msg.which == all_types.GroupedUnionAllTypes.Union.G2
 
     msg = all_types.UnionAllTypes.new_message()
@@ -248,9 +236,7 @@ def isstr(s):
 
 
 def test_to_dict_enum(addressbook):
-    person = addressbook.Person.new_message(
+    person = addressbook.Person.new_message(**{"phones": [{"number": "999-9999", "type": "mobile"}]})
-        **{"phones": [{"number": "999-9999", "type": "mobile"}]}
-    )
 
     field = person.to_dict()["phones"][0]["type"]
     assert isstr(field)