diff --git a/blueman/Functions.py b/blueman/Functions.py index 2e48e8aab..f4d70ded9 100644 --- a/blueman/Functions.py +++ b/blueman/Functions.py @@ -209,7 +209,7 @@ def create_menuitem( def have(t: str) -> pathlib.Path | None: pathstr = os.environ['PATH'] + ':/sbin:/usr/sbin' for path in [pathlib.Path(p, t) for p in pathstr.split(":")]: - if path.exists() and os.access(path, os.EX_OK): + if path.exists() and os.access(path, os.X_OK): return path return None @@ -278,14 +278,15 @@ def create_parser( return parser -def open_rfcomm(file: str, mode: int) -> int: +def open_rfcomm(file: str, rw: bool = False) -> int: + mode = os.O_RDWR if rw else os.O_RDONLY try: return os.open(file, mode | os.O_EXCL | os.O_NONBLOCK | os.O_NOCTTY) except OSError as err: if err.errno == errno.EBUSY: logging.warning(f"{file} is busy, delaying 2 seconds") sleep(2) - return open_rfcomm(file, mode) + return open_rfcomm(file, rw) else: raise diff --git a/blueman/main/PPPConnection.py b/blueman/main/PPPConnection.py index 566c6dae8..c432b809c 100644 --- a/blueman/main/PPPConnection.py +++ b/blueman/main/PPPConnection.py @@ -125,7 +125,7 @@ def send_commands(self, start: int = 0) -> None: def connect_rfcomm(self) -> None: - self.file = open_rfcomm(self.port, os.O_RDWR) + self.file = open_rfcomm(self.port, rw=True) tty.setraw(self.file) diff --git a/test/Makefile.am b/test/Makefile.am index 0333b925e..4b10d3d8e 100644 --- a/test/Makefile.am +++ b/test/Makefile.am @@ -10,5 +10,6 @@ SUBDIRS = \ EXTRA_DIST = \ __init__.py \ - test_imports.py \ - test_gobject.py + test_functions.py \ + test_gobject.py \ + test_imports.py diff --git a/test/test_functions.py b/test/test_functions.py new file mode 100644 index 000000000..50c81605c --- /dev/null +++ b/test/test_functions.py @@ -0,0 +1,159 @@ +import errno +import os + +from pathlib import Path +from tempfile import TemporaryDirectory, TemporaryFile +from unittest import TestCase +from unittest.mock import patch + +import blueman.Functions as Functions + + +class TestAdapterPathToName(TestCase): + def test_matches_hci(self): + self.assertEqual(Functions.adapter_path_to_name("/path/hci9"), "hci9") + + def test_returns_none_for_empty_or_none(self): + self.assertIsNone(Functions.adapter_path_to_name(None)) + self.assertIsNone(Functions.adapter_path_to_name("")) + + def test_no_match(self): + self.assertIsNone(Functions.adapter_path_to_name("regular/path")) + + +class TestE_(TestCase): + def test_string_input(self): + msg, tb = Functions.e_("error: some message") + self.assertEqual(msg, "some message") + self.assertIsNone(tb) + + def test_simple_string(self): + msg, tb = Functions.e_("simple error") + self.assertEqual(msg, "simple error") + self.assertIsNone(tb) + + def test_exception_input(self): + exc = ValueError("test error") + msg, tb = Functions.e_(exc) + self.assertEqual(msg, "test error") + self.assertIsNotNone(tb) + + +class TestFormatBytes(TestCase): + def test_bytes(self): + val, suffix = Functions.format_bytes(500) + self.assertEqual(val, 500.0) + self.assertEqual(suffix, "B") + + def test_kb(self): + val, suffix = Functions.format_bytes(2048) + self.assertAlmostEqual(val, 2.0) + self.assertEqual(suffix, "KB") + + def test_mb(self): + val, suffix = Functions.format_bytes(1024 * 1024 + 1024) + self.assertGreater(val, 1.0) + self.assertEqual(suffix, "MB") + + def test_gb(self): + val, suffix = Functions.format_bytes(1024 ** 3 + 1024 ** 2) + self.assertGreater(val, 1.0) + self.assertEqual(suffix, "GB") + + +class TestHave(TestCase): + def setUp(self): + self.env_path = {'PATH': '/usr/bin:/sbin'} + + def test_path_exists(self): + with patch.dict(os.environ, self.env_path), \ + patch.object(Path, "exists", return_value=True), \ + patch.object(os, "access", return_value=True): + + result = Functions.have("executable") + self.assertIsNotNone(result) + self.assertEqual(result, Path("/usr/bin/executable")) + + def test_path_does_not_exist(self): + with patch.dict(os.environ, self.env_path), \ + patch.object(Path, "exists", return_value=False): + + result = Functions.have("executable") + self.assertIsNone(result) + + def test_path_not_executable(self): + with patch.dict(os.environ, self.env_path), \ + patch.object(Path, "exists", return_value=True), \ + patch.object(os, "access", return_value=False): + + result = Functions.have("executable") + self.assertIsNone(result) + + +class TestBmexit(TestCase): + def test_exit_with_message(self): + with self.assertRaises(SystemExit) as ctx: + Functions.bmexit("test exit") + self.assertEqual(ctx.exception.code, "test exit") + + def test_exit_none(self): + with self.assertRaises(SystemExit) as ctx: + Functions.bmexit(None) + self.assertIsNone(ctx.exception.code) + + +class TestPluginNames(TestCase): + def setUp(self) -> None: + self.tempdir = TemporaryDirectory() + + def tearDown(self) -> None: + self.tempdir.cleanup() + + def test_plugins(self): + for name in ("plugin_a.py", "plugin_b.py", "__init__.py"): + Path(self.tempdir.name).joinpath(name).touch() + + result = Functions.plugin_names(Path(self.tempdir.name) / "__init__.py") + # names can be in any order + self.assertIn("plugin_a", result) + self.assertIn("plugin_b", result) + self.assertNotIn("__init__", result) + self.assertCountEqual(result, ["plugin_a", "plugin_b"]) + + def test_no_plugins(self): + result = Functions.plugin_names(Path(self.tempdir.name) / "__init__.py") + self.assertEqual(result, []) + + +class TestOpenRfcomm(TestCase): + def setUp(self): + self.fake_rfcomm = TemporaryFile() + + def tearDown(self): + self.fake_rfcomm.close() + + @patch("blueman.Functions.sleep") + def test_open_success(self, mock_sleep): + with patch('os.open', return_value=42) as mock_open: + fd = Functions.open_rfcomm(self.fake_rfcomm.name, rw=True) + self.assertEqual(fd, 42) + mock_open.assert_called_once() + + @patch("blueman.Functions.sleep") + def test_open_busy_then_success(self, mock_sleep): + with patch('os.open') as mock_open: + # First call raises EBUSY, second succeeds + mock_open.side_effect = [OSError(errno.EBUSY, "file busy"), 42] + + fd = Functions.open_rfcomm(self.fake_rfcomm.name) + + self.assertEqual(fd, 42) + self.assertEqual(mock_sleep.call_count, 1) + mock_sleep.assert_called_with(2) + + @patch("blueman.Functions.sleep") + def test_open_other_error(self, mock_sleep): + with patch('os.open', side_effect=OSError(errno.ENOENT, "no such file")) as mock_open: + with self.assertRaises(OSError) as context: + Functions.open_rfcomm(self.fake_rfcomm.name) + self.assertEqual(context.exception.errno, errno.ENOENT)