haystack/test/components/caching/test_cache_checker.py at 003cb43a29029d0eed4d4e6ce96ff3f96586bbee · deepset-ai/haystack · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
# SPDX-FileCopyrightText: 2022-present deepset GmbH <info@deepset.ai>
#
# SPDX-License-Identifier: Apache-2.0

from unittest.mock import MagicMock, Mock

import pytest

from haystack import Document
from haystack.components.caching.cache_checker import CacheChecker
from haystack.document_stores.in_memory import InMemoryDocumentStore
from haystack.testing.factory import document_store_class


class TestCacheChecker:
    def test_to_dict(self):
        mocked_docstore_class = document_store_class("MockedDocumentStore")
        component = CacheChecker(document_store=mocked_docstore_class(), cache_field="url")
        data = component.to_dict()
        assert data == {
            "type": "haystack.components.caching.cache_checker.CacheChecker",
            "init_parameters": {
                "document_store": {"type": "haystack.testing.factory.MockedDocumentStore", "init_parameters": {}},
                "cache_field": "url",
            },
        }

    def test_to_dict_with_custom_init_parameters(self):
        mocked_docstore_class = document_store_class("MockedDocumentStore")
        component = CacheChecker(document_store=mocked_docstore_class(), cache_field="my_url_field")
        data = component.to_dict()
        assert data == {
            "type": "haystack.components.caching.cache_checker.CacheChecker",
            "init_parameters": {
                "document_store": {"type": "haystack.testing.factory.MockedDocumentStore", "init_parameters": {}},
                "cache_field": "my_url_field",
            },
        }

    def test_from_dict(self):
        data = {
            "type": "haystack.components.caching.cache_checker.CacheChecker",
            "init_parameters": {
                "document_store": {
                    "type": "haystack.document_stores.in_memory.document_store.InMemoryDocumentStore",
                    "init_parameters": {},
                },
                "cache_field": "my_url_field",
            },
        }
        component = CacheChecker.from_dict(data)
        assert isinstance(component.document_store, InMemoryDocumentStore)
        assert component.cache_field == "my_url_field"

    def test_from_dict_without_docstore(self):
        data = {"type": "haystack.components.caching.cache_checker.CacheChecker", "init_parameters": {}}
        with pytest.raises(
            TypeError, match="missing 2 required positional arguments: 'document_store' and 'cache_field'"
        ):
            CacheChecker.from_dict(data)

    def test_from_dict_nonexisting_docstore(self):
        # Use a type whose module passes the deserialization allowlist (haystack.*) but cannot be
        # resolved, so we still exercise the "import failed" code path rather than the allowlist gate.
        data = {
            "type": "haystack.components.caching.cache_checker.CacheChecker",
            "init_parameters": {
                "document_store": {"type": "haystack.does.not.exist.DocumentStore", "init_parameters": {}}
            },
        }
        with pytest.raises(
            ImportError, match=r"Failed to deserialize 'document_store':.*haystack\.does\.not\.exist\.DocumentStore"
        ):
            CacheChecker.from_dict(data)

    def test_run(self, in_memory_doc_store):
        documents = [
            Document(content="doc1", meta={"url": "https://example.com/1"}),
            Document(content="doc2", meta={"url": "https://example.com/2"}),
            Document(content="doc3", meta={"url": "https://example.com/1"}),
            Document(content="doc4", meta={"url": "https://example.com/2"}),
        ]
        in_memory_doc_store.write_documents(documents)
        checker = CacheChecker(in_memory_doc_store, cache_field="url")
        results = checker.run(items=["https://example.com/1", "https://example.com/5"])
        assert results == {"hits": [documents[0], documents[2]], "misses": ["https://example.com/5"]}

    def test_filters_syntax(self):
        mocked_docstore_class = document_store_class("MockedDocumentStore")
        mocked_docstore_class.filter_documents = MagicMock()
        checker = CacheChecker(document_store=mocked_docstore_class(), cache_field="url")
        checker.run(items=["https://example.com/1"])
        valid_filters_syntax = {"field": "url", "operator": "==", "value": "https://example.com/1"}
        mocked_docstore_class.filter_documents.assert_any_call(filters=valid_filters_syntax)

    def test_close(self):
        closable_document_store = Mock(spec=["close"])
        checker = CacheChecker(document_store=closable_document_store, cache_field="url")
        checker.close()
        closable_document_store.close.assert_called_once_with()

        nonclosable_document_store = Mock(spec=[])
        checker = CacheChecker(document_store=nonclosable_document_store, cache_field="url")
        checker.close()
        assert nonclosable_document_store.mock_calls == []