diff --git a/test/conftest.py b/test/conftest.py index 4887fa777..c88ee3406 100644 --- a/test/conftest.py +++ b/test/conftest.py @@ -21,3 +21,20 @@ def mock_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) diff --git a/test/pipelines/test_indexing_pipeline.py b/test/pipelines/test_indexing_pipeline.py index 1b77e7a27..c7be20c2f 100644 --- a/test/pipelines/test_indexing_pipeline.py +++ b/test/pipelines/test_indexing_pipeline.py @@ -6,6 +6,7 @@ from haystack.document_stores import InMemoryDocumentStore class TestIndexingPipeline: # indexing files without embeddings + @pytest.mark.integration def test_indexing_files_without_embeddings(self, test_files_path): file_paths = [test_files_path / "txt" / "doc_1.txt", test_files_path / "txt" / "doc_2.txt"] document_store = InMemoryDocumentStore() @@ -36,6 +37,7 @@ class TestIndexingPipeline: assert result["documents_written"] >= 3 # indexing multiple files + @pytest.mark.integration def test_indexing_multiple_file_types(self, test_files_path): document_store = InMemoryDocumentStore() pipeline = build_indexing_pipeline( diff --git a/test/pipelines/test_rag_pipelines.py b/test/pipelines/test_rag_pipelines.py index 1b9dfbbd2..e3ffd9051 100644 --- a/test/pipelines/test_rag_pipelines.py +++ b/test/pipelines/test_rag_pipelines.py @@ -37,6 +37,7 @@ def mock_chat_completion(): yield mock_chat_completion_create +@pytest.mark.integration def test_rag_pipeline(mock_chat_completion): rag_pipe = build_rag_pipeline(document_store=InMemoryDocumentStore()) answer = rag_pipe.run(query="question")