took the tracetools_read and tracetools_trace from upstream (rolling)

tracetools_trace/test/test_flake8.py

@@ -12,12 +12,14 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-from ament_flake8.main import main
+from ament_flake8.main import main_with_errors
 import pytest
 
 
 @pytest.mark.flake8
 @pytest.mark.linter
 def test_flake8():
-    rc = main(argv=[])
-    assert rc == 0, 'Found errors'
+    rc, errors = main_with_errors(argv=[])
+    assert rc == 0, \
+        'Found %d code style errors / warnings:\n' % len(errors) + \
+        '\n'.join(errors)

tracetools_trace/test/tracetools_trace/test_lttng_tracing.py

@@ -0,0 +1,167 @@
+# Copyright 2021 Christophe Bedard
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import os
+import tempfile
+import unittest
+from unittest import mock
+
+from packaging.version import Version
+from tracetools_trace.tools.lttng import is_lttng_installed
+
+
+@unittest.skipIf(not is_lttng_installed(), 'LTTng is required')
+class TestLttngTracing(unittest.TestCase):
+
+    def __init__(self, *args) -> None:
+        super().__init__(
+            *args,
+        )
+
+    def test_is_lttng_installed(self):
+        # Different OS
+        with mock.patch('platform.system', return_value='Windows'):
+            self.assertFalse(is_lttng_installed())
+
+        # lttng-ctl or version not found
+        with mock.patch('tracetools_trace.tools.lttng.get_lttng_version', return_value=None):
+            self.assertFalse(is_lttng_installed())
+
+        # Minimum version requirement
+        with mock.patch(
+            'tracetools_trace.tools.lttng.get_lttng_version',
+            return_value=Version('1.2.3'),
+        ):
+            self.assertFalse(is_lttng_installed(minimum_version='1.2.4'))
+            self.assertTrue(is_lttng_installed(minimum_version='1.2.3'))
+            self.assertTrue(is_lttng_installed())
+
+    def test_lttng_not_installed(self):
+        from tracetools_trace.tools.lttng import lttng_init
+        with mock.patch('tracetools_trace.tools.lttng.is_lttng_installed', return_value=False):
+            self.assertIsNone(lttng_init(session_name='test-session', base_path='/tmp'))
+
+    def test_no_kernel_tracer(self):
+        from tracetools_trace.tools.lttng_impl import setup
+        with (
+            mock.patch(
+                'tracetools_trace.tools.lttng_impl.is_session_daemon_not_alive',
+                return_value=False,
+            ),
+            mock.patch(
+                'tracetools_trace.tools.lttng_impl.is_session_daemon_unreachable',
+                return_value=False,
+            ),
+            mock.patch(
+                'tracetools_trace.tools.lttng_impl.is_kernel_tracer_available',
+                return_value=False,
+            ),
+        ):
+            with self.assertRaises(RuntimeError):
+                setup(
+                    session_name='test-session',
+                    base_path='/tmp',
+                    kernel_events=['sched_switch'],
+                )
+            with self.assertRaises(RuntimeError):
+                setup(
+                    session_name='test-session',
+                    base_path='/tmp',
+                    syscalls=['open'],
+                )
+
+    def test_get_lttng_home(self):
+        from tracetools_trace.tools.lttng_impl import get_lttng_home
+        # Uses $LTTNG_HOME if set
+        environ = {'LTTNG_HOME': 'the_lttng_home', 'HOME': 'the_home'}
+        with mock.patch.dict(os.environ, environ, clear=True):
+            self.assertEqual('the_lttng_home', get_lttng_home())
+        # Defaults to $HOME if LTTNG_HOME is unset
+        environ = {'HOME': 'the_home'}
+        with mock.patch.dict(os.environ, environ, clear=True):
+            self.assertEqual('the_home', get_lttng_home())
+        # Returns `None` otherwise
+        with mock.patch.dict(os.environ, {}, clear=True):
+            self.assertIsNone(get_lttng_home())
+
+    def test_get_session_daemon_pid(self):
+        from tracetools_trace.tools.lttng_impl import get_session_daemon_pid
+        # No PID if there is no LTTng home
+        with mock.patch('tracetools_trace.tools.lttng_impl.get_lttng_home', return_value=None):
+            self.assertIsNone(get_session_daemon_pid())
+        # No PID if the PID file doesn't exist
+        with mock.patch(
+            'tracetools_trace.tools.lttng_impl.get_lttng_home',
+            return_value=os.path.join(tempfile.gettempdir(), 'doesnt_exist'),
+        ):
+            self.assertIsNone(get_session_daemon_pid())
+        # PID file exists...
+        with (
+            mock.patch(
+                'tracetools_trace.tools.lttng_impl.get_lttng_home',
+                return_value='some_non-None_value',
+            ),
+            mock.patch('os.path.isfile', return_value=True),
+        ):
+            # ...but is not a valid int
+            with mock.patch('builtins.open', mock.mock_open(read_data='')):
+                self.assertIsNone(get_session_daemon_pid())
+            with mock.patch('builtins.open', mock.mock_open(read_data='abc')):
+                self.assertIsNone(get_session_daemon_pid())
+            # ...and has a valid int when stripped
+            with mock.patch('builtins.open', mock.mock_open(read_data='123\n')):
+                self.assertEqual(123, get_session_daemon_pid())
+
+    def test_is_session_daemon_unreachable(self):
+        from tracetools_trace.tools.lttng_impl import is_session_daemon_unreachable
+        # All good if we can't get the session daemon PID
+        with mock.patch(
+            'tracetools_trace.tools.lttng_impl.get_session_daemon_pid',
+            return_value=None,
+        ):
+            self.assertFalse(is_session_daemon_unreachable())
+        # If we can get the session daemon PID...
+        with mock.patch(
+            'tracetools_trace.tools.lttng_impl.get_session_daemon_pid',
+            return_value=123,
+        ):
+            # Unreachable if we can't find the process with the PID
+            with mock.patch('subprocess.run') as patched_subprocess_run:
+                patched_subprocess_run.return_value.returncode = 1
+                self.assertTrue(is_session_daemon_unreachable())
+            # Unreachable if we can find the process with the PID, but it is not a session daemon
+            with mock.patch('subprocess.run') as patched_subprocess_run:
+                patched_subprocess_run.return_value.returncode = 0
+                patched_subprocess_run.return_value.stdout = 'some-random-command\n'
+                self.assertTrue(is_session_daemon_unreachable())
+            # All good if we can find the process with the PID and it is a session daemon
+            with mock.patch('subprocess.run') as patched_subprocess_run:
+                patched_subprocess_run.return_value.returncode = 0
+                patched_subprocess_run.return_value.stdout = 'lttng-sessiond\n'
+                self.assertFalse(is_session_daemon_unreachable())
+
+    def test_unreachable_session_daemon(self):
+        from tracetools_trace.tools.lttng_impl import setup
+        with (
+            mock.patch(
+                'tracetools_trace.tools.lttng_impl.is_session_daemon_not_alive',
+                return_value=False,
+            ),
+            mock.patch(
+                'tracetools_trace.tools.lttng_impl.is_session_daemon_unreachable',
+                return_value=True,
+            ),
+        ):
+            with self.assertRaises(RuntimeError):
+                setup(session_name='test-session', base_path='/tmp')

