7a0da7932b
Backwards Compatibility / Verify Encryption Constants (push) Waiting to run
Backwards Compatibility / PyPI Version Compatibility (push) Waiting to run
Backwards Compatibility / Database Migration Tests (push) Waiting to run
CodeQL Advanced / Analyze (javascript-typescript) (push) Waiting to run
CodeQL Advanced / Analyze (python) (push) Waiting to run
Docker Tests (Consolidated) / UI Tests (Puppeteer) [research-form] (push) Blocked by required conditions
Docker Tests (Consolidated) / UI Tests (Puppeteer) [research-metrics] (push) Blocked by required conditions
Docker Tests (Consolidated) / UI Tests (Puppeteer) [research-workflow] (push) Blocked by required conditions
Docker Tests (Consolidated) / UI Tests (Puppeteer) [settings-core] (push) Blocked by required conditions
Docker Tests (Consolidated) / UI Tests (Puppeteer) [settings-pages] (push) Blocked by required conditions
Docker Tests (Consolidated) / UI Tests (Puppeteer) [history-news] (push) Blocked by required conditions
Docker Tests (Consolidated) / UI Tests (Puppeteer) [library] (push) Blocked by required conditions
Docker Tests (Consolidated) / UI Tests (Puppeteer) [link-analytics] (push) Blocked by required conditions
Docker Tests (Consolidated) / UI Tests (Puppeteer) [mobile] (push) Blocked by required conditions
Docker Tests (Consolidated) / detect-changes (push) Waiting to run
Docker Tests (Consolidated) / Build Test Image (push) Waiting to run
Docker Tests (Consolidated) / All Pytest Tests + Coverage (push) Blocked by required conditions
Docker Tests (Consolidated) / UI Tests (Puppeteer) [accessibility] (push) Blocked by required conditions
Docker Tests (Consolidated) / UI Tests (Puppeteer) [api-crud] (push) Blocked by required conditions
Docker Tests (Consolidated) / UI Tests (Puppeteer) [auth-login] (push) Blocked by required conditions
Docker Tests (Consolidated) / UI Tests (Puppeteer) [auth-pages] (push) Blocked by required conditions
Docker Tests (Consolidated) / UI Tests (Puppeteer) [auth-register] (push) Blocked by required conditions
Docker Tests (Consolidated) / UI Tests (Puppeteer) [chat-core] (push) Blocked by required conditions
Docker Tests (Consolidated) / UI Tests (Puppeteer) [chat-lifecycle] (push) Blocked by required conditions
Docker Tests (Consolidated) / UI Tests (Puppeteer) [error-benchmark] (push) Blocked by required conditions
Docker Tests (Consolidated) / UI Tests (Puppeteer) (push) Blocked by required conditions
Docker Tests (Consolidated) / Accessibility Tests (push) Blocked by required conditions
Docker Tests (Consolidated) / LLM Unit Tests (push) Blocked by required conditions
Docker Tests (Consolidated) / LLM Example Tests (push) Blocked by required conditions
Docker Tests (Consolidated) / Production Image Smoke Test (push) Blocked by required conditions
Docker Tests (Consolidated) / Infrastructure Tests (push) Blocked by required conditions
OSSF Scorecard / OSSF Security Scorecard Analysis (push) Waiting to run
OSV-Scanner (Scheduled) / scan-scheduled (push) Failing after 0s
Create Release / test-gate (push) Has been cancelled
Create Release / release-gate (push) Has been cancelled
Create Release / ci-gate (push) Has been cancelled
Create Release / version-check (push) Has been cancelled
Create Release / e2e-test-gate (push) Has been cancelled
Create Release / responsive-test-gate (push) Has been cancelled
Create Release / compat-test-gate (push) Has been cancelled
Create Release / compose-integration-gate (push) Has been cancelled
Create Release / vulture-gate (push) Has been cancelled
Create Release / build (push) Has been cancelled
Create Release / provenance (push) Has been cancelled
Create Release / prerelease-docker (push) Has been cancelled
Create Release / publish-docker (push) Has been cancelled
Create Release / create-release (push) Has been cancelled
Create Release / cleanup-changelog (push) Has been cancelled
Create Release / trigger-pypi (push) Has been cancelled
Create Release / monitor-pypi (push) Has been cancelled
Create Release / Clean up orphan prerelease tags and signatures (push) Has been cancelled
277 lines
9.3 KiB
Python
277 lines
9.3 KiB
Python
"""Tests for SimpleJSONLoader."""
|
|
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from local_deep_research.document_loaders.json_loader import (
|
|
SimpleJSONLoader,
|
|
extract_strings_from_json,
|
|
extract_text_from_json,
|
|
)
|
|
|
|
|
|
class TestExtractStringsFromJson:
|
|
"""Tests for extract_strings_from_json helper function."""
|
|
|
|
def test_extract_simple_string(self) -> None:
|
|
"""Test extracting a simple string value."""
|
|
data = "hello"
|
|
result = extract_strings_from_json(data)
|
|
|
|
assert result == [("", "hello")]
|
|
|
|
def test_extract_from_dict(self) -> None:
|
|
"""Test extracting strings from a dictionary."""
|
|
data = {"name": "Alice", "city": "Boston"}
|
|
result = extract_strings_from_json(data)
|
|
|
|
assert ("name", "Alice") in result
|
|
assert ("city", "Boston") in result
|
|
|
|
def test_extract_from_nested_dict(self) -> None:
|
|
"""Test extracting strings from nested dictionary."""
|
|
data = {"user": {"name": "Bob", "email": "bob@example.com"}}
|
|
result = extract_strings_from_json(data)
|
|
|
|
assert ("user.name", "Bob") in result
|
|
assert ("user.email", "bob@example.com") in result
|
|
|
|
def test_extract_from_list(self) -> None:
|
|
"""Test extracting strings from a list."""
|
|
data = ["apple", "banana", "cherry"]
|
|
result = extract_strings_from_json(data)
|
|
|
|
assert ("[0]", "apple") in result
|
|
assert ("[1]", "banana") in result
|
|
assert ("[2]", "cherry") in result
|
|
|
|
def test_extract_from_nested_list(self) -> None:
|
|
"""Test extracting strings from nested list in dict."""
|
|
data = {"fruits": ["apple", "banana"]}
|
|
result = extract_strings_from_json(data)
|
|
|
|
assert ("fruits[0]", "apple") in result
|
|
assert ("fruits[1]", "banana") in result
|
|
|
|
def test_skip_numbers_and_booleans(self) -> None:
|
|
"""Test that numbers and booleans are skipped."""
|
|
data = {"count": 42, "active": True, "name": "test"}
|
|
result = extract_strings_from_json(data)
|
|
|
|
# Only name should be extracted
|
|
assert len(result) == 1
|
|
assert ("name", "test") in result
|
|
|
|
def test_skip_none(self) -> None:
|
|
"""Test that None values are skipped."""
|
|
data = {"value": None, "name": "test"}
|
|
result = extract_strings_from_json(data)
|
|
|
|
assert len(result) == 1
|
|
assert ("name", "test") in result
|
|
|
|
def test_empty_dict(self) -> None:
|
|
"""Test with empty dictionary."""
|
|
data = {}
|
|
result = extract_strings_from_json(data)
|
|
|
|
assert result == []
|
|
|
|
def test_empty_list(self) -> None:
|
|
"""Test with empty list."""
|
|
data = []
|
|
result = extract_strings_from_json(data)
|
|
|
|
assert result == []
|
|
|
|
def test_deeply_nested(self) -> None:
|
|
"""Test deeply nested structure."""
|
|
data = {"a": {"b": {"c": {"d": "deep"}}}}
|
|
result = extract_strings_from_json(data)
|
|
|
|
assert ("a.b.c.d", "deep") in result
|
|
|
|
|
|
class TestSimpleJSONLoader:
|
|
"""Tests for SimpleJSONLoader class."""
|
|
|
|
def test_load_simple_json(self, tmp_path: Path) -> None:
|
|
"""Test loading a simple JSON file."""
|
|
json_file = tmp_path / "test.json"
|
|
json_file.write_text('{"name": "test", "value": "hello"}')
|
|
|
|
loader = SimpleJSONLoader(json_file)
|
|
docs = loader.load()
|
|
|
|
assert len(docs) == 1
|
|
assert "name: test" in docs[0].page_content
|
|
assert "value: hello" in docs[0].page_content
|
|
assert docs[0].metadata["file_type"] == "json"
|
|
|
|
def test_load_nested_json(self, tmp_path: Path) -> None:
|
|
"""Test loading nested JSON structure."""
|
|
json_content = (
|
|
'{"user": {"name": "Alice", "email": "alice@example.com"}}'
|
|
)
|
|
json_file = tmp_path / "nested.json"
|
|
json_file.write_text(json_content)
|
|
|
|
loader = SimpleJSONLoader(json_file)
|
|
docs = loader.load()
|
|
|
|
assert len(docs) == 1
|
|
assert "user.name: Alice" in docs[0].page_content
|
|
assert "user.email: alice@example.com" in docs[0].page_content
|
|
|
|
def test_load_json_array(self, tmp_path: Path) -> None:
|
|
"""Test loading JSON with array."""
|
|
json_content = '{"items": ["one", "two", "three"]}'
|
|
json_file = tmp_path / "array.json"
|
|
json_file.write_text(json_content)
|
|
|
|
loader = SimpleJSONLoader(json_file)
|
|
docs = loader.load()
|
|
|
|
assert len(docs) == 1
|
|
assert "items[0]: one" in docs[0].page_content
|
|
assert "items[1]: two" in docs[0].page_content
|
|
|
|
def test_load_invalid_json_returns_raw_content(
|
|
self, tmp_path: Path
|
|
) -> None:
|
|
"""Test that invalid JSON returns raw content with parse_error flag."""
|
|
json_file = tmp_path / "invalid.json"
|
|
json_file.write_text('{"invalid": json content')
|
|
|
|
loader = SimpleJSONLoader(json_file)
|
|
docs = loader.load()
|
|
|
|
assert len(docs) == 1
|
|
assert docs[0].metadata.get("parse_error") is True
|
|
assert '{"invalid":' in docs[0].page_content
|
|
|
|
def test_load_without_paths(self, tmp_path: Path) -> None:
|
|
"""Test loading with include_paths=False."""
|
|
json_file = tmp_path / "test.json"
|
|
json_file.write_text('{"name": "test", "city": "Boston"}')
|
|
|
|
loader = SimpleJSONLoader(json_file, include_paths=False)
|
|
docs = loader.load()
|
|
|
|
assert len(docs) == 1
|
|
# Should contain values but not paths
|
|
assert "test" in docs[0].page_content
|
|
assert "Boston" in docs[0].page_content
|
|
# Should NOT contain "name:" prefix
|
|
assert "name:" not in docs[0].page_content
|
|
|
|
def test_load_json_with_unicode(self, tmp_path: Path) -> None:
|
|
"""Test loading JSON with unicode characters."""
|
|
json_content = '{"greeting": "こんにちは", "emoji": "🎉"}'
|
|
json_file = tmp_path / "unicode.json"
|
|
json_file.write_text(json_content, encoding="utf-8")
|
|
|
|
loader = SimpleJSONLoader(json_file)
|
|
docs = loader.load()
|
|
|
|
assert len(docs) == 1
|
|
assert "こんにちは" in docs[0].page_content
|
|
assert "🎉" in docs[0].page_content
|
|
|
|
def test_string_count_metadata(self, tmp_path: Path) -> None:
|
|
"""Test that string_count metadata is set correctly."""
|
|
json_file = tmp_path / "test.json"
|
|
json_file.write_text('{"a": "one", "b": "two", "c": "three"}')
|
|
|
|
loader = SimpleJSONLoader(json_file)
|
|
docs = loader.load()
|
|
|
|
assert docs[0].metadata["string_count"] == 3
|
|
|
|
def test_lazy_load(self, tmp_path: Path) -> None:
|
|
"""Test lazy_load returns an iterator."""
|
|
json_file = tmp_path / "test.json"
|
|
json_file.write_text('{"key": "value"}')
|
|
|
|
loader = SimpleJSONLoader(json_file)
|
|
result = loader.lazy_load()
|
|
|
|
assert hasattr(result, "__iter__")
|
|
assert hasattr(result, "__next__")
|
|
|
|
docs = list(result)
|
|
assert len(docs) == 1
|
|
|
|
def test_file_not_found_raises_error(self) -> None:
|
|
"""Test that missing file raises appropriate error."""
|
|
loader = SimpleJSONLoader("/nonexistent/path/file.json")
|
|
|
|
with pytest.raises(Exception):
|
|
loader.load()
|
|
|
|
def test_load_json_with_duplicate_keys(self, tmp_path: Path) -> None:
|
|
"""JSON with duplicate keys - last value wins per RFC 8259."""
|
|
json_file = tmp_path / "duplicates.json"
|
|
json_file.write_text('{"key": "first", "key": "second"}')
|
|
|
|
loader = SimpleJSONLoader(json_file)
|
|
docs = loader.load()
|
|
|
|
assert len(docs) == 1
|
|
# Python's json.loads keeps last value for duplicate keys
|
|
assert "second" in docs[0].page_content
|
|
|
|
|
|
class TestExtractTextFromJson:
|
|
"""Tests for extract_text_from_json function."""
|
|
|
|
def test_extract_simple_json(self) -> None:
|
|
"""Test extracting text from simple JSON bytes."""
|
|
content = b'{"name": "test", "value": "hello"}'
|
|
result = extract_text_from_json(content)
|
|
|
|
assert "name: test" in result
|
|
assert "value: hello" in result
|
|
|
|
def test_extract_nested_json(self) -> None:
|
|
"""Test extracting text from nested JSON bytes."""
|
|
content = b'{"user": {"name": "Alice"}}'
|
|
result = extract_text_from_json(content)
|
|
|
|
assert "user.name: Alice" in result
|
|
|
|
def test_extract_invalid_json_returns_raw(self) -> None:
|
|
"""Test that invalid JSON returns raw content."""
|
|
content = b'{"invalid": json'
|
|
result = extract_text_from_json(content)
|
|
|
|
# Should return the raw content
|
|
assert '{"invalid":' in result
|
|
|
|
def test_extract_with_encoding_errors(self) -> None:
|
|
"""Test handling of encoding errors."""
|
|
# Invalid UTF-8 bytes mixed with valid JSON
|
|
content = b'{"name": "test\xff\xfe"}'
|
|
result = extract_text_from_json(content)
|
|
|
|
# Should still return something
|
|
assert "name" in result
|
|
|
|
def test_extract_empty_object(self) -> None:
|
|
"""Test extracting from empty JSON object."""
|
|
content = b"{}"
|
|
result = extract_text_from_json(content)
|
|
|
|
assert result == ""
|
|
|
|
def test_extract_filters_empty_strings(self) -> None:
|
|
"""Test that empty strings are filtered out."""
|
|
content = b'{"empty": "", "valid": "value"}'
|
|
result = extract_text_from_json(content)
|
|
|
|
assert "valid: value" in result
|
|
# Empty string should not create a line
|
|
lines = [line for line in result.split("\n") if line.strip()]
|
|
assert len(lines) == 1
|