"""Tests for database storage implementation.""" from unittest.mock import MagicMock from local_deep_research.storage.database import DatabaseReportStorage class TestDatabaseReportStorageInit: """Tests for DatabaseReportStorage initialization.""" def test_stores_session(self, mock_session): """Should store provided session.""" storage = DatabaseReportStorage(mock_session) assert storage.session is mock_session class TestSaveReport: """Tests for save_report method.""" def test_saves_content_to_existing_record( self, mock_session, mock_research_history, sample_report_content ): """Should save content to existing research record.""" mock_session.query.return_value.filter_by.return_value.first.return_value = mock_research_history storage = DatabaseReportStorage(mock_session) result = storage.save_report("test-uuid", sample_report_content) assert result is True assert mock_research_history.report_content == sample_report_content mock_session.commit.assert_called_once() def test_updates_metadata_when_provided( self, mock_session, mock_research_history, sample_report_content, sample_metadata, ): """Should update metadata when provided.""" mock_research_history.research_meta = {"existing": "value"} mock_session.query.return_value.filter_by.return_value.first.return_value = mock_research_history storage = DatabaseReportStorage(mock_session) storage.save_report( "test-uuid", sample_report_content, metadata=sample_metadata ) # Should merge metadata assert "existing" in mock_research_history.research_meta assert "query" in mock_research_history.research_meta def test_sets_metadata_when_none_exists( self, mock_session, mock_research_history, sample_report_content, sample_metadata, ): """Should set metadata when none exists.""" mock_research_history.research_meta = None mock_session.query.return_value.filter_by.return_value.first.return_value = mock_research_history storage = DatabaseReportStorage(mock_session) storage.save_report( "test-uuid", sample_report_content, metadata=sample_metadata ) assert mock_research_history.research_meta == sample_metadata def test_returns_false_when_record_not_found( self, mock_session, sample_report_content ): """Should return False when research record not found.""" mock_session.query.return_value.filter_by.return_value.first.return_value = None storage = DatabaseReportStorage(mock_session) result = storage.save_report("nonexistent", sample_report_content) assert result is False def test_returns_false_on_error( self, mock_session, mock_research_history, sample_report_content ): """Should return False on error and rollback.""" mock_session.query.return_value.filter_by.return_value.first.return_value = mock_research_history mock_session.commit.side_effect = Exception("commit error") storage = DatabaseReportStorage(mock_session) result = storage.save_report("test-uuid", sample_report_content) assert result is False mock_session.rollback.assert_called_once() class TestGetReport: """Tests for get_report method.""" def test_returns_content_when_exists( self, mock_session, mock_research_history ): """Should return report content when exists.""" mock_session.query.return_value.filter_by.return_value.first.return_value = mock_research_history storage = DatabaseReportStorage(mock_session) result = storage.get_report("test-uuid") assert result == mock_research_history.report_content def test_returns_none_when_not_found(self, mock_session): """Should return None when record not found.""" mock_session.query.return_value.filter_by.return_value.first.return_value = None storage = DatabaseReportStorage(mock_session) result = storage.get_report("nonexistent") assert result is None def test_returns_none_when_content_is_none( self, mock_session, mock_research_history ): """Should return None when report_content is None.""" mock_research_history.report_content = None mock_session.query.return_value.filter_by.return_value.first.return_value = mock_research_history storage = DatabaseReportStorage(mock_session) result = storage.get_report("test-uuid") assert result is None def test_returns_none_on_error(self, mock_session): """Should return None on error.""" mock_session.query.side_effect = Exception("query error") storage = DatabaseReportStorage(mock_session) result = storage.get_report("test-uuid") assert result is None class TestGetReportWithMetadata: """Tests for get_report_with_metadata method.""" def test_returns_full_record(self, mock_session, mock_research_history): """Should return full record with metadata.""" mock_session.query.return_value.filter_by.return_value.first.return_value = mock_research_history storage = DatabaseReportStorage(mock_session) result = storage.get_report_with_metadata("test-uuid") assert result["content"] == mock_research_history.report_content assert result["metadata"] == mock_research_history.research_meta assert result["query"] == mock_research_history.query assert result["mode"] == mock_research_history.mode assert "created_at" in result assert "completed_at" in result assert "duration_seconds" in result def test_returns_empty_metadata_when_none( self, mock_session, mock_research_history ): """Should return empty dict when research_meta is None.""" mock_research_history.research_meta = None mock_session.query.return_value.filter_by.return_value.first.return_value = mock_research_history storage = DatabaseReportStorage(mock_session) result = storage.get_report_with_metadata("test-uuid") assert result["metadata"] == {} def test_returns_none_when_not_found(self, mock_session): """Should return None when not found.""" mock_session.query.return_value.filter_by.return_value.first.return_value = None storage = DatabaseReportStorage(mock_session) result = storage.get_report_with_metadata("nonexistent") assert result is None def test_returns_none_when_no_content( self, mock_session, mock_research_history ): """Should return None when report_content is None.""" mock_research_history.report_content = None mock_session.query.return_value.filter_by.return_value.first.return_value = mock_research_history storage = DatabaseReportStorage(mock_session) result = storage.get_report_with_metadata("test-uuid") assert result is None def test_strips_settings_snapshot_from_metadata( self, mock_session, mock_research_history ): """settings_snapshot (API keys/tokens) must be stripped from the returned metadata — defence-in-depth at the source so a future route wiring this method to a response cannot leak it (CWE-200). Other metadata fields are preserved.""" mock_research_history.research_meta = { "iterations": 2, "settings_snapshot": {"llm.openai.api_key": "sk-SECRET-KEY"}, } mock_session.query.return_value.filter_by.return_value.first.return_value = mock_research_history storage = DatabaseReportStorage(mock_session) result = storage.get_report_with_metadata("test-uuid") assert "settings_snapshot" not in result["metadata"] assert "sk-SECRET-KEY" not in str(result["metadata"]) assert result["metadata"]["iterations"] == 2 class TestDeleteReport: """Tests for delete_report method.""" def test_sets_content_to_none(self, mock_session, mock_research_history): """Should set report_content to None.""" mock_session.query.return_value.filter_by.return_value.first.return_value = mock_research_history storage = DatabaseReportStorage(mock_session) result = storage.delete_report("test-uuid") assert result is True assert mock_research_history.report_content is None mock_session.commit.assert_called_once() def test_returns_false_when_not_found(self, mock_session): """Should return False when not found.""" mock_session.query.return_value.filter_by.return_value.first.return_value = None storage = DatabaseReportStorage(mock_session) result = storage.delete_report("nonexistent") assert result is False def test_returns_false_on_error(self, mock_session, mock_research_history): """Should return False on error and rollback.""" mock_session.query.return_value.filter_by.return_value.first.return_value = mock_research_history mock_session.commit.side_effect = Exception("commit error") storage = DatabaseReportStorage(mock_session) result = storage.delete_report("test-uuid") assert result is False mock_session.rollback.assert_called_once() class TestReportExists: """Tests for report_exists method.""" def test_returns_true_when_content_exists( self, mock_session, mock_research_history ): """Should return True when report content exists.""" mock_session.query.return_value.filter_by.return_value.first.return_value = mock_research_history storage = DatabaseReportStorage(mock_session) result = storage.report_exists("test-uuid") assert result is True def test_returns_false_when_not_found(self, mock_session): """Should return False when not found.""" mock_session.query.return_value.filter_by.return_value.first.return_value = None storage = DatabaseReportStorage(mock_session) result = storage.report_exists("nonexistent") assert result is False def test_returns_false_when_content_is_none( self, mock_session, mock_research_history ): """Should return False when report_content is None.""" mock_research_history.report_content = None mock_session.query.return_value.filter_by.return_value.first.return_value = mock_research_history storage = DatabaseReportStorage(mock_session) result = storage.report_exists("test-uuid") assert result is False def test_returns_false_on_error(self, mock_session): """Should return False on error.""" mock_session.query.side_effect = Exception("query error") storage = DatabaseReportStorage(mock_session) result = storage.report_exists("test-uuid") assert result is False class TestListReports: """Tests for list_reports method.""" def test_returns_list_of_report_dicts(self, mock_session): """Should return list of dicts with report metadata.""" record1 = MagicMock() record1.id = "uuid-1" record1.query = "query one" record1.mode = "quick" record1.created_at = "2024-01-01T10:00:00" record1.completed_at = "2024-01-01T10:05:00" record2 = MagicMock() record2.id = "uuid-2" record2.query = "query two" record2.mode = "detailed" record2.created_at = "2024-01-02T10:00:00" record2.completed_at = "2024-01-02T10:30:00" mock_session.query.return_value.filter.return_value.all.return_value = [ record1, record2, ] storage = DatabaseReportStorage(mock_session) result = storage.list_reports() assert len(result) == 2 assert result[0]["id"] == "uuid-1" assert result[0]["query"] == "query one" assert result[0]["mode"] == "quick" assert result[0]["created_at"] == "2024-01-01T10:00:00" assert result[0]["completed_at"] == "2024-01-01T10:05:00" assert result[1]["id"] == "uuid-2" def test_returns_empty_list_when_no_reports(self, mock_session): """Should return empty list when no reports with content exist.""" mock_session.query.return_value.filter.return_value.all.return_value = [] storage = DatabaseReportStorage(mock_session) result = storage.list_reports() assert result == [] def test_returns_empty_list_on_error(self, mock_session): """Should return empty list on database error.""" mock_session.query.side_effect = Exception("db error") storage = DatabaseReportStorage(mock_session) result = storage.list_reports() assert result == [] def test_accepts_username_parameter(self, mock_session): """Should accept username parameter without error.""" mock_session.query.return_value.filter.return_value.all.return_value = [] storage = DatabaseReportStorage(mock_session) result = storage.list_reports(username="testuser") assert result == [] def test_projects_columns_not_full_entity(self, mock_session): """list_reports must select only metadata columns, never the full ResearchHistory entity — querying the entity eagerly loads the large report_content body into memory. Regression guard for #4560 (a revert to query(ResearchHistory) is output-identical and would otherwise pass silently).""" from local_deep_research.database.models import ResearchHistory mock_session.query.return_value.filter.return_value.all.return_value = [] storage = DatabaseReportStorage(mock_session) storage.list_reports() # Identity checks: a SQLAlchemy column's __eq__ builds a SQL # clause, so `in`/`==` membership tests are unsafe here. selected = mock_session.query.call_args.args assert not any(arg is ResearchHistory for arg in selected), ( "list_reports must not query the full ResearchHistory entity" ) assert not any( arg is ResearchHistory.report_content for arg in selected ), "list_reports must not load the report_content body" def test_filters_records_with_report_content(self, mock_session): """Should query with isnot(None) filter on report_content.""" mock_session.query.return_value.filter.return_value.all.return_value = [] storage = DatabaseReportStorage(mock_session) storage.list_reports() # Verify filter was called (the .filter(ResearchHistory.report_content.isnot(None))) mock_session.query.return_value.filter.assert_called_once() def test_single_report_returns_correct_keys(self, mock_session): """Should return dict with all expected keys for a single report.""" record = MagicMock() record.id = "uuid-1" record.query = "single query" record.mode = "quick" record.created_at = "2024-01-01T10:00:00" record.completed_at = None mock_session.query.return_value.filter.return_value.all.return_value = [ record ] storage = DatabaseReportStorage(mock_session) result = storage.list_reports() assert len(result) == 1 assert set(result[0].keys()) == { "id", "query", "mode", "created_at", "completed_at", } assert result[0]["completed_at"] is None def test_default_username_is_none(self, mock_session): """Should default username to None when not provided.""" mock_session.query.return_value.filter.return_value.all.return_value = [] storage = DatabaseReportStorage(mock_session) storage.list_reports() # Should not raise; username is ignored in the implementation