Pytest: more post-test path cleanup (#30556)

* test cleanup

* missed this one

* remove that

* static class

common/api/__init__.py

@@ -2,7 +2,7 @@ import jwt
 import os
 import requests
 from datetime import datetime, timedelta
-from openpilot.common.basedir import PERSIST
+from openpilot.system.hardware.hw import Paths
 from openpilot.system.version import get_version
 
 API_HOST = os.getenv('API_HOST', 'https://api.commadotai.com')
@@ -10,7 +10,7 @@ API_HOST = os.getenv('API_HOST', 'https://api.commadotai.com')
 class Api():
   def __init__(self, dongle_id):
     self.dongle_id = dongle_id
-    with open(PERSIST+'/comma/id_rsa') as f:
+    with open(Paths.persist_root()+'/comma/id_rsa') as f:
       self.private_key = f.read()
 
   def get(self, *args, **kwargs):

common/basedir.py

@@ -1,11 +1,4 @@
 import os
-from pathlib import Path
 
-from openpilot.system.hardware import PC
 
 BASEDIR = os.path.abspath(os.path.join(os.path.dirname(os.path.realpath(__file__)), "../"))
-
-if PC:
-  PERSIST = os.path.join(str(Path.home()), ".comma", "persist")
-else:
-  PERSIST = "/persist"

selfdrive/athena/athenad.py

@@ -31,13 +31,11 @@ import cereal.messaging as messaging
 from cereal import log
 from cereal.services import SERVICE_LIST
 from openpilot.common.api import Api
-from openpilot.common.basedir import PERSIST
 from openpilot.common.file_helpers import CallbackReader
 from openpilot.common.params import Params
 from openpilot.common.realtime import set_core_affinity
 from openpilot.system.hardware import HARDWARE, PC, AGNOS
 from openpilot.system.loggerd.xattr_cache import getxattr, setxattr
-from openpilot.selfdrive.statsd import STATS_DIR
 from openpilot.system.swaglog import cloudlog
 from openpilot.system.version import get_commit, get_origin, get_short_branch, get_version
 from openpilot.system.hardware.hw import Paths
@@ -502,10 +500,10 @@ def startLocalProxy(global_end_event: threading.Event, remote_ws_uri: str, local
 
 @dispatcher.add_method
 def getPublicKey() -> Optional[str]:
-  if not os.path.isfile(PERSIST + '/comma/id_rsa.pub'):
+  if not os.path.isfile(Paths.persist_root() + '/comma/id_rsa.pub'):
     return None
 
-  with open(PERSIST + '/comma/id_rsa.pub') as f:
+  with open(Paths.persist_root() + '/comma/id_rsa.pub') as f:
     return f.read()
 
 
@@ -641,6 +639,7 @@ def log_handler(end_event: threading.Event) -> None:
 
 
 def stat_handler(end_event: threading.Event) -> None:
+  STATS_DIR = Paths.stats_root()
   while not end_event.is_set():
     last_scan = 0.
     curr_scan = time.monotonic()

selfdrive/athena/registration.py

@@ -9,9 +9,9 @@ from datetime import datetime, timedelta
 from openpilot.common.api import api_get
 from openpilot.common.params import Params
 from openpilot.common.spinner import Spinner
-from openpilot.common.basedir import PERSIST
 from openpilot.selfdrive.controls.lib.alertmanager import set_offroad_alert
 from openpilot.system.hardware import HARDWARE, PC
+from openpilot.system.hardware.hw import Paths
 from openpilot.system.swaglog import cloudlog
 
 
@@ -32,7 +32,7 @@ def register(show_spinner=False) -> Optional[str]:
   dongle_id: Optional[str] = params.get("DongleId", encoding='utf8')
   needs_registration = None in (IMEI, HardwareSerial, dongle_id)
 
-  pubkey = Path(PERSIST+"/comma/id_rsa.pub")
+  pubkey = Path(Paths.persist_root()+"/comma/id_rsa.pub")
   if not pubkey.is_file():
     dongle_id = UNREGISTERED_DONGLE_ID
     cloudlog.warning(f"missing public key: {pubkey}")
@@ -42,7 +42,7 @@ def register(show_spinner=False) -> Optional[str]:
       spinner.update("registering device")
 
     # Create registration token, in the future, this key will make JWTs directly
-    with open(PERSIST+"/comma/id_rsa.pub") as f1, open(PERSIST+"/comma/id_rsa") as f2:
+    with open(Paths.persist_root()+"/comma/id_rsa.pub") as f1, open(Paths.persist_root()+"/comma/id_rsa") as f2:
       public_key = f1.read()
       private_key = f2.read()
 

selfdrive/athena/tests/test_athenad_ping.py

@@ -40,8 +40,6 @@ class TestAthenadPing(unittest.TestCase):
 
   @classmethod
   def setUpClass(cls) -> None:
-    cls.params = Params()
-    cls.dongle_id = cls.params.get("DongleId", encoding="utf-8")
     cls._create_connection = athenad.create_connection
     athenad.create_connection = MagicMock(wraps=cls._create_connection)
 
@@ -51,6 +49,9 @@ class TestAthenadPing(unittest.TestCase):
     athenad.create_connection = cls._create_connection
 
   def setUp(self) -> None:
+    self.params = Params()
+    self.dongle_id = self.params.get("DongleId", encoding="utf-8")
+
     wifi_radio(True)
     self._clear_ping_time()
 

selfdrive/athena/tests/test_registration.py

@@ -1,7 +1,5 @@
 #!/usr/bin/env python3
 import json
-import os
-import tempfile
 import unittest
 from Crypto.PublicKey import RSA
 from pathlib import Path
@@ -10,6 +8,7 @@ from unittest import mock
 from openpilot.common.params import Params
 from openpilot.selfdrive.athena.registration import register, UNREGISTERED_DONGLE_ID
 from openpilot.selfdrive.athena.tests.helpers import MockResponse
+from openpilot.system.hardware.hw import Paths
 
 
 class TestRegistration(unittest.TestCase):
@@ -19,16 +18,11 @@ class TestRegistration(unittest.TestCase):
     self.params = Params()
     self.params.clear_all()
 
-    self.persist = tempfile.TemporaryDirectory()
-    os.mkdir(os.path.join(self.persist.name, "comma"))
-    self.priv_key = Path(os.path.join(self.persist.name, "comma/id_rsa"))
-    self.pub_key = Path(os.path.join(self.persist.name, "comma/id_rsa.pub"))
-    self.persist_patcher = mock.patch("openpilot.selfdrive.athena.registration.PERSIST", self.persist.name)
-    self.persist_patcher.start()
+    persist_dir = Path(Paths.persist_root()) / "comma"
+    persist_dir.mkdir(parents=True, exist_ok=True)
 
-  def tearDown(self):
-    self.persist_patcher.stop()
-    self.persist.cleanup()
+    self.priv_key = persist_dir / "id_rsa"
+    self.pub_key = persist_dir / "id_rsa.pub"
 
   def _generate_keys(self):
     self.pub_key.touch()

selfdrive/statsd.py

@@ -9,11 +9,12 @@ from typing import NoReturn, Union, List, Dict
 
 from openpilot.common.params import Params
 from cereal.messaging import SubMaster
+from openpilot.system.hardware.hw import Paths
 from openpilot.system.swaglog import cloudlog
 from openpilot.system.hardware import HARDWARE
 from openpilot.common.file_helpers import atomic_write_in_dir
 from openpilot.system.version import get_normalized_origin, get_short_branch, get_short_version, is_dirty
-from openpilot.system.loggerd.config import STATS_DIR, STATS_DIR_FILE_LIMIT, STATS_SOCKET, STATS_FLUSH_TIME_S
+from openpilot.system.loggerd.config import STATS_DIR_FILE_LIMIT, STATS_SOCKET, STATS_FLUSH_TIME_S
 
 
 class METRIC_TYPE:
@@ -80,6 +81,8 @@ def main() -> NoReturn:
   sock = ctx.socket(zmq.PULL)
   sock.bind(STATS_SOCKET)
 
+  STATS_DIR = Paths.stats_root()
+
   # initialize stats directory
   Path(STATS_DIR).mkdir(parents=True, exist_ok=True)
 

selfdrive/test/longitudinal_maneuvers/test_longitudinal.py

@@ -156,6 +156,7 @@ class LongitudinalControl(unittest.TestCase):
     os.environ['SKIP_FW_QUERY'] = "1"
     os.environ['NO_CAN_TIMEOUT'] = "1"
 
+  def setUp(self):
     params = Params()
     params.clear_all()
     params.put_bool("Passive", bool(os.getenv("PASSIVE")))

system/hardware/hw.py

@@ -33,3 +33,24 @@ class Paths:
     if os.environ.get('COMMA_CACHE', False):
       return os.environ['COMMA_CACHE']
     return "/tmp/comma_download_cache" + os.environ.get("OPENPILOT_PREFIX", "") + "/"
+
+  @staticmethod
+  def persist_root() -> str:
+    if PC:
+      return os.path.join(Paths.comma_home(), "persist")
+    else:
+      return "/persist/"
+
+  @staticmethod
+  def stats_root() -> str:
+    if PC:
+      return str(Path(Paths.comma_home()) / "stats")
+    else:
+      return "/data/stats/"
+
+  @staticmethod
+  def config_root() -> str:
+    if PC:
+      return Paths.comma_home()
+    else:
+      return "/tmp/.comma"

system/loggerd/config.py

@@ -1,6 +1,4 @@
 import os
-from pathlib import Path
-from openpilot.system.hardware import PC
 from openpilot.system.hardware.hw import Paths
 
 
@@ -9,10 +7,6 @@ SEGMENT_LENGTH = 60
 
 STATS_DIR_FILE_LIMIT = 10000
 STATS_SOCKET = "ipc:///tmp/stats"
-if PC:
-  STATS_DIR = str(Path.home() / ".comma" / "stats")
-else:
-  STATS_DIR = "/data/stats/"
 STATS_FLUSH_TIME_S = 60
 
 def get_available_percent(default=None):

tools/lib/auth_config.py

@@ -1,21 +1,16 @@
 import json
 import os
 from openpilot.common.file_helpers import mkdirs_exists_ok
-from openpilot.system.hardware import PC
+from openpilot.system.hardware.hw import Paths
 
 
 class MissingAuthConfigError(Exception):
   pass
 
 
-if PC:
-  CONFIG_DIR = os.path.expanduser('~/.comma')
-else:
-  CONFIG_DIR = "/tmp/.comma"
-
 def get_token():
   try:
-    with open(os.path.join(CONFIG_DIR, 'auth.json')) as f:
+    with open(os.path.join(Paths.config_root(), 'auth.json')) as f:
       auth = json.load(f)
       return auth['access_token']
   except Exception:
@@ -23,13 +18,13 @@ def get_token():
 
 
 def set_token(token):
-  mkdirs_exists_ok(CONFIG_DIR)
-  with open(os.path.join(CONFIG_DIR, 'auth.json'), 'w') as f:
+  mkdirs_exists_ok(Paths.config_root())
+  with open(os.path.join(Paths.config_root(), 'auth.json'), 'w') as f:
     json.dump({'access_token': token}, f)
 
 
 def clear_token():
   try:
-    os.unlink(os.path.join(CONFIG_DIR, 'auth.json'))
+    os.unlink(os.path.join(Paths.config_root(), 'auth.json'))
   except FileNotFoundError:
     pass