fix(serialization-text): preserve empty string and bytes values and handle malformed inputs

packages/abstractions/kiota_abstractions/serialization/parse_node_factory_registry.py

@@ -33,7 +33,7 @@ def get_valid_content_type(self) -> str:
     def get_root_parse_node(self, content_type: str, content: bytes) -> ParseNode:
         if not content_type:
             raise Exception("Content type cannot be null")
-        if not content:
+        if content is None:
             raise Exception("Content cannot be null")
 
         vendor_specific_content_type = content_type.split(';')[0]

packages/abstractions/tests/test_parse_node_factory_registry.py

@@ -0,0 +1,38 @@
+# ------------------------------------------------------------------------------
+# Copyright (c) Microsoft Corporation.  All Rights Reserved.
+# Licensed under the MIT License.
+# See License in the project root for license information.
+# ------------------------------------------------------------------------------
+
+from unittest.mock import Mock
+
+import pytest
+
+from kiota_abstractions.serialization import ParseNode, ParseNodeFactory, ParseNodeFactoryRegistry
+
+
+def test_registry_dispatches_empty_content(monkeypatch):
+    registry = ParseNodeFactoryRegistry()
+    mock_node = Mock(spec=ParseNode)
+    mock_factory = Mock(spec=ParseNodeFactory)
+    mock_factory.get_root_parse_node.return_value = mock_node
+
+    monkeypatch.setitem(registry.CONTENT_TYPE_ASSOCIATED_FACTORIES, "text/plain", mock_factory)
+
+    result = registry.get_root_parse_node("text/plain", b"")
+    assert result is mock_node
+    mock_factory.get_root_parse_node.assert_called_once_with("text/plain", b"")
+
+
+def test_registry_raises_on_none_content():
+    registry = ParseNodeFactoryRegistry()
+    with pytest.raises(Exception) as exc_info:
+        registry.get_root_parse_node("text/plain", None)
+    assert "Content cannot be null" in str(exc_info.value)
+
+
+def test_registry_raises_on_empty_content_type():
+    registry = ParseNodeFactoryRegistry()
+    with pytest.raises(Exception) as exc_info:
+        registry.get_root_parse_node("", b"hello")
+    assert "Content type cannot be null" in str(exc_info.value)

packages/serialization/text/kiota_serialization_text/text_parse_node.py

@@ -8,7 +8,9 @@
 from uuid import UUID
 
 from kiota_abstractions.date_utils import (
-    parse_timedelta_string, datetime_from_iso_format_compat, time_from_iso_format_compat
+    datetime_from_iso_format_compat,
+    parse_timedelta_string,
+    time_from_iso_format_compat,
 )
 from kiota_abstractions.serialization import Parsable, ParsableFactory, ParseNode
 
@@ -40,7 +42,7 @@ def get_str_value(self) -> Optional[str]:
         Returns:
             str: The string value of the node
         """
-        if self._text:
+        if self._text is not None:
             return str(self._text)
         return None
 
@@ -185,7 +187,7 @@ def get_bytes_value(self) -> Optional[bytes]:
             bytes: The bytes value from the node
         """
         base64_string = self.get_str_value()
-        if not base64_string:
+        if base64_string is None:
             return None
         base64_bytes = base64_string.encode("utf-8")
         return base64.b64decode(base64_bytes)

packages/serialization/text/kiota_serialization_text/text_parse_node_factory.py

@@ -28,7 +28,7 @@ def get_root_parse_node(self, content_type: str, content: bytes) -> ParseNode:
         if valid_content_type.casefold() != content_type.casefold():
             raise TypeError(f"Expected {valid_content_type} as content type")
 
-        if not content:
+        if content is None:
             raise TypeError("Content cannot be null")
 
         content_as_str = content.decode('utf-8')

packages/serialization/text/kiota_serialization_text/text_serialization_writer.py

@@ -34,7 +34,7 @@ def write_str_value(self, key: Optional[str], value: Optional[str]) -> None:
         if isinstance(value, str):
             if key:
                 raise Exception(self.NO_STRUCTURED_DATA_MESSAGE)
