""" Comprehensive tests for loader_registry module. Covers registry lookup logic, loader selection by file type, optional-loader availability flags, loader kwargs, error handling, and edge cases not covered in existing test files. """ from pathlib import Path from unittest.mock import patch import pytest 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, ) # --------------------------------------------------------------------------- # Registry structure and content # --------------------------------------------------------------------------- class TestRegistryStructure: """Verify the structure and invariants of LOADER_REGISTRY.""" def test_every_loader_class_is_a_type(self): """Every loader_class value must be a class (type), not an instance.""" for ext, entry in LOADER_REGISTRY.items(): assert isinstance(entry["loader_class"], type), ( f"{ext}: loader_class should be a type, got {type(entry['loader_class'])}" ) def test_loader_kwargs_defaults_to_empty_dict(self): """loader_kwargs should always be a dict (may be empty).""" for ext, entry in LOADER_REGISTRY.items(): kwargs = entry.get("loader_kwargs", {}) assert isinstance(kwargs, dict), ( f"{ext}: loader_kwargs is not a dict" ) def test_minimum_number_of_registered_extensions(self): """Registry should have a reasonable minimum number of extensions.""" # Core set: pdf, txt, md, markdown, docx, doc, csv, xlsx, xls, # html, htm, ppt, pptx, xml, eml, tsv, json, yaml, yml, # ipynb, enex, toml, mhtml, mht = 24 assert len(LOADER_REGISTRY) >= 24 class TestRegistrySpecificLoaderClasses: """Verify that specific extensions map to the expected loader class.""" @pytest.mark.parametrize( "ext, expected_class_name", [ (".pdf", "PyPDFLoader"), (".txt", "TextLoader"), (".md", "UnstructuredMarkdownLoader"), (".markdown", "UnstructuredMarkdownLoader"), (".docx", "UnstructuredWordDocumentLoader"), (".csv", "CSVLoader"), (".xlsx", "UnstructuredExcelLoader"), (".xls", "XLSLoader"), (".html", "UnstructuredHTMLLoader"), (".htm", "UnstructuredHTMLLoader"), (".pptx", "UnstructuredPowerPointLoader"), (".xml", "UnstructuredXMLLoader"), (".eml", "UnstructuredEmailLoader"), (".tsv", "CSVLoader"), (".json", "SimpleJSONLoader"), (".yaml", "YAMLLoader"), (".yml", "YAMLLoader"), (".ipynb", "NotebookLoader"), (".enex", "EverNoteLoader"), (".toml", "TomlLoader"), (".mhtml", "MHTMLLoader"), (".mht", "MHTMLLoader"), ], ) def test_extension_maps_to_expected_class(self, ext, expected_class_name): entry = LOADER_REGISTRY[ext] assert entry["loader_class"].__name__ == expected_class_name class TestRegistrySpecificKwargs: """Test that specific extensions carry the expected kwargs.""" def test_txt_encoding_kwargs(self): entry = LOADER_REGISTRY[".txt"] assert entry["loader_kwargs"]["encoding"] == "utf-8" assert entry["loader_kwargs"]["autodetect_encoding"] is True def test_tsv_delimiter_kwargs(self): entry = LOADER_REGISTRY[".tsv"] assert "csv_args" in entry["loader_kwargs"] assert entry["loader_kwargs"]["csv_args"]["delimiter"] == "\t" def test_ipynb_include_outputs(self): entry = LOADER_REGISTRY[".ipynb"] assert entry["loader_kwargs"]["include_outputs"] is True assert entry["loader_kwargs"]["remove_newline"] is True def test_enex_load_single_document_false(self): entry = LOADER_REGISTRY[".enex"] assert entry["loader_kwargs"]["load_single_document"] is False def test_pdf_kwargs_empty(self): entry = LOADER_REGISTRY[".pdf"] assert entry["loader_kwargs"] == {} # --------------------------------------------------------------------------- # Optional loader availability flags # --------------------------------------------------------------------------- class TestOptionalLoaderFlags: """Test that optional loaders are conditionally registered.""" def test_odt_conditional_registration(self): from local_deep_research.document_loaders import loader_registry as mod if mod.HAS_ODT_LOADER: assert ".odt" in LOADER_REGISTRY else: assert ".odt" not in LOADER_REGISTRY def test_epub_conditional_registration(self): from local_deep_research.document_loaders import loader_registry as mod if mod.HAS_EPUB_LOADER: assert ".epub" in LOADER_REGISTRY else: assert ".epub" not in LOADER_REGISTRY def test_rtf_conditional_registration(self): from local_deep_research.document_loaders import loader_registry as mod if mod.HAS_RTF_LOADER: assert ".rtf" in LOADER_REGISTRY else: assert ".rtf" not in LOADER_REGISTRY def test_rst_conditional_registration(self): from local_deep_research.document_loaders import loader_registry as mod if mod.HAS_RST_LOADER: assert ".rst" in LOADER_REGISTRY else: assert ".rst" not in LOADER_REGISTRY def test_org_conditional_registration(self): from local_deep_research.document_loaders import loader_registry as mod if mod.HAS_ORG_LOADER: assert ".org" in LOADER_REGISTRY else: assert ".org" not in LOADER_REGISTRY def test_image_conditional_registration(self): from local_deep_research.document_loaders import loader_registry as mod image_exts = [".png", ".jpg", ".jpeg", ".tiff", ".tif", ".bmp", ".heic"] # Images require both the loader class and the OCR runtime dep # (pytesseract + the tesseract binary). if mod.HAS_IMAGE_LOADER and mod.HAS_OCR_DEP: for ext in image_exts: assert ext in LOADER_REGISTRY, ( f"{ext} missing despite image loader + OCR dep" ) else: for ext in image_exts: assert ext not in LOADER_REGISTRY, ( f"{ext} present despite missing image loader or OCR dep" ) # --------------------------------------------------------------------------- # get_supported_extensions # --------------------------------------------------------------------------- class TestGetSupportedExtensionsCoverage: def test_matches_registry_keys(self): """Return value must match LOADER_REGISTRY keys exactly.""" assert set(get_supported_extensions()) == set(LOADER_REGISTRY.keys()) def test_returns_fresh_list_each_call(self): """Each call returns a new list object (no shared mutable state).""" a = get_supported_extensions() b = get_supported_extensions() assert a is not b # --------------------------------------------------------------------------- # is_extension_supported - additional edge cases # --------------------------------------------------------------------------- class TestIsExtensionSupportedEdgeCases: def test_empty_string(self): assert is_extension_supported("") is False def test_dot_only(self): assert is_extension_supported(".") is False def test_multiple_dots(self): """Extension like '.tar.gz' -- only last suffix matters in Path, but the function receives raw string; '.tar.gz' starts with dot so treated as-is and should not match.""" assert is_extension_supported(".tar.gz") is False def test_whitespace_extension(self): assert is_extension_supported(" ") is False assert is_extension_supported(" .pdf") is False def test_numeric_extension(self): assert is_extension_supported(".123") is False def test_all_registered_extensions_report_supported(self): """Cross-check: every key in LOADER_REGISTRY should be supported.""" for ext in LOADER_REGISTRY: assert is_extension_supported(ext) is True, ( f"{ext} not reported as supported" ) def test_without_dot_case_insensitive(self): """Without-dot + uppercase should still resolve.""" assert is_extension_supported("JSON") is True assert is_extension_supported("Csv") is True # --------------------------------------------------------------------------- # get_loader_class_for_extension - additional coverage # --------------------------------------------------------------------------- class TestGetLoaderClassForExtensionCoverage: def test_returns_none_for_empty_string(self): assert get_loader_class_for_extension("") is None def test_returns_none_for_dot_only(self): assert get_loader_class_for_extension(".") is None def test_yml_same_class_as_yaml(self): r1 = get_loader_class_for_extension(".yml") r2 = get_loader_class_for_extension(".yaml") assert r1 is not None and r2 is not None assert r1[0] is r2[0] def test_htm_same_class_as_html(self): r1 = get_loader_class_for_extension(".htm") r2 = get_loader_class_for_extension(".html") assert r1 is not None and r2 is not None assert r1[0] is r2[0] def test_markdown_same_class_as_md(self): r1 = get_loader_class_for_extension(".markdown") r2 = get_loader_class_for_extension(".md") assert r1 is not None and r2 is not None assert r1[0] is r2[0] def test_mht_same_class_as_mhtml(self): r1 = get_loader_class_for_extension(".mht") r2 = get_loader_class_for_extension(".mhtml") assert r1 is not None and r2 is not None assert r1[0] is r2[0] def test_xls_and_xlsx_use_distinct_loaders(self): # .xlsx uses the unstructured Excel loader; legacy .xls uses our own # XLSLoader (pandas + xlrd) to avoid the msoffcrypto pre-check crash. r_xls = get_loader_class_for_extension(".xls") r_xlsx = get_loader_class_for_extension(".xlsx") assert r_xls is not None and r_xlsx is not None assert r_xls[0].__name__ == "XLSLoader" assert r_xlsx[0].__name__ == "UnstructuredExcelLoader" def test_doc_same_class_as_docx_when_registered(self): # .doc is only registered when LibreOffice is available; when it is, # it must share the Word loader class with .docx. r1 = get_loader_class_for_extension(".doc") r2 = get_loader_class_for_extension(".docx") assert r2 is not None if r1 is not None: assert r1[0] is r2[0] def test_ppt_same_class_as_pptx_when_registered(self): r1 = get_loader_class_for_extension(".ppt") r2 = get_loader_class_for_extension(".pptx") assert r2 is not None if r1 is not None: assert r1[0] is r2[0] @pytest.mark.parametrize("ext", list(LOADER_REGISTRY.keys())) def test_all_registry_extensions_resolve(self, ext): """Every registered extension must resolve to a non-None result.""" result = get_loader_class_for_extension(ext) assert result is not None, f"{ext} returned None" assert len(result) == 2 def test_tsv_kwargs_include_tab_delimiter(self): result = get_loader_class_for_extension(".tsv") assert result is not None kwargs = result[1] assert kwargs["csv_args"]["delimiter"] == "\t" # --------------------------------------------------------------------------- # get_loader_for_path - comprehensive coverage # --------------------------------------------------------------------------- class TestGetLoaderForPathCoverage: def test_returns_base_loader_subclass_for_txt(self, tmp_path: Path): f = tmp_path / "sample.txt" f.write_text("hello") loader = get_loader_for_path(f) assert loader is not None assert isinstance(loader, BaseLoader) def test_returns_base_loader_subclass_for_csv(self, tmp_path: Path): f = tmp_path / "data.csv" f.write_text("a,b\n1,2\n") loader = get_loader_for_path(f) assert loader is not None assert isinstance(loader, BaseLoader) def test_returns_none_for_unsupported_extension(self, tmp_path: Path): f = tmp_path / "file.zzz" f.write_text("data") assert get_loader_for_path(f) is None def test_accepts_pathlib_path(self, tmp_path: Path): f = tmp_path / "test.txt" f.write_text("x") loader = get_loader_for_path(f) assert loader is not None def test_accepts_string_path(self, tmp_path: Path): f = tmp_path / "test.txt" f.write_text("x") loader = get_loader_for_path(str(f)) assert loader is not None def test_case_insensitive_extension(self, tmp_path: Path): f = tmp_path / "TEST.TXT" f.write_text("x") loader = get_loader_for_path(f) assert loader is not None def test_mixed_case_extension(self, tmp_path: Path): f = tmp_path / "readme.Md" f.write_text("# Title") loader = get_loader_for_path(f) assert loader is not None def test_returns_none_on_loader_constructor_exception(self, tmp_path: Path): """When the loader constructor raises, get_loader_for_path returns None.""" f = tmp_path / "broken.txt" f.write_text("content") with patch.dict( LOADER_REGISTRY, {".txt": {"loader_class": _RaisingLoader, "loader_kwargs": {}}}, ): result = get_loader_for_path(f) assert result is None def test_no_suffix_returns_none(self, tmp_path: Path): """File without extension should return None.""" f = tmp_path / "Makefile" f.write_text("all:") assert get_loader_for_path(f) is None def test_hidden_file_with_extension(self, tmp_path: Path): """Hidden file like .gitignore has suffix '' on some systems; ensure no crash even if suffix is empty.""" f = tmp_path / ".hidden" f.write_text("secret") # .hidden has suffix '' -- should return None gracefully result = get_loader_for_path(f) # Just ensure no exception; result depends on whether '.hidden' is registered assert result is None or isinstance(result, BaseLoader) def test_double_extension_uses_last_suffix(self, tmp_path: Path): """Path('a.backup.txt').suffix == '.txt', so it should resolve to TextLoader.""" f = tmp_path / "notes.backup.txt" f.write_text("content") loader = get_loader_for_path(f) assert loader is not None def test_json_loader_for_json_file(self, tmp_path: Path): from local_deep_research.document_loaders.json_loader import ( SimpleJSONLoader, ) f = tmp_path / "data.json" f.write_text('{"key": "value"}') loader = get_loader_for_path(f) assert loader is not None assert isinstance(loader, SimpleJSONLoader) def test_yaml_loader_for_yaml_file(self, tmp_path: Path): from local_deep_research.document_loaders.yaml_loader import YAMLLoader f = tmp_path / "config.yaml" f.write_text("key: value") loader = get_loader_for_path(f) assert loader is not None assert isinstance(loader, YAMLLoader) def test_yaml_loader_for_yml_file(self, tmp_path: Path): from local_deep_research.document_loaders.yaml_loader import YAMLLoader f = tmp_path / "config.yml" f.write_text("key: value") loader = get_loader_for_path(f) assert loader is not None assert isinstance(loader, YAMLLoader) # --------------------------------------------------------------------------- # Consistency checks # --------------------------------------------------------------------------- class TestConsistencyBetweenFunctions: """Ensure is_extension_supported, get_loader_class_for_extension, and get_supported_extensions agree.""" @pytest.mark.parametrize("ext", list(LOADER_REGISTRY.keys())) def test_supported_extension_resolves_to_loader_class(self, ext): assert is_extension_supported(ext) is True assert get_loader_class_for_extension(ext) is not None def test_unsupported_extension_consistent_across_functions(self): ext = ".this_does_not_exist_42" assert is_extension_supported(ext) is False assert get_loader_class_for_extension(ext) is None assert ext not in get_supported_extensions() # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- class _RaisingLoader: """Fake loader whose __init__ always raises, used to test error handling.""" def __init__(self, *args, **kwargs): raise RuntimeError("Intentional test error")