feat: Record by default

appmap/_implementation/detect_enabled.py

@@ -0,0 +1,182 @@
+"""
+Detect if AppMap is enabled.
+"""
+
+import importlib
+import logging
+import os
+from textwrap import dedent
+
+logger = logging.getLogger(__name__)
+
+
+class UnrecognizedRecordingMethodException(Exception):
+    pass
+
+
+RECORDING_METHODS = ["pytest", "unittest", "remote", "requests"]
+
+# Detects whether AppMap recording should be enabled. This test can be
+# performed generally, or for a particular recording method. Recording
+# can be enabled explicitly, for example via APPMAP=true, or it can be
+# enabled implicitly, by running in a dev or test web application
+# environment. Recording can also be disabled explicitly, using
+# environment variables.
+class DetectEnabled:
+    _instance = None
+
+    def __new__(cls):
+        if cls._instance is None:
+            logger.debug("Creating the DetectEnabled object")
+            cls._instance = super(DetectEnabled, cls).__new__(cls)
+            cls._instance._initialized = False
+        return cls._instance
+
+    def __init__(self):
+        if self._initialized:
+            return
+
+        self._initialized = True
+        self._detected_for_method = {}
+
+    @classmethod
+    def initialize(cls):
+        cls._instance = None
+        # because apparently __new__ and __init__ don't get called
+        cls._detected_for_method = {}
+
+    @classmethod
+    def clear_cache(cls):
+        cls._detected_for_method = {}
+
+    @classmethod
+    def is_appmap_repo(cls):
+        return os.path.exists("appmap/_implementation/event.py") and os.path.exists(
+            "appmap/_implementation/recording.py"
+        )
+
+    @classmethod
+    def should_enable(cls, recording_method):
+        """
+        Should recording be enabled for the current recording method?
+        """
+        if recording_method in cls._detected_for_method:
+            return cls._detected_for_method[recording_method]
+        else:
+            message, enabled = cls.detect_should_enable(recording_method)
+            cls._detected_for_method[recording_method] = enabled
+            if enabled:
+                logger.warning(dedent(f"AppMap recording is enabled because {message}"))
+            return enabled
+
+    @classmethod
+    def detect_should_enable(cls, recording_method):
+        if not recording_method:
+            return ["no recording method is set", False]
+
+        if recording_method not in RECORDING_METHODS:
+            return ["invalid recording method", False]
+
+        # explicitly disabled or enabled
+        if "APPMAP" in os.environ:
+            if os.environ["APPMAP"] == "false":
+                return ["APPMAP=false", False]
+            elif os.environ["APPMAP"] == "true":
+                return ["APPMAP=true", True]
+
+        # recording method explicitly disabled or enabled
+        if recording_method:
+            for one_recording_method in RECORDING_METHODS:
+                if one_recording_method == recording_method.lower():
+                    env_var = "_".join(["APPMAP", "RECORD", recording_method.upper()])
+                    if env_var in os.environ:
+                        if os.environ[env_var] == "false":
+                            return [f"{env_var}=false", False]
+                        elif os.environ[env_var] == "true":
+                            return [f"{env_var}=true", True]
+
+        # it's flask
+        message, should_enable = cls.is_flask_and_should_enable()
+        if should_enable == True or should_enable == False:
+            return [message, should_enable]
+
+        # it's django
+        message, should_enable = cls.is_django_and_should_enable()
+        if should_enable == True or should_enable == False:
+            return [message, should_enable]
+
+        if recording_method in RECORDING_METHODS:
+            return ["will record by default", True]
+
+        return ["it's not enabled by any configuration or framework", False]
+
+    @classmethod
+    def is_flask_and_should_enable(cls):
+        if "FLASK_DEBUG" in os.environ:
+            if os.environ["FLASK_DEBUG"] == "1":
+                return [f"FLASK_DEBUG={os.environ['FLASK_DEBUG']}", True]
+            elif os.environ["FLASK_DEBUG"] == "0":
+                return [f"FLASK_DEBUG={os.environ['FLASK_DEBUG']}", False]
+
+        if "FLASK_ENV" in os.environ:
+            if os.environ["FLASK_ENV"] == "development":
+                return [f"FLASK_ENV={os.environ['FLASK_ENV']}", True]
+            elif os.environ["FLASK_ENV"] == "production":
+                return [f"FLASK_ENV={os.environ['FLASK_ENV']}", False]
+
+        return ["it's not Flask", None]
+
+    @classmethod
+    def is_django_and_should_enable(cls):
+        if (
+            "DJANGO_SETTINGS_MODULE" in os.environ
+            and os.environ["DJANGO_SETTINGS_MODULE"] != ""
+        ):
+            try:
+                settings = importlib.import_module(os.environ["DJANGO_SETTINGS_MODULE"])
+            except Exception as exn:
+                settings = None
+                return [
+                    "couldn't load DJANGO_SETTINGS_MODULE={os.environ['DJANGO_SETTINGS_MODULE']}",
+                    False,
+                ]
+
+            if settings:
+                try:
+                    if settings.DEBUG == True:
+                        return [
+                            f"{os.environ['DJANGO_SETTINGS_MODULE']}.DEBUG={settings.DEBUG}",
+                            True,
+                        ]
+                    elif settings.DEBUG == False:
+                        return [
+                            f"{os.environ['DJANGO_SETTINGS_MODULE']}.DEBUG={settings.DEBUG}",
+                            False,
+                        ]
+                except Exception as exn:
+                    # it wasn't set. it's ok
+                    pass
+
+            if settings:
+                try:
+                    if (
+                        settings.APPMAP == True
+                        or str(settings.APPMAP).upper() == "true".upper()
+                    ):
+                        return [
+                            f"{os.environ['DJANGO_SETTINGS_MODULE']}.APPMAP={settings.APPMAP}",
+                            True,
+                        ]
+                    elif (
+                        settings.APPMAP == False
+                        or str(settings.APPMAP).upper() == "false".upper()
+                    ):
+                        return [
+                            f"{os.environ['DJANGO_SETTINGS_MODULE']}.APPMAP={settings.APPMAP}",
+                            False,
+                        ]
+                except Exception as exn:
+                    # it wasn't set. it's ok
+                    pass
+
+        return ["it's not Django", None]

