""" Behavioral tests for yaml_loader module. Tests the YAMLLoader and extract_text_from_yaml function. """ import tempfile from pathlib import Path import yaml class TestYAMLLoaderInit: """Tests for YAMLLoader initialization.""" def test_accepts_string_path(self): """Accepts string path.""" from local_deep_research.document_loaders.yaml_loader import YAMLLoader loader = YAMLLoader("/path/to/file.yaml") assert loader.file_path == Path("/path/to/file.yaml") def test_accepts_path_object(self): """Accepts Path object.""" from local_deep_research.document_loaders.yaml_loader import YAMLLoader loader = YAMLLoader(Path("/path/to/file.yaml")) assert loader.file_path == Path("/path/to/file.yaml") def test_default_encoding_is_utf8(self): """Default encoding is utf-8.""" from local_deep_research.document_loaders.yaml_loader import YAMLLoader loader = YAMLLoader("/path/to/file.yaml") assert loader.encoding == "utf-8" def test_accepts_custom_encoding(self): """Accepts custom encoding.""" from local_deep_research.document_loaders.yaml_loader import YAMLLoader loader = YAMLLoader("/path/to/file.yaml", encoding="latin-1") assert loader.encoding == "latin-1" class TestYAMLLoaderLoad: """Tests for YAMLLoader load functionality.""" def test_loads_valid_yaml(self): """Loads valid YAML file.""" from local_deep_research.document_loaders.yaml_loader import YAMLLoader with tempfile.NamedTemporaryFile( mode="w", suffix=".yaml", delete=False ) as f: yaml.dump({"name": "Test", "value": 42}, f) f.flush() path = f.name try: loader = YAMLLoader(path) docs = loader.load() assert len(docs) == 1 assert "name" in docs[0].page_content assert "Test" in docs[0].page_content finally: Path(path).unlink() def test_includes_source_in_metadata(self): """Includes source path in metadata.""" from local_deep_research.document_loaders.yaml_loader import YAMLLoader with tempfile.NamedTemporaryFile( mode="w", suffix=".yaml", delete=False ) as f: yaml.dump({"key": "value"}, f) f.flush() path = f.name try: loader = YAMLLoader(path) docs = loader.load() assert "source" in docs[0].metadata assert path in docs[0].metadata["source"] finally: Path(path).unlink() def test_includes_file_type_in_metadata(self): """Includes file_type in metadata.""" from local_deep_research.document_loaders.yaml_loader import YAMLLoader with tempfile.NamedTemporaryFile( mode="w", suffix=".yaml", delete=False ) as f: yaml.dump({"key": "value"}, f) f.flush() path = f.name try: loader = YAMLLoader(path) docs = loader.load() assert docs[0].metadata.get("file_type") == "yaml" finally: Path(path).unlink() def test_handles_invalid_yaml(self): """Handles invalid YAML gracefully.""" from local_deep_research.document_loaders.yaml_loader import YAMLLoader with tempfile.NamedTemporaryFile( mode="w", suffix=".yaml", delete=False ) as f: f.write("invalid: yaml: content:") f.flush() path = f.name try: loader = YAMLLoader(path) docs = loader.load() assert len(docs) == 1 assert docs[0].metadata.get("parse_error") is True finally: Path(path).unlink() def test_skips_empty_yaml(self): """Skips empty YAML documents.""" from local_deep_research.document_loaders.yaml_loader import YAMLLoader with tempfile.NamedTemporaryFile( mode="w", suffix=".yaml", delete=False ) as f: f.write("") f.flush() path = f.name try: loader = YAMLLoader(path) docs = loader.load() assert len(docs) == 0 finally: Path(path).unlink() def test_handles_list_yaml(self): """Handles YAML with list as root.""" from local_deep_research.document_loaders.yaml_loader import YAMLLoader with tempfile.NamedTemporaryFile( mode="w", suffix=".yaml", delete=False ) as f: yaml.dump(["item1", "item2", "item3"], f) f.flush() path = f.name try: loader = YAMLLoader(path) docs = loader.load() assert len(docs) == 1 assert "item1" in docs[0].page_content finally: Path(path).unlink() def test_handles_nested_yaml(self): """Handles nested YAML structure.""" from local_deep_research.document_loaders.yaml_loader import YAMLLoader with tempfile.NamedTemporaryFile( mode="w", suffix=".yaml", delete=False ) as f: yaml.dump({"level1": {"level2": {"level3": "deep value"}}}, f) f.flush() path = f.name try: loader = YAMLLoader(path) docs = loader.load() assert len(docs) == 1 assert "level1" in docs[0].page_content assert "deep value" in docs[0].page_content finally: Path(path).unlink() class TestYAMLLoaderLazyLoad: """Tests for YAMLLoader lazy_load functionality.""" def test_lazy_load_yields_documents(self): """lazy_load yields Document objects.""" from local_deep_research.document_loaders.yaml_loader import YAMLLoader with tempfile.NamedTemporaryFile( mode="w", suffix=".yaml", delete=False ) as f: yaml.dump({"key": "value"}, f) f.flush() path = f.name try: loader = YAMLLoader(path) docs = list(loader.lazy_load()) assert len(docs) == 1 finally: Path(path).unlink() def test_lazy_load_is_iterator(self): """lazy_load returns an iterator.""" from local_deep_research.document_loaders.yaml_loader import YAMLLoader with tempfile.NamedTemporaryFile( mode="w", suffix=".yaml", delete=False ) as f: yaml.dump({"key": "value"}, f) f.flush() path = f.name try: loader = YAMLLoader(path) result = loader.lazy_load() # Should be an iterator assert hasattr(result, "__iter__") assert hasattr(result, "__next__") finally: Path(path).unlink() class TestExtractTextFromYaml: """Tests for extract_text_from_yaml function.""" def test_extracts_text_from_bytes(self): """Extracts text from bytes content.""" from local_deep_research.document_loaders.yaml_loader import ( extract_text_from_yaml, ) content = b"name: John\ncity: NYC" result = extract_text_from_yaml(content) assert "name" in result assert "John" in result def test_handles_unicode(self): """Handles Unicode content.""" from local_deep_research.document_loaders.yaml_loader import ( extract_text_from_yaml, ) content = "greeting: 你好".encode("utf-8") result = extract_text_from_yaml(content) assert "你好" in result def test_handles_invalid_yaml_bytes(self): """Returns raw text for invalid YAML.""" from local_deep_research.document_loaders.yaml_loader import ( extract_text_from_yaml, ) content = b"invalid: yaml: content:" result = extract_text_from_yaml(content) assert "invalid" in result def test_returns_empty_for_empty_yaml(self): """Returns empty string for empty YAML.""" from local_deep_research.document_loaders.yaml_loader import ( extract_text_from_yaml, ) content = b"" result = extract_text_from_yaml(content) assert result == "" def test_uses_custom_encoding(self): """Uses custom encoding when specified.""" from local_deep_research.document_loaders.yaml_loader import ( extract_text_from_yaml, ) # Latin-1 encoded content content = "text: café".encode("latin-1") result = extract_text_from_yaml(content, encoding="latin-1") assert "café" in result def test_formats_as_block_style(self): """Formats output as block-style YAML.""" from local_deep_research.document_loaders.yaml_loader import ( extract_text_from_yaml, ) content = b"items:\n - one\n - two\n - three" result = extract_text_from_yaml(content) # Should have proper block formatting assert "items:" in result assert "one" in result