import pytest from langchain_core.documents import Document from langflow.schema import Data def test_data_initialization(): record = Data(text_key="msg", data={"msg": "Hello, World!", "extra": "value"}) assert record.msg == "Hello, World!" assert record.extra == "value" def test_validate_data_with_extra_keys(): record = Data(dummy_key="dummy", data={"key": "value"}) assert record.data["dummy_key"] == "dummy" assert "dummy_key" in record.data assert record.key == "value" def test_conversion_to_document(): record = Data(data={"text": "Sample text", "meta": "data"}) document = record.to_lc_document() assert document.page_content == "Sample text" assert document.metadata == {"meta": "data"} def test_conversion_from_document(): document = Document(page_content="Doc content", metadata={"meta": "info"}) record = Data.from_document(document) assert record.text == "Doc content" assert record.meta == "info" def test_add_method_for_strings(): record1 = Data(data={"text": "Hello"}) record2 = Data(data={"text": " World"}) combined = record1 + record2 assert combined.text == "Hello World" def test_add_method_for_integers(): record1 = Data(data={"number": 5}) record2 = Data(data={"number": 10}) combined = record1 + record2 assert combined.number == 15 def test_add_method_with_non_overlapping_keys(): record1 = Data(data={"text": "Hello"}) record2 = Data(data={"number": 10}) combined = record1 + record2 assert combined.text == "Hello" assert combined.number == 10 def test_custom_attribute_get_set_del(): record = Data() record.custom_attr = "custom_value" assert record.custom_attr == "custom_value" del record.custom_attr with pytest.raises(AttributeError): _ = record.custom_attr def test_deep_copy(): import copy record1 = Data(data={"text": "Hello", "number": 10}) record2 = copy.deepcopy(record1) assert record2.text == "Hello" assert record2.number == 10 record2.text = "World" assert record1.text == "Hello" # Ensure original is unchanged def test_custom_attribute_setting_and_getting(): record = Data() record.dynamic_attribute = "Dynamic Value" assert record.dynamic_attribute == "Dynamic Value" def test_str_and_dir_methods(): record = Data(text_key="text", data={"text": "Test Text", "key": "value"}) assert "Test Text" in str(record) assert "key" in dir(record) assert "data" in dir(record) def test_dir_includes_data_keys(): record = Data(data={"text": "Hello", "new_attr": "value"}) dir_output = dir(record) # Check for standard attributes assert "data" in dir_output assert "text_key" in dir_output assert "__add__" in dir_output # Checking for a method # Check for dynamic attributes from data assert "text" in dir_output assert "new_attr" in dir_output # Optionally, verify that dynamically added attributes are listed record.dynamic_attr = "dynamic" assert "dynamic_attr" in dir_output or "dynamic_attr" in dir(record) # To account for the change def test_dir_reflects_attribute_deletion(): record = Data(data={"removable": "I can be removed"}) assert "removable" in dir(record) # Delete the attribute and check again del record.removable assert "removable" not in dir(record) def test_get_text_with_text_key(): data = {"text": "Hello, World!"} schema = Data(data=data, text_key="text", default_value="default") result = schema.get_text() assert result == "Hello, World!" def test_get_text_without_text_key(): data = {"other_key": "Hello, World!"} schema = Data(data=data, text_key="text", default_value="default") result = schema.get_text() assert result == "default" def test_get_text_with_empty_data(): data = {} schema = Data(data=data, text_key="text", default_value="default") result = schema.get_text() assert result == "default" def test_get_text_with_none_data(): data = None schema = Data(data=data, text_key="text", default_value="default") result = schema.get_text() assert result == "default"