diff --git a/pifpaf/drivers/__init__.py b/pifpaf/drivers/__init__.py index 54f9564..f7a32f0 100644 --- a/pifpaf/drivers/__init__.py +++ b/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,7 +55,7 @@ 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) @@ -63,7 +63,7 @@ def _setUp(self): 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)) diff --git a/pifpaf/tests/test_drivers.py b/pifpaf/tests/test_drivers.py index 457591e..95ca166 100644 --- a/pifpaf/tests/test_drivers.py +++ b/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")