tests/test_app.py

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()