appmap/_implementation/testing_framework.py

@@ -1,5 +1,6 @@
 """Shared infrastructure for testing framework integration."""
 
+import os
 import re
 from contextlib import contextmanager
 
@@ -104,10 +105,6 @@ def __init__(self, name, recorder_type, version=None):
 
     @contextmanager
     def record(self, klass, method, **kwds):
-        if not env.Env.current.enabled:
-            yield
-            return
-
         Metadata.add_framework(self.name, self.version)
 
         item = FuncItem(klass, method, **kwds)
@@ -147,3 +144,16 @@ def collect_result_metadata(metadata):
         metadata["test_status"] = "failed"
         metadata["exception"] = {"class": exn.__class__.__name__, "message": str(exn)}
         raise
+
+
+def file_write(filename, contents):
+    f = open(filename, "w")
+    f.write(contents)
+    f.close()
+
+
+def file_delete(filename):
+    try:
+        os.remove(filename)
+    except FileNotFoundError:
+        pass

appmap/_implementation/web_framework.py

@@ -10,6 +10,7 @@
 from tempfile import NamedTemporaryFile
 
 from appmap._implementation import generation
+from appmap._implementation.detect_enabled import DetectEnabled
 from appmap._implementation.env import Env
 from appmap._implementation.event import Event, ReturnEvent, _EventIds, describe_value
 from appmap._implementation.recording import Recorder
