learningcircuit--local-deep-research/tests/research_library/utils/test_utils.py

"""Tests for research_library/utils/__init__.py — utility functions."""

import hashlib
from unittest.mock import MagicMock, patch


# ---------------------------------------------------------------------------
# get_url_hash
# ---------------------------------------------------------------------------


class TestGetUrlHash:
    """Tests for get_url_hash()."""

    def test_returns_sha256_hex(self):
        from local_deep_research.research_library.utils import get_url_hash

        url = "https://example.com/page"
        expected = hashlib.sha256(url.lower().encode()).hexdigest()
        assert get_url_hash(url) == expected

    def test_case_insensitive(self):
        from local_deep_research.research_library.utils import get_url_hash

        assert get_url_hash("HTTPS://EXAMPLE.COM") == get_url_hash(
            "https://example.com"
        )

    def test_different_urls_different_hashes(self):
        from local_deep_research.research_library.utils import get_url_hash

        assert get_url_hash("https://a.com") != get_url_hash("https://b.com")

    def test_empty_url(self):
        from local_deep_research.research_library.utils import get_url_hash

        result = get_url_hash("")
        assert isinstance(result, str)
        assert len(result) == 64  # SHA256 hex length


# ---------------------------------------------------------------------------
# get_library_storage_path
# ---------------------------------------------------------------------------


class TestGetLibraryStoragePath:
    """Tests for get_library_storage_path()."""

    @patch("local_deep_research.utilities.db_utils.get_settings_manager")
    @patch("local_deep_research.research_library.utils.get_library_directory")
    def test_user_isolation_mode(self, mock_lib_dir, mock_get_sm, tmp_path):
        mock_lib_dir.return_value = tmp_path / "library"
        mock_sm = MagicMock()
        mock_sm.get_setting.side_effect = lambda key, default: {
            "research_library.storage_path": str(tmp_path / "library"),
            "research_library.shared_library": False,
        }.get(key, default)
        mock_get_sm.return_value = mock_sm

        from local_deep_research.research_library.utils import (
            get_library_storage_path,
        )

        result = get_library_storage_path("alice")
        assert result == tmp_path / "library" / "alice"
        assert result.exists()

    @patch("local_deep_research.utilities.db_utils.get_settings_manager")
    @patch("local_deep_research.research_library.utils.get_library_directory")
    def test_shared_library_mode(self, mock_lib_dir, mock_get_sm, tmp_path):
        mock_lib_dir.return_value = tmp_path / "library"
        mock_sm = MagicMock()
        mock_sm.get_setting.side_effect = lambda key, default: {
            "research_library.storage_path": str(tmp_path / "library"),
            "research_library.shared_library": True,
        }.get(key, default)
        mock_get_sm.return_value = mock_sm

        from local_deep_research.research_library.utils import (
            get_library_storage_path,
        )

        result = get_library_storage_path("alice")
        # Shared mode: no username subdirectory
        assert result == tmp_path / "library"
        assert result.exists()

    @patch("local_deep_research.utilities.db_utils.get_settings_manager")
    @patch("local_deep_research.research_library.utils.get_library_directory")
    def test_creates_directory_if_not_exists(
        self, mock_lib_dir, mock_get_sm, tmp_path
    ):
        base = tmp_path / "new_dir"
        mock_lib_dir.return_value = base
        mock_sm = MagicMock()
        mock_sm.get_setting.side_effect = lambda key, default: {
            "research_library.storage_path": str(base),
            "research_library.shared_library": False,
        }.get(key, default)
        mock_get_sm.return_value = mock_sm

        from local_deep_research.research_library.utils import (
            get_library_storage_path,
        )

        result = get_library_storage_path("bob")
        assert result.exists()
        assert result.is_dir()


# ---------------------------------------------------------------------------
# open_file_location
# ---------------------------------------------------------------------------


class TestOpenFileLocation:
    """Tests for open_file_location()."""

    @patch("local_deep_research.research_library.utils.PathValidator")
    @patch("local_deep_research.research_library.utils.subprocess.run")
    @patch("local_deep_research.research_library.utils.sys")
    def test_linux_success(self, mock_sys, mock_run, mock_validator, tmp_path):
        mock_sys.platform = "linux"
        mock_validator.validate_local_filesystem_path.return_value = (
            tmp_path / "file.txt"
        )
        mock_run.return_value = MagicMock(returncode=0)

        from local_deep_research.research_library.utils import (
            open_file_location,
        )

        assert open_file_location(str(tmp_path / "file.txt")) is True
        mock_run.assert_called_once_with(
            ["xdg-open", str(tmp_path)],
            capture_output=True,
            text=True,
            shell=False,
        )

    @patch("local_deep_research.research_library.utils.PathValidator")
    @patch("local_deep_research.research_library.utils.subprocess.run")
    @patch("local_deep_research.research_library.utils.sys")
    def test_linux_failure(self, mock_sys, mock_run, mock_validator, tmp_path):
        mock_sys.platform = "linux"
        mock_validator.validate_local_filesystem_path.return_value = (
            tmp_path / "file.txt"
        )
        mock_run.return_value = MagicMock(returncode=1, stderr="error")

        from local_deep_research.research_library.utils import (
            open_file_location,
        )

        assert open_file_location(str(tmp_path / "file.txt")) is False

    @patch("local_deep_research.research_library.utils.PathValidator")
    @patch("local_deep_research.research_library.utils.subprocess.run")
    @patch("local_deep_research.research_library.utils.sys")
    def test_macos_success(self, mock_sys, mock_run, mock_validator, tmp_path):
        mock_sys.platform = "darwin"
        mock_validator.validate_local_filesystem_path.return_value = (
            tmp_path / "file.txt"
        )
        mock_run.return_value = MagicMock(returncode=0)

        from local_deep_research.research_library.utils import (
            open_file_location,
        )

        assert open_file_location(str(tmp_path / "file.txt")) is True
        mock_run.assert_called_once_with(
            ["open", str(tmp_path)],
            capture_output=True,
            text=True,
            shell=False,
        )

    @patch("local_deep_research.research_library.utils.PathValidator")
    def test_path_validation_failure(self, mock_validator):
        mock_validator.validate_local_filesystem_path.side_effect = ValueError(
            "blocked"
        )

        from local_deep_research.research_library.utils import (
            open_file_location,
        )

        assert open_file_location("/etc/passwd") is False


