"""Tests for loader_registry module.""" from pathlib import Path from langchain_core.document_loaders import BaseLoader from local_deep_research.document_loaders.loader_registry import ( LOADER_REGISTRY, get_loader_class_for_extension, get_loader_for_path, get_supported_extensions, is_extension_supported, ) class TestGetSupportedExtensions: """Tests for get_supported_extensions function.""" def test_returns_list(self) -> None: """Test that function returns a list.""" result = get_supported_extensions() assert isinstance(result, list) def test_includes_common_extensions(self) -> None: """Test that common extensions are included.""" extensions = get_supported_extensions() # Core document formats assert ".pdf" in extensions assert ".txt" in extensions assert ".md" in extensions assert ".docx" in extensions # .doc (legacy OLE binary) is only registered when LibreOffice is # installed, so it is intentionally not asserted here. # Spreadsheets assert ".csv" in extensions assert ".xlsx" in extensions # Web assert ".html" in extensions # Data formats assert ".json" in extensions assert ".yaml" in extensions assert ".yml" in extensions def test_includes_yaml_extensions(self) -> None: """Test that YAML extensions are included.""" extensions = get_supported_extensions() assert ".yaml" in extensions assert ".yml" in extensions def test_no_duplicates(self) -> None: """Test that there are no duplicate extensions.""" extensions = get_supported_extensions() assert len(extensions) == len(set(extensions)) class TestIsExtensionSupported: """Tests for is_extension_supported function.""" def test_supported_with_dot(self) -> None: """Test checking supported extension with leading dot.""" assert is_extension_supported(".pdf") is True assert is_extension_supported(".txt") is True assert is_extension_supported(".json") is True def test_supported_without_dot(self) -> None: """Test checking supported extension without leading dot.""" assert is_extension_supported("pdf") is True assert is_extension_supported("txt") is True assert is_extension_supported("json") is True def test_unsupported_extension(self) -> None: """Test checking unsupported extension.""" assert is_extension_supported(".xyz") is False assert is_extension_supported(".unknown") is False assert is_extension_supported("notreal") is False def test_case_insensitive(self) -> None: """Test that extension check is case insensitive.""" assert is_extension_supported(".PDF") is True assert is_extension_supported(".Pdf") is True assert is_extension_supported("PDF") is True def test_yaml_extensions(self) -> None: """Test YAML extensions are supported.""" assert is_extension_supported(".yaml") is True assert is_extension_supported(".yml") is True assert is_extension_supported("yaml") is True assert is_extension_supported("yml") is True class TestGetLoaderForPath: """Tests for get_loader_for_path function.""" def test_returns_loader_for_txt(self, tmp_path: Path) -> None: """Test getting loader for text file.""" txt_file = tmp_path / "test.txt" txt_file.write_text("Hello world") loader = get_loader_for_path(txt_file) assert loader is not None assert isinstance(loader, BaseLoader) def test_returns_loader_for_json(self, tmp_path: Path) -> None: """Test getting loader for JSON file.""" json_file = tmp_path / "test.json" json_file.write_text('{"key": "value"}') loader = get_loader_for_path(json_file) assert loader is not None assert isinstance(loader, BaseLoader) def test_returns_loader_for_yaml(self, tmp_path: Path) -> None: """Test getting loader for YAML file.""" yaml_file = tmp_path / "test.yaml" yaml_file.write_text("key: value") loader = get_loader_for_path(yaml_file) assert loader is not None assert isinstance(loader, BaseLoader) def test_returns_loader_for_yml(self, tmp_path: Path) -> None: """Test getting loader for .yml file.""" yml_file = tmp_path / "test.yml" yml_file.write_text("key: value") loader = get_loader_for_path(yml_file) assert loader is not None assert isinstance(loader, BaseLoader) def test_returns_none_for_unsupported(self, tmp_path: Path) -> None: """Test that None is returned for unsupported extension.""" unknown_file = tmp_path / "test.xyz" unknown_file.write_text("content") loader = get_loader_for_path(unknown_file) assert loader is None def test_accepts_string_path(self, tmp_path: Path) -> None: """Test that function accepts string path.""" txt_file = tmp_path / "test.txt" txt_file.write_text("content") loader = get_loader_for_path(str(txt_file)) assert loader is not None def test_case_insensitive_extension(self, tmp_path: Path) -> None: """Test that extension matching is case insensitive.""" txt_file = tmp_path / "test.TXT" txt_file.write_text("content") loader = get_loader_for_path(txt_file) assert loader is not None class TestGetLoaderClassForExtension: """Tests for get_loader_class_for_extension function.""" def test_returns_tuple_for_supported(self) -> None: """Test that function returns (class, kwargs) tuple.""" result = get_loader_class_for_extension(".txt") assert result is not None assert isinstance(result, tuple) assert len(result) == 2 loader_class, kwargs = result assert callable(loader_class) assert isinstance(kwargs, dict) def test_returns_none_for_unsupported(self) -> None: """Test that function returns None for unsupported extension.""" result = get_loader_class_for_extension(".xyz") assert result is None def test_with_dot_prefix(self) -> None: """Test with extension that has dot prefix.""" result = get_loader_class_for_extension(".pdf") assert result is not None def test_without_dot_prefix(self) -> None: """Test with extension without dot prefix.""" result = get_loader_class_for_extension("pdf") assert result is not None def test_case_insensitive(self) -> None: """Test case insensitivity.""" result1 = get_loader_class_for_extension(".PDF") result2 = get_loader_class_for_extension(".pdf") assert result1 is not None assert result2 is not None assert result1[0] == result2[0] def test_json_returns_simple_json_loader(self) -> None: """Test that JSON extension returns SimpleJSONLoader.""" from local_deep_research.document_loaders.json_loader import ( SimpleJSONLoader, ) result = get_loader_class_for_extension(".json") assert result is not None assert result[0] == SimpleJSONLoader class TestLoaderRegistry: """Tests for LOADER_REGISTRY constant.""" def test_registry_is_dict(self) -> None: """Test that registry is a dictionary.""" assert isinstance(LOADER_REGISTRY, dict) def test_registry_entries_have_loader_class(self) -> None: """Test that all entries have loader_class.""" for ext, entry in LOADER_REGISTRY.items(): assert "loader_class" in entry, f"Missing loader_class for {ext}" assert callable(entry["loader_class"]), ( f"loader_class not callable for {ext}" ) def test_registry_entries_have_loader_kwargs(self) -> None: """Test that all entries have loader_kwargs (even if empty).""" for ext, entry in LOADER_REGISTRY.items(): # loader_kwargs should exist or default to empty dict kwargs = entry.get("loader_kwargs", {}) assert isinstance(kwargs, dict), f"loader_kwargs not dict for {ext}" def test_registry_keys_start_with_dot(self) -> None: """Test that all registry keys start with a dot.""" for ext in LOADER_REGISTRY.keys(): assert ext.startswith("."), ( f"Extension {ext} doesn't start with dot" ) def test_registry_keys_are_lowercase(self) -> None: """Test that all registry keys are lowercase.""" for ext in LOADER_REGISTRY.keys(): assert ext == ext.lower(), f"Extension {ext} is not lowercase"