learningcircuit--local-deep-research/tests/storage/test_file_storage.py

"""Tests for file storage implementation."""

import json
from pathlib import Path
from unittest.mock import patch

import pytest

from local_deep_research.storage.file import FileReportStorage


class TestFileReportStorageInit:
    """Tests for FileReportStorage initialization."""

    def test_uses_default_directory_when_none(self, tmp_path):
        """Should use default research outputs directory when none provided."""
        with patch(
            "local_deep_research.storage.file.get_research_outputs_directory",
            return_value=tmp_path,
        ):
            storage = FileReportStorage()
            assert storage.base_dir == tmp_path

    def test_uses_provided_directory(self, tmp_path):
        """Should use provided directory."""
        custom_dir = tmp_path / "custom"
        storage = FileReportStorage(base_dir=custom_dir)
        assert storage.base_dir == custom_dir

    def test_creates_directory_if_not_exists(self, tmp_path):
        """Should create directory if it doesn't exist."""
        new_dir = tmp_path / "new_reports"
        storage = FileReportStorage(base_dir=new_dir)
        assert storage.base_dir.exists()


class TestGetReportPath:
    """Tests for _get_report_path method."""

    def test_returns_md_file_path(self, tmp_path):
        """Should return path with .md extension."""
        storage = FileReportStorage(base_dir=tmp_path)
        path = storage._get_report_path("test-uuid")
        assert path == tmp_path / "test-uuid.md"

    def test_returns_path_object(self, tmp_path):
        """Should return Path object."""
        storage = FileReportStorage(base_dir=tmp_path)
        path = storage._get_report_path("test-uuid")
        assert isinstance(path, Path)


class TestGetMetadataPath:
    """Tests for _get_metadata_path method."""

    def test_returns_metadata_json_path(self, tmp_path):
        """Should return path with _metadata.json suffix."""
        storage = FileReportStorage(base_dir=tmp_path)
        path = storage._get_metadata_path("test-uuid")
        assert path == tmp_path / "test-uuid_metadata.json"


class TestPathTraversalContainment:
    """Regression tests for the #3090 traversal guard: a research_id that
    escapes base_dir must be rejected, not resolved to an arbitrary on-disk
    path. (research_ids are normally server-generated UUIDs; this is
    defense-in-depth.)"""

    def test_report_path_rejects_parent_traversal(self, tmp_path):
        storage = FileReportStorage(base_dir=tmp_path)
        with pytest.raises(ValueError, match="Path traversal attempt"):
            storage._get_report_path("../../etc/passwd")

    def test_metadata_path_rejects_parent_traversal(self, tmp_path):
        storage = FileReportStorage(base_dir=tmp_path)
        with pytest.raises(ValueError, match="Path traversal attempt"):
            storage._get_metadata_path("../../etc/passwd")

    def test_report_path_rejects_single_level_escape(self, tmp_path):
        storage = FileReportStorage(base_dir=tmp_path / "reports")
        with pytest.raises(ValueError, match="Path traversal attempt"):
            storage._get_report_path("../escapee")

    def test_normal_id_stays_within_base_dir(self, tmp_path):
        storage = FileReportStorage(base_dir=tmp_path)
        path = storage._get_report_path("9f8e-uuid-1234")
        assert path.is_relative_to(tmp_path.resolve())
        assert path.name == "9f8e-uuid-1234.md"


class TestSaveReport:
    """Tests for save_report method."""

    def test_saves_report_content(self, tmp_path, sample_report_content):
        """Should save report content to file."""
        storage = FileReportStorage(base_dir=tmp_path)

        # Patch at source location since it's imported inside the method
        with patch(
            "local_deep_research.security.file_write_verifier.write_file_verified"
        ) as mock_write:
            with patch(
                "local_deep_research.security.file_write_verifier.write_json_verified"
            ):
                result = storage.save_report("test-uuid", sample_report_content)

        assert result is True
        mock_write.assert_called_once()

    def test_saves_metadata_when_provided(
        self, tmp_path, sample_report_content, sample_metadata
    ):
        """Should save metadata when provided."""
        storage = FileReportStorage(base_dir=tmp_path)

        with patch(
            "local_deep_research.security.file_write_verifier.write_file_verified"
        ):
            with patch(
                "local_deep_research.security.file_write_verifier.write_json_verified"
            ) as mock_json:
                storage.save_report(
                    "test-uuid", sample_report_content, metadata=sample_metadata
                )

        mock_json.assert_called_once()

    def test_skips_metadata_when_not_provided(
        self, tmp_path, sample_report_content
    ):
        """Should skip metadata save when not provided."""
        storage = FileReportStorage(base_dir=tmp_path)

        with patch(
            "local_deep_research.security.file_write_verifier.write_file_verified"
        ):
            with patch(
                "local_deep_research.security.file_write_verifier.write_json_verified"
            ) as mock_json:
                storage.save_report("test-uuid", sample_report_content)

        mock_json.assert_not_called()

    def test_returns_false_on_error(self, tmp_path, sample_report_content):
        """Should return False on error."""
        storage = FileReportStorage(base_dir=tmp_path)

        with patch(
            "local_deep_research.security.file_write_verifier.write_file_verified",
            side_effect=Exception("write error"),
        ):
            result = storage.save_report("test-uuid", sample_report_content)

        assert result is False

    def test_save_report_gates_on_report_enable_file_backup(
        self, tmp_path, sample_report_content
    ):
        """Regression: the verifier gate must consult the same setting key
        that the UI toggle writes (``report.enable_file_backup``). The
        previous literal ``storage.allow_file_backup`` was never registered
        in ``default_settings.json``, so every save raised
        ``FileWriteSecurityError`` regardless of the user's toggle.
        """
        from local_deep_research.security.file_write_verifier import (
            FileWriteSecurityError,
        )

        captured = {}

        def fake_write_file_verified(path, content, setting_name, **_kwargs):
            captured["setting_name"] = setting_name
            raise FileWriteSecurityError("ignored")

        storage = FileReportStorage(base_dir=tmp_path)

        with patch(
            "local_deep_research.security.file_write_verifier.write_file_verified",
            fake_write_file_verified,
        ):
            result = storage.save_report("abc", sample_report_content)

        # The exception is swallowed by FileReportStorage.save_report, so the
        # outer result is False -- the bug we care about is the *key* passed
        # to the verifier, not the return value.
        assert result is False
        assert captured["setting_name"] == "report.enable_file_backup"

    def test_save_metadata_gates_on_report_enable_file_backup(
        self, tmp_path, sample_report_content, sample_metadata
    ):
        """The metadata write must consult the same UI-visible key as the
        body write. A divergence here would silently drop the
        ``_metadata.json`` sidecar even when the body lands on disk.
        """
        from local_deep_research.security.file_write_verifier import (
            FileWriteSecurityError,
        )

        captured = {}

        def fake_write_file_verified(path, content, setting_name, **_kwargs):
            # Allow the body write through so the metadata branch runs.
            captured.setdefault("body_setting_name", setting_name)

        def fake_write_json_verified(path, payload, setting_name, **_kwargs):
            captured["metadata_setting_name"] = setting_name
            raise FileWriteSecurityError("ignored")

        storage = FileReportStorage(base_dir=tmp_path)

        with (
            patch(
                "local_deep_research.security.file_write_verifier.write_file_verified",
                fake_write_file_verified,
            ),
            patch(
                "local_deep_research.security.file_write_verifier.write_json_verified",
                fake_write_json_verified,
            ),
        ):
            storage.save_report(
                "abc", sample_report_content, metadata=sample_metadata
            )

        assert captured["body_setting_name"] == "report.enable_file_backup"
        assert captured["metadata_setting_name"] == "report.enable_file_backup"

    def test_save_report_and_factory_agree_on_setting_key(
        self, tmp_path, sample_report_content, mock_session
    ):
        """End-to-end key-alignment check: the key consulted by
        ``FileReportStorage.save_report`` must match the key consulted by
        ``get_report_storage``. If either side drifts, a user who flipped
        the UI toggle would still see no files on disk.
        """
        from local_deep_research.storage.factory import get_report_storage

        factory_calls = []

        def fake_get_setting(name, settings_snapshot=None):
            factory_calls.append(name)
            return True

        file_calls = []

        from local_deep_research.security.file_write_verifier import (
            FileWriteSecurityError,
        )

        def fake_write_file_verified(path, content, setting_name, **_kwargs):
            file_calls.append(setting_name)
            raise FileWriteSecurityError("ignored")

        with (
            patch(
                "local_deep_research.storage.factory.get_setting_from_snapshot",
                fake_get_setting,
            ),
            patch(
                "local_deep_research.security.file_write_verifier.write_file_verified",
                fake_write_file_verified,
            ),
        ):
            # Factory side: which key unlocks file backup?
            factory = get_report_storage(session=mock_session)
            assert factory.enable_file_storage is True

            # Storage side: which key gates the disk write?
            fs = FileReportStorage(base_dir=tmp_path)
            fs.save_report("abc", sample_report_content)

        assert factory_calls == ["report.enable_file_backup"]
        assert file_calls == ["report.enable_file_backup"]
        assert factory_calls == file_calls


class TestGetReport:
    """Tests for get_report method."""

    def test_returns_content_when_file_exists(
        self, tmp_path, sample_report_content
    ):
        """Should return content when file exists."""
        storage = FileReportStorage(base_dir=tmp_path)
        report_path = tmp_path / "test-uuid.md"
        report_path.write_text(sample_report_content, encoding="utf-8")

        result = storage.get_report("test-uuid")

        assert result == sample_report_content

    def test_returns_none_when_file_not_exists(self, tmp_path):
        """Should return None when file doesn't exist."""
        storage = FileReportStorage(base_dir=tmp_path)

        result = storage.get_report("nonexistent-uuid")

        assert result is None

    def test_returns_none_on_error(self, tmp_path):
        """Should return None on read error."""
        storage = FileReportStorage(base_dir=tmp_path)
        report_path = tmp_path / "test-uuid.md"
        report_path.write_text("content", encoding="utf-8")

        with patch("builtins.open", side_effect=Exception("read error")):
            result = storage.get_report("test-uuid")

        assert result is None


class TestGetReportWithMetadata:
    """Tests for get_report_with_metadata method."""

    def test_returns_content_and_metadata(
        self, tmp_path, sample_report_content, sample_metadata
    ):
        """Should return both content and metadata."""
        storage = FileReportStorage(base_dir=tmp_path)

        # Create files
        report_path = tmp_path / "test-uuid.md"
        report_path.write_text(sample_report_content, encoding="utf-8")

        metadata_path = tmp_path / "test-uuid_metadata.json"
        metadata_path.write_text(json.dumps(sample_metadata), encoding="utf-8")

        result = storage.get_report_with_metadata("test-uuid")

        assert result["content"] == sample_report_content
        assert result["metadata"] == sample_metadata

    def test_returns_empty_metadata_when_missing(
        self, tmp_path, sample_report_content
    ):
        """Should return empty metadata when file missing."""
        storage = FileReportStorage(base_dir=tmp_path)

        report_path = tmp_path / "test-uuid.md"
        report_path.write_text(sample_report_content, encoding="utf-8")

        result = storage.get_report_with_metadata("test-uuid")

        assert result["content"] == sample_report_content
        assert result["metadata"] == {}

    def test_returns_none_when_report_missing(self, tmp_path):
        """Should return None when report doesn't exist."""
        storage = FileReportStorage(base_dir=tmp_path)

        result = storage.get_report_with_metadata("nonexistent")

        assert result is None


class TestDeleteReport:
    """Tests for delete_report method."""

    def test_deletes_report_file(self, tmp_path, sample_report_content):
        """Should delete report file."""
        storage = FileReportStorage(base_dir=tmp_path)

        report_path = tmp_path / "test-uuid.md"
        report_path.write_text(sample_report_content, encoding="utf-8")

        result = storage.delete_report("test-uuid")

        assert result is True
        assert not report_path.exists()

    def test_deletes_metadata_file(
        self, tmp_path, sample_report_content, sample_metadata
    ):
        """Should delete metadata file."""
        storage = FileReportStorage(base_dir=tmp_path)

        report_path = tmp_path / "test-uuid.md"
        report_path.write_text(sample_report_content, encoding="utf-8")

        metadata_path = tmp_path / "test-uuid_metadata.json"
        metadata_path.write_text(json.dumps(sample_metadata), encoding="utf-8")

        storage.delete_report("test-uuid")

        assert not metadata_path.exists()

    def test_returns_false_when_files_not_exist(self, tmp_path):
        """Should return False when no files to delete."""
        storage = FileReportStorage(base_dir=tmp_path)

        result = storage.delete_report("nonexistent")

        assert result is False

    def test_returns_false_on_error(self, tmp_path, sample_report_content):
        """Should return False on delete error."""
        storage = FileReportStorage(base_dir=tmp_path)

        report_path = tmp_path / "test-uuid.md"
        report_path.write_text(sample_report_content, encoding="utf-8")

        with patch.object(
            Path, "unlink", side_effect=Exception("delete error")
        ):
            result = storage.delete_report("test-uuid")

        assert result is False


class TestReportExists:
    """Tests for report_exists method."""

    def test_returns_true_when_file_exists(
        self, tmp_path, sample_report_content
    ):
        """Should return True when file exists."""
        storage = FileReportStorage(base_dir=tmp_path)

        report_path = tmp_path / "test-uuid.md"
        report_path.write_text(sample_report_content, encoding="utf-8")

        result = storage.report_exists("test-uuid")

        assert result is True

    def test_returns_false_when_file_not_exists(self, tmp_path):
        """Should return False when file doesn't exist."""
        storage = FileReportStorage(base_dir=tmp_path)

        result = storage.report_exists("nonexistent")

        assert result is False


class TestListReports:
    """Tests for list_reports method."""

    def test_returns_list_of_report_dicts(self, tmp_path):
        """Should return list of dicts for each .md file."""
        (tmp_path / "report-1.md").write_text("# Report 1", encoding="utf-8")
        (tmp_path / "report-2.md").write_text("# Report 2", encoding="utf-8")

        storage = FileReportStorage(base_dir=tmp_path)
        result = storage.list_reports()

        assert len(result) == 2
        ids = {r["id"] for r in result}
        assert ids == {"report-1", "report-2"}
        for r in result:
            assert "id" in r
            assert "query" in r
            assert "mode" in r
            assert "created_at" in r
            assert "completed_at" in r

    def test_returns_empty_list_when_no_reports(self, tmp_path):
        """Should return empty list when directory has no .md files."""
        storage = FileReportStorage(base_dir=tmp_path)
        result = storage.list_reports()

        assert result == []

    def test_loads_metadata_from_json(self, tmp_path):
        """Should populate query and mode from metadata JSON when available."""
        (tmp_path / "report-1.md").write_text("# Report", encoding="utf-8")
        metadata = {"query": "test query", "mode": "detailed"}
        (tmp_path / "report-1_metadata.json").write_text(
            json.dumps(metadata), encoding="utf-8"
        )

        storage = FileReportStorage(base_dir=tmp_path)
        result = storage.list_reports()

        assert len(result) == 1
        assert result[0]["query"] == "test query"
        assert result[0]["mode"] == "detailed"

    def test_handles_missing_metadata(self, tmp_path):
        """Should return None query and 'unknown' mode when no metadata file."""
        (tmp_path / "report-1.md").write_text("# Report", encoding="utf-8")

        storage = FileReportStorage(base_dir=tmp_path)
        result = storage.list_reports()

        assert len(result) == 1
        assert result[0]["query"] is None
        assert result[0]["mode"] == "unknown"

    def test_returns_empty_list_on_error(self, tmp_path):
        """Should return empty list on error."""
        storage = FileReportStorage(base_dir=tmp_path)

        with patch.object(Path, "glob", side_effect=Exception("glob error")):
            result = storage.list_reports()

        assert result == []

    def test_accepts_username_parameter(self, tmp_path):
        """Should accept username parameter without error."""
        storage = FileReportStorage(base_dir=tmp_path)
        result = storage.list_reports(username="testuser")

        assert result == []

    def test_ignores_non_md_files(self, tmp_path):
        """Should only list .md files, ignoring other file types."""
        (tmp_path / "report-1.md").write_text("# Report", encoding="utf-8")
        (tmp_path / "report-1_metadata.json").write_text("{}", encoding="utf-8")
        (tmp_path / "notes.txt").write_text("notes", encoding="utf-8")
        (tmp_path / "data.csv").write_text("a,b", encoding="utf-8")

        storage = FileReportStorage(base_dir=tmp_path)
        result = storage.list_reports()

        assert len(result) == 1
        assert result[0]["id"] == "report-1"

    def test_created_at_is_iso_timestamp(self, tmp_path):
        """Should return created_at as ISO 8601 string, not raw float."""
        (tmp_path / "report-1.md").write_text("# Report", encoding="utf-8")

        storage = FileReportStorage(base_dir=tmp_path)
        result = storage.list_reports()

        created_at = result[0]["created_at"]
        assert isinstance(created_at, str)
        # ISO format should contain 'T' separator and timezone info
        assert "T" in created_at

    def test_handles_corrupt_metadata_json(self, tmp_path):
        """Should gracefully handle corrupt metadata JSON."""
        (tmp_path / "report-1.md").write_text("# Report", encoding="utf-8")
        (tmp_path / "report-1_metadata.json").write_text(
            "not valid json{{{", encoding="utf-8"
        )

        storage = FileReportStorage(base_dir=tmp_path)
        result = storage.list_reports()

        assert len(result) == 1
        assert result[0]["query"] is None
        assert result[0]["mode"] == "unknown"

    def test_metadata_with_extra_fields_only_uses_query_and_mode(
        self, tmp_path
    ):
        """Should extract only query and mode from metadata."""
        (tmp_path / "report-1.md").write_text("# Report", encoding="utf-8")
        metadata = {
            "query": "my query",
            "mode": "detailed",
            "extra_field": "ignored",
            "sources_count": 42,
        }
        (tmp_path / "report-1_metadata.json").write_text(
            json.dumps(metadata), encoding="utf-8"
        )

        storage = FileReportStorage(base_dir=tmp_path)
        result = storage.list_reports()

        assert set(result[0].keys()) == {
            "id",
            "query",
            "mode",
            "created_at",
            "completed_at",
        }


class TestLoadMetadata:
    """Tests for _load_metadata helper method."""

    def test_returns_dict_when_metadata_exists(self, tmp_path):
        """Should return parsed dict from metadata JSON."""
        metadata = {"query": "test", "mode": "quick"}
        (tmp_path / "r1_metadata.json").write_text(
            json.dumps(metadata), encoding="utf-8"
        )

        storage = FileReportStorage(base_dir=tmp_path)
        result = storage._load_metadata("r1")

        assert result == metadata

    def test_returns_none_when_no_metadata_file(self, tmp_path):
        """Should return None when metadata file doesn't exist."""
        storage = FileReportStorage(base_dir=tmp_path)
        result = storage._load_metadata("nonexistent")

        assert result is None

    def test_returns_none_on_invalid_json(self, tmp_path):
        """Should return None when metadata file has invalid JSON."""
        (tmp_path / "r1_metadata.json").write_text("{{bad", encoding="utf-8")

        storage = FileReportStorage(base_dir=tmp_path)
        result = storage._load_metadata("r1")

        assert result is None