mirror of
https://github.com/microsoft/graphrag.git
synced 2026-02-22 02:35:34 +08:00
4404668aa8
Some checks are pending
Python Build and Type Check / python-ci (ubuntu-latest, 3.11) (push) Waiting to run
Python Build and Type Check / python-ci (ubuntu-latest, 3.12) (push) Waiting to run
Python Build and Type Check / python-ci (windows-latest, 3.11) (push) Waiting to run
Python Build and Type Check / python-ci (windows-latest, 3.12) (push) Waiting to run
Python Integration Tests / python-ci (ubuntu-latest, 3.12) (push) Waiting to run
Python Integration Tests / python-ci (windows-latest, 3.12) (push) Waiting to run
Python Notebook Tests / python-ci (ubuntu-latest, 3.12) (push) Waiting to run
Python Notebook Tests / python-ci (windows-latest, 3.12) (push) Waiting to run
Python Smoke Tests / python-ci (ubuntu-latest, 3.12) (push) Waiting to run
Python Smoke Tests / python-ci (windows-latest, 3.12) (push) Waiting to run
Python Unit Tests / python-ci (ubuntu-latest, 3.12) (push) Waiting to run
Python Unit Tests / python-ci (windows-latest, 3.12) (push) Waiting to run
* Add graphrag-storage.
75 lines
2.2 KiB
Python
75 lines
2.2 KiB
Python
# Copyright (c) 2024 Microsoft Corporation.
|
|
# Licensed under the MIT License
|
|
import asyncio
|
|
import os
|
|
import unittest
|
|
|
|
from graphrag.cache.json_pipeline_cache import JsonPipelineCache
|
|
from graphrag_storage.file_storage import (
|
|
FileStorage,
|
|
)
|
|
|
|
TEMP_DIR = "./.tmp"
|
|
|
|
|
|
def create_cache():
|
|
storage = FileStorage(base_dir=os.path.join(os.getcwd(), ".tmp"))
|
|
return JsonPipelineCache(storage)
|
|
|
|
|
|
class TestFilePipelineCache(unittest.IsolatedAsyncioTestCase):
|
|
def setUp(self):
|
|
self.cache = create_cache()
|
|
|
|
def tearDown(self):
|
|
asyncio.run(self.cache.clear())
|
|
|
|
async def test_cache_clear(self):
|
|
# Create a cache directory
|
|
if not os.path.exists(TEMP_DIR):
|
|
os.mkdir(TEMP_DIR)
|
|
with open(f"{TEMP_DIR}/test1", "w") as f:
|
|
f.write("This is test1 file.")
|
|
with open(f"{TEMP_DIR}/test2", "w") as f:
|
|
f.write("This is test2 file.")
|
|
|
|
# this invokes cache.clear()
|
|
await self.cache.clear()
|
|
|
|
# Check if the cache directory is empty
|
|
files = os.listdir(TEMP_DIR)
|
|
assert len(files) == 0
|
|
|
|
async def test_child_cache(self):
|
|
await self.cache.set("test1", "test1")
|
|
assert os.path.exists(f"{TEMP_DIR}/test1")
|
|
|
|
child = self.cache.child("test")
|
|
assert os.path.exists(f"{TEMP_DIR}/test")
|
|
|
|
await child.set("test2", "test2")
|
|
assert os.path.exists(f"{TEMP_DIR}/test/test2")
|
|
|
|
await self.cache.set("test1", "test1")
|
|
await self.cache.delete("test1")
|
|
assert not os.path.exists(f"{TEMP_DIR}/test1")
|
|
|
|
async def test_cache_has(self):
|
|
test1 = "this is a test file"
|
|
await self.cache.set("test1", test1)
|
|
|
|
assert await self.cache.has("test1")
|
|
assert not await self.cache.has("NON_EXISTENT")
|
|
assert await self.cache.get("NON_EXISTENT") is None
|
|
|
|
async def test_get_set(self):
|
|
test1 = "this is a test file"
|
|
test2 = "\\n test"
|
|
test3 = "\\\\\\"
|
|
await self.cache.set("test1", test1)
|
|
await self.cache.set("test2", test2)
|
|
await self.cache.set("test3", test3)
|
|
assert await self.cache.get("test1") == test1
|
|
assert await self.cache.get("test2") == test2
|
|
assert await self.cache.get("test3") == test3
|