fix: Allow for blob & timer unit testing

azure/functions/__init__.py

@@ -1,7 +1,8 @@
 # Copyright (c) Microsoft Corporation. All rights reserved.
 # Licensed under the MIT License.
 
-from ._abc import TimerRequest, InputStream, Context, Out
+from ._abc import Context, Out
+from ._blob import InputStream
 from ._eventhub import EventHubEvent
 from ._eventgrid import EventGridEvent, EventGridOutputEvent
 from ._cosmosdb import Document, DocumentList
@@ -24,6 +25,7 @@
 from ._servicebus import ServiceBusMessage
 from ._sql import SqlRow, SqlRowList
 from ._mysql import MySqlRow, MySqlRowList
+from ._timer import TimerRequest
 
 # Import binding implementations to register them
 from . import blob  # NoQA

azure/functions/_blob.py

@@ -0,0 +1,55 @@
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License.
+
+from typing import Optional
+
+from azure.functions import _abc as azf_abc
+
+
+class InputStream(azf_abc.InputStream):
+    """An InputStream object.
+
+    :param str name:
+        An optional str specifying the name of the blob.
+
+    :param str uri:
+        An optional str specifying the uri of the blob.
+
+    :param str length:
+        An optional int specifying the length of the blob.
+    """
+    def __init__(self, *,
+                 name: Optional[str] = None,
+                 uri: Optional[str] = None,
+                 length: Optional[int] = None) -> None:
+        self._name = name
+        self._length = length
+        self._uri = uri
+
+    @property
+    def name(self) -> Optional[str]:
+        """The name of the blob."""
+        return self._name
+
+    @property
+    def length(self) -> Optional[int]:
+        """The size of the blob in bytes."""
+        return self._length
+
+    @property
+    def uri(self) -> Optional[str]:
+        """The blob's primary location URI."""
+        return self._uri
+
+    def read(self, size=-1) -> bytes:
+        """Return and read up to *size* bytes.
+
+        :param int size:
+            The number of bytes to read.  If the argument is omitted,
+            ``None``, or negative, data is read and returned until
+            EOF is reached.
+
+        :return:
+            Bytes read from the input stream.
+        """
+        return self._io.read(size)

azure/functions/_timer.py

@@ -0,0 +1,20 @@
+# Copyright (c) Microsoft Corporation. All rights reserved.
+# Licensed under the MIT License.
+
+from . import _abc
+
+
+class TimerRequest(_abc.TimerRequest):
+    """A Timer Request object.
+
+    :param bool past_due:
+        An optional boolean specifying if the timer is past due.
+    """
+
+    def __init__(self, *, past_due: bool = False) -> None:
+        self.__past_due = past_due
+
+    @property
+    def past_due(self) -> bool:
+        """Whether the timer is past due."""
+        return self.__past_due

azure/functions/blob.py

@@ -5,20 +5,19 @@
 from typing import Any, Optional, Union
 
 from azure.functions import _abc as azf_abc
+from azure.functions import _blob as azf_blob
 from . import meta
 
 
-class InputStream(azf_abc.InputStream):
+class InputStream(azf_blob.InputStream):
     def __init__(self, *, data: Union[bytes, meta.Datum],
                  name: Optional[str] = None,
                  uri: Optional[str] = None,
                  length: Optional[int] = None,
                  blob_properties: Optional[dict] = None,
                  metadata: Optional[dict] = None) -> None:
+        super().__init__(name=name, length=length, uri=uri)
         self._io = io.BytesIO(data)  # type: ignore
-        self._name = name
-        self._length = length
-        self._uri = uri
         self._blob_properties = blob_properties
         self._metadata = metadata
 

azure/functions/queue.py

@@ -13,7 +13,7 @@
 
 
 class QueueMessage(azf_queue.QueueMessage):
-    """An HTTP response object."""
+    """A Queue message object."""
 
     def __init__(self, *,
                  id=None, body=None,

azure/functions/timer.py

@@ -5,22 +5,20 @@
 import typing
 
 from azure.functions import _abc as azf_abc
+from azure.functions import _timer as azf_timer
 from . import meta
 
 
-class TimerRequest(azf_abc.TimerRequest):
+class TimerRequest(azf_timer.TimerRequest):
+    """A Timer request object."""
 
     def __init__(self, *, past_due: bool = False,
                  schedule_status: typing.Optional[dict] = None,
                  schedule: typing.Optional[dict] = None) -> None:
-        self.__past_due = past_due
+        super().__init__(past_due=past_due)
         self.__schedule_status = schedule_status if schedule_status else {}
         self.__schedule = schedule if schedule else {}
 
-    @property
-    def past_due(self) -> bool:
-        return self.__past_due
-
     @property
     def schedule_status(self) -> dict:
         return self.__schedule_status