@@ -130,35 +131,33 @@ def create_appmap_file(
 
 
 class AppmapMiddleware:
-    def before_request_hook(
-        self, request, request_path, record_url, recording_is_running
-    ):
-        if request_path == record_url:
+    def __init__(self):
+        self.record_url = "/_appmap/record"
+        DetectEnabled.initialize()
+
+    def before_request_hook(self, request, request_path, recording_is_running):
+        if request_path == self.record_url:
             return None, None, None
 
+        rec = None
         start = None
         call_event_id = None
-        if Env.current.enabled or recording_is_running:
-            # It should be recording or it's currently recording.  The
-            # recording is either
-            # a) remote, enabled by POST to /_appmap/record, which set
-            #    recording_is_running, or
-            # b) requests, set by Env.current.record_all_requests, or
-            # c) both remote and requests; there are multiple active recorders.
-            if not Env.current.record_all_requests and recording_is_running:
-                # a)
-                rec = Recorder()
-            else:
-                # b) or c)
-                rec = Recorder(_EventIds.get_thread_id())
-                rec.start_recording()
-                # Each time an event is added for a thread_id it's also
-                # added to the global Recorder().  So don't add the event
-                # to the global Recorder() explicitly because that would
-                # add the event in it twice.
-
-            if rec.enabled:
-                start, call_event_id = self.before_request_main(rec, request)
+        if DetectEnabled.should_enable("requests"):
+            # a) requests
+            rec = Recorder(_EventIds.get_thread_id())
+            rec.start_recording()
+            # Each time an event is added for a thread_id it's also
+            # added to the global Recorder().  So don't add the event
+            # to the global Recorder() explicitly because that would
+            # add the event in it twice.
+        elif DetectEnabled.should_enable("remote") or recording_is_running:
+            # b) APPMAP=true, or
+            # c) remote, enabled by POST to /_appmap/record, which set
+            #    recording_is_running
+            rec = Recorder()
+
+        if rec and rec.enabled:
+            start, call_event_id = self.before_request_main(rec, request)
 
         return rec, start, call_event_id
 
@@ -170,7 +169,6 @@ def after_request_hook(
         self,
         request,
         request_path,
-        record_url,
         recording_is_running,
         request_method,
         request_base_url,
@@ -179,44 +177,39 @@ def after_request_hook(
         start,
         call_event_id,
     ):
-        if request_path == record_url:
+        if request_path == self.record_url:
             return response
 
-        if Env.current.enabled or recording_is_running:
-            # It should be recording or it's currently recording.  The
-            # recording is either
-            # a) remote, enabled by POST to /_appmap/record, which set
-            #    self.recording.is_running, or
-            # b) requests, set by Env.current.record_all_requests, or
-            # c) both remote and requests; there are multiple active recorders.
-            if not Env.current.record_all_requests and recording_is_running:
-                # a)
-                rec = Recorder()
+        if DetectEnabled.should_enable("requests"):
+            # a) requests
+            rec = Recorder(_EventIds.get_thread_id())
+            # Each time an event is added for a thread_id it's also
+            # added to the global Recorder().  So don't add the event
+            # to the global Recorder() explicitly because that would
+            # add the event in it twice.
+            try:
                 if rec.enabled:
                     self.after_request_main(rec, response, start, call_event_id)
-            else:
-                # b) or c)
-                rec = Recorder(_EventIds.get_thread_id())
-                # Each time an event is added for a thread_id it's also
-                # added to the global Recorder().  So don't add the event
-                # to the global Recorder() explicitly because that would
-                # add the event in it twice.
-                try:
-                    if rec.enabled:
-                        self.after_request_main(rec, response, start, call_event_id)
-
-                    output_dir = Env.current.output_dir / "requests"
-                    create_appmap_file(
-                        output_dir,
-                        request_method,
-                        request_path,
-                        request_base_url,
-                        response,
-                        response_headers,
-                        rec,
-                    )
-                finally:
-                    rec.stop_recording()
+
+                output_dir = Env.current.output_dir / "requests"
+                create_appmap_file(
+                    output_dir,
+                    request_method,
+                    request_path,
+                    request_base_url,
+                    response,
+                    response_headers,
+                    rec,
+                )
+            finally:
+                rec.stop_recording()
+        elif DetectEnabled.should_enable("remote") or recording_is_running:
+            # b) APPMAP=true, or
+            # c) remote, enabled by POST to /_appmap/record, which set
+            #    recording_is_running
+            rec = Recorder()
+            if rec.enabled:
+                self.after_request_main(rec, response, start, call_event_id)
 
         return response