chore(tests): add more tests and enforce test coverage

.flake8

@@ -12,16 +12,14 @@ per-file-ignores =
     # ignore unused imports in __init__ files
     */__init__.py: F401
     # supress some docstring requirements in tests
-    test/__init__.py: D104
-    test/test_deep/__init__.py: D104
-    test/test_deep/*/__init__.py: D104,D107
-    test/test_deep/*/test_*.py: D101,D102,D107,D100,D105
-    test/test_deep/test_*.py: D101,D102,D107,D100,D105
-    test/test_deep/test_target.py: D102,D107,D103
+    tests/unit_tests/*.py: D
+    tests/unit_tests/**/*.py: D
     # these files are from OTEL so should use OTEL license.
     */deep/api/types.py: NCF102
     */deep/api/resource/__init__.py: NCF102
     */deep/api/attributes/__init__.py: NCF102
+    tests/unit_tests/api/attributes/*.py: NCF102,D
+    tests/unit_tests/api/resource/*.py: NCF102,D
 
 detailed-output = True
 copyright-regex =

.github/workflows/on_push.yaml

@@ -22,8 +22,26 @@ jobs:
           python -m pip install --upgrade pip
           pip install -r requirements.txt
           pip install -r dev-requirements.txt
+
       - name: Flake8
-        run: flake8
+        run: make lint
+
+  coverage:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v3
+      - name: Setup Python # Set Python version
+        uses: actions/setup-python@v4
+        with:
+          python-version: 3.12
+      # Install pip and pytest
+      - name: Install dependencies
+        run: |
+          python -m pip install --upgrade pip
+          pip install -r requirements.txt
+          pip install -r dev-requirements.txt
+      - run: |
+          make coverage
 
   tests:
 
@@ -46,7 +64,7 @@ jobs:
           pip install -r dev-requirements.txt
           pip install .
       - name: Test with pytest
-        run: pytest test --doctest-modules --junitxml=junit/test-results-${{ matrix.python-version }}.xml
+        run: pytest tests/unit_tests --doctest-modules --junitxml=junit/test-results-${{ matrix.python-version }}.xml
       - name: Upload pytest test results
         uses: actions/upload-artifact@v3
         with:

Makefile

@@ -16,7 +16,11 @@ endif
 
 .PHONY: test
 test:
-	pytest test
+	pytest tests/unit_tests
+
+.PHONY: coverage
+coverage:
+	pytest tests/unit_tests --cov=deep  --cov-report term --cov-fail-under=77 --cov-report html --cov-branch
 
 .PHONY: lint
 lint:

deep-python-client.iml

@@ -7,10 +7,13 @@
       <sourceFolder url="file://$MODULE_DIR$/deep-extractor/src" isTestSource="false" />
       <sourceFolder url="file://$MODULE_DIR$/examples/simple-app/src" isTestSource="false" />
       <sourceFolder url="file://$MODULE_DIR$/src" isTestSource="false" />
-      <sourceFolder url="file://$MODULE_DIR$/test" isTestSource="true" />
+      <sourceFolder url="file://$MODULE_DIR$/tests" isTestSource="true" />
       <excludeFolder url="file://$MODULE_DIR$/venv/lib/python3.10/site-packages/deep" />
     </content>
-    <orderEntry type="jdk" jdkName="Python 3.10 (deep-python-client)" jdkType="Python SDK" />
+    <orderEntry type="inheritedJdk" />
     <orderEntry type="sourceFolder" forTests="false" />
   </component>
+  <component name="PackageRequirementsSettings">
+    <option name="requirementsPath" value="requirements.txt, dev-requirments.txt" />
+  </component>
 </module>

dev-requirements.txt

@@ -10,3 +10,7 @@ flake8-docstrings
 certifi>=2023.7.22 # not directly required, pinned by Snyk to avoid a vulnerability
 flake8-header-validator>=0.0.3
 setuptools>=65.5.1 # not directly required, pinned by Snyk to avoid a vulnerability
+pytest-cov
+mockito
+opentelemetry-api
+opentelemetry-sdk

pyproject.toml

@@ -46,3 +46,15 @@ path = "src/deep/version.py"
 # read dependencies from reuirements.txt
 [tool.setuptools.dynamic]
 dependencies = {file = ["requirements.txt"]}
+
+[tool.pytest.ini_options]
+pythonpath = [
+    "./src",
+    "./tests"
+]
+
+[tool.coverage.report]
+exclude_lines = [
+    "if TYPE_CHECKING:",
+    "@abc.abstractmethod"
+]

src/deep/api/attributes/__init__.py

@@ -43,7 +43,7 @@ def _clean_attribute_value(
 
 
 def _clean_attribute(
-        key: str, value: types.AttributeValue, max_len: Optional[int]
+        key: str, value: Union[types.AttributeValue, Sequence[types.AttributeValue]], max_len: Optional[int]
 ) -> Optional[types.AttributeValue]:
     """
     Check if attribute value is valid and cleans it if required.

src/deep/api/deep.py

@@ -68,6 +68,7 @@ def shutdown(self):
         if not self.started:
             return
         self.task_handler.flush()
+        self.poll.shutdown()
         self.started = False
 
     def register_tracepoint(self, path: str, line: int, args: Dict[str, str] = None,

src/deep/api/plugin/__init__.py

@@ -38,24 +38,26 @@ def __plugin_generator(configured):
         try:
             module, cls = plugin.rsplit(".", 1)
             yield getattr(import_module(module), cls)
-            logging.debug('Did import default integration %s', plugin)
-        except (DidNotEnable, SyntaxError) as e:
+            logging.debug('Did import integration %s', plugin)
+        except (DidNotEnable, Exception) as e:
             logging.debug(
-                "Did not import default integration %s: %s", plugin, e
+                "Did not import integration %s: %s", plugin, e
             )
 
 
-def load_plugins() -> 'Tuple[list[Plugin], BoundedAttributes]':
+def load_plugins(custom=None) -> 'Tuple[list[Plugin], BoundedAttributes]':
     """
     Load all the deep plugins.
 
     Attempt to load each plugin, if successful merge a attributes list of each plugin.
 
     :return: the loaded plugins and attributes.
     """
+    if custom is None:
+        custom = []
     bounded_attributes = BoundedAttributes(immutable=False)
     loaded = []
-    for plugin in __plugin_generator(DEEP_PLUGINS):
+    for plugin in __plugin_generator(DEEP_PLUGINS + custom):
         try:
             plugin_instance = plugin()
             if not plugin_instance.is_active():

src/deep/api/plugin/otel.py

@@ -71,12 +71,12 @@ def __span_name(span):
     @staticmethod
     def __span_id(span):
         # type: (_Span)-> Optional[str]
-        return (OTelPlugin.__format_span_id(span.context.__span_id)) if span else None
+        return (OTelPlugin.__format_span_id(span.context.span_id)) if span else None
 
     @staticmethod
     def __trace_id(span):
         # type: (_Span)-> Optional[str]
-        return (OTelPlugin.__format_trace_id(span.context.__trace_id)) if span else None
+        return (OTelPlugin.__format_trace_id(span.context.trace_id)) if span else None
 
     @staticmethod
     def __get_span():

src/deep/api/resource/__init__.py

@@ -142,12 +142,14 @@ def create(
             DeepResourceDetector().detect()
         ).merge(Resource(attributes, schema_url))
         if not resource.attributes.get(SERVICE_NAME, None):
-            default_service_name = "unknown_service:python"
+            default_service_name = "unknown_service"
             process_executable_name = resource.attributes.get(
                 PROCESS_EXECUTABLE_NAME, None
             )
             if process_executable_name:
                 default_service_name += ":" + process_executable_name
+            else:
+                default_service_name += ":python"
             resource = resource.merge(
                 Resource({SERVICE_NAME: default_service_name}, schema_url)
             )
@@ -294,3 +296,40 @@ def detect(self) -> "Resource":
         if service_name:
             env_resource_map[SERVICE_NAME] = service_name
         return Resource(env_resource_map)
+
+
+def get_aggregated_resources(
+        detectors: typing.List["ResourceDetector"],
+        initial_resource: typing.Optional[Resource] = None,
+        timeout=5,
+) -> "Resource":
+    """Retrieve resources from detectors in the order that they were passed.
+
+    :param detectors: List of resources in order of priority
+    :param initial_resource: Static resource. This has the highest priority
+    :param timeout: Number of seconds to wait for each detector to return
+    :return:
+    """
+    detectors_merged_resource = initial_resource or Resource.create()
+    import concurrent.futures
+
+    with concurrent.futures.ThreadPoolExecutor(max_workers=4) as executor:
+        futures = [executor.submit(detector.detect) for detector in detectors]
+        for detector_ind, future in enumerate(futures):
+            detector = detectors[detector_ind]
+            try:
+                detected_resource = future.result(timeout=timeout)
+            # pylint: disable=broad-except
+            except Exception as ex:
+                detected_resource = _EMPTY_RESOURCE
+                if detector.raise_on_error:
+                    raise ex
+                logging.warning(
+                    "Exception %s in detector %s, ignoring", ex, detector
+                )
+            finally:
+                detectors_merged_resource = detectors_merged_resource.merge(
+                    detected_resource
+                )
+
+    return detectors_merged_resource

src/deep/api/tracepoint/eventsnapshot.py

@@ -38,7 +38,7 @@ def __init__(self, tracepoint, ts, resource, frames, var_lookup: Dict[str, 'Vari
         """
         self._id = random.getrandbits(128)
         self._tracepoint = tracepoint
-        self._var_lookup: dict[str, 'Variable'] = var_lookup
+        self._var_lookup: Dict[str, 'Variable'] = var_lookup
         self._ts_nanos = ts
         self._frames = frames
         self._watches = []

src/deep/config/config_service.py

@@ -16,7 +16,7 @@
 """Service for handling deep config."""
 
 import os
-from typing import Any, List, Dict
+from typing import Any, List
 
 from deep import logging
 from deep.api.plugin import Plugin
@@ -28,16 +28,18 @@
 class ConfigService:
     """This is the main service that handles config for DEEP."""
 
-    def __init__(self, custom: Dict[str, any]):
+    def __init__(self, custom=None, tracepoints=TracepointConfigService()):
         """
         Create a new config object.
 
         :param custom: any custom values that are passed to DEEP
         """
+        if custom is None:
+            custom = {}
         self._plugins = []
         self.__custom = custom
         self._resource = None
-        self._tracepoint_config = TracepointConfigService()
+        self._tracepoint_config = tracepoints
         self._tracepoint_logger: 'TracepointLogger' = DefaultLogger()
 
     def __getattribute__(self, name: str) -> Any:

src/deep/config/tracepoint_config.py

@@ -33,7 +33,7 @@ def __init__(self) -> None:
         self._current_hash = None
         self._last_update = 0
         self._task_handler = None
-        self._listeners: list[ConfigUpdateListener] = []
+        self._listeners: List[ConfigUpdateListener] = []
 
     def update_no_change(self, ts):
         """
@@ -149,6 +149,7 @@ def remove_custom(self, config: TracePointConfig):
         for idx, cfg in enumerate(self._custom):
             if cfg.id == config.id:
                 del self._custom[idx]
+                self.__trigger_update(None, None)
                 return
 
 

src/deep/logging/__init__.py

@@ -61,7 +61,7 @@ def error(msg, *args, **kwargs):
     :param args:  the args for the log
     :param kwargs: the kwargs
     """
-    logging.getLogger("deep").debug(msg, *args, **kwargs)
+    logging.getLogger("deep").error(msg, *args, **kwargs)
 
 
 def exception(msg, *args, exc_info=True, **kwargs):

src/deep/poll/poll.py

@@ -47,8 +47,6 @@ def __init__(self, config: ConfigService, grpc: GRPCService):
     def start(self):
         """Start the long poll service."""
         logging.info("Starting Long Poll system")
-        if self.timer is not None:
-            self.timer.stop()
         self.timer = RepeatedTimer("Tracepoint Long Poll", self.config.POLL_TIMER, self.poll)
         self.__initial_poll()
         self.timer.start()
@@ -72,3 +70,9 @@ def poll(self):
         else:
             self.config.tracepoints.update_new_config(response.ts_nanos, response.current_hash,
                                                       convert_response(response.response))
+
+    def shutdown(self):
+        """Shutdown the timer."""
+        if self.timer:
+            self.timer.stop()
+        self.timer = None

src/deep/push/__init__.py

@@ -98,4 +98,4 @@ def convert_snapshot(snapshot: EventSnapshot) -> Snapshot:
                         log_msg=snapshot.log_msg)
     except Exception:
         logging.exception("Error converting to protobuf")
-        return Snapshot()
+        return None

src/deep/push/push_service.py

@@ -47,6 +47,9 @@ def push_snapshot(self, snapshot: EventSnapshot):
     def _push_task(self, snapshot):
         from deep.push import convert_snapshot
         converted = convert_snapshot(snapshot)
+        if converted is None:
+            return
+
         logging.debug("Uploading snapshot: %s", snapshot_id_as_hex_str(snapshot.id))
 
         stub = SnapshotServiceStub(self.grpc.channel)

src/deep/task/__init__.py

@@ -63,8 +63,8 @@ def submit_task(self, task, *args) -> Future:
         self._pending[next_id] = future
 
         # cannot use 'del' in lambda: https://stackoverflow.com/a/41953232/5151254
-        def callback(future: Future):
-            if future.exception() is not None:
+        def callback(_future: Future):
+            if _future.exception() is not None:
                 logging.exception("Submitted task failed %s", task)
             if next_id in self._pending:
                 del self._pending[next_id]

src/deep/utils.py

@@ -35,32 +35,14 @@ def time_ns():
     return time.time_ns()
 
 
-def reduce_list(key, update_value, default_value, lst):
-    """Reduce a list to a dict.
-
-    key :: list_item -> dict_key
-    update_value :: key * existing_value -> updated_value
-    default_value :: initial value passed to update_value
-    lst :: The list
-
-    default_value comes before l. This is different from functools.reduce,
-    because functools.reduce's order is wrong.
-    """
-    d = {}
-    for k in lst:
-        j = key(k)
-        d[j] = update_value(k, d.get(j, default_value))
-    return d
-
-
 def str2bool(string):
     """
     Convert a string to a boolean.
 
     :param string: the string to convert
     :return: True, if string is yes, true, t or 1. (case insensitive)
     """
-    return string.lower() in ("yes", "true", "t", "1")
+    return string.lower() in ("yes", "true", "t", "1", "y")
 
 
 class RepeatedTimer: