Fixture testdatapath, refactor and unify testdatapath handling (#147)

Fixes #146 (see there for details).

Author: bmerkle

src/typeagent/mcp/server.py

@@ -6,6 +6,7 @@
 
 import argparse
 from dataclasses import dataclass
+import os
 import time
 from typing import Any
 
@@ -28,6 +29,9 @@
 from typeagent.storage.memory.semrefindex import TermToSemanticRefIndex
 from typeagent.storage.utils import create_storage_provider
 
+# Example podcast index path for documentation and error messages
+_EXAMPLE_PODCAST_INDEX = "tests/testdata/Episode_53_AdrianTchaikovsky_index"
+
 
 class MCPTypeChatModel(typechat.TypeChatLanguageModel):
     """TypeChat language model that uses MCP sampling API instead of direct API calls."""
@@ -142,7 +146,9 @@ async def make_context(
         entities_top_k=50, topics_top_k=50, messages_top_k=None, chunking=None
     )
 
-    query_context = await load_podcast_index_or_database(settings, dbname)
+    query_context = await load_podcast_database_or_index(
+        settings, dbname, _podcast_index
+    )
 
     # Use MCP-based model instead of one that requires API keys
     model = MCPTypeChatModel(session)
@@ -161,24 +167,33 @@ async def make_context(
     return context
 
 
-async def load_podcast_index_or_database(
+async def load_podcast_database_or_index(
     settings: ConversationSettings,
     dbname: str | None = None,
+    podcast_index: str | None = None,
 ) -> query.QueryEvalContext[podcast.PodcastMessage, Any]:
-    if dbname is None:
-        conversation = await podcast.Podcast.read_from_file(
-            "tests/testdata/Episode_53_AdrianTchaikovsky_index", settings
-        )
-    else:
+    if dbname is not None:
+        # Load from SQLite database
         conversation = await podcast.Podcast.create(settings)
+    elif podcast_index is not None:
+        # Load from JSON index files
+        conversation = await podcast.Podcast.read_from_file(podcast_index, settings)
+    else:
+        raise ValueError(
+            "Either --database or --podcast-index must be specified. "
+            "Use --podcast-index to specify the path to podcast index files "
+            f"(e.g., '{_EXAMPLE_PODCAST_INDEX}')."
+        )
     return query.QueryEvalContext(conversation)
 
 
 # Create an MCP server
 mcp = FastMCP("typagent")
 
-# Global variable to store database path (set via command-line argument)
+# Global variables to store command-line arguments
+# (no other straightforward way to pass to tool handlers)
 _dbname: str | None = None
+_podcast_index: str | None = None
 
 
 @dataclass
@@ -245,12 +260,49 @@ async def query_conversation(
         "--database",
         type=str,
         default=None,
-        help="Path to the SQLite database file (default: load from JSON file)",
+        help="Path to a SQLite database file with pre-indexed podcast data",
+    )
+    parser.add_argument(
+        "-p",
+        "--podcast-index",
+        type=str,
+        default=None,
+        help="Path to podcast index files (excluding '_data.json' suffix), "
+        f"e.g., '{_EXAMPLE_PODCAST_INDEX}'",
     )
     args = parser.parse_args()
 
-    # Store database path in global variable (no other straightforward way to pass to tool)
+    # Validate arguments
+    if args.database is None and args.podcast_index is None:
+        parser.error(
+            "Either --database or --podcast-index is required.\n"
+            "Example: python -m typeagent.mcp.server "
+            f"--podcast-index {_EXAMPLE_PODCAST_INDEX}"
+        )
+
+    if args.database is not None and args.podcast_index is not None:
+        parser.error("Cannot specify both --database and --podcast-index")
+
+    # Validate file existence
+    if args.database is not None and not os.path.exists(args.database):
+        parser.error(
+            f"Database file not found: {args.database}\n"
+            "Please provide a valid path to an existing SQLite database."
+        )
+
+    if args.podcast_index is not None:
+        data_file = args.podcast_index + "_data.json"
+        if not os.path.exists(data_file):
+            parser.error(
+                f"Podcast index file not found: {data_file}\n"
+                "Please provide a valid path to podcast index files "
+                "(without the '_data.json' suffix).\n"
+                f"Example: {_EXAMPLE_PODCAST_INDEX}"
+            )
+
+    # Store in global variables for tool handlers
     _dbname = args.database
+    _podcast_index = args.podcast_index
 
     # Use stdio transport for simplicity
     mcp.run(transport="stdio")

tests/test/conftest.py tests/conftest.pytests/test/conftest.py renamed to tests/conftest.py

@@ -1,8 +1,9 @@
 # Copyright (c) Microsoft Corporation.
 # Licensed under the MIT License.
 
-from collections.abc import AsyncGenerator, Iterator
+from collections.abc import AsyncGenerator, Callable, Iterator
 import os
+from pathlib import Path
 import tempfile
 from typing import Any
 
@@ -43,6 +44,37 @@
     MemorySemanticRefCollection,
 )
 
+# --- Testdata path utilities ---
+# Locate the tests directory relative to this file
+_TESTS_DIR = Path(__file__).resolve().parent  # tests/
+_TESTDATA_DIR = _TESTS_DIR / "testdata"
+_REPO_ROOT = _TESTS_DIR.parent
+
+
+def get_testdata_path(filename: str) -> str:
+    """Return absolute path to a file in tests/testdata/."""
+    return str(_TESTDATA_DIR / filename)
+
+
+def get_repo_root() -> Path:
+    """Return the repository root path."""
+    return _REPO_ROOT
+
+
+def has_testdata_file(filename: str) -> bool:
+    """Check if a testdata file exists (for use in skipif conditions)."""
+    return (_TESTDATA_DIR / filename).exists()
+
+
+# Commonly used test files as constants
+CONFUSE_A_CAT_VTT = get_testdata_path("Confuse-A-Cat.vtt")
+PARROT_SKETCH_VTT = get_testdata_path("Parrot_Sketch.vtt")
+FAKE_PODCAST_TXT = get_testdata_path("FakePodcast.txt")
+EPISODE_53_INDEX = get_testdata_path("Episode_53_AdrianTchaikovsky_index")
+EPISODE_53_TRANSCRIPT = get_testdata_path("Episode_53_AdrianTchaikovsky.txt")
+EPISODE_53_ANSWERS = get_testdata_path("Episode_53_Answer_results.json")
+EPISODE_53_SEARCH = get_testdata_path("Episode_53_Search_results.json")
+
 
 @pytest.fixture(scope="session")
 def needs_auth() -> None:
@@ -63,6 +95,17 @@ def embedding_model() -> AsyncEmbeddingModel:
     return AsyncEmbeddingModel(model_name=TEST_MODEL_NAME)
 
 
+@pytest.fixture(scope="session")
+def testdata_path() -> Callable[[str], str]:
+    """Fixture returning a function to get absolute paths to testdata files.
+
+    Usage:
+        def test_something(testdata_path):
+            path = testdata_path("Confuse-A-Cat.vtt")
+    """
+    return get_testdata_path
+
+
 @pytest.fixture
 def temp_dir() -> Iterator[str]:
     with tempfile.TemporaryDirectory() as dir:

…/test/test_add_messages_with_indexing.py tests/test_add_messages_with_indexing.pytests/test/test_add_messages_with_indexing.py renamed to tests/test_add_messages_with_indexing.py tests/test/test_add_messages_with_indexing.py renamed to tests/test_add_messages_with_indexing.py

@@ -3,7 +3,6 @@
 
 import argparse
 import asyncio
-import os
 import textwrap
 import time
 
@@ -12,23 +11,21 @@
 from typeagent.knowpro.interfaces import ScoredSemanticRefOrdinal
 from typeagent.podcasts import podcast
 
-tests_dir = os.path.dirname(__file__)
-root_dir = os.path.dirname(tests_dir)
-DEFAULT_FILE = os.path.join(root_dir, "testdata", "Episode_53_AdrianTchaikovsky_index")
+from conftest import EPISODE_53_INDEX
 
 parser = argparse.ArgumentParser()
 parser.add_argument(
     "filename",
     nargs="?",
     type=str,
-    default=DEFAULT_FILE,
+    default=EPISODE_53_INDEX,
 )
 
 
 def test_main(really_needs_auth: None):
     # auth is needed because we use embeddings.
     # TODO: Only use the embeddings loaded from the file and cached.
-    asyncio.run(main(DEFAULT_FILE))
+    asyncio.run(main(EPISODE_53_INDEX))
 
 
 async def main(filename_prefix: str):