python: add pytest checks to ruff config

...and do some suggested cleanups.

pyproject.toml

@@ -30,6 +30,7 @@ select = [
     "F",       # pyflakes
     "G",       # flake8-logging-format
     "I",       # isort
+    "PT",      # flake8-pytest-style
 ]
 exclude = [
     ".git/",
@@ -44,9 +45,14 @@ ignore = [
     "A003",  # Class attribute is shadowing a python builtin
     "B011",  # Do not `assert False` (`python -O` removes these calls), raise `AssertionError()`
     "E731",  # Do not assign a `lambda` expression, use a `def`
+    "PT011", # `pytest.raises(OSError)` is too broad
 ]
 line-length = 118
 
+[tool.ruff.flake8-pytest-style]
+fixture-parentheses = false
+mark-parentheses = false
+
 [tool.ruff.isort]
 known-first-party = ["cockpit"]
 

src/cockpit/transports.py

@@ -324,7 +324,8 @@ def _exited(self, pid: int, code: int) -> None:
         # zero.  For that reason, we need to store our own copy of the return
         # status.  See https://github.com/python/cpython/issues/59960
         assert isinstance(self._protocol, SubprocessProtocol)
-        assert self._process is not None and self._process.pid == pid
+        assert self._process is not None
+        assert self._process.pid == pid
         self._returncode = code
         logger.debug('Process exited with status %d', self._returncode)
         if not self._closing:
@@ -360,7 +361,8 @@ def preexec_fn():
         else:
             self._process = subprocess.Popen(args, stdin=subprocess.PIPE, stdout=subprocess.PIPE,
                                              preexec_fn=preexec_fn, **kwargs)
-            assert self._process.stdin and self._process.stdout
+            assert self._process.stdin
+            assert self._process.stdout
             in_fd = self._process.stdout.fileno()
             out_fd = self._process.stdin.fileno()
 

test/pytest/mocktransport.py

@@ -4,6 +4,8 @@
 import subprocess
 from typing import Any, Dict, Iterable, Optional, Tuple
 
+import pytest
+
 from cockpit.router import Router
 
 MOCK_HOSTNAME = 'mockbox'
@@ -27,7 +29,7 @@ def assert_no_subprocesses():
     subprocess.call(['pkill', '-9', '-P', f'{os.getpid()}'])
     for _ in range(100):  # zombie vacuum
         os.waitpid(-1, os.WNOHANG)  # will eventually throw, failing the test (which we want)
-    assert False  # more than 100?
+    pytest.fail('failed to reap all children')
 
 
 def assert_no_tasks(allow_tasks=0):

test/pytest/test_bridge.py

@@ -497,7 +497,7 @@ async def serve_page(reader, writer):
                 await settle_down()
                 return
             else:
-                assert False, (payload, args, control)
+                pytest.fail('unexpected event', (payload, args, control))
         else:
             saw_data = True
 

test/pytest/test_code.py

@@ -35,4 +35,4 @@ def test_linter(command, pytestconfig):
     except FileNotFoundError as exc:
         pytest.skip(f'{exc.filename} not installed')
     except subprocess.CalledProcessError:
-        pytest.fail()
+        pytest.fail('linting failed')

test/pytest/test_packages.py

@@ -23,10 +23,11 @@
 from cockpit.packages import Packages, parse_accept_language
 
 
-@pytest.mark.parametrize("test_input,expected", [
-                         ('de-at, zh-CH, en,', ['de-at', 'zh-ch', 'en']),
-                         ('es-es, nl;q=0.8, fr;q=0.9', ['es-es', 'fr', 'nl']),
-                         ('fr-CH, fr;q=0.9, en;q=0.8, de;q=0.7, *;q=0.5', ['fr-ch', 'fr', 'en', 'de', '*'])])
+@pytest.mark.parametrize(("test_input", "expected"), [
+    ('de-at, zh-CH, en,', ['de-at', 'zh-ch', 'en']),
+    ('es-es, nl;q=0.8, fr;q=0.9', ['es-es', 'fr', 'nl']),
+    ('fr-CH, fr;q=0.9, en;q=0.8, de;q=0.7, *;q=0.5', ['fr-ch', 'fr', 'en', 'de', '*'])
+])
 def test_parse_accept_language(test_input, expected):
     assert parse_accept_language({'Accept-Language': test_input}) == expected
 

test/pytest/test_transport.py

@@ -25,6 +25,8 @@
 import unittest.mock
 from typing import Any, List, Optional, Tuple
 
+import pytest
+
 import cockpit.transports
 
 
@@ -96,10 +98,9 @@ class TestSpooler(unittest.IsolatedAsyncioTestCase):
     async def test_bad_fd(self) -> None:
         # Make sure failing to construct succeeds without further failures
         loop = asyncio.get_running_loop()
-        try:
+        with pytest.raises(OSError) as raises:
             cockpit.transports.Spooler(loop, -1)
-        except OSError as exc:
-            assert exc.errno == errno.EBADF
+        assert raises.value.errno == errno.EBADF
 
     def create_spooler(self, to_write: bytes = b'') -> cockpit.transports.Spooler:
         loop = asyncio.get_running_loop()