system independent path separators; pytest for windows fixes

Author: shukari

meshtastic/analysis/__main__.py

@@ -2,6 +2,7 @@
 
 import argparse
 import logging
+from pathlib import Path
 from typing import cast, List
 
 import dash_bootstrap_components as dbc  # type: ignore[import-untyped]
@@ -190,7 +191,7 @@ def main():
     parser = create_argparser()
     args = parser.parse_args()
     if not args.slog:
-        args.slog = f"{root_dir()}/latest"
+        args.slog = str(Path(root_dir(), "latest"))
 
     app = create_dash(slog_path=args.slog)
     port = 8051

meshtastic/slog/slog.py

@@ -3,13 +3,13 @@
 import atexit
 import io
 import logging
-import os
 import re
 import threading
 import time
 from dataclasses import dataclass
 from datetime import datetime
 from functools import reduce
+from pathlib import Path
 from typing import Optional, List, Tuple
 
 import parse  # type: ignore[import-untyped]
@@ -22,16 +22,17 @@
 
 from .arrow import FeatherWriter
 
+logger = logging.getLogger(__name__)
 
 def root_dir() -> str:
     """Return the root directory for slog files."""
 
     app_name = "meshtastic"
     app_author = "meshtastic"
     app_dir = platformdirs.user_data_dir(app_name, app_author)
-    dir_name = f"{app_dir}/slogs"
-    os.makedirs(dir_name, exist_ok=True)
-    return dir_name
+    dir_name = Path(app_dir, "slogs")
+    dir_name.mkdir(exist_ok=True)
+    return str(dir_name)
 
 
 @dataclass(init=False)
@@ -197,7 +198,7 @@ def _onLogMessage(self, line: str) -> None:
         if m:
             src = m.group(1)
             args = m.group(2)
-            logging.debug(f"SLog {src}, args: {args}")
+            logger.debug(f"SLog {src}, args: {args}")
 
             d = log_defs.get(src)
             if d:
@@ -219,9 +220,9 @@ def _onLogMessage(self, line: str) -> None:
                             # If the last field is an empty string, remove it
                             del di[last_field[0]]
                 else:
-                    logging.warning(f"Failed to parse slog {line} with {d.format}")
+                    logger.warning(f"Failed to parse slog {line} with {d.format}")
             else:
-                logging.warning(f"Unknown Structured Log: {line}")
+                logger.warning(f"Unknown Structured Log: {line}")
 
         # Store our structured log record
         if di or self.include_raw:
@@ -256,18 +257,22 @@ def __init__(
 
         if not dir_name:
             app_dir = root_dir()
-            dir_name = f"{app_dir}/{datetime.now().strftime('%Y%m%d-%H%M%S')}"
-            os.makedirs(dir_name, exist_ok=True)
+            dir_name = Path(app_dir, datetime.now().strftime('%Y%m%d-%H%M%S'))
+            dir_name.mkdir(exist_ok=True)
 
             # Also make a 'latest' directory that always points to the most recent logs
+            latest_dir = Path(app_dir, "latest")
+            latest_dir.unlink(missing_ok=True)
+
             # symlink might fail on some platforms, if it does fail silently
-            if os.path.exists(f"{app_dir}/latest"):
-                os.unlink(f"{app_dir}/latest")
-            os.symlink(dir_name, f"{app_dir}/latest", target_is_directory=True)
+            try:
+                latest_dir.symlink_to(dir_name, target_is_directory=True)
+            except OSError as e:
+                pass
 
         self.dir_name = dir_name
 
-        logging.info(f"Writing slogs to {dir_name}")
+        logger.info(f"Writing slogs to {dir_name}")
 
         self.power_logger: Optional[PowerLogger] = (
             None
@@ -286,7 +291,7 @@ def close(self) -> None:
         """Close the log set."""
 
         if self.slog_logger:
-            logging.info(f"Closing slogs in {self.dir_name}")
+            logger.info(f"Closing slogs in {self.dir_name}")
             atexit.unregister(
                 self.atexit_handler
             )  # docs say it will silently ignore if not found

meshtastic/tests/test_main.py

@@ -2683,7 +2683,6 @@ def test_tunnel_no_args(capsys):
     assert re.search(r"usage: ", err, re.MULTILINE)
 
 
-@pytest.mark.skipif(sys.platform == "win32", reason="Linux is forced in test and no termios")
 @pytest.mark.unit
 @pytest.mark.usefixtures("reset_mt_config")
 @patch("meshtastic.util.findPorts", return_value=[])
@@ -2707,7 +2706,6 @@ def test_tunnel_tunnel_arg_with_no_devices(mock_platform_system, caplog, capsys)
         assert err == ""
 
 
-@pytest.mark.skipif(sys.platform == "win32", reason="Linux is forced in test and no termios")
 @pytest.mark.unit
 @pytest.mark.usefixtures("reset_mt_config")
 @patch("meshtastic.util.findPorts", return_value=[])

meshtastic/tests/test_serial_interface.py

@@ -1,4 +1,5 @@
 """Meshtastic unit tests for serial_interface.py"""
+import platform
 # pylint: disable=R0917
 
 import re
@@ -28,9 +29,13 @@ def test_SerialInterface_single_port(
     iface.close()
     mocked_findPorts.assert_called()
     mocked_serial.assert_called()
-    mocked_open.assert_called()
-    mock_get.assert_called()
-    mock_set.assert_called()
+
+    # doesn't get called in SerialInterface.__init__ on windows
+    if platform.system() != "Windows":
+        mocked_open.assert_called()
+        mock_get.assert_called()
+        mock_set.assert_called()
+
     mock_sleep.assert_called()
     out, err = capsys.readouterr()
     assert re.search(r"Nodes in mesh", out, re.MULTILINE)