From a25a18648d4cce18ff793b5d1102069be16effff Mon Sep 17 00:00:00 2001 From: "Hanzhang Zeng (Roger)" Date: Tue, 18 Aug 2020 13:39:47 -0700 Subject: [PATCH 1/2] Added test for blob trigger --- azure/functions/blob.py | 3 ++ tests/test_blob.py | 104 ++++++++++++++++++++++++++++++++++++++++ 2 files changed, 107 insertions(+) create mode 100644 tests/test_blob.py diff --git a/azure/functions/blob.py b/azure/functions/blob.py index 69c6e410..ce47ee98 100644 --- a/azure/functions/blob.py +++ b/azure/functions/blob.py @@ -78,6 +78,9 @@ def encode(cls, obj: Any, *, @classmethod def decode(cls, data: meta.Datum, *, trigger_metadata) -> Any: + if data is None or data.type is None: + return None + data_type = data.type if data_type == 'string': diff --git a/tests/test_blob.py b/tests/test_blob.py new file mode 100644 index 00000000..ca3c0a19 --- /dev/null +++ b/tests/test_blob.py @@ -0,0 +1,104 @@ +# Copyright (c) Microsoft Corporation. All rights reserved. +# Licensed under the MIT License. + +from typing import Dict, Any +import unittest + +import azure.functions as func +import azure.functions.blob as afb +from azure.functions.meta import Datum +from azure.functions.blob import InputStream + + +class TestBlob(unittest.TestCase): + def test_blob_input_type(self): + check_input_type = afb.BlobConverter.check_input_type_annotation + self.assertTrue(check_input_type(str)) + self.assertTrue(check_input_type(bytes)) + self.assertTrue(check_input_type(InputStream)) + self.assertFalse(check_input_type(bytearray)) + + def test_blob_input_none(self): + result: func.DocumentList = afb.BlobConverter.decode( + data=None, trigger_metadata=None) + self.assertIsNone(result) + + def test_blob_input_string_no_metadata(self): + datum: Datum = Datum(value='string_content', type='string') + result: InputStream = afb.BlobConverter.decode( + data=datum, trigger_metadata=None) + self.assertIsNotNone(result) + + # Verify result metadata + self.assertIsInstance(result, InputStream) + self.assertIsNone(result.name) + self.assertIsNone(result.length) + self.assertIsNone(result.uri) + self.assertTrue(result.readable()) + self.assertFalse(result.seekable()) + self.assertFalse(result.writable()) + + # Verify result content + content: bytes = result.read() + self.assertEqual(content, b'string_content') + + def test_blob_input_bytes_no_metadata(self): + datum: Datum = Datum(value=b'bytes_content', type='bytes') + result: InputStream = afb.BlobConverter.decode( + data=datum, trigger_metadata=None) + self.assertIsNotNone(result) + + # Verify result metadata + self.assertIsInstance(result, InputStream) + self.assertIsNone(result.name) + self.assertIsNone(result.length) + self.assertIsNone(result.uri) + self.assertTrue(result.readable()) + self.assertFalse(result.seekable()) + self.assertFalse(result.writable()) + + # Verify result content + content: bytes = result.read() + self.assertEqual(content, b'bytes_content') + + def test_blob_input_with_metadata(self): + datum: Datum = Datum(value=b'blob_content', type='bytes') + metadata: Dict[str, Any] = { + 'Properties': Datum('{"Length": "12"}', 'json'), + 'BlobTrigger': Datum('blob_trigger_name', 'string'), + 'Uri': Datum('https://test.io/blob_trigger', 'string') + } + result: InputStream = afb.BlobConverter.decode( + data=datum, trigger_metadata=metadata) + + # Verify result metadata + self.assertIsInstance(result, InputStream) + self.assertEqual(result.name, 'blob_trigger_name') + self.assertEqual(result.length, len(b'blob_content')) + self.assertEqual(result.uri, 'https://test.io/blob_trigger') + + def test_blob_incomplete_read(self): + datum: Datum = Datum(value=b'blob_content', type='bytes') + result: InputStream = afb.BlobConverter.decode( + data=datum, trigger_metadata=None) + + self.assertEqual(result.read(size=3), b'blo') + + def test_blob_output_type(self): + check_output_type = afb.BlobConverter.check_output_type_annotation + self.assertTrue(check_output_type(str)) + self.assertTrue(check_output_type(bytes)) + self.assertTrue(check_output_type(bytearray)) + self.assertTrue(check_output_type(InputStream)) + + def test_blob_output_custom_type(self): + class CustomOutput: + def read(self) -> bytes: + return b'custom_output_content' + + check_output_type = afb.BlobConverter.check_output_type_annotation + self.assertTrue(check_output_type(CustomOutput)) + + def test_blob_output_custom_output_content(self): + class CustomOutput: + def read(self) -> bytes: From 189526fd0517d46f9efc1e43c2e0b2bfcf05b3cf Mon Sep 17 00:00:00 2001 From: "Hanzhang Zeng (Roger)" Date: Tue, 18 Aug 2020 14:14:58 -0700 Subject: [PATCH 2/2] Added more test cases for blob trigger --- tests/test_blob.py | 42 ++++++++++++++++++++++++++++++++++++------ 1 file changed, 36 insertions(+), 6 deletions(-) diff --git a/tests/test_blob.py b/tests/test_blob.py index ca3c0a19..150951d2 100644 --- a/tests/test_blob.py +++ b/tests/test_blob.py @@ -84,6 +84,40 @@ def test_blob_incomplete_read(self): self.assertEqual(result.read(size=3), b'blo') + def test_blob_output_custom_output_content(self): + class CustomOutput: + def read(self) -> bytes: + return b'custom_output_content' + + # Try encoding a custom instance as an output return + out = CustomOutput() + result: Datum = afb.BlobConverter.encode(obj=out, expected_type=None) + self.assertEqual(result.value, b'custom_output_content') + self.assertEqual(result.type, 'bytes') + + def test_blob_output_custom_output_without_read_method(self): + class CustomOutput: + def _read(self) -> bytes: + return b'should_not_be_called' + + # Try encoding a custom instance without read() method + # This should raise an error when an unknown output is returned + out = CustomOutput() + with self.assertRaises(NotImplementedError): + afb.BlobConverter.encode(obj=out, expected_type=None) + + def test_blob_output_string(self): + out: str = 'blob_output_string' + result: Datum = afb.BlobConverter.encode(obj=out, expected_type=None) + self.assertEqual(result.value, 'blob_output_string') + self.assertEqual(result.type, 'string') + + def test_blob_output_bytes(self): + out: bytes = b'blob_output_bytes' + result: Datum = afb.BlobConverter.encode(obj=out, expected_type=None) + self.assertEqual(result.value, b'blob_output_bytes') + self.assertEqual(result.type, 'bytes') + def test_blob_output_type(self): check_output_type = afb.BlobConverter.check_output_type_annotation self.assertTrue(check_output_type(str)) @@ -93,12 +127,8 @@ def test_blob_output_type(self): def test_blob_output_custom_type(self): class CustomOutput: - def read(self) -> bytes: - return b'custom_output_content' + def read(self) -> Datum: + return Datum(b'custom_output_content', 'types') check_output_type = afb.BlobConverter.check_output_type_annotation self.assertTrue(check_output_type(CustomOutput)) - - def test_blob_output_custom_output_content(self): - class CustomOutput: - def read(self) -> bytes: