Add more tests and fix some of the code that wasn't working.

opentelemetry-instrumentation/src/opentelemetry/instrumentation/_blobupload/api/blob.py

@@ -1,5 +1,8 @@
 import base64
-from typing import Dict, Optional
+import json
+
+from types import MappingProxyType as _frozendict
+from typing import Mapping, Dict, Optional
 
 
 class Blob(object):
@@ -17,7 +20,7 @@ def __init__(
         self,
         raw_bytes: bytes,
         content_type: Optional[str] = None,
-        labels: Optional[Dict[str, str]] = None,
+        labels: Optional[Mapping[str, str]] = None,
     ):
         """Initialize the blob with an explicit set of properties.
 
@@ -26,12 +29,18 @@ def __init__(
           content_type: the MIME type describing the type of data in the payload
           labels: additional key/value data about the Blob
         """
-        self._raw_bytes = _raw_bytes
+        self._raw_bytes = raw_bytes
         self._content_type = content_type
-        self._labels = labels or {}
+        self._labels = {}
+        if labels is not None:
+            if isinstance(labels, dict):
+                self._labels.update(labels)
+            else:
+                for k in labels:
+                    self._labels[k] = labels[k]
 
     @staticmethod
-    def from_data_uri(cls, uri: str, labels: Optional[dict] = None) -> "Blob":
+    def from_data_uri(uri: str, labels: Optional[dict] = None) -> "Blob":
         """Instantiate a blob from a 'data:...' URI.
 
         Args:
@@ -67,10 +76,7 @@ def from_data_uri(cls, uri: str, labels: Optional[dict] = None) -> "Blob":
         assert remaining.startswith("base64,")
         base64_len = len("base64,")
         base64_encoded_content = remaining[base64_len:]
-        try:
-            raw_bytes = base64.standard_b64decode(base64_encoded_content)
-        except ValueError:
-            raw_bytes = base64.urlsafe_b64decode(base64_encoded_content)
+        raw_bytes = base64.b64decode(base64_encoded_content)
         return Blob(raw_bytes, content_type=content_type, labels=labels)
 
     @property
@@ -84,6 +90,23 @@ def content_type(self) -> Optional[str]:
         return self._content_type
 
     @property
-    def labels(self) -> Dict[str, str]:
+    def labels(self) -> Mapping[str, str]:
         """Returns the key/value metadata of this Blob."""
         return _frozendict(self._labels)
+
+    def __eq__(self, o):
+        return (
+            (isinstance(o, Blob)) and
+            (self.raw_bytes == o.raw_bytes) and
+            (self.content_type == o.content_type) and
+            (self.labels == o.labels)
+        )
+
+    def __repr__(self):
+        params = [repr(self._raw_bytes)]
+        if self._content_type is not None:
+            params.append('content_type={}'.format(repr(self._content_type)))
+        if self._labels:
+            params.append('labels={}'.format(json.dumps(self._labels, sort_keys=True)))
+        params_string = ', '.join(params)
+        return 'Blob({})'.format(params_string)

opentelemetry-instrumentation/src/opentelemetry/instrumentation/_blobupload/api/content_type.py

@@ -23,4 +23,6 @@ def from_buffer(self, raw_bytes, mime=True):
 
 def detect_content_type(raw_bytes: bytes) -> str:
     """Attempts to infer the content type of the specified data."""
+    if not raw_bytes:
+        return 'application/octet-stream'
     return _module.from_buffer(raw_bytes, mime=True)

opentelemetry-instrumentation/src/opentelemetry/instrumentation/_blobupload/api/provider.py

@@ -6,6 +6,8 @@
 from opentelemetry.instrumentation._blobupload.api.blob_uploader import (
     BlobUploader,
 )
+from opentelemetry.instrumentation._blobupload.api.constants import NOT_UPLOADED
+
 
 _logger = logging.getLogger(__name__)
 
@@ -42,7 +44,7 @@ def get_blob_uploader(self, use_case: Optional[str]) -> BlobUploader:
         if use_case:
             use_case_formatted = use_case
         _logger.warning(
-            "No BlobUploaderProvider configured; returning a no-op for use case {}".format(
+            "No BlobUploaderProvider configured; returning a no-op for use case \"{}\". Use 'set_blob_uploader_provider()' to configure.".format(
                 use_case_formatted
             )
         )
@@ -52,10 +54,12 @@ def get_blob_uploader(self, use_case: Optional[str]) -> BlobUploader:
 _blob_uploader_provider = _DefaultBlobUploaderProvider()
 
 
-def set_blob_uploader_provider(provider: BlobUploaderProvider):
+def set_blob_uploader_provider(provider: BlobUploaderProvider) -> BlobUploaderProvider:
     """Allows configuring the behavior of 'get_blob_uploader."""
     global _blob_uploader_provider
+    old_provider = _blob_uploader_provider
     _blob_uploader_provider = provider
+    return old_provider
 
 
 def get_blob_uploader(use_case: Optional[str] = None) -> BlobUploader:

opentelemetry-instrumentation/src/opentelemetry/instrumentation/_blobupload/backend/google/gcs/__init__.py

@@ -1 +1,5 @@
 from opentelemetry.instrumentation._blobupload.backend.google.gcs._gcs_impl import GcsBlobUploader
+
+__all__ = [
+    GcsBlobUploader
+]

opentelemetry-instrumentation/src/opentelemetry/instrumentation/_blobupload/utils/__init__.py

@@ -0,0 +1,9 @@
+"""Exposes API methods to callers from the package name."""
+
+from opentelemetry.instrumentation._blobupload.utils.simple_blob_uploader_adaptor import blob_uploader_from_simple_blob_uploader
+from opentelemetry.instrumentation._blobupload.utils.simple_blob_uploader import SimpleBlobUploader
+
+__all__ = [
+    blob_uploader_from_simple_blob_uploader,
+    SimpleBlobUploader,
+]

opentelemetry-instrumentation/src/opentelemetry/instrumentation/_blobupload/utils/simple_blob_uploader.py

@@ -1,11 +1,14 @@
 """Defines a simple, synchronous interface for providing a backend implementation."""
 
+import abc
 
-class SimpleBlobUploader(ABC):
+from opentelemetry.instrumentation._blobupload.api import Blob
+
+class SimpleBlobUploader(abc.ABC):
     """Pure abstract base class of a backend implementation that is synchronous."""
 
-   @abstractmethod
-   def generate_destination_uri(self, blob: Blob) -> str:
+    @abc.abstractmethod
+    def generate_destination_uri(self, blob: Blob) -> str:
        """Generates a URI of where the blob will get written.
 
        Args:
@@ -16,8 +19,8 @@ def generate_destination_uri(self, blob: Blob) -> str:
        """
        raise NotImplementedError('SimpleBlobUploader.generate_destination_uri')
 
-   @abstractmethod
-   def upload_sync(self, uri: str, blob: Blob):
+    @abc.abstractmethod
+    def upload_sync(self, uri: str, blob: Blob):
        """Synchronously writes the blob to the specified destination URI.
 
        Args:
@@ -28,4 +31,4 @@ def upload_sync(self, uri: str, blob: Blob):
        Effects:
          Attempts to upload/write the Blob to the specified destination URI.
        """
-       raise NotImplementedError('SimpleBlobUploader.upload_sync')
+       raise NotImplementedError('SimpleBlobUploader.upload_sync')

opentelemetry-instrumentation/src/opentelemetry/instrumentation/_blobupload/utils/simple_blob_uploader_adaptor.py

@@ -1,13 +1,19 @@
 import atexit
+import logging
 
-from concurrent.futures import Executor, ProcessPoolExecutor
+from typing import Optional
+from concurrent.futures import Executor, ThreadPoolExecutor
 
-from opentelemetry.instrumentation._blobupload.api import Blob
-from opentelemetry.instrumentation._blobupload.api import BlobUploader
-from opentelemetry.instrumentation._blobupload.api import detect_content_type
+from opentelemetry.instrumentation._blobupload.api import (
+    Blob,
+    BlobUploader,
+    detect_content_type)
 from opentelemetry.instrumentation._blobupload.utils.simple_blob_uploader import SimpleBlobUploader
 
 
+_logger = logging.getLogger(__name__)
+
+
 def _with_content_type(blob: Blob) -> Blob:
     """Returns a variant of the Blob with the content type auto-detected if needed."""
     if blob.content_type is not None:
@@ -16,7 +22,7 @@ def _with_content_type(blob: Blob) -> Blob:
     return Blob(blob.raw_bytes, content_type=content_type, labels=blob.labels)
 
 
-def _UploadAction(object):
+class _UploadAction(object):
     """Represents the work to be done in the background to upload a blob."""
 
     def __init__(self, simple_uploader, uri, blob):
@@ -25,7 +31,11 @@ def __init__(self, simple_uploader, uri, blob):
         self._blob = blob
 
     def __call__(self):
-        self._simple_uploader.upload_sync(self._uri, self._blob)
+        _logger.debug('Uploading blob to "{}".'.format(self._uri))
+        try:
+            self._simple_uploader.upload_sync(self._uri, self._blob)
+        except:
+            _logger.error('Failed to upload blob to "{}".'.format(self._uri))
 
 
 def _create_default_executor_no_cleanup():
@@ -37,14 +47,16 @@ def _create_default_executor_no_cleanup():
     # It is because of this potential future enhancement, that we
     # have moved this logic into a separate function despite it
     # being currently logically quite simple.
-    return ProcessPoolExecutor()
+    _logger.debug('Creating thread pool executor')
+    return ThreadPoolExecutor()
 
 
 def _create_default_executor():
     """Creates an executor and registers appropriate cleanup."""
     result = _create_default_executor_no_cleanup()
     def _cleanup():
         result.shutdown()
+    _logger.debug('Registering cleanup for the pool')
     atexit.register(_cleanup)
     return result
 
@@ -58,7 +70,10 @@ def _get_or_create_default_executor():
     """Return or lazily instantiate a shared default executor."""
     global _default_executor
     if _default_executor is None:
+        _logger.debug('No existing executor found; creating one lazily.')
         _default_executor = _create_default_executor()
+    else:
+        _logger.debug('Reusing existing executor.')
     return _default_executor
 
 
@@ -79,6 +94,7 @@ def upload_async(self, blob: Blob) -> str:
         return uri
 
     def _do_in_background(self, action):
+        _logger.debug('Scheduling background upload.')
         self._executor.submit(action)
 
 

opentelemetry-instrumentation/tests/_blobupload/api/test_blob.py

