Merge pull request getappmap#344 from getappmap/props_20240626

feat: instrument properties

Author: apotterri

_appmap/event.py

@@ -176,7 +176,7 @@ def to_dict(self, value):
 
 class CallEvent(Event):
     # pylint: disable=method-cache-max-size-none
-    __slots__ = ["_fn", "_fqfn", "static", "receiver", "parameters", "labels"]
+    __slots__ = ["_fn", "_fqfn", "static", "receiver", "parameters", "labels", "auxtype"]
 
     @staticmethod
     def make(fn, fntype):
@@ -283,7 +283,10 @@ def defined_class(self):
     @property
     @lru_cache(maxsize=None)
     def method_id(self):
-        return self._fqfn.fqfn[1]
+        ret = self._fqfn.fqfn[1]
+        if self.auxtype is not None:
+            ret = f"{ret} ({self.auxtype})"
+        return ret
 
     @property
     @lru_cache(maxsize=None)
@@ -319,6 +322,13 @@ def __init__(self, fn, fntype, parameters, labels):
             parameters = parameters[1:]
         self.parameters = parameters
         self.labels = labels
+        self.auxtype = None
+        if fntype & FnType.GET:
+            self.auxtype = "get"
+        elif fntype & FnType.SET:
+            self.auxtype = "set"
+        elif fntype & FnType.DEL:
+            self.auxtype = "del"
 
     def to_dict(self, attrs=None):
         ret = super().to_dict()  # get the attrs defined in __slots__

_appmap/importer.py

@@ -37,22 +37,25 @@ def __new__(cls, clazz):
 class FilterableFn(
     namedtuple(
         "FilterableFn",
-        Filterable._fields + ("static_fn",),
+        Filterable._fields + ("static_fn", "auxtype"),
     )
 ):
     __slots__ = ()
 
-    def __new__(cls, scope, fn, static_fn):
+    def __new__(cls, scope, fn, static_fn, auxtype=None):
         fqname = "%s.%s" % (scope.fqname, fn.__name__)
-        self = super(FilterableFn, cls).__new__(cls, scope.scope, fqname, fn, static_fn)
+        self = super(FilterableFn, cls).__new__(cls, scope.scope, fqname, fn, static_fn, auxtype)
         return self
 
     @property
     def fntype(self):
         if self.scope == Scope.MODULE:
             return FnType.MODULE
 
-        return FnType.classify(self.static_fn)
+        ret = FnType.classify(self.static_fn)
+        if self.auxtype is not None:
+            ret |= self.auxtype
+        return ret
 
 
 class Filter(ABC):  # pylint: disable=too-few-public-methods
@@ -122,19 +125,31 @@ def is_member_func(m):
     # instead iterate over dir(cls), we would see functions from
     # superclasses, too. Those don't need to be instrumented here,
     # they'll get taken care of when the superclass is imported.
-    ret = []
+    functions = []
+    properties = {}
     modname = cls.__module__ if hasattr(cls, "__module__") else cls.__name__
     for key in cls.__dict__:
         if key.startswith("__"):
             continue
         static_value = inspect.getattr_static(cls, key)
-        if not is_member_func(static_value):
-            continue
-        value = getattr(cls, key)
-        if value.__module__ != modname:
-            continue
-        ret.append((key, static_value, value))
-    return ret
+        if isinstance(static_value, property):
+            properties[key] = (
+                static_value,
+                {
+                    "fget": (static_value.fget, FnType.GET),
+                    "fset": (static_value.fset, FnType.SET),
+                    "fdel": (static_value.fdel, FnType.DEL),
+                },
+            )
+        else:
+            if not is_member_func(static_value):
+                continue
+            value = getattr(cls, key)
+            if value.__module__ != modname:
+                continue
+            functions.append((key, static_value, value))
+
+    return (functions, properties)
 
 
 class Importer:
@@ -177,14 +192,28 @@ def do_import(cls, *args, **kwargs):
 
         def instrument_functions(filterable, selected_functions=None):
             logger.trace("  looking for members of %s", filterable.obj)
-            functions = get_members(filterable.obj)
+            functions, properties = get_members(filterable.obj)
             logger.trace("  functions %s", functions)
 
             for fn_name, static_fn, fn in functions:
                 filterableFn = FilterableFn(filterable, fn, static_fn)
                 new_fn = cls.instrument_function(fn_name, filterableFn, selected_functions)
                 if new_fn != fn:
                     wrapt.wrap_function_wrapper(filterable.obj, fn_name, new_fn)
+            # Now that we've instrumented all the functions, go through the properties and update
+            # them
+            for prop_name, (prop, prop_fns) in properties.items():
+                instrumented_fns = {}
+                for k, (fn, auxtype) in prop_fns.items():
+                    if fn is None:
+                        continue
+                    filterableFn = FilterableFn(filterable, fn, fn, auxtype)
+                    new_fn = cls.instrument_function(fn.__name__, filterableFn, selected_functions)
+                    if new_fn != fn:
+                        new_fn = wrapt.FunctionWrapper(fn, new_fn)
+                    instrumented_fns[k] = new_fn
+                instrumented_fns["doc"] = prop.__doc__
+                setattr(filterable.obj, prop_name, property(**instrumented_fns))
 
         # Import Config here, to avoid circular top-level imports.
         from .configuration import Config  # pylint: disable=import-outside-toplevel

