pytest: use a clean environment for all tests (#29788)

* pytest: use a clean environment for all tests

* rm that

* fix pj

* put build back

* fix params

* fix that

* handle no key

* that was removed

---------

Co-authored-by: Justin Newberry <justin@comma.ai>
old-commit-hash: 7b6afbc1623b8c159a92f6a5a951fb4d688045ee

common/prefix.py

@@ -0,0 +1,38 @@
+import os
+import shutil
+import uuid
+
+from typing import Optional
+
+from openpilot.common.params import Params
+
+class OpenpilotPrefix:
+  def __init__(self, prefix: Optional[str] = None, clean_dirs_on_exit: bool = True):
+    self.prefix = prefix if prefix else str(uuid.uuid4())
+    self.msgq_path = os.path.join('/dev/shm', self.prefix)
+    self.clean_dirs_on_exit = clean_dirs_on_exit
+
+  def __enter__(self):
+    os.environ['OPENPILOT_PREFIX'] = self.prefix
+    try:
+      os.mkdir(self.msgq_path)
+    except FileExistsError:
+      pass
+
+    return self
+
+  def __exit__(self, exc_type, exc_obj, exc_tb):
+    if self.clean_dirs_on_exit:
+      self.clean_dirs()
+    try:
+      del os.environ['OPENPILOT_PREFIX']
+    except KeyError:
+      pass
+    return False
+
+  def clean_dirs(self):
+    symlink_path = Params().get_param_path()
+    if os.path.exists(symlink_path):
+      shutil.rmtree(os.path.realpath(symlink_path), ignore_errors=True)
+      os.remove(symlink_path)
+    shutil.rmtree(self.msgq_path, ignore_errors=True)

common/tests/test_params.py

@@ -1,8 +1,6 @@
 import os
 import threading
 import time
-import tempfile
-import shutil
 import uuid
 import unittest
 
@@ -10,12 +8,7 @@ from openpilot.common.params import Params, ParamKeyType, UnknownKeyName, put_no
 
 class TestParams(unittest.TestCase):
   def setUp(self):
-    self.tmpdir = tempfile.mkdtemp()
-    print("using", self.tmpdir)
-    self.params = Params(self.tmpdir)
-
-  def tearDown(self):
-    shutil.rmtree(self.tmpdir)
+    self.params = Params()
 
   def test_params_put_and_get(self):
     self.params.put("DongleId", "cb38263377b873ee")
@@ -90,19 +83,19 @@ class TestParams(unittest.TestCase):
     self.assertFalse(self.params.get_bool("IsMetric"))
 
   def test_put_non_blocking_with_get_block(self):
-    q = Params(self.tmpdir)
+    q = Params()
     def _delayed_writer():
       time.sleep(0.1)
-      put_nonblocking("CarParams", "test", self.tmpdir)
+      put_nonblocking("CarParams", "test")
     threading.Thread(target=_delayed_writer).start()
     assert q.get("CarParams") is None
     assert q.get("CarParams", True) == b"test"
 
   def test_put_bool_non_blocking_with_get_block(self):
-    q = Params(self.tmpdir)
+    q = Params()
     def _delayed_writer():
       time.sleep(0.1)
-      put_bool_nonblocking("CarParams", True, self.tmpdir)
+      put_bool_nonblocking("CarParams", True)
     threading.Thread(target=_delayed_writer).start()
     assert q.get("CarParams") is None
     assert q.get("CarParams", True) == b"1"