Function test_memory_query_cache

tests/test_query_cache.py:64–98

(processed_queries)

def test_memory_query_cache(processed_queries):
    environ = {
        "MM_QUERY_CACHE_IN_MEMORY": "1",
        "MM_QUERY_CACHE_WRITE_SIZE": "10"
    }

    with TemporaryDirectory() as tmpdir, patch.dict(os.environ, environ):
        text_prep_pipeline = TextPreparationPipelineFactory.create_from_app_path(tmpdir)
        cache = QueryCache(tmpdir, text_prep_pipeline.get_hashid())

        # Verify that there is in-memory caching
        assert cache.memory_connection is not None
        for q in processed_queries[:9]:
            key = QueryCache.get_key(q.domain, q.intent, q.query.text)
            cache.put(key, q)
            # Verify that the first 9 queries are not written to disk
            assert not get_query_from_disk(tmpdir, key)

        # Verify that the 10th query triggers a flush to disk
        q = processed_queries[9]
        key = QueryCache.get_key(q.domain, q.intent, q.query.text)
        cache.put(key, q)
        for q in processed_queries[:10]:
            key = QueryCache.get_key(q.domain, q.intent, q.query.text)
            assert get_query_from_disk(tmpdir, key) == (q.domain, q.intent, q.query.text)

        # Verify that a GC triggers a flush to disk
        for q in processed_queries[10:15]:
            key = QueryCache.get_key(q.domain, q.intent, q.query.text)
            cache.put(key, q)
            assert not get_query_from_disk(tmpdir, key)
        cache = None
        for q in processed_queries[10:15]:
            key = QueryCache.get_key(q.domain, q.intent, q.query.text)
            assert get_query_from_disk(tmpdir, key)