(processed_queries)
| 46 | |
| 47 | |
| 48 | def test_disk_query_cache(processed_queries): |
| 49 | environ = {"MM_QUERY_CACHE_IN_MEMORY": "0"} |
| 50 | |
| 51 | with TemporaryDirectory() as tmpdir, patch.dict(os.environ, environ): |
| 52 | text_prep_pipeline = TextPreparationPipelineFactory.create_from_app_path(tmpdir) |
| 53 | cache = QueryCache(tmpdir, text_prep_pipeline.get_hashid()) |
| 54 | |
| 55 | # Verify that there is no in-memory caching |
| 56 | assert cache.memory_connection is None |
| 57 | for q in processed_queries: |
| 58 | key = QueryCache.get_key(q.domain, q.intent, q.query.text) |
| 59 | cache.put(key, q) |
| 60 | # Verify that queries are written to disk immediately |
| 61 | assert get_query_from_disk(tmpdir, key) == (q.domain, q.intent, q.query.text) |
| 62 | |
| 63 | |
| 64 | def test_memory_query_cache(processed_queries): |
nothing calls this directly
no test coverage detected