Add 'really_needs_auth' fixture that skips the test if no API key env var is found

Author: gvanrossum-ms

test/fixtures.py

@@ -44,6 +44,13 @@
 def needs_auth() -> None:
     utils.load_dotenv()
 
+@pytest.fixture(scope="session")
+def really_needs_auth() -> None:
+    utils.load_dotenv()
+    # Check if any of the supported API keys is set
+    if not (os.getenv("OPENAI_API_KEY") or os.getenv("AZURE_OPENAI_API_KEY")):
+        pytest.skip("No API key found")
+
 
 @pytest.fixture(scope="session")
 def embedding_model() -> AsyncEmbeddingModel:

test/test_demo.py

@@ -12,7 +12,7 @@
 from typeagent.knowpro.interfaces import ScoredSemanticRefOrdinal
 from typeagent.podcasts import podcast
 
-from fixtures import needs_auth
+from fixtures import really_needs_auth
 
 tests_dir = os.path.dirname(__file__)
 root_dir = os.path.dirname(tests_dir)
@@ -27,7 +27,7 @@
 )
 
 
-def test_main(needs_auth: None):
+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))

test/test_knowledge.py

@@ -14,7 +14,7 @@
 )
 from typeagent.knowpro import convknowledge, kplib
 
-from fixtures import needs_auth  # type: ignore  # Used!
+from fixtures import really_needs_auth
 
 
 class MockKnowledgeExtractor:
@@ -34,7 +34,7 @@ def mock_knowledge_extractor() -> convknowledge.KnowledgeExtractor:
     return MockKnowledgeExtractor()  # type: ignore
 
 
-def test_create_knowledge_extractor(needs_auth: None):
+def test_create_knowledge_extractor(really_needs_auth: None):
     """Test creating a knowledge extractor."""
     extractor = create_knowledge_extractor()
     assert isinstance(extractor, convknowledge.KnowledgeExtractor)

test/test_property_index_population.py

@@ -21,6 +21,8 @@
 from typeagent.podcasts.podcast import PodcastMessage
 from typeagent.storage import SqliteStorageProvider
 
+from fixtures import really_needs_auth
+
 
 class MockEmbeddingModel(AsyncEmbeddingModel):
     def __init__(self):
@@ -33,7 +35,7 @@ async def get_embeddings(self, keys: list[str]) -> np.ndarray:
 
 
 @pytest.mark.asyncio
-async def test_property_index_population_from_database():
+async def test_property_index_population_from_database(really_needs_auth):
     """Test that property index is correctly populated when reopening a database."""
     load_dotenv()
     temp_db_path = tempfile.mktemp(suffix=".sqlite")

test/test_query_method.py

@@ -11,15 +11,11 @@
 from typeagent.knowpro.convsettings import ConversationSettings
 from typeagent.transcripts.transcript import TranscriptMessage, TranscriptMessageMeta
 
-
-@pytest.fixture(scope="session")
-def needs_auth() -> None:
-    """Load environment variables for authentication."""
-    load_dotenv()
+from fixtures import really_needs_auth
 
 
 @pytest.mark.asyncio
-async def test_query_method_basic(needs_auth: None):
+async def test_query_method_basic(really_needs_auth: None):
     """Test the basic query method workflow."""
     # Create a conversation with some test data
     test_model = AsyncEmbeddingModel(model_name=TEST_MODEL_NAME)
@@ -65,7 +61,7 @@ async def test_query_method_basic(needs_auth: None):
 
 
 @pytest.mark.asyncio
-async def test_query_method_empty_conversation(needs_auth: None):
+async def test_query_method_empty_conversation(really_needs_auth: None):
     """Test query method on an empty conversation."""
     test_model = AsyncEmbeddingModel(model_name=TEST_MODEL_NAME)
     settings = ConversationSettings(model=test_model)

test/test_related_terms_index_population.py

@@ -4,7 +4,6 @@
 
 """Test to verify related terms index population in storage providers."""
 
-import asyncio
 import tempfile
 import os
 import pytest
@@ -18,13 +17,13 @@
     MessageTextIndexSettings,
     RelatedTermIndexSettings,
 )
-from typeagent.storage.memory.reltermsindex import RelatedTermsIndex
 from typeagent.podcasts.podcast import PodcastMessage, PodcastMessageMeta
 from typeagent.storage import SqliteStorageProvider
 
+from fixtures import really_needs_auth
 
 @pytest.mark.asyncio
-async def test_related_terms_index_population_from_database():
+async def test_related_terms_index_population_from_database(really_needs_auth):
     """Test that related terms index is correctly populated when reopening a database."""
     load_dotenv()
     temp_db_path = tempfile.mktemp(suffix=".sqlite")

test/test_reltermsindex.py

@@ -26,7 +26,7 @@
 from typeagent.storage import SqliteStorageProvider
 
 # Test fixtures
-from fixtures import needs_auth, embedding_model, temp_db_path
+from fixtures import needs_auth, really_needs_auth, embedding_model, temp_db_path
 
 
 @pytest_asyncio.fixture(params=["memory", "sqlite"])
@@ -122,7 +122,7 @@ async def test_serialize_and_deserialize(
 
 
 @pytest.mark.asyncio
-async def test_related_terms_index_basic(needs_auth: None) -> None:
+async def test_related_terms_index_basic(really_needs_auth: None) -> None:
     settings = RelatedTermIndexSettings(TextEmbeddingIndexSettings())
     index = RelatedTermsIndex(settings)
     assert isinstance(index.aliases, TermToRelatedTermsMap)

test/test_transcripts.py

@@ -3,12 +3,9 @@
 
 import pytest
 import os
-import tempfile
 from datetime import timedelta
-from typing import AsyncGenerator
 
 from typeagent.transcripts.transcript_ingest import (
-    ingest_vtt_transcript,
     get_transcript_speakers,
     get_transcript_duration,
     extract_speaker_from_text,
@@ -24,10 +21,9 @@
     format_timestamp_utc,
 )
 from typeagent.knowpro.convsettings import ConversationSettings
-from typeagent.knowpro.interfaces import Datetime
 from typeagent.aitools.embeddings import AsyncEmbeddingModel
 
-from fixtures import needs_auth, temp_dir, embedding_model  # type: ignore
+from fixtures import needs_auth, really_needs_auth, temp_dir, embedding_model  # type: ignore
 
 
 def test_extract_speaker_from_text():
@@ -245,7 +241,7 @@ async def test_transcript_creation():
 
 @pytest.mark.asyncio
 async def test_transcript_knowledge_extraction_slow(
-    needs_auth: None, embedding_model: AsyncEmbeddingModel
+    really_needs_auth: None, embedding_model: AsyncEmbeddingModel
 ):
     """
     Test that knowledge extraction works during transcript ingestion.

test/test_utils.py

@@ -2,13 +2,15 @@
 # Licensed under the MIT License.
 
 import os
-import pytest
 from contextlib import redirect_stdout
 from io import StringIO
 
 import typeagent.aitools.utils as utils
 
 
+from fixtures import really_needs_auth
+
+
 def test_timelog():
     buf = StringIO()
     with redirect_stdout(buf):
@@ -29,7 +31,7 @@ def test_pretty_print():
     assert out == '{"a": 1}\n', out
 
 
-def test_load_dotenv():
+def test_load_dotenv(really_needs_auth):
     # Call load_dotenv and check for at least one expected key
     utils.load_dotenv()
     assert "OPENAI_API_KEY" in os.environ or "AZURE_OPENAI_API_KEY" in os.environ