test(logs): add it tests for injected log messages

src/deep/api/deep.py

@@ -14,9 +14,9 @@
 #      along with this program.  If not, see <https://www.gnu.org/licenses/>.from typing import Dict, List
 
 """The main services for Deep."""
-
 from typing import Dict, List
 
+import deep.logging
 from deep.api.plugin import load_plugins
 from deep.api.resource import Resource
 from deep.config import ConfigService
@@ -70,6 +70,7 @@ def shutdown(self):
         self.trigger_handler.shutdown()
         self.task_handler.flush()
         self.poll.shutdown()
+        deep.logging.info("Deep is shutdown.")
         self.started = False
 
     def register_tracepoint(self, path: str, line: int, args: Dict[str, str] = None,

tests/it_tests/it_utils.py

@@ -99,10 +99,10 @@ def await_poll(self):
         with self.__poll_service.condition:
             self.__poll_service.condition.wait(10)
 
-    def await_snapshot(self):
+    def await_snapshot(self, timeout=10):
         """Await for the next snapshot to be received. Time out after 10 seconds."""
         with self.__snapshot_service.condition:
-            self.__snapshot_service.condition.wait(10)
+            self.__snapshot_service.condition.wait(timeout)
         return self.__snapshot_service.snapshot
 
 

tests/it_tests/test_it_basic.py

@@ -14,18 +14,23 @@
 #      along with this program.  If not, see <https://www.gnu.org/licenses/>.
 
 """A collection of IT tests that simulate user usage of Deep."""
-
+import logging
 import unittest
 
 import deep
 import it_tests
+from deep.api.tracepoint.constants import LOG_MSG, SNAPSHOT, NO_COLLECT
 from it_tests.it_utils import start_server, MockServer
 from it_tests.test_target import BPTargetTest
 from test_utils import find_var_in_snap_by_name, find_var_in_snap_by_path
 
 
 class BasicITTest(unittest.TestCase):
-    """These tests are intended to simulate a real user installing and using Deep."""
+    """
+    These tests are intended to simulate a real user installing and using Deep.
+
+    These tests cannot be debugged as we use the same approach.
+    """
 
     def test_simple_it(self):
         server: MockServer
@@ -56,3 +61,63 @@ def test_simple_it(self):
 
             var_not_on_super = find_var_in_snap_by_path(snapshot, "self._BPSuperClass__not_on_super")
             self.assertIsNotNone(var_not_on_super)
+
+    def test_log_action(self):
+        server: MockServer
+
+        with start_server() as server:
+            server.add_tp("test_target.py", 40, {LOG_MSG: 'test log {name}'}, [], [])
+            _deep = deep.start(server.config())
+            server.await_poll()
+            with self.assertLogs('deep', level=logging.INFO) as logs:
+                test = BPTargetTest("name", 123)
+                _ = test.name
+                snapshot = server.await_snapshot()
+                _deep.shutdown()
+
+                self.assertIsNotNone(snapshot)
+
+                self.assertEqual("[deep] test log name", snapshot.log_msg)
+                self.assertIn("[deep] test log name", logs.output[0])
+
+                frames = snapshot.frames
+                self.assertEqual(it_tests.test_target.__file__, frames[0].file_name)
+                self.assertEqual("/it_tests/test_target.py", frames[0].short_path)
+                self.assertEqual(40, frames[0].line_number)
+                self.assertEqual(4, len(frames[0].variables))
+                self.assertEqual(6, len(snapshot.var_lookup))
+
+                var_name = find_var_in_snap_by_name(snapshot, "name")
+                self.assertIsNotNone(var_name)
+
+                var_i = find_var_in_snap_by_name(snapshot, "i")
+                self.assertIsNotNone(var_i)
+
+                var_self = find_var_in_snap_by_name(snapshot, "self")
+                self.assertIsNotNone(var_self)
+
+                var_not_on_super = find_var_in_snap_by_path(snapshot, "self._BPSuperClass__not_on_super")
+                self.assertIsNotNone(var_not_on_super)
+
+    def test_log_only_action(self):
+        """
+        For some reason the log message doesn't appear in the output, but it is logged.
+
+        This can be verified by changing the assertion, or looking at the tracepoint handler output.
+        """
+        server: MockServer
+
+        with start_server() as server:
+            server.add_tp("test_target.py", 40, {LOG_MSG: 'test log {name}', SNAPSHOT: NO_COLLECT}, [], [])
+            _deep = deep.start(server.config())
+            server.await_poll()
+            with self.assertLogs('deep', level=logging.INFO) as logs:
+                test = BPTargetTest("name", 123)
+                _ = test.name
+                # we do not want a snapshot, but we have to await to see if one is sent. So we just wait 5 seconds,
+                # as it should not take this long for a snapshot to be sent if it was triggered.
+                snapshot = server.await_snapshot(timeout=5)
+
+                self.assertIsNone(snapshot)
+                self.assertIn("[deep] test log name", logs.output[0])
+            _deep.shutdown()