-            if self._writer:
+            if self._writer is not None:
                 raise Exception(
                     'A value was already written for this serialization writer,'
                     'text content only supports a single value'
@@ -152,9 +152,9 @@ def write_bytes_value(self, key: Optional[str], value: Optional[bytes]) -> None:
             key (Optional[str]): The key to be used for the written value. May be null.
             value (bytes): The byte array to be written.
         """
-        if key and value:
+        if key and value is not None:
             raise Exception(self.NO_STRUCTURED_DATA_MESSAGE)
-        if value and not key:
+        if value is not None and not key:
             base64_bytes = base64.b64encode(value)
             base64_string = base64_bytes.decode('utf-8')
             self.write_str_value(key, base64_string)
@@ -201,7 +201,7 @@ def get_serialized_content(self) -> bytes:
         """
         text_string = self._writer
         self._writer = None
-        if text_string:
+        if text_string is not None:
             stream = text_string.encode('utf-8')
             return stream
 

packages/serialization/text/tests/unit/test_text_parse_node.py

@@ -4,7 +4,6 @@
 from uuid import UUID
 
 import pytest
-
 from kiota_serialization_text.text_parse_node import TextParseNode
 
 from ..helpers import OfficeLocation, User
@@ -116,6 +115,7 @@
 
 
 def test_get_enum_value_none_member():
+
     class RiskLevel(Enum):
         Low = "low"
         None_ = "none"
@@ -135,3 +135,98 @@
     with pytest.raises(Exception) as e_info:
         parse_node = TextParseNode(sample_structured_data)
         result = parse_node.get_collection_of_object_values(User)
+
+
+def test_get_empty_str_value():
+    parse_node = TextParseNode("")
+    assert parse_node.get_str_value() == ""
+
+
+def test_get_empty_bytes_value():
+    parse_node = TextParseNode("")
+    assert parse_node.get_bytes_value() == b""
+
+
+def test_get_bytes_value_none():
+    parse_node = TextParseNode(None)
+    assert parse_node.get_bytes_value() is None
+
+
+def test_get_bytes_value_invalid_base64_raises():
+    parse_node = TextParseNode("not-base64!!!")
+    with pytest.raises(Exception):
+        parse_node.get_bytes_value()
+
+
+def test_get_int_value_invalid_raises():
+    parse_node = TextParseNode("not-an-int")
+    with pytest.raises(ValueError):
+        parse_node.get_int_value()
+
+
+def test_get_float_value_invalid_raises():
+    parse_node = TextParseNode("not-a-float")
+    with pytest.raises(ValueError):
+        parse_node.get_float_value()
+
+
+def test_get_uuid_value_invalid_raises():
+    parse_node = TextParseNode("not-a-uuid")
+    with pytest.raises(ValueError):
+        parse_node.get_uuid_value()
+
+
+def test_get_datetime_value_invalid_raises():
+    parse_node = TextParseNode("not-a-datetime")
+    with pytest.raises(ValueError):
+        parse_node.get_datetime_value()
+
+
+def test_get_timedelta_value_invalid_raises():
+    parse_node = TextParseNode("not-a-timedelta")
+    with pytest.raises(Exception):
+        parse_node.get_timedelta_value()
+
+
+def test_get_date_value_invalid_raises():
+    parse_node = TextParseNode("not-a-date")
+    with pytest.raises(ValueError):
+        parse_node.get_date_value()
+
+
+def test_get_time_value_invalid_raises():
+    parse_node = TextParseNode("not-a-time")
+    with pytest.raises(ValueError):
+        parse_node.get_time_value()
+
+
+def test_get_enum_value_unknown_raises():
+    parse_node = TextParseNode("unknown_location")
+    with pytest.raises(Exception):
+        parse_node.get_enum_value(OfficeLocation)
+
+
+def test_get_scalars_empty_text_returns_none():
+    parse_node = TextParseNode("")
+    assert parse_node.get_int_value() is None
+    assert parse_node.get_float_value() is None
+    assert parse_node.get_uuid_value() is None
+    assert parse_node.get_datetime_value() is None
+    assert parse_node.get_timedelta_value() is None
+    assert parse_node.get_date_value() is None
+    assert parse_node.get_time_value() is None
+    assert parse_node.get_enum_value(OfficeLocation) is None
+
+
+def test_get_scalars_none_text_returns_none():
+    parse_node = TextParseNode(None)
+    assert parse_node.get_str_value() is None
+    assert parse_node.get_bytes_value() is None
+    assert parse_node.get_int_value() is None
+    assert parse_node.get_float_value() is None
+    assert parse_node.get_uuid_value() is None
+    assert parse_node.get_datetime_value() is None
+    assert parse_node.get_timedelta_value() is None
+    assert parse_node.get_date_value() is None
+    assert parse_node.get_time_value() is None
+    assert parse_node.get_enum_value(OfficeLocation) is None

packages/serialization/text/tests/unit/test_text_parse_node_factory.py

@@ -2,6 +2,7 @@
 
 from kiota_serialization_text.text_parse_node import TextParseNode
 from kiota_serialization_text.text_parse_node_factory import TextParseNodeFactory
+from kiota_serialization_text.text_serialization_writer import TextSerializationWriter
 
 
 @pytest.fixture
@@ -31,13 +32,39 @@
 
 
 def test_get_root_parse_node_empty_text():
-    with pytest.raises(TypeError) as e_info:
+    factory = TextParseNodeFactory()
+    sample_string_bytes = ''.encode('utf-8')
+    root = factory.get_root_parse_node('text/plain', sample_string_bytes)
+    assert isinstance(root, TextParseNode)
+    assert root.get_str_value() == ''
+    assert root.get_bytes_value() == b''
+
+
+def test_get_root_parse_node_none_content():
+    with pytest.raises(TypeError):
         factory = TextParseNodeFactory()
-        sample_string_bytes = ''.encode('utf-8')
-        root = factory.get_root_parse_node('text/plain', sample_string_bytes)
+        factory.get_root_parse_node('text/plain', None)
 
 
 def test_get_valid_content_type():
     factory = TextParseNodeFactory()
     content_type = factory.get_valid_content_type()
     assert content_type == 'text/plain'
+
+
+def test_empty_string_round_trip():
+    writer = TextSerializationWriter()
+    writer.write_str_value(None, '')
+    content = writer.get_serialized_content()
+    factory = TextParseNodeFactory()
+    node = factory.get_root_parse_node('text/plain', content)
+    assert node.get_str_value() == ''
+
+
+def test_empty_bytes_round_trip():
+    writer = TextSerializationWriter()
+    writer.write_bytes_value(None, b'')
+    content = writer.get_serialized_content()
+    factory = TextParseNodeFactory()
+    node = factory.get_root_parse_node('text/plain', content)
+    assert node.get_bytes_value() == b''

packages/serialization/text/tests/unit/test_text_serialization_writer.py

@@ -249,3 +249,32 @@
                 "businessPhones": ["+1 205 555 0108"],
             }
         )
+
+
+def test_write_empty_str_value():
+    text_serialization_writer = TextSerializationWriter()
+    text_serialization_writer.write_str_value(None, "")
+    content = text_serialization_writer.get_serialized_content()
+    assert content == b""
+
+
+def test_write_empty_str_value_duplicate_fails():
+    text_serialization_writer = TextSerializationWriter()
+    text_serialization_writer.write_str_value(None, "")
+    with pytest.raises(Exception) as exc_info:
+        text_serialization_writer.write_str_value(None, "second")
+    assert "A value was already written" in str(exc_info.value)
+
+
+def test_write_empty_bytes_value():
+    text_serialization_writer = TextSerializationWriter()
+    text_serialization_writer.write_bytes_value(None, b"")
+    content = text_serialization_writer.get_serialized_content()
+    assert content == b""
+
+
+def test_write_empty_bytes_value_with_key():
+    text_serialization_writer = TextSerializationWriter()
+    with pytest.raises(Exception) as exc_info:
+        text_serialization_writer.write_bytes_value("key", b"")
+    assert "Text does not support structured data" in str(exc_info.value)