mirror of
https://github.com/deepset-ai/haystack.git
synced 2025-07-27 02:40:41 +00:00
88 lines
2.4 KiB
Python
88 lines
2.4 KiB
Python
# SPDX-FileCopyrightText: 2022-present deepset GmbH <info@deepset.ai>
|
|
#
|
|
# SPDX-License-Identifier: Apache-2.0
|
|
|
|
import asyncio
|
|
import time
|
|
from pathlib import Path
|
|
from typing import Dict, Generator
|
|
from unittest.mock import Mock
|
|
|
|
import pytest
|
|
|
|
from haystack import component, tracing
|
|
from haystack.testing.test_utils import set_all_seeds
|
|
from test.tracing.utils import SpyingTracer
|
|
|
|
set_all_seeds(0)
|
|
|
|
# Tracing is disable by default to avoid failures in CI
|
|
tracing.disable_tracing()
|
|
|
|
|
|
@pytest.fixture()
|
|
def waiting_component():
|
|
@component
|
|
class Waiter:
|
|
@component.output_types(waited_for=int)
|
|
def run(self, wait_for: int) -> Dict[str, int]:
|
|
time.sleep(wait_for)
|
|
return {"waited_for": wait_for}
|
|
|
|
@component.output_types(waited_for=int)
|
|
async def run_async(self, wait_for: int) -> Dict[str, int]:
|
|
await asyncio.sleep(wait_for)
|
|
return {"waited_for": wait_for}
|
|
|
|
return Waiter
|
|
|
|
|
|
@pytest.fixture()
|
|
def mock_tokenizer():
|
|
"""
|
|
Tokenizes the string by splitting on spaces.
|
|
"""
|
|
tokenizer = Mock()
|
|
tokenizer.encode = lambda text: text.split()
|
|
tokenizer.decode = lambda tokens: " ".join(tokens)
|
|
return tokenizer
|
|
|
|
|
|
@pytest.fixture()
|
|
def test_files_path():
|
|
return Path(__file__).parent / "test_files"
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def request_blocker(request: pytest.FixtureRequest, monkeypatch):
|
|
"""
|
|
This fixture is applied automatically to all tests.
|
|
Those that are not marked as integration will have the requests module
|
|
monkeypatched to avoid making HTTP requests by mistake.
|
|
"""
|
|
marker = request.node.get_closest_marker("integration")
|
|
if marker is not None:
|
|
return
|
|
|
|
def urlopen_mock(self, method, url, *args, **kwargs):
|
|
raise RuntimeError(f"The test was about to {method} {self.scheme}://{self.host}{url}")
|
|
|
|
monkeypatch.setattr("urllib3.connectionpool.HTTPConnectionPool.urlopen", urlopen_mock)
|
|
|
|
|
|
@pytest.fixture()
|
|
def spying_tracer() -> Generator[SpyingTracer, None, None]:
|
|
tracer = SpyingTracer()
|
|
tracing.enable_tracing(tracer)
|
|
tracer.is_content_tracing_enabled = True
|
|
|
|
yield tracer
|
|
|
|
# Make sure to disable tracing after the test to avoid affecting other tests
|
|
tracing.disable_tracing()
|
|
|
|
|
|
@pytest.fixture()
|
|
def base64_image_string():
|
|
return "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mP8/x8AAwMCAO+ip1sAAAAASUVORK5CYII="
|