_appmap/test/data/example_class.py

@@ -113,6 +113,53 @@ def with_comment(self):
     def return_self(self):
         return self
 
+    def __init__(self):
+        self._read_only = "read only"
+        self._fully_accessible = "fully accessible"
+        self._undecorated = "undecorated"
+
+    @property
+    def read_only(self):
+        """Read-only"""
+        return self._read_only
+
+    @property
+    def fully_accessible(self):
+        """Fully-accessible"""
+        return self._fully_accessible
+
+    @fully_accessible.setter
+    def fully_accessible(self, v):
+        self._fully_accessible = v
+
+    @fully_accessible.deleter
+    def fully_accessible(self):
+        del self._fully_accessible
+
+    def get_undecorated(self):
+        return self._undecorated
+
+    def set_undecorated(self, value):
+        self._undecorated = value
+
+    def delete_undecorated(self):
+        del self._undecorated
+
+    undecorated_property = property(get_undecorated, set_undecorated, delete_undecorated)
+
+    def set_write_only(self, v):
+        self._write_only = v
+
+    def del_write_only(self):
+        del self._write_only
+
+    write_only = property(None, set_write_only, del_write_only, "Write-only")
+
 
 def modfunc():
     return "Hello world!"
+
+if __name__ == "__main__":
+    ec = ExampleClass()
+    ec.fully_accessible = "updated"
+    print(ec.fully_accessible)

_appmap/test/test_params.py

@@ -1,4 +1,4 @@
-"""Tests for the function parameter handling"""
+"""Tests for function parameter handling"""
 
 # pylint: disable=missing-function-docstring
 

_appmap/test/test_properties.py

@@ -0,0 +1,105 @@
+"""Tests for methods decorated with @property"""
+
+# pyright: reportMissingImports=false
+# pylint: disable=import-error,import-outside-toplevel
+import pytest
+from _appmap.test.helpers import DictIncluding
+
+pytestmark = [
+    pytest.mark.appmap_enabled,
+]
+
+
+@pytest.fixture(autouse=True)
+def setup(with_data_dir):  # pylint: disable=unused-argument
+    # with_data_dir sets up sys.path so example_class can be imported
+    pass
+
+
+def test_getter_instrumented(events):
+    from example_class import ExampleClass
+
+    ec = ExampleClass()
+
+    actual = ExampleClass.read_only.__doc__
+    assert actual == "Read-only"
+
+    assert ec.read_only == "read only"
+
+    with pytest.raises(AttributeError, match=r".*(has no setter|can't set attribute).*"):
+        # E           AttributeError: can't set attribute
+
+        ec.read_only = "not allowed"
+
+    with pytest.raises(AttributeError, match=r".*(has no deleter|can't delete attribute).*"):
+        del ec.read_only
+
+    assert len(events) == 2
+    assert events[0].to_dict() == DictIncluding(
+        {
+            "event": "call",
+            "defined_class": "example_class.ExampleClass",
+            "method_id": "read_only (get)",
+        }
+    )
+
+
+def test_accessible_instrumented(events):
+    from example_class import ExampleClass
+
+    ec = ExampleClass()
+    assert ExampleClass.fully_accessible.__doc__ == "Fully-accessible"
+
+    assert ec.fully_accessible == "fully accessible"
+
+    ec.fully_accessible = "updated"
+    # Check the value of the attribute directly, to avoid extra events
+    assert ec._fully_accessible == "updated"  # pylint: disable=protected-access
+
+    del ec.fully_accessible
+
+    # assert len(events) == 6
+    assert events[0].to_dict() == DictIncluding(
+        {
+            "event": "call",
+            "defined_class": "example_class.ExampleClass",
+            "method_id": "fully_accessible (get)",
+        }
+    )
+
+    assert events[2].to_dict() == DictIncluding(
+        {
+            "event": "call",
+            "defined_class": "example_class.ExampleClass",
+            "method_id": "fully_accessible (set)",
+        }
+    )
+
+    assert events[4].to_dict() == DictIncluding(
+        {
+            "event": "call",
+            "defined_class": "example_class.ExampleClass",
+            "method_id": "fully_accessible (del)",
+        }
+    )
+
+
+def test_writable_instrumented(events):
+    from example_class import ExampleClass
+
+    ec = ExampleClass()
+    assert ExampleClass.write_only.__doc__ == "Write-only"
+
+    with pytest.raises(AttributeError, match=r".*(has no getter|unreadable attribute).*"):
+        _ = ec.write_only
+
+    ec.write_only = "updated example"
+
+    assert len(events) == 2
+    assert events[0].to_dict() == DictIncluding(
+        {
+            "event": "call",
+            "defined_class": "example_class.ExampleClass",
+            "method_id": "set_write_only (set)",
+        }
+    )

_appmap/utils.py

@@ -35,6 +35,10 @@ class FnType(IntFlag):
     CLASS = auto()
     INSTANCE = auto()
     MODULE = auto()
+    # auxtypes
+    GET = auto()
+    SET = auto()
+    DEL = auto()
 
     @staticmethod
     def classify(fn):