learningcircuit--local-deep-research/tests/document_loaders/test_bytes_loader.py

"""Tests for bytes_loader module."""

import pytest

from local_deep_research.document_loaders.bytes_loader import (
    extract_text_from_bytes,
    load_from_bytes,
)


class TestLoadFromBytes:
    """Tests for load_from_bytes function."""

    def test_load_txt_bytes(self) -> None:
        """Test loading text content from bytes."""
        content = b"Hello, world!"
        docs = load_from_bytes(content, ".txt", "test.txt")

        assert len(docs) >= 1
        assert "Hello, world!" in docs[0].page_content
        assert docs[0].metadata["original_filename"] == "test.txt"

    def test_load_json_bytes(self) -> None:
        """Test loading JSON content from bytes."""
        content = b'{"name": "test", "value": "hello"}'
        docs = load_from_bytes(content, ".json", "test.json")

        assert len(docs) >= 1
        assert "name" in docs[0].page_content
        assert "test" in docs[0].page_content

    def test_load_yaml_bytes(self) -> None:
        """Test loading YAML content from bytes."""
        content = b"name: test\nvalue: 123"
        docs = load_from_bytes(content, ".yaml", "test.yaml")

        assert len(docs) >= 1
        assert "name" in docs[0].page_content
        assert "test" in docs[0].page_content

    def test_load_yml_bytes(self) -> None:
        """Test loading .yml content from bytes."""
        content = b"key: value"
        docs = load_from_bytes(content, ".yml", "test.yml")

        assert len(docs) >= 1
        assert "key" in docs[0].page_content

    def test_extension_without_dot(self) -> None:
        """Test that extension without dot is handled."""
        content = b"Hello, world!"
        docs = load_from_bytes(content, "txt", "test.txt")

        assert len(docs) >= 1
        assert "Hello, world!" in docs[0].page_content

    def test_extension_case_insensitive(self) -> None:
        """Test that extension is case insensitive."""
        content = b"Hello, world!"
        docs = load_from_bytes(content, ".TXT", "test.txt")

        assert len(docs) >= 1

    def test_unsupported_extension_raises_error(self) -> None:
        """Test that unsupported extension raises ValueError."""
        content = b"content"

        with pytest.raises(ValueError, match="Unsupported file extension"):
            load_from_bytes(content, ".xyz", "test.xyz")

    def test_source_url_in_metadata(self) -> None:
        """Test that source_url is added to metadata when provided."""
        content = b"Hello, world!"
        docs = load_from_bytes(
            content,
            ".txt",
            "test.txt",
            source_url="https://example.com/test.txt",
        )

        assert len(docs) >= 1
        assert docs[0].metadata["source_url"] == "https://example.com/test.txt"

    def test_no_source_url_when_not_provided(self) -> None:
        """Test that source_url is not in metadata when not provided."""
        content = b"Hello, world!"
        docs = load_from_bytes(content, ".txt", "test.txt")

        assert "source_url" not in docs[0].metadata

    def test_original_filename_in_metadata(self) -> None:
        """Test that original filename is in metadata."""
        content = b"Hello, world!"
        docs = load_from_bytes(content, ".txt", "my_document.txt")

        assert docs[0].metadata["original_filename"] == "my_document.txt"


