mirror of
https://github.com/microsoft/agent-framework.git
synced 2026-06-16 21:04:09 +08:00
Python: Samples Integration Tests (#615)
* Samples Tests * small fixes * job fix * telemetry dependency fix * job error fix * sorting provider specific tests * telemetry fixes * openai file search fix --------- Co-authored-by: Giles Odigwe <gilesodigwe@microsoft.com>
This commit is contained in:
committed by
GitHub
Unverified
parent
240edb00cd
commit
ee56314a26
@@ -0,0 +1,54 @@
|
||||
# Copyright (c) Microsoft. All rights reserved.
|
||||
|
||||
import copy
|
||||
import os
|
||||
from collections.abc import Awaitable, Callable
|
||||
from typing import Any
|
||||
|
||||
import pytest
|
||||
from pytest import MonkeyPatch, mark, param
|
||||
|
||||
from samples.getting_started.threads.custom_chat_message_store_thread import main as threads_custom_store
|
||||
from samples.getting_started.threads.suspend_resume_thread import main as threads_suspend_resume
|
||||
from tests.sample_utils import retry
|
||||
|
||||
# Environment variable for controlling sample tests
|
||||
RUN_SAMPLES_TESTS = "RUN_SAMPLES_TESTS"
|
||||
|
||||
# All thread samples
|
||||
thread_samples = [
|
||||
param(
|
||||
threads_custom_store,
|
||||
[], # Non-interactive sample
|
||||
id="threads_custom_store",
|
||||
marks=[
|
||||
pytest.mark.openai,
|
||||
pytest.mark.skipif(os.getenv(RUN_SAMPLES_TESTS, None) is None, reason="Not running sample tests."),
|
||||
],
|
||||
),
|
||||
param(
|
||||
threads_suspend_resume,
|
||||
[], # Non-interactive sample
|
||||
id="threads_suspend_resume",
|
||||
marks=[
|
||||
pytest.mark.openai,
|
||||
pytest.mark.skipif(os.getenv(RUN_SAMPLES_TESTS, None) is None, reason="Not running sample tests."),
|
||||
],
|
||||
),
|
||||
]
|
||||
|
||||
|
||||
@mark.parametrize("sample, responses", thread_samples)
|
||||
async def test_thread_samples(sample: Callable[..., Awaitable[Any]], responses: list[str], monkeypatch: MonkeyPatch):
|
||||
"""Test thread samples with input mocking and retry logic."""
|
||||
saved_responses = copy.deepcopy(responses)
|
||||
|
||||
def reset():
|
||||
responses.clear()
|
||||
responses.extend(saved_responses)
|
||||
|
||||
def mock_input(prompt: str = "") -> str:
|
||||
return responses.pop(0) if responses else "exit"
|
||||
|
||||
monkeypatch.setattr("builtins.input", mock_input)
|
||||
await retry(sample, retries=3, reset=reset)
|
||||
Reference in New Issue
Block a user