chore: improve pifpaf maintenance path

pifpaf/drivers/__init__.py

@@ -42,8 +42,8 @@
 
 
 class Driver(fixtures.Fixture):
-    def __init__(self, env_prefix="PIFPAF", templatedir=".", debug=False,
-                 tmp_rootdir=None):
+    def __init__(self, env_prefix: str = "PIFPAF", templatedir: str = ".",
+                 debug: bool = False, tmp_rootdir: str | None = None) -> None:
         """Create a new driver."""
         super(Driver, self).__init__()
         self.env_prefix = env_prefix
@@ -55,15 +55,15 @@ def __init__(self, env_prefix="PIFPAF", templatedir=".", debug=False,
         self.template_env = jinja2.Environment(
             loader=jinja2.PackageLoader('pifpaf', templatedir))
 
-    def _setUp(self):
+    def _setUp(self) -> None:
         self.tempdir = self.useFixture(fixtures.TempDir(self.tmp_rootdir)).path
         self.putenv("DATA", self.tempdir)
 
     @staticmethod
     def get_options():
         return []
 
-    def putenv(self, key, value, raw=False):
+    def putenv(self, key: str, value: str, raw: bool = False) -> fixtures.Fixture:
         if not raw:
             key = self.env_prefix + "_" + key
         self.env[key] = value
@@ -139,11 +139,15 @@ def _log_output(appname, pid, data):
         data = os.fsdecode(data)
         LOG.debug("%s[%d] output: %s", appname, pid, data.rstrip())
 
-    def _exec(self, command, stdout=False, ignore_failure=False,
-              stdin=None, wait_for_line=None, wait_for_port=None,
-              path=[], env=None,
-              forbidden_line_after_start=None,
-              allow_debug=True):
+    def _exec(self, command: list[str], stdout: bool = False,
+              ignore_failure: bool = False,
+              stdin: bytes | None = None,
+              wait_for_line: str | None = None,
+              wait_for_port: int | None = None,
+              path: list[str] = [],
+              env: dict[str, str] | None = None,
+              forbidden_line_after_start: tuple[float, str] | None = None,
+              allow_debug: bool = True) -> tuple[psutil.Popen, bytes | None]:
         LOG.debug("executing: %s", command)
 
         app = command[0]
@@ -264,7 +268,7 @@ def _touch(self, fname):
         open(fname, 'a').close()
         os.utime(fname, None)
 
-    def template(self, resource, env, dest):
+    def template(self, resource: str, env: dict[str, str], dest: str) -> None:
         template = self.template_env.get_template(resource)
         with open(dest, 'w') as f:
             f.write(template.render(**env))

pifpaf/tests/test_drivers.py

@@ -120,6 +120,15 @@ def test_stuck_simple(self):
         self._do_test_stuck(["bash", "-c",
                              "trap ':' TERM ; echo started; sleep 10000"])
 
+    def test_putenv_empty_and_special(self):
+        d = self.useFixture(drivers.Driver())
+        d.putenv("EMPTY", "")
+        d.putenv("SPECIAL", "foo=bar;baz")
+        self.assertEqual("", d.env["PIFPAF_EMPTY"])
+        self.assertEqual("foo=bar;baz", d.env["PIFPAF_SPECIAL"])
+        self.assertEqual("", os.environ["PIFPAF_EMPTY"])
+        self.assertEqual("foo=bar;baz", os.environ["PIFPAF_SPECIAL"])
+
     @testtools.skip("Driver need rework")
     @testtools.skipUnless(shutil.which("elasticsearch"),
                           "elasticsearch not found")