# ---------------------------------------------------------------------------
# get_absolute_library_path
# ---------------------------------------------------------------------------


class TestLibraryPathConversions:
    """Tests for get_absolute_library_path."""

    @patch(
        "local_deep_research.research_library.utils.get_library_storage_path"
    )
    def test_absolute_path_from_relative(self, mock_storage, tmp_path):
        library_root = tmp_path / "library" / "alice"
        mock_storage.return_value = library_root

        from local_deep_research.research_library.utils import (
            get_absolute_library_path,
        )

        result = get_absolute_library_path("subdir/file.pdf", "alice")
        assert result == library_root / "subdir" / "file.pdf"


# ---------------------------------------------------------------------------
# get_absolute_path_from_settings
# ---------------------------------------------------------------------------


class TestGetAbsolutePathFromSettings:
    """Tests for get_absolute_path_from_settings."""

    @patch("local_deep_research.utilities.db_utils.get_settings_manager")
    @patch("local_deep_research.research_library.utils.get_library_directory")
    def test_with_relative_path(self, mock_lib_dir, mock_get_sm, tmp_path):
        mock_lib_dir.return_value = tmp_path / "library"
        mock_sm = MagicMock()
        mock_sm.get_setting.return_value = str(tmp_path / "library")
        mock_get_sm.return_value = mock_sm

        from local_deep_research.research_library.utils import (
            get_absolute_path_from_settings,
        )

        result = get_absolute_path_from_settings("docs/file.pdf")
        assert result == tmp_path / "library" / "docs" / "file.pdf"

    @patch("local_deep_research.utilities.db_utils.get_settings_manager")
    @patch("local_deep_research.research_library.utils.get_library_directory")
    def test_empty_relative_path_returns_root(
        self, mock_lib_dir, mock_get_sm, tmp_path
    ):
        mock_lib_dir.return_value = tmp_path / "library"
        mock_sm = MagicMock()
        mock_sm.get_setting.return_value = str(tmp_path / "library")
        mock_get_sm.return_value = mock_sm

        from local_deep_research.research_library.utils import (
            get_absolute_path_from_settings,
        )

        result = get_absolute_path_from_settings("")
        assert result == tmp_path / "library"


# ---------------------------------------------------------------------------
# handle_api_error
# ---------------------------------------------------------------------------


class TestHandleApiError:
    """Tests for handle_api_error."""

    def test_returns_generic_error_message(self, app):
        from local_deep_research.research_library.utils import handle_api_error

        with app.app_context():
            response, status_code = handle_api_error(
                "test operation", ValueError("secret details"), 500
            )
            data = response.get_json()
            assert status_code == 500
            assert data["success"] is False
            # Must NOT leak the exception message
            assert "secret details" not in data["error"]
            assert "internal error" in data["error"].lower()

    def test_custom_status_code(self, app):
        from local_deep_research.research_library.utils import handle_api_error

        with app.app_context():
            _, status_code = handle_api_error("op", RuntimeError("err"), 503)
            assert status_code == 503

    def test_default_status_code_is_500(self, app):
        from local_deep_research.research_library.utils import handle_api_error

        with app.app_context():
            _, status_code = handle_api_error("op", RuntimeError("err"))
            assert status_code == 500


# ---------------------------------------------------------------------------
# ensure_in_collection
# ---------------------------------------------------------------------------


class TestEnsureInCollection:
    """Tests for ensure_in_collection()."""

    def test_returns_existing_row_without_adding(self):
        from local_deep_research.research_library.utils import (
            ensure_in_collection,
        )

        existing = MagicMock(name="existing_DocumentCollection")
        session = MagicMock()
        session.query.return_value.filter_by.return_value.first.return_value = (
            existing
        )

        result = ensure_in_collection(session, "doc-1", "coll-1")

        assert result is existing
        session.query.return_value.filter_by.assert_called_once_with(
            document_id="doc-1", collection_id="coll-1"
        )
        session.add.assert_not_called()

    def test_creates_and_adds_when_missing(self):
        from local_deep_research.research_library.utils import (
            ensure_in_collection,
        )

        session = MagicMock()
        session.query.return_value.filter_by.return_value.first.return_value = (
            None
        )

        result = ensure_in_collection(session, "doc-2", "coll-2")

        session.add.assert_called_once_with(result)
        assert result.document_id == "doc-2"
        assert result.collection_id == "coll-2"
        assert result.indexed is False