diff --git a/src/deep/api/deep.py b/src/deep/api/deep.py
index 5a8f2db..20f0994 100644
--- a/src/deep/api/deep.py
+++ b/src/deep/api/deep.py
@@ -14,9 +14,9 @@
# along with this program. If not, see .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,
diff --git a/tests/it_tests/it_utils.py b/tests/it_tests/it_utils.py
index 70b075b..dc08047 100644
--- a/tests/it_tests/it_utils.py
+++ b/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
diff --git a/tests/it_tests/test_it_basic.py b/tests/it_tests/test_it_basic.py
index bb381b0..91e1e6a 100644
--- a/tests/it_tests/test_it_basic.py
+++ b/tests/it_tests/test_it_basic.py
@@ -14,18 +14,23 @@
# along with this program. If not, see .
"""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()