tracetools_trace/test/tracetools_trace/test_names.py

@@ -0,0 +1,38 @@
+# Copyright 2021 Christophe Bedard
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import unittest
+
+from tracetools_trace.tools import names
+from tracetools_trace.tools import tracepoints
+
+
+class TestNames(unittest.TestCase):
+
+    def __init__(self, *args) -> None:
+        super().__init__(
+            *args,
+        )
+
+    def test_tracepoint_names(self) -> None:
+        # Make sure the list of default ROS events contains exactly all the
+        # tracepoints defined as constants, otherwise something might have been forgotten
+        tp_constant_names = {name for name in dir(tracepoints) if not name.startswith('__')}
+        tp_names = {getattr(tracepoints, name) for name in tp_constant_names}
+        self.assertTrue(all(name.startswith('ros2:') for name in tp_names), tp_names)
+        self.assertSetEqual(set(names.DEFAULT_EVENTS_ROS), tp_names)
+
+
+if __name__ == '__main__':
+    unittest.main()

tracetools_trace/test/tracetools_trace/test_trace_directory.py

@@ -0,0 +1,71 @@
+# Copyright 2020 Christophe Bedard
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Tests for the trace directory logic."""
+
+import os
+import pathlib
+
+from tracetools_trace.tools.path import get_tracing_directory
+
+
+def test_get_trace_directory():
+    os.environ.pop('ROS_TRACE_DIR', None)
+    os.environ.pop('ROS_HOME', None)
+    home = pathlib.Path.home()
+    assert str(home)
+
+    # Default case without ROS_TRACE_DIR or ROS_HOME being set (but with HOME)
+    default_dir = str(home / '.ros/tracing')
+    assert get_tracing_directory() == default_dir
+
+    # Use $ROS_TRACE_DIR if it is set
+    my_trace_dir_raw = '/my/ROS_TRACE_DIR'
+    my_trace_dir = str(pathlib.Path(my_trace_dir_raw))
+    os.environ['ROS_TRACE_DIR'] = my_trace_dir
+    assert get_tracing_directory() == my_trace_dir
+    # Make sure it converts path separators when necessary
+    os.environ['ROS_TRACE_DIR'] = my_trace_dir_raw
+    assert get_tracing_directory() == my_trace_dir
+    # Setting ROS_HOME won't change anything since ROS_TRACE_DIR is used first
+    os.environ['ROS_HOME'] = '/this/wont/be/used'
+    assert get_tracing_directory() == my_trace_dir
+    os.environ.pop('ROS_HOME', None)
+    # Empty is considered unset
+    os.environ['ROS_TRACE_DIR'] = ''
+    assert get_tracing_directory() == default_dir
+    # Make sure '~' is expanded to the home directory
+    os.environ['ROS_TRACE_DIR'] = '~/tracedir'
+    assert get_tracing_directory() == str(home / 'tracedir')
+
+    os.environ.pop('ROS_TRACE_DIR', None)
+
+    # Without ROS_TRACE_DIR, use $ROS_HOME/tracing
+    fake_ros_home = home / '.fakeroshome'
+    fake_ros_home_trace_dir = str(fake_ros_home / 'tracing')
+    os.environ['ROS_HOME'] = str(fake_ros_home)
+    assert get_tracing_directory() == fake_ros_home_trace_dir
+    # Make sure it converts path separators when necessary
+    my_ros_home_raw = '/my/ros/home'
+    my_ros_home_trace_dir = str(pathlib.Path(my_ros_home_raw) / 'tracing')
+    os.environ['ROS_HOME'] = my_ros_home_raw
+    assert get_tracing_directory() == my_ros_home_trace_dir
+    # Empty is considered unset
+    os.environ['ROS_HOME'] = ''
+    assert get_tracing_directory() == default_dir
+    # Make sure '~' is expanded to the home directory
+    os.environ['ROS_HOME'] = '~/.fakeroshome'
+    assert get_tracing_directory() == fake_ros_home_trace_dir
+
+    os.environ.pop('ROS_HOME', None)