class TestExtractTextFromBytes:
    """Tests for extract_text_from_bytes function."""

    def test_extract_txt(self) -> None:
        """Test extracting text from TXT bytes."""
        content = b"Hello, world!"
        result = extract_text_from_bytes(content, ".txt", "test.txt")

        assert result == "Hello, world!"

    def test_extract_json(self) -> None:
        """Test extracting text from JSON bytes."""
        content = b'{"name": "Alice", "city": "Boston"}'
        result = extract_text_from_bytes(content, ".json", "test.json")

        assert result is not None
        assert "Alice" in result
        assert "Boston" in result

    def test_extract_yaml(self) -> None:
        """Test extracting text from YAML bytes."""
        content = b"name: test\nvalue: hello"
        result = extract_text_from_bytes(content, ".yaml", "test.yaml")

        assert result is not None
        assert "name" in result
        assert "test" in result

    def test_extract_unsupported_returns_none(self) -> None:
        """Test that unsupported extension returns None."""
        content = b"content"
        result = extract_text_from_bytes(content, ".xyz", "test.xyz")

        assert result is None

    def test_extract_extension_without_dot(self) -> None:
        """Test extraction with extension without dot."""
        content = b"Hello!"
        result = extract_text_from_bytes(content, "txt", "test.txt")

        assert result == "Hello!"

    def test_extract_case_insensitive(self) -> None:
        """Test case insensitive extension handling."""
        content = b"Hello!"
        result = extract_text_from_bytes(content, ".TXT", "test.txt")

        assert result == "Hello!"

    def test_extract_empty_content(self) -> None:
        """Empty input must not crash and must return an empty string
        (not None — None would signal extraction failure)."""
        content = b""
        result = extract_text_from_bytes(content, ".txt", "empty.txt")

        assert result == ""

    def test_extract_unicode_content(self) -> None:
        """Test extracting unicode content."""
        content = "Hello, 世界! 🌍".encode("utf-8")
        result = extract_text_from_bytes(content, ".txt", "unicode.txt")

        assert result is not None
        assert "世界" in result
        assert "🌍" in result

    def test_extract_joins_multiple_documents(self) -> None:
        """Test that multiple documents are joined with double newline."""
        # Most loaders return single document, but test the join logic
        content = b"Hello, world!"
        result = extract_text_from_bytes(content, ".txt", "test.txt")

        # Should be string (joined if multiple docs)
        assert isinstance(result, str)

    def test_extract_whitespace_only_txt(self) -> None:
        """Whitespace-only TXT files should return whitespace or empty string."""
        content = b"   \n\t\n   "
        result = extract_text_from_bytes(content, ".txt", "whitespace.txt")

        # Should not raise, result may be whitespace or empty
        assert result is not None or result == ""

    def test_extract_json_with_utf8_bom(self) -> None:
        """Handle UTF-8 BOM (Byte Order Mark) prefix in JSON."""
        # UTF-8 BOM: \xef\xbb\xbf
        content = b'\xef\xbb\xbf{"name": "test"}'
        result = extract_text_from_bytes(content, ".json", "bom.json")

        assert result is not None
        assert "test" in result


class TestTempFileCleanup:
    """Tests for temporary file cleanup."""

    def test_temp_file_cleaned_up_on_success(self, tmp_path) -> None:
        """Test that temp file is cleaned up after successful load."""
        import os
        import tempfile

        content = b"Hello, world!"

        # Get initial temp file count
        temp_dir = tempfile.gettempdir()
        initial_files = set(os.listdir(temp_dir))

        # Load content
        _docs = load_from_bytes(content, ".txt", "test.txt")
        assert _docs is not None  # Ensure load succeeded

        # Check temp file was cleaned up
        final_files = set(os.listdir(temp_dir))
        new_ldr_files = [
            f
            for f in (final_files - initial_files)
            if f.startswith("ldr_upload_")
        ]

        assert len(new_ldr_files) == 0, "Temp file was not cleaned up"

    def test_temp_file_cleaned_up_on_error(self) -> None:
        """Test that temp file is cleaned up even on parse error."""
        import json
        import os
        import tempfile

        # Invalid JSON that will cause parsing error
        content = b'{"invalid": json content'

        temp_dir = tempfile.gettempdir()
        initial_files = set(os.listdir(temp_dir))

        # The JSON loader currently handles parse errors by returning a
        # Document with parse_error=True. If a future implementation switches
        # to raising, only the expected parse-error exception types should be
        # swallowed — anything else is a real bug.
        try:
            load_from_bytes(content, ".json", "invalid.json")
        except (json.JSONDecodeError, ValueError):
            pass

        final_files = set(os.listdir(temp_dir))
        new_ldr_files = [
            f
            for f in (final_files - initial_files)
            if f.startswith("ldr_upload_")
        ]

        assert len(new_ldr_files) == 0, (
            "Temp file was not cleaned up after error"
        )