@@ -0,0 +1,108 @@
+#! /usr/bin/env python3
+
+if __name__ == "__main__":
+    import sys
+    sys.path.append("../../../src")
+
+import base64
+import unittest
+
+
+from opentelemetry.instrumentation._blobupload.api import Blob
+
+
+class TestBlob(unittest.TestCase):
+
+    def test_construction_with_just_bytes(self):
+        data = 'some string'.encode()
+        blob = Blob(data)
+        self.assertEqual(blob.raw_bytes, data)
+        self.assertIsNone(blob.content_type)
+        self.assertIsNotNone(blob.labels)
+        self.assertEqual(len(blob.labels), 0)
+
+    def test_construction_with_bytes_and_content_type(self):
+        data = 'some string'.encode()
+        content_type = 'text/plain'
+        blob = Blob(data, content_type=content_type)
+        self.assertEqual(blob.raw_bytes, data)
+        self.assertEqual(blob.content_type, content_type)
+        self.assertIsNotNone(blob.labels)
+        self.assertEqual(len(blob.labels), 0)
+
+    def test_construction_with_bytes_and_labels(self):
+        data = 'some string'.encode()
+        labels = {'key1': 'value1', 'key2': 'value2'}
+        blob = Blob(data, labels=labels)
+        self.assertEqual(blob.raw_bytes, data)
+        self.assertIsNone(blob.content_type)
+        self.assert_labels_equal(blob.labels, labels)
+
+    def test_construction_with_all_fields(self):
+        data = 'some string'.encode()
+        content_type = 'text/plain'
+        labels = {'key1': 'value1', 'key2': 'value2'}
+        blob = Blob(data, content_type=content_type, labels=labels)
+        self.assertEqual(blob.raw_bytes, data)
+        self.assertEqual(blob.content_type, content_type)
+        self.assert_labels_equal(blob.labels, labels)
+
+    def test_from_data_uri_without_labels(self):
+        data = 'some string'.encode()
+        content_type = 'text/plain'
+        encoded_data = base64.b64encode(data).decode()
+        uri = 'data:{};base64,{}'.format(content_type, encoded_data)
+        blob = Blob.from_data_uri(uri)
+        self.assertEqual(blob.raw_bytes, data)
+        self.assertEqual(blob.content_type, content_type)
+        self.assertIsNotNone(blob.labels)
+        self.assertEqual(len(blob.labels), 0)
+
+    def test_from_data_uri_with_labels(self):
+        data = 'some string'.encode()
+        content_type = 'text/plain'
+        encoded_data = base64.b64encode(data).decode()
+        uri = 'data:{};base64,{}'.format(content_type, encoded_data)
+        labels = {'key1': 'value1', 'key2': 'value2'}
+        blob = Blob.from_data_uri(uri, labels=labels)
+        self.assertEqual(blob.raw_bytes, data)
+        self.assertEqual(blob.content_type, content_type)
+        self.assert_labels_equal(blob.labels, labels)
+
+    def test_from_data_uri_with_valid_standard_base64(self):
+        data = 'some string'.encode()
+        content_type = 'text/plain'
+        encoded_data = base64.standard_b64encode(data).decode()
+        uri = 'data:{};base64,{}'.format(content_type, encoded_data)
+        blob = Blob.from_data_uri(uri)
+        self.assertEqual(blob.raw_bytes, data)
+        self.assertEqual(blob.content_type, content_type)
+
+    def test_from_data_uri_with_valid_websafe_base64(self):
+        data = 'some string'.encode()
+        content_type = 'text/plain'
+        encoded_data = base64.urlsafe_b64encode(data).decode()
+        uri = 'data:{};base64,{}'.format(content_type, encoded_data)
+        blob = Blob.from_data_uri(uri)
+        self.assertEqual(blob.raw_bytes, data)
+        self.assertEqual(blob.content_type, content_type)
+
+    def test_from_data_uri_with_non_data_uri_content(self):
+        with self.assertRaisesRegex(ValueError, 'expected "data:" prefix'):
+            Blob.from_data_uri('not a valid data uri')
+
+    def test_from_data_uri_with_non_base64_content(self):
+        with self.assertRaisesRegex(ValueError, 'expected ";base64," section'):
+            Blob.from_data_uri('data:text/plain,validifpercentencoded')
+
+    def assert_labels_equal(self, a, b):
+        self.assertEqual(len(a), len(b), msg='Different sizes: {} vs {}; a={}, b={}'.format(len(a), len(b), a, b))
+        for k in a:
+            self.assertTrue(k in b, msg='Key {} found in a but not b'.format(k))
+            va = a[k]
+            vb = b[k]
+            self.assertEqual(va, vb, msg='Values for key {} different for a vs b: {} vs {}'.format(k, va, vb))
+
+
+if __name__ == "__main__":
+    unittest.main()

opentelemetry-instrumentation/tests/_blobupload/api/test_content_type.py

@@ -2,7 +2,6 @@
 
 if __name__ == "__main__":
     import sys
-
     sys.path.append("../../../src")
 
 import io
@@ -25,6 +24,12 @@ def create_test_image(format):
 
 
 class TestContentType(unittest.TestCase):
+
+    def test_handles_empty_correctly(self):
+        input = bytes()
+        output = detect_content_type(input)
+        self.assertEqual(output, "application/octet-stream")
+
     def test_detects_plaintext(self):
         input = "this is just regular text"
         output = detect_content_type(input.encode())