update tests for python

Author: HomelessDinosaur

.coveragerc

@@ -0,0 +1,3 @@
+[run]
+ omit =
+  ./nitric/proto/*

nitric/api/storage.py

@@ -31,6 +31,7 @@
     StorageListFilesRequest,
 )
 from enum import Enum
+from warnings import warn
 
 
 class Storage(object):
@@ -140,6 +141,7 @@ async def download_url(self, expiry: int = 600):
 
     async def sign_url(self, mode: FileMode = FileMode.READ, expiry: int = 3600):
         """Generate a signed URL for reading or writing to a file."""
+        warn("File.sign_url() is deprecated, use upload_url() or download_url() instead", DeprecationWarning)
         try:
             response = await self._storage._storage_stub.pre_sign_url(
                 storage_pre_sign_url_request=StoragePreSignUrlRequest(

nitric/application.py

@@ -67,9 +67,9 @@ def _create_resource(cls, resource: Type[BT], name: str, *args, **kwargs) -> BT:
             )
 
     @classmethod
-    def _create_tracer(cls) -> TracerProvider:
-        local_run = "OTELCOL_BIN" not in environ
-        samplePercent = int(getenv("NITRIC_TRACE_SAMPLE_PERCENT", "100")) / 100.0
+    def _create_tracer(cls, local: bool = True, sampler: int = 100) -> TracerProvider:
+        local_run = local or "OTELCOL_BIN" not in environ
+        samplePercent = int(getenv("NITRIC_TRACE_SAMPLE_PERCENT", sampler)) / 100.0
 
         # If its a local run use a console exporter, otherwise export using OTEL Protocol
         exporter = OTLPSpanExporter(endpoint="http://localhost:4317", insecure=True)

nitric/faas.py

@@ -395,7 +395,10 @@ class Frequency(Enum):
     @staticmethod
     def from_str(value: str) -> Frequency:
         """Convert a string frequency value to a Frequency."""
-        return Frequency[value.strip().lower()]
+        try:
+            return Frequency[value.strip().lower()]
+        except Exception:
+            raise ValueError(f"{value} is not valid frequency")
 
     @staticmethod
     def as_str_list() -> List[str]:

nitric/resources/apis.py

@@ -52,7 +52,12 @@ class ApiDetails:
 
 @dataclass
 class JwtSecurityDefinition:
-    """Represents the JWT security definition for an API."""
+    """
+    Represents the JWT security definition for an API.
+
+    issuer (str): the JWT issuer
+    audiences (List[str]): a list of the allowed audiences for the API
+    """
 
     issuer: str
     audiences: List[str]
@@ -68,16 +73,16 @@ class ApiOptions:
     """Represents options when creating an API, such as middleware to be applied to all HTTP request to the API."""
 
     path: str
-    middleware: Union[HttpMiddleware, List[HttpMiddleware], None]
-    security_definitions: Union[dict[str, SecurityDefinition], None]
-    security: Union[dict[str, List[str]], None]
+    middleware: Union[HttpMiddleware, List[HttpMiddleware]]
+    security_definitions: dict[str, SecurityDefinition]
+    security: dict[str, List[str]]
 
     def __init__(
         self,
         path: str = "",
         middleware: List[Middleware] = [],
-        security_definitions: dict[str, SecurityDefinition] = None,
-        security: dict[str, List[str]] = None,
+        security_definitions: dict[str, SecurityDefinition] = {},
+        security: dict[str, List[str]] = {},
     ):
         """Construct a new API options object."""
         self.middleware = middleware
@@ -102,18 +107,18 @@ def _to_resource(b: Api) -> Resource:
 
 def _security_definition_to_grpc_declaration(
     security_definitions: dict[str, SecurityDefinition]
-) -> Union[dict[str, ApiSecurityDefinition], None]:
+) -> dict[str, ApiSecurityDefinition]:
     if security_definitions is None or len(security_definitions) == 0:
-        return None
+        return {}
     return {
         k: ApiSecurityDefinition(jwt=ApiSecurityDefinitionJwt(issuer=v.issuer, audiences=v.audiences))
         for k, v in security_definitions.items()
     }
 
 
-def _security_to_grpc_declaration(security: dict[str, List[str]]) -> dict[str, ApiScopes] | None:
+def _security_to_grpc_declaration(security: dict[str, List[str]]) -> dict[str, ApiScopes]:
     if security is None or len(security) == 0:
-        return None
+        return {}
     return {k: ApiScopes(v) for k, v in security.items()}
 
 
@@ -187,7 +192,7 @@ def decorator(function: HttpMiddleware):
         return decorator
 
     def methods(self, methods: List[HttpMethod], match: str, opts: MethodOptions = None):
-        """Define an HTTP route which will respond to HTTP GET requests."""
+        """Define an HTTP route which will respond to specific HTTP requests defined by a list of verbs."""
         if opts is None:
             opts = MethodOptions()
 
@@ -275,7 +280,7 @@ async def _details(self) -> ApiDetails:
         except GRPCError as grpc_err:
             raise exception_from_grpc_error(grpc_err)
 
-    async def URL(self) -> str:
+    async def url(self) -> str:
         """Get the APIs live URL."""
         details = await self._details()
         return details.url
@@ -291,7 +296,7 @@ class Route:
     def __init__(self, api: Api, path: str, opts: RouteOptions):
         """Define a route to be handled by the provided API."""
         self.api = api
-        self.path = api.path.join(path)
+        self.path = (api.path + path).replace("//", "/")
         self.middleware = opts.middleware if opts.middleware is not None else []
 
     def method(self, methods: List[HttpMethod], *middleware: HttpMiddleware, opts: MethodOptions = None):

nitric/resources/buckets.py

@@ -62,7 +62,7 @@ async def _register(self):
         except GRPCError as grpc_err:
             raise exception_from_grpc_error(grpc_err)
 
-    def _perms_to_actions(self, *args: [Union[BucketPermission, str]]) -> List[Action]:
+    def _perms_to_actions(self, *args: List[Union[BucketPermission, str]]) -> List[Action]:
         permission_actions_map = {
             BucketPermission.reading: [Action.BucketFileGet, Action.BucketFileList],
             BucketPermission.writing: [Action.BucketFilePut],

nitric/resources/schedules.py

@@ -47,17 +47,15 @@ def every(self, rate_description: str, *middleware: EventMiddleware):
             # handle singular frequencies. e.g. every('day')
             rate_description = f"1 {rate_description}s"  # 'day' becomes '1 days'
 
-        rate, freq_str = rate_description.split(" ")
-        freq = Frequency.from_str(freq_str)
+        try:
+            rate, freq_str = rate_description.split(" ")
+            freq = Frequency.from_str(freq_str)
+        except Exception:
+            raise Exception(f"invalid rate expression, frequency must be one of {Frequency.as_str_list()}")
 
         if not rate.isdigit():
             raise Exception("invalid rate expression, expression must begin with a positive integer")
 
-        if not freq:
-            raise Exception(
-                f"invalid rate expression, frequency must be one of ${Frequency.as_str_list()}, received ${freq_str}"
-            )
-
         opts = RateWorkerOptions(self.description, int(rate), freq)
 
         self.server = FunctionServer(opts)
@@ -73,4 +71,4 @@ def decorator(func: EventMiddleware):
         r.every(every, func)
         return r
 
-    return decorator
+    return decorator

tests/api/test_documents.py

@@ -50,7 +50,6 @@
     DocumentQueryStreamRequest,
 )
 from nitric.proto.nitric.event.v1 import TopicListResponse, NitricTopic
-from nitric.utils import _struct_from_dict
 
 
 class Object(object):

tests/api/test_exception.py

@@ -66,7 +66,7 @@
 
 
 class TestException:
-    @pytest.yield_fixture(autouse=True)
+    @pytest.fixture(autouse=True)
     def init_exceptions(self):
         # Status codes that can be automatically converted to exceptions
         self.accepted_status_codes = set(_exception_code_map.keys())

tests/api/test_queues.py

@@ -63,6 +63,24 @@ async def test_send(self):
             )
         )
 
+    async def test_send_with_only_payload(self):
+        mock_send = AsyncMock()
+        mock_response = Object()
+        mock_send.return_value = mock_response
+
+        payload = {"content": "of task"}
+
+        with patch("nitric.proto.nitric.queue.v1.QueueServiceStub.send", mock_send):
+            queue = Queues().queue("test-queue")
+            await queue.send(payload)
+
+        # Check expected values were passed to Stub
+        mock_send.assert_called_once_with(
+            queue_send_request=QueueSendRequest(
+                queue="test-queue", task=NitricTask(id=None, payload_type=None, payload=_struct_from_dict(payload))
+            )
+        )
+
     async def test_send_with_failed(self):
         payload = {"content": "of task"}