Spaces:
Running
Running
| import os | |
| import tempfile | |
| import threading | |
| import unittest | |
| from concurrent.futures import ThreadPoolExecutor | |
| from pathlib import Path | |
| from unittest.mock import Mock, patch | |
| import pandas as pd | |
| os.environ.setdefault("APP_STORAGE_DIR", tempfile.mkdtemp(prefix="llm_screener_tests_")) | |
| import app | |
| class ReviewFieldsTests(unittest.TestCase): | |
| def test_user_storage_ids_cannot_escape_the_users_directory(self): | |
| self.assertEqual(app.sanitize_user_id(".."), "default") | |
| self.assertEqual(app.sanitize_user_id("research/group"), "research_group") | |
| def test_concurrent_session_updates_preserve_every_field(self): | |
| user_id = "concurrent-save-user" | |
| app.save_session_meta(user_id, {"df_path": "/data/original.xlsx"}) | |
| worker_count = 24 | |
| start = threading.Barrier(worker_count) | |
| def save_field(index): | |
| start.wait() | |
| app.save_session_meta(user_id, {f"field_{index}": index}) | |
| with ThreadPoolExecutor(max_workers=worker_count) as executor: | |
| list(executor.map(save_field, range(worker_count))) | |
| saved = app.load_session_meta(user_id) | |
| self.assertEqual(saved["df_path"], "/data/original.xlsx") | |
| for index in range(worker_count): | |
| self.assertEqual(saved[f"field_{index}"], index) | |
| def test_invalid_session_metadata_is_not_overwritten(self): | |
| user_id = "invalid-session-user" | |
| session_path = app.get_user_session_meta_path(user_id) | |
| app._ensure_session_dirs(user_id) | |
| invalid_contents = '{"df_path": "unterminated' | |
| session_path.write_text(invalid_contents, encoding="utf-8") | |
| with self.assertRaisesRegex(RuntimeError, "refusing to overwrite"): | |
| app.save_session_meta(user_id, {"position": 2}) | |
| self.assertEqual(session_path.read_text(encoding="utf-8"), invalid_contents) | |
| def test_config_editor_updates_target_state_used_by_actions(self): | |
| df = pd.DataFrame( | |
| { | |
| "URL": ["https://example.test"], | |
| "Outcome": [None], | |
| "Population": [None], | |
| } | |
| ) | |
| with patch.object(app, "save_session_meta") as save_meta: | |
| result = app.sync_extraction_config_inputs( | |
| "Population", | |
| "URL", | |
| df, | |
| "sync-user", | |
| ["Outcome", "Population"], | |
| "Outcome description", | |
| "Population description", | |
| ) | |
| self.assertEqual(result[0], ["Population"]) | |
| self.assertEqual(result[1], ["Population"]) | |
| self.assertEqual(result[2]["label"], "Description: Population") | |
| self.assertEqual(result[2]["value"], "Population description") | |
| saved = save_meta.call_args.args[1] | |
| self.assertEqual(saved["target_columns_text"], "Population") | |
| self.assertEqual(saved["description_values"], ["Population description"]) | |
| def test_reserved_review_columns_are_not_extraction_targets(self): | |
| df = pd.DataFrame( | |
| { | |
| "URL": ["https://example.test"], | |
| "Outcome": [None], | |
| app.DECISION_COLUMN: [""], | |
| app.INCLUSION_RATIONALE_COLUMN: [""], | |
| } | |
| ) | |
| self.assertEqual(app._parse_target_columns_for_ui("", "URL", df), ["Outcome"]) | |
| self.assertEqual( | |
| app._parse_target_columns_for_ui( | |
| "Outcome, Decision, Inclusion Rationale", "URL", df | |
| ), | |
| ["Outcome"], | |
| ) | |
| def test_review_columns_are_added_without_changing_existing_values(self): | |
| df = pd.DataFrame( | |
| { | |
| "URL": ["https://example.test"], | |
| "Outcome": [None], | |
| app.DECISION_COLUMN: ["exclude"], | |
| } | |
| ) | |
| result = app.ensure_review_output_columns(df) | |
| self.assertIs(result, df) | |
| self.assertEqual(result.loc[0, "URL"], "https://example.test") | |
| self.assertEqual(result.loc[0, app.DECISION_COLUMN], "exclude") | |
| self.assertIn(app.DECISION_COLUMN, result.columns) | |
| self.assertIn(app.INCLUSION_RATIONALE_COLUMN, result.columns) | |
| def test_accept_saves_edited_fields_decision_and_rationale(self): | |
| df = pd.DataFrame( | |
| { | |
| "URL": ["https://example.test/1", "https://example.test/2"], | |
| "Outcome": [None, "Existing outcome"], | |
| app.DECISION_COLUMN: ["", "include"], | |
| app.INCLUSION_RATIONALE_COLUMN: ["", "Existing rationale"], | |
| } | |
| ) | |
| with ( | |
| patch.object(app, "persist_downloadable_dataframe", return_value=None), | |
| patch.object(app, "persist_dataframe", return_value="/tmp/working.xlsx"), | |
| patch.object(app, "save_session_meta"), | |
| patch.object(app, "load_session_meta", return_value={}), | |
| ): | |
| result = app.accept_extraction( | |
| ["Outcome"], | |
| "test-user", | |
| df, | |
| 0, | |
| [0], | |
| 0, | |
| "URL", | |
| ["Outcome"], | |
| "exclude", | |
| "The population does not meet the inclusion criteria.", | |
| "Edited outcome", | |
| ) | |
| accepted_df = result[0] | |
| self.assertEqual(accepted_df.at[0, "Outcome"], "Edited outcome") | |
| self.assertEqual(accepted_df.at[0, app.DECISION_COLUMN], "exclude") | |
| self.assertEqual( | |
| accepted_df.at[0, app.INCLUSION_RATIONALE_COLUMN], | |
| "The population does not meet the inclusion criteria.", | |
| ) | |
| self.assertEqual(accepted_df.at[1, "Outcome"], "Existing outcome") | |
| self.assertEqual(accepted_df.at[1, app.DECISION_COLUMN], "include") | |
| self.assertEqual( | |
| accepted_df.at[1, app.INCLUSION_RATIONALE_COLUMN], "Existing rationale" | |
| ) | |
| def test_accept_saves_to_displayed_all_table_position_when_row_state_is_stale(self): | |
| df = pd.DataFrame( | |
| { | |
| "URL": ["zero", "one", "two", "three"], | |
| "Outcome": ["Original zero", "Original one", None, "Original three"], | |
| } | |
| ) | |
| user_id = "accept-position-user" | |
| result = app.accept_extraction( | |
| ["Outcome"], | |
| user_id, | |
| df, | |
| 0, # Simulate the stale pre-navigation row-index state. | |
| [2], | |
| 2, # The displayed all-table row is position 2 (Row 3). | |
| "URL", | |
| ["Outcome"], | |
| "include", | |
| "Displayed row rationale", | |
| "Saved to displayed row", | |
| ) | |
| accepted_df = result[0] | |
| self.assertEqual(accepted_df.at[0, "Outcome"], "Original zero") | |
| self.assertEqual(accepted_df.at[2, "Outcome"], "Saved to displayed row") | |
| persisted_df = pd.read_excel(app.load_session_meta(user_id)["df_path"]) | |
| self.assertEqual(persisted_df.at[0, "Outcome"], "Original zero") | |
| self.assertEqual(persisted_df.at[2, "Outcome"], "Saved to displayed row") | |
| def test_gradio_review_fields_have_expected_interactivity(self): | |
| self.assertFalse(app.evidence_box.interactive) | |
| self.assertTrue(app.decision_box.interactive) | |
| self.assertTrue(app.inclusion_rationale_box.interactive) | |
| self.assertTrue(all(component.interactive for component in app.extracted_inputs)) | |
| self.assertFalse(app.row_number_input.show_label) | |
| self.assertFalse(app.row_number_input.container) | |
| self.assertEqual(app.row_number_input.min_width, 70) | |
| self.assertEqual(app.row_number_input.elem_id, "row-navigation-input") | |
| self.assertIn("grid-template-columns", app.ROW_NAVIGATION_CSS) | |
| button_values = { | |
| component.get("props", {}).get("value") | |
| for component in app.demo.config["components"] | |
| if component.get("type") == "button" | |
| } | |
| self.assertNotIn("Start screening", button_values) | |
| load_excel_events = [ | |
| dependency | |
| for dependency in app.demo.config["dependencies"] | |
| if dependency.get("api_name") == "load_excel" | |
| ] | |
| self.assertTrue(load_excel_events) | |
| self.assertTrue( | |
| any( | |
| event_name == "upload" | |
| for dependency in load_excel_events | |
| for _, event_name in dependency["targets"] | |
| ) | |
| ) | |
| labels = { | |
| component.get("props", {}).get("label") | |
| for component in app.demo.config["components"] | |
| } | |
| self.assertNotIn("Confidence", labels) | |
| self.assertFalse(any(label and "RAYYAN" in label for label in labels)) | |
| def test_table_preview_toggle_switches_full_app_pages(self): | |
| opened = app.toggle_table_preview(False) | |
| self.assertEqual(len(opened), len(app.preview_toggle_outputs)) | |
| self.assertTrue(opened[0]) | |
| self.assertFalse(opened[1]["visible"]) | |
| self.assertFalse(opened[2]["visible"]) | |
| self.assertFalse(opened[3]["visible"]) | |
| self.assertFalse(opened[4]["visible"]) | |
| self.assertTrue(opened[5]["visible"]) | |
| self.assertEqual(opened[6]["value"], "Close table preview") | |
| closed = app.toggle_table_preview(True) | |
| self.assertFalse(closed[0]) | |
| self.assertTrue(closed[1]["visible"]) | |
| self.assertFalse(closed[5]["visible"]) | |
| self.assertEqual(closed[6]["value"], "Open table preview") | |
| def test_table_preview_is_editable_and_preloadable(self): | |
| df = pd.DataFrame({"URL": ["one"], "Outcome": [None]}) | |
| update, status = app.sync_table_preview(df) | |
| self.assertTrue(app.working_table_preview.interactive) | |
| self.assertEqual(app.working_table_preview.show_search, "filter") | |
| self.assertFalse(app.working_table_preview.show_row_numbers) | |
| self.assertEqual(app.working_table_preview.static_columns, [0]) | |
| self.assertEqual(app.working_table_preview.buttons, ["fullscreen", "copy"]) | |
| pd.testing.assert_frame_equal(update["value"], df) | |
| self.assertEqual(update["headers"], ["URL", "Outcome"]) | |
| self.assertIn("1 rows × 2 columns", status) | |
| def test_large_table_preview_centers_on_current_row(self): | |
| df = pd.DataFrame({"URL": [f"row-{index}" for index in range(50)]}) | |
| preview, is_limited = app.build_table_preview_frame(df, 25) | |
| self.assertTrue(is_limited) | |
| self.assertEqual(len(preview), app.TABLE_PREVIEW_MAX_ROWS) | |
| self.assertEqual(preview.index[0], 15) | |
| self.assertEqual(preview.index[-1], 35) | |
| def test_table_preview_selects_metadata_targets_and_keeps_working_row_ids(self): | |
| df = pd.DataFrame( | |
| { | |
| "URL": [f"url-{index}" for index in range(50)], | |
| "Title": [f"Title {index}" for index in range(50)], | |
| "Authors": [f"Author {index}" for index in range(50)], | |
| "Year": list(range(2000, 2050)), | |
| "Outcome": [None] * 50, | |
| "Hidden column": [index for index in range(50)], | |
| } | |
| ) | |
| preview, _ = app.build_table_preview_frame(df, 25, ["Outcome"]) | |
| self.assertEqual( | |
| list(preview.columns), | |
| [app.TABLE_PREVIEW_ROW_ID_COLUMN, "Title", "Authors", "Year", "Outcome"], | |
| ) | |
| self.assertEqual(preview.iloc[0][app.TABLE_PREVIEW_ROW_ID_COLUMN], 16) | |
| self.assertEqual(preview.iloc[-1][app.TABLE_PREVIEW_ROW_ID_COLUMN], 36) | |
| self.assertNotIn("URL", preview.columns) | |
| self.assertNotIn("Hidden column", preview.columns) | |
| def test_large_table_edit_updates_only_displayed_rows(self): | |
| current_df = pd.DataFrame( | |
| { | |
| "URL": [f"row-{index}" for index in range(50)], | |
| "Outcome": [None] * 50, | |
| } | |
| ) | |
| edited_table, _ = app.build_table_preview_frame(current_df, 25, ["Outcome"]) | |
| edited_table = edited_table.copy() | |
| edited_table.at[15, "Outcome"] = "first displayed edit" | |
| edited_table.at[25, "Outcome"] = "active row edit" | |
| edited_table.at[35, "Outcome"] = "last displayed edit" | |
| with ( | |
| patch.object(app, "persist_dataframe", return_value="/tmp/working.xlsx") as persist, | |
| patch.object(app, "save_session_meta"), | |
| ): | |
| result = app.apply_table_edits( | |
| edited_table, | |
| current_df, | |
| "large-table-user", | |
| 25, | |
| 25, | |
| "URL", | |
| ["Outcome"], | |
| ) | |
| updated_df = result[0] | |
| self.assertEqual(updated_df.at[15, "Outcome"], "first displayed edit") | |
| self.assertEqual(updated_df.at[25, "Outcome"], "active row edit") | |
| self.assertEqual(updated_df.at[35, "Outcome"], "last displayed edit") | |
| self.assertTrue(updated_df.loc[~updated_df.index.isin([15, 25, 35]), "Outcome"].isna().all()) | |
| pd.testing.assert_frame_equal(persist.call_args.args[1], updated_df) | |
| def test_table_preview_sync_skips_when_preview_closed(self): | |
| update, status = app.sync_table_preview_if_open(pd.DataFrame({"URL": ["one"]}), False) | |
| skip_type = type(app.gr.skip()) | |
| self.assertIsInstance(update, skip_type) | |
| self.assertIsInstance(status, skip_type) | |
| def test_table_cell_edit_saves_and_marks_download_stale(self): | |
| current_df = pd.DataFrame( | |
| { | |
| "URL": ["one", "two"], | |
| "Title": ["Old title", "Second"], | |
| "Outcome": [None, None], | |
| } | |
| ) | |
| edited_df = current_df.copy() | |
| edited_df.at[0, "Title"] = "Edited title" | |
| edited_df.at[0, "Outcome"] = "Complete" | |
| with ( | |
| patch.object(app, "persist_dataframe", return_value="/tmp/working.xlsx") as persist, | |
| patch.object(app, "save_session_meta") as save_meta, | |
| ): | |
| result = app.apply_table_edits( | |
| edited_df, | |
| current_df, | |
| "table-user", | |
| 0, | |
| 0, | |
| "URL", | |
| ["Outcome"], | |
| ) | |
| self.assertEqual(len(result), len(app.table_edit_outputs)) | |
| pd.testing.assert_frame_equal(result[0], edited_df) | |
| self.assertEqual(result[1], [1]) | |
| self.assertIn("Edited title", result[2]) | |
| self.assertEqual(result[6], 1) | |
| self.assertEqual(result[7], "of 2 (index: 0)") | |
| self.assertIn("Download is out of date", result[9]) | |
| self.assertIn("Working table edit saved", result[10]) | |
| pd.testing.assert_frame_equal(persist.call_args.args[1], edited_df) | |
| self.assertEqual(save_meta.call_args.args[1]["df_path"], "/tmp/working.xlsx") | |
| self.assertEqual(save_meta.call_args.args[1]["download_path"], "") | |
| def test_refresh_download_export_generates_file_on_demand(self): | |
| df = pd.DataFrame({"URL": ["one"], "Outcome": ["done"]}) | |
| with ( | |
| patch.object(app, "persist_downloadable_dataframe", return_value="/tmp/export.xlsx"), | |
| patch.object(app, "build_inline_download_html", return_value="download-html"), | |
| patch.object(app, "save_session_meta") as save_meta, | |
| ): | |
| result = app.refresh_download_export(df, "download-user") | |
| self.assertEqual(result, ("download-html", "Download export refreshed.", "Download export refreshed.")) | |
| self.assertEqual(save_meta.call_args.args[1]["download_path"], "/tmp/export.xlsx") | |
| def test_table_structural_edit_is_rejected_and_reverted(self): | |
| current_df = pd.DataFrame({"URL": ["one"], "Outcome": [None]}) | |
| edited_df = pd.concat( | |
| [current_df, pd.DataFrame({"URL": ["two"], "Outcome": [None]})], | |
| ignore_index=True, | |
| ) | |
| with ( | |
| patch.object(app, "persist_dataframe") as persist, | |
| patch.object(app, "save_session_meta") as save_meta, | |
| ): | |
| result = app.apply_table_edits( | |
| edited_df, | |
| current_df, | |
| "table-user", | |
| 0, | |
| 0, | |
| "URL", | |
| ["Outcome"], | |
| ) | |
| self.assertEqual(len(result), len(app.table_edit_outputs)) | |
| pd.testing.assert_frame_equal(result[0], current_df) | |
| pd.testing.assert_frame_equal(result[8]["value"], current_df) | |
| self.assertIn("rows cannot be added or removed", result[10]) | |
| persist.assert_not_called() | |
| save_meta.assert_not_called() | |
| def test_navigation_opens_exact_completed_row_and_loads_review_fields(self): | |
| df = pd.DataFrame( | |
| { | |
| "URL": ["https://example.test/1", "https://example.test/2"], | |
| "Title": ["First", "Completed row"], | |
| "Outcome": [None, "Existing outcome"], | |
| app.DECISION_COLUMN: ["", "exclude"], | |
| app.INCLUSION_RATIONALE_COLUMN: ["", "Existing rationale"], | |
| } | |
| ) | |
| with patch.object(app, "save_session_meta") as save_meta: | |
| result = app.navigate_to_row( | |
| 2, | |
| "navigation-user", | |
| df, | |
| 0, | |
| 0, | |
| "URL", | |
| ["Outcome"], | |
| ) | |
| self.assertEqual(len(result), len(app.navigation_outputs)) | |
| self.assertEqual(result[0], 1) | |
| self.assertEqual(result[1], 1) | |
| self.assertIn("Completed row", result[2]) | |
| self.assertEqual(result[6], 2) | |
| self.assertEqual(result[7], "of 2 (index: 1)") | |
| self.assertEqual(result[9], ["Outcome"]) | |
| self.assertEqual(result[10]["value"], "Existing outcome") | |
| self.assertEqual(result[-5], "") | |
| self.assertEqual(result[-4], "exclude") | |
| self.assertEqual(result[-3], "Existing rationale") | |
| self.assertEqual(result[-2]["fields"], {"Outcome": "Existing outcome"}) | |
| saved = save_meta.call_args.args[1] | |
| self.assertEqual(saved["position"], 1) | |
| self.assertEqual(saved["current_row_index"], 1) | |
| self.assertEqual(saved["extracted_columns"], ["Outcome"]) | |
| self.assertEqual(saved["extracted_values"], ["Existing outcome"]) | |
| self.assertEqual(saved["decision"], "exclude") | |
| self.assertEqual(saved["inclusion_rationale"], "Existing rationale") | |
| self.assertEqual(saved["pdf_path"], "") | |
| def test_invalid_navigation_keeps_current_row_and_review(self): | |
| df = pd.DataFrame({"URL": ["one", "two"], "Outcome": [None, None]}) | |
| for requested in (None, 0, 1.5, 3): | |
| with self.subTest(requested=requested), patch.object( | |
| app, "save_session_meta" | |
| ) as save_meta: | |
| result = app.navigate_to_row( | |
| requested, | |
| "navigation-user", | |
| df, | |
| 1, | |
| 1, | |
| "URL", | |
| ["Outcome"], | |
| ) | |
| self.assertEqual(result[0], 1) | |
| self.assertEqual(result[1], 1) | |
| self.assertEqual(result[6], 2) | |
| self.assertIn("row number", result[8].lower()) | |
| save_meta.assert_not_called() | |
| def test_next_incomplete_row_scans_forward_by_excel_position(self): | |
| df = pd.DataFrame( | |
| { | |
| "Outcome": [None, "complete", None, "complete", None], | |
| } | |
| ) | |
| self.assertEqual(app.find_next_incomplete_row(df, 1, ["Outcome"]), (2, 2)) | |
| self.assertEqual(app.find_next_incomplete_row(df, 3, ["Outcome"]), (4, 4)) | |
| self.assertEqual(app.find_next_incomplete_row(df, 5, ["Outcome"]), (5, None)) | |
| def test_skip_advances_to_next_forward_incomplete_excel_row(self): | |
| df = pd.DataFrame( | |
| { | |
| "URL": ["one", "two", "three"], | |
| "Outcome": [None, "complete", None], | |
| } | |
| ) | |
| with ( | |
| patch.object(app, "persist_downloadable_dataframe", return_value=None), | |
| patch.object(app, "persist_dataframe", return_value="/tmp/working.xlsx"), | |
| patch.object(app, "save_session_meta"), | |
| patch.object(app, "load_session_meta", return_value={}), | |
| ): | |
| result = app.skip_row( | |
| "skip-user", | |
| df, | |
| [0, 2], | |
| 0, | |
| "URL", | |
| ["Outcome"], | |
| ) | |
| self.assertEqual(len(result), len(app.skip_outputs)) | |
| self.assertEqual(result[1], 2) | |
| self.assertEqual(result[2], 2) | |
| self.assertEqual(result[7], 3) | |
| self.assertEqual(result[8], "of 3 (index: 2)") | |
| def test_successful_load_adds_review_columns_and_matches_output_count(self): | |
| source_path = os.path.join(os.environ["APP_STORAGE_DIR"], "source.xlsx") | |
| pd.DataFrame( | |
| {"URL": ["https://example.test"], "Outcome": [None]} | |
| ).to_excel(source_path, index=False) | |
| result = app.load_excel( | |
| source_path, | |
| None, | |
| "load-test-user", | |
| "Outcome", | |
| "URL", | |
| [], | |
| ) | |
| self.assertEqual(len(result), len(app.start_outputs)) | |
| self.assertIn(app.DECISION_COLUMN, result[0].columns) | |
| self.assertIn(app.INCLUSION_RATIONALE_COLUMN, result[0].columns) | |
| self.assertEqual(result[4], ["Outcome"]) | |
| self.assertEqual(result[6]["fields"], {"Outcome": ""}) | |
| self.assertEqual(result[8], ["Outcome"]) | |
| def test_successful_processing_matches_simplified_output_count(self): | |
| df = app.ensure_review_output_columns( | |
| pd.DataFrame({"URL": ["https://example.test"], "Outcome": [None]}) | |
| ) | |
| extracted = { | |
| "fields": {"Outcome": "Generated outcome"}, | |
| "evidence": "- Outcome: supporting text", | |
| "evidence_snippets": [ | |
| {"field": "Outcome", "snippet": "supporting text"} | |
| ], | |
| "confidence": 0.9, | |
| "decision": "include", | |
| } | |
| with ( | |
| patch.object(app, "parse_pdf", return_value="article text"), | |
| patch.object(app, "extract_with_llm", return_value=extracted), | |
| patch.object(app, "persist_uploaded_file", return_value=None), | |
| patch.object(app, "save_session_meta"), | |
| patch.object(app, "load_session_meta", return_value={}), | |
| ): | |
| result = app.process_pdf_and_extract( | |
| "article.pdf", | |
| None, | |
| "process-test-user", | |
| df, | |
| 0, | |
| 0, | |
| ["Outcome"], | |
| ["Outcome"], | |
| "Extract the outcome", | |
| progress=lambda *args, **kwargs: None, | |
| ) | |
| self.assertEqual(len(result), len(app.process_outputs)) | |
| self.assertEqual(result[-4], "- Outcome: supporting text") | |
| self.assertEqual(result[-3], "include") | |
| self.assertEqual(result[-2], "") | |
| def test_processing_uses_displayed_position_when_row_state_is_stale(self): | |
| df = app.ensure_review_output_columns( | |
| pd.DataFrame( | |
| { | |
| "URL": ["zero", "one", "two"], | |
| "Outcome": ["Complete", "Complete", None], | |
| "Population": ["Complete", "Complete", "Complete"], | |
| } | |
| ) | |
| ) | |
| extracted = { | |
| "fields": {"Outcome": "Generated outcome"}, | |
| "evidence": "- Outcome: supporting text", | |
| "evidence_snippets": [], | |
| "confidence": 0.9, | |
| "decision": "include", | |
| } | |
| with ( | |
| patch.object(app, "parse_pdf", return_value="article text"), | |
| patch.object(app, "extract_with_llm", return_value=extracted) as extract, | |
| patch.object(app, "persist_uploaded_file", return_value=None), | |
| patch.object(app, "save_session_meta"), | |
| patch.object(app, "load_session_meta", return_value={}), | |
| ): | |
| result = app.process_pdf_and_extract( | |
| "article.pdf", | |
| None, | |
| "process-position-user", | |
| df, | |
| 0, # Stale row-index state. | |
| 2, # Displayed all-table position (Row 3). | |
| ["Outcome", "Population"], | |
| ["Outcome", "Population"], | |
| "Extract outcome", | |
| "Extract population", | |
| progress=lambda *args, **kwargs: None, | |
| ) | |
| self.assertEqual(extract.call_args.args[1], ["Outcome"]) | |
| self.assertEqual(result[1], ["Outcome", "Population"]) | |
| self.assertEqual(result[2]["value"], "Generated outcome") | |
| self.assertEqual(result[3]["value"], "Complete") | |
| def test_completed_row_reprocesses_all_target_fields(self): | |
| df = app.ensure_review_output_columns( | |
| pd.DataFrame( | |
| { | |
| "URL": ["https://example.test"], | |
| "Outcome": ["Existing outcome"], | |
| "Population": ["Existing population"], | |
| } | |
| ) | |
| ) | |
| extracted = { | |
| "fields": { | |
| "Outcome": "Regenerated outcome", | |
| "Population": "Regenerated population", | |
| }, | |
| "evidence": "- Outcome: supporting text", | |
| "evidence_snippets": [], | |
| "confidence": 0.9, | |
| "decision": "include", | |
| } | |
| with ( | |
| patch.object(app, "parse_pdf", return_value="article text"), | |
| patch.object(app, "extract_with_llm", return_value=extracted) as extract, | |
| patch.object(app, "persist_uploaded_file", return_value=None), | |
| patch.object(app, "save_session_meta"), | |
| patch.object(app, "load_session_meta", return_value={}), | |
| ): | |
| result = app.process_pdf_and_extract( | |
| "article.pdf", | |
| None, | |
| "process-test-user", | |
| df, | |
| 0, | |
| 0, | |
| ["Outcome", "Population"], | |
| ["Outcome", "Population"], | |
| "Extract outcome", | |
| "Extract population", | |
| progress=lambda *args, **kwargs: None, | |
| ) | |
| self.assertEqual( | |
| extract.call_args.args[1], | |
| ["Outcome", "Population"], | |
| ) | |
| self.assertIn("Reprocessing completed row", result[-1]) | |
| def test_processing_uses_target_columns_from_latest_config_sync(self): | |
| df = app.ensure_review_output_columns( | |
| pd.DataFrame( | |
| { | |
| "URL": ["https://example.test"], | |
| "Outcome": ["Existing outcome"], | |
| "Population": ["Existing population"], | |
| } | |
| ) | |
| ) | |
| extracted = { | |
| "fields": {"Population": "Generated population"}, | |
| "evidence": "- Population: supporting text", | |
| "evidence_snippets": [], | |
| "confidence": 0.9, | |
| "decision": "include", | |
| } | |
| with ( | |
| patch.object(app, "parse_pdf", return_value="article text"), | |
| patch.object(app, "extract_with_llm", return_value=extracted) as extract, | |
| patch.object(app, "persist_uploaded_file", return_value=None), | |
| patch.object(app, "save_session_meta"), | |
| patch.object(app, "load_session_meta", return_value={}), | |
| ): | |
| synced = app.sync_extraction_config_inputs( | |
| "Population", | |
| "URL", | |
| df, | |
| "process-test-user", | |
| ["Outcome", "Population"], | |
| "Outcome description", | |
| "Population description", | |
| ) | |
| app.process_pdf_and_extract( | |
| "article.pdf", | |
| None, | |
| "process-test-user", | |
| df, | |
| 0, | |
| 0, | |
| synced[0], | |
| synced[1], | |
| *[update.get("value", "") for update in synced[2:]], | |
| progress=lambda *args, **kwargs: None, | |
| ) | |
| self.assertEqual(extract.call_args.args[1], ["Population"]) | |
| def test_restore_uses_saved_exact_row_and_legacy_queue_position(self): | |
| source_path = os.path.join(os.environ["APP_STORAGE_DIR"], "restore_source.xlsx") | |
| pd.DataFrame( | |
| { | |
| "URL": ["one", "two", "three"], | |
| "Outcome": [None, "complete", None], | |
| } | |
| ).to_excel(source_path, index=False) | |
| user_id = "restore-navigation-user" | |
| app.load_excel(source_path, None, user_id, "Outcome", "URL", []) | |
| app.save_session_meta( | |
| user_id, | |
| { | |
| "position": 1, | |
| "position_mode": "all_rows", | |
| "current_row_index": 1, | |
| "extracted_columns": ["Outcome"], | |
| "extracted_values": ["complete"], | |
| }, | |
| ) | |
| restored = app.restore_saved_session(user_id) | |
| self.assertEqual(len(restored), len(app.demo_load_outputs)) | |
| self.assertEqual(restored[8], 1) | |
| self.assertEqual(restored[9], 1) | |
| self.assertEqual(restored[12]["fields"], {"Outcome": "complete"}) | |
| self.assertEqual(restored[14], ["Outcome"]) | |
| self.assertEqual(restored[20], 2) | |
| self.assertEqual(restored[21], "of 3 (index: 1)") | |
| app.save_session_meta( | |
| user_id, | |
| { | |
| "position": 1, | |
| "position_mode": "", | |
| "current_row_index": None, | |
| "extracted_columns": [], | |
| "extracted_values": [], | |
| }, | |
| ) | |
| legacy_restored = app.restore_saved_session(user_id) | |
| self.assertEqual(legacy_restored[8], 2) | |
| self.assertEqual(legacy_restored[9], 2) | |
| def test_event_return_counts_match_gradio_outputs(self): | |
| self.assertEqual( | |
| len(app.empty_extracted_state("test")), | |
| 2 + len(app.extraction_outputs) + 1, | |
| ) | |
| self.assertEqual(len(app.reject_extraction("test-user")), len(app.reject_outputs)) | |
| self.assertEqual( | |
| len( | |
| app.accept_extraction( | |
| [], "test-user", None, None, [], 0, "", [], "include", "" | |
| ) | |
| ), | |
| len(app.accept_outputs), | |
| ) | |
| self.assertEqual( | |
| len(app.skip_row("test-user", None, [], 0, "", [])), | |
| len(app.skip_outputs), | |
| ) | |
| self.assertEqual( | |
| len(app.load_excel(None, None, "test-user", "", "", [])), | |
| len(app.start_outputs), | |
| ) | |
| self.assertEqual( | |
| len(app.navigate_to_row(1, "test-user", None, 0, None, "", [])), | |
| len(app.navigation_outputs), | |
| ) | |
| self.assertEqual( | |
| len(app.clear_user_session("clear-output-test")), | |
| len(app.clear_session_outputs), | |
| ) | |
| def test_metadata_detection_supports_common_column_names_and_missing_values(self): | |
| df = pd.DataFrame( | |
| {"Article title": ["A study"], "First author": ["Researcher"], "Publication year": [2024]} | |
| ) | |
| self.assertEqual( | |
| app.identify_article_metadata_columns(df), | |
| {"title": "Article title", "authors": "First author", "year": "Publication year"}, | |
| ) | |
| details = app.article_details_markdown(df, 0) | |
| self.assertIn("A study", details) | |
| self.assertIn("Researcher", details) | |
| self.assertIn("2024", details) | |
| def test_metadata_rendering_does_not_require_title_or_author_columns(self): | |
| df = pd.DataFrame({"Record ID": [1], "Outcome": [None]}) | |
| details = app.article_details_markdown(df, 0) | |
| self.assertIn("Title:** Not available", details) | |
| self.assertIn("Author(s):** Not available", details) | |
| def test_explicit_unknown_target_columns_are_reported(self): | |
| df = pd.DataFrame({"URL": ["url"], "Outcome": [None]}) | |
| selected, unknown = app.target_column_validation("Outcome, Missing", "URL", df) | |
| self.assertEqual(selected, ["Outcome"]) | |
| self.assertEqual(unknown, ["Missing"]) | |
| def test_review_summary_counts_remaining_rows(self): | |
| df = pd.DataFrame( | |
| { | |
| "Outcome": ["yes", None, "no"], | |
| app.DECISION_COLUMN: ["include", "", "exclude"], | |
| } | |
| ) | |
| summary = app.build_review_summary(df, ["Outcome"]) | |
| self.assertIn("2/3 rows complete", summary) | |
| self.assertIn("1 included", summary) | |
| self.assertIn("1 excluded", summary) | |
| self.assertIn("1 remaining", summary) | |
| class AutoFetchPdfTests(unittest.TestCase): | |
| def test_doi_is_extracted_from_a_doi_org_url(self): | |
| self.assertEqual( | |
| app.extract_doi_from_text("https://doi.org/10.1234/abc.def-5678"), | |
| "10.1234/abc.def-5678", | |
| ) | |
| def test_doi_extraction_returns_none_without_a_doi(self): | |
| self.assertIsNone(app.extract_doi_from_text("https://example.com/articles/42")) | |
| self.assertIsNone(app.extract_doi_from_text("")) | |
| def test_auto_fetch_skips_when_disabled(self): | |
| df = pd.DataFrame({"URL": ["https://doi.org/10.1234/abc"]}) | |
| with patch.object(app, "AUTO_FETCH_PDF_ENABLED", False): | |
| path, message = app.auto_fetch_pdf_for_row(df, 0, "URL", "auto-fetch-user") | |
| self.assertIsNone(path) | |
| self.assertEqual(message, "") | |
| def test_auto_fetch_skips_rows_without_a_usable_url(self): | |
| df = pd.DataFrame({"URL": [None]}) | |
| path, message = app.auto_fetch_pdf_for_row(df, 0, "URL", "auto-fetch-user") | |
| self.assertIsNone(path) | |
| self.assertEqual(message, "") | |
| def test_auto_fetch_uses_direct_url_when_it_serves_a_pdf(self): | |
| df = pd.DataFrame({"URL": ["https://example.com/article.pdf"]}) | |
| user_id = "auto-fetch-direct-user" | |
| with ( | |
| patch.object(app, "_download_if_pdf", return_value=b"%PDF-1.4 fake") as download, | |
| patch.object(app, "_fetch_pdf_via_unpaywall") as unpaywall, | |
| ): | |
| path, message = app.auto_fetch_pdf_for_row(df, 0, "URL", user_id) | |
| download.assert_called_once_with("https://example.com/article.pdf") | |
| unpaywall.assert_not_called() | |
| self.assertIsNotNone(path) | |
| self.assertTrue(Path(path).exists()) | |
| self.assertEqual(Path(path).read_bytes(), b"%PDF-1.4 fake") | |
| self.assertIn("article URL", message) | |
| def test_auto_fetch_falls_back_to_unpaywall_by_doi(self): | |
| df = pd.DataFrame({"URL": ["https://doi.org/10.1234/abc"]}) | |
| user_id = "auto-fetch-unpaywall-user" | |
| with ( | |
| patch.object(app, "_download_if_pdf", return_value=None) as download, | |
| patch.object(app, "_fetch_pdf_via_unpaywall", return_value=b"%PDF-1.4 oa") as unpaywall, | |
| ): | |
| path, message = app.auto_fetch_pdf_for_row(df, 0, "URL", user_id) | |
| download.assert_called_once_with("https://doi.org/10.1234/abc") | |
| unpaywall.assert_called_once_with("10.1234/abc") | |
| self.assertIsNotNone(path) | |
| self.assertEqual(Path(path).read_bytes(), b"%PDF-1.4 oa") | |
| self.assertIn("Unpaywall", message) | |
| def test_unpaywall_lookup_is_skipped_without_a_configured_contact_email(self): | |
| # Unpaywall's API rejects placeholder/example.com addresses with a 422, so | |
| # sending one is worse than useless. Regression test for that bug. | |
| with ( | |
| patch.object(app, "UNPAYWALL_EMAIL", ""), | |
| patch("requests.get") as mocked_get, | |
| ): | |
| result = app._fetch_pdf_via_unpaywall("10.1234/abc") | |
| self.assertIsNone(result) | |
| mocked_get.assert_not_called() | |
| def test_doi_column_is_found_by_common_header_names(self): | |
| df = pd.DataFrame({"URL": ["https://example.com/x"], "DOI": ["10.1234/abc"]}) | |
| self.assertEqual(app.find_doi_column(df), "DOI") | |
| df2 = pd.DataFrame({"URL": ["https://example.com/x"], "Article DOI": ["10.1234/abc"]}) | |
| self.assertEqual(app.find_doi_column(df2), "Article DOI") | |
| df3 = pd.DataFrame({"URL": ["https://example.com/x"], "Outcome": [""]}) | |
| self.assertIsNone(app.find_doi_column(df3)) | |
| def test_resolve_doi_from_columns_extracts_bare_doi_from_dedicated_column(self): | |
| df = pd.DataFrame({"URL": ["https://example.com/x"], "DOI": ["10.1234/abc"]}) | |
| self.assertEqual(app.resolve_doi_from_columns(df, 0), "10.1234/abc") | |
| def test_resolve_doi_from_columns_returns_none_without_a_doi_column_or_value(self): | |
| df = pd.DataFrame({"URL": ["https://example.com/x"], "Outcome": [""]}) | |
| self.assertIsNone(app.resolve_doi_from_columns(df, 0)) | |
| df2 = pd.DataFrame({"URL": ["https://example.com/x"], "DOI": [None]}) | |
| self.assertIsNone(app.resolve_doi_from_columns(df2, 0)) | |
| def test_auto_fetch_falls_back_to_doi_column_when_url_has_no_doi(self): | |
| df = pd.DataFrame( | |
| {"URL": ["https://example.com/articles/42"], "DOI": ["10.1234/from-column"]} | |
| ) | |
| user_id = "auto-fetch-doi-column-user" | |
| with ( | |
| patch.object(app, "_download_if_pdf", return_value=None) as download, | |
| patch.object(app, "_fetch_pdf_via_unpaywall", return_value=b"%PDF-1.4 col") as unpaywall, | |
| ): | |
| path, message = app.auto_fetch_pdf_for_row(df, 0, "URL", user_id) | |
| download.assert_called_once_with("https://example.com/articles/42") | |
| unpaywall.assert_called_once_with("10.1234/from-column") | |
| self.assertIsNotNone(path) | |
| self.assertEqual(Path(path).read_bytes(), b"%PDF-1.4 col") | |
| self.assertIn("Unpaywall", message) | |
| def test_auto_fetch_prefers_doi_in_url_over_doi_column(self): | |
| df = pd.DataFrame( | |
| {"URL": ["https://doi.org/10.1234/from-url"], "DOI": ["10.1234/from-column"]} | |
| ) | |
| with ( | |
| patch.object(app, "_download_if_pdf", return_value=None), | |
| patch.object(app, "_fetch_pdf_via_unpaywall", return_value=None) as unpaywall, | |
| patch.object(app, "_resolve_pmcid", return_value=None), | |
| ): | |
| app.auto_fetch_pdf_for_row(df, 0, "URL", "auto-fetch-priority-user") | |
| unpaywall.assert_called_once_with("10.1234/from-url") | |
| def test_auto_fetch_returns_nothing_when_no_open_access_copy_exists(self): | |
| df = pd.DataFrame({"URL": ["https://doi.org/10.1234/paywalled"]}) | |
| with ( | |
| patch.object(app, "_download_if_pdf", return_value=None), | |
| patch.object(app, "_fetch_pdf_via_unpaywall", return_value=None), | |
| patch.object(app, "_resolve_pmcid", return_value=None) as resolve_pmcid, | |
| ): | |
| path, message = app.auto_fetch_pdf_for_row(df, 0, "URL", "auto-fetch-fail-user") | |
| resolve_pmcid.assert_called_once_with("10.1234/paywalled") | |
| self.assertIsNone(path) | |
| self.assertEqual(message, "") | |
| def test_auto_fetch_falls_back_to_pmc_when_unpaywall_has_no_copy(self): | |
| # This is the JMIR case: the publisher's PDF endpoint is behind a bot | |
| # challenge (returns 202/403 instead of a PDF), and Unpaywall's OA location | |
| # points at that same blocked URL, so PMC is the only path left. | |
| df = pd.DataFrame({"URL": ["https://www.jmir.org/2021/6/e18167/PDF"], "DOI": ["10.2196/18167"]}) | |
| with ( | |
| patch.object(app, "_download_if_pdf", return_value=None), | |
| patch.object(app, "_fetch_pdf_via_unpaywall", return_value=None), | |
| patch.object(app, "_resolve_pmcid", return_value="PMC8209530") as resolve_pmcid, | |
| patch.object(app, "_fetch_pdf_via_pmc", return_value=b"%PDF-1.4 pmc") as fetch_pmc, | |
| ): | |
| path, message = app.auto_fetch_pdf_for_row(df, 0, "URL", "auto-fetch-pmc-user") | |
| resolve_pmcid.assert_called_once_with("10.2196/18167") | |
| fetch_pmc.assert_called_once_with("PMC8209530") | |
| self.assertIsNotNone(path) | |
| self.assertEqual(Path(path).read_bytes(), b"%PDF-1.4 pmc") | |
| self.assertIn("PubMed Central", message) | |
| def test_auto_fetch_uses_pmcid_column_without_resolving_doi(self): | |
| df = pd.DataFrame({"URL": ["https://example.com/x"], "PMCID": ["PMC8209530"]}) | |
| with ( | |
| patch.object(app, "_download_if_pdf", return_value=None), | |
| patch.object(app, "_resolve_pmcid") as resolve_pmcid, | |
| patch.object(app, "_fetch_pdf_via_pmc", return_value=b"%PDF-1.4 pmc") as fetch_pmc, | |
| ): | |
| path, message = app.auto_fetch_pdf_for_row(df, 0, "URL", "auto-fetch-pmcid-col-user") | |
| resolve_pmcid.assert_not_called() | |
| fetch_pmc.assert_called_once_with("PMC8209530") | |
| self.assertIsNotNone(path) | |
| def test_auto_fetch_pmc_is_skipped_when_disabled(self): | |
| df = pd.DataFrame({"URL": ["https://example.com/x"], "PMCID": ["PMC8209530"]}) | |
| with ( | |
| patch.object(app, "AUTO_FETCH_PMC_ENABLED", False), | |
| patch.object(app, "_download_if_pdf", return_value=None), | |
| patch.object(app, "_fetch_pdf_via_pmc") as fetch_pmc, | |
| ): | |
| path, message = app.auto_fetch_pdf_for_row(df, 0, "URL", "auto-fetch-pmc-off-user") | |
| fetch_pmc.assert_not_called() | |
| self.assertIsNone(path) | |
| def test_resolve_pmcid_from_columns_finds_pmcid_or_pmid_column(self): | |
| df = pd.DataFrame({"URL": ["x"], "PMCID": ["PMC8209530"]}) | |
| self.assertEqual(app.resolve_pmcid_from_columns(df, 0), "PMC8209530") | |
| with patch.object(app, "_resolve_pmcid", return_value="PMC999") as resolve_pmcid: | |
| df2 = pd.DataFrame({"URL": ["x"], "PMID": ["34076579"]}) | |
| self.assertEqual(app.resolve_pmcid_from_columns(df2, 0), "PMC999") | |
| resolve_pmcid.assert_called_once_with("34076579") | |
| df3 = pd.DataFrame({"URL": ["x"], "Outcome": [""]}) | |
| self.assertIsNone(app.resolve_pmcid_from_columns(df3, 0)) | |
| def test_extract_text_from_jats_xml_flattens_title_abstract_and_body(self): | |
| xml_bytes = b""" | |
| <article> | |
| <front><article-meta><title-group><article-title>A Study</article-title></title-group> | |
| <abstract><p>Background text.</p></abstract></article-meta></front> | |
| <body><sec><p>First paragraph.</p><p>Second paragraph.</p></sec></body> | |
| </article> | |
| """ | |
| text = app._extract_text_from_jats_xml(xml_bytes) | |
| self.assertIn("A Study", text) | |
| self.assertIn("Background text.", text) | |
| self.assertIn("First paragraph.", text) | |
| self.assertIn("Second paragraph.", text) | |
| def test_extract_text_from_jats_xml_returns_empty_on_malformed_xml(self): | |
| self.assertEqual(app._extract_text_from_jats_xml(b"<not-xml"), "") | |
| def test_pmc_xml_retries_server_errors_then_uses_ncbi_oai_fallback(self): | |
| europe_error = Mock(status_code=500, content=b"temporary error") | |
| oai_xml = b"<GetRecord><article><article-title>Fallback article</article-title></article></GetRecord>" | |
| oai_success = Mock(status_code=200, content=oai_xml) | |
| with ( | |
| patch("requests.get", side_effect=[europe_error, europe_error, oai_success]) as get, | |
| patch.object(app, "AUTO_FETCH_PMC_RETRIES", 1), | |
| patch.object(app, "AUTO_FETCH_RETRY_BACKOFF_SECONDS", 0), | |
| ): | |
| result = app._fetch_pmc_fulltext_xml("PMC5126964") | |
| self.assertEqual(result, oai_xml) | |
| self.assertEqual(get.call_count, 3) | |
| self.assertEqual( | |
| get.call_args_list[-1].kwargs["params"], | |
| { | |
| "verb": "GetRecord", | |
| "identifier": "oai:pubmedcentral.nih.gov:5126964", | |
| "metadataPrefix": "pmc", | |
| }, | |
| ) | |
| def test_pmc_xml_rejects_invalid_pmcid_before_network_call(self): | |
| with patch("requests.get") as get: | |
| self.assertIsNone(app._fetch_pmc_fulltext_xml("PMC/not-an-id")) | |
| get.assert_not_called() | |
| def test_render_text_as_pdf_bytes_produces_a_valid_multi_page_pdf(self): | |
| text = "\n\n".join([f"Paragraph {i} " + ("word " * 200) for i in range(10)]) | |
| pdf_bytes = app._render_text_as_pdf_bytes(text, title="Test Article") | |
| self.assertTrue(pdf_bytes.startswith(b"%PDF-")) | |
| import fitz | |
| doc = fitz.open(stream=pdf_bytes, filetype="pdf") | |
| try: | |
| self.assertGreater(doc.page_count, 1) | |
| self.assertIn("Test Article", doc[0].get_text()) | |
| finally: | |
| doc.close() | |
| def test_auto_fetch_update_returns_gradio_update_with_fetched_path(self): | |
| df = pd.DataFrame({"URL": ["https://example.com/article.pdf"]}) | |
| with patch.object( | |
| app, | |
| "auto_fetch_pdf_for_row", | |
| return_value=("/tmp/fetched.pdf", "Auto-fetched PDF from the article URL."), | |
| ): | |
| update, message = app.auto_fetch_pdf_update(df, 0, "URL", "auto-fetch-update-user") | |
| self.assertEqual(update["value"], "/tmp/fetched.pdf") | |
| self.assertIn("Auto-fetched", message) | |
| if __name__ == "__main__": | |
| unittest.main() | |