wehub-resource-sync
6b7e6b44f1
Python Build and Type Check / python-ci (ubuntu-latest, 3.11) (push) Has been cancelled
Python Build and Type Check / python-ci (ubuntu-latest, 3.13) (push) Has been cancelled
Python Build and Type Check / python-ci (windows-latest, 3.11) (push) Has been cancelled
Python Build and Type Check / python-ci (windows-latest, 3.13) (push) Has been cancelled
Python Integration Tests / python-ci (ubuntu-latest, 3.13) (push) Has been cancelled
Python Integration Tests / python-ci (windows-latest, 3.13) (push) Has been cancelled
Python Notebook Tests / python-ci (ubuntu-latest, 3.13) (push) Has been cancelled
Python Notebook Tests / python-ci (windows-latest, 3.13) (push) Has been cancelled
Python Smoke Tests / python-ci (ubuntu-latest, 3.13) (push) Has been cancelled
Python Smoke Tests / python-ci (windows-latest, 3.13) (push) Has been cancelled
Python Unit Tests / python-ci (ubuntu-latest, 3.13) (push) Has been cancelled
Python Unit Tests / python-ci (windows-latest, 3.13) (push) Has been cancelled
gh-pages / build (push) Has been cancelled
Python Publish (pypi) / Upload release to PyPI (push) Has been cancelled
Spellcheck / spellcheck (push) Has been cancelled
81 lines
2.4 KiB
Python
81 lines
2.4 KiB
Python
# Copyright (c) 2024 Microsoft Corporation.
|
|
# Licensed under the MIT License
|
|
import asyncio
|
|
import os
|
|
import unittest
|
|
|
|
from graphrag_cache import CacheConfig, CacheType
|
|
from graphrag_cache import create_cache as cc
|
|
from graphrag_storage import StorageConfig, StorageType
|
|
|
|
TEMP_DIR = "./.tmp"
|
|
|
|
|
|
def create_cache():
|
|
return cc(
|
|
CacheConfig(
|
|
type=CacheType.Json,
|
|
storage=StorageConfig(
|
|
type=StorageType.File,
|
|
base_dir=os.path.join(os.getcwd(), ".tmp"),
|
|
),
|
|
),
|
|
)
|
|
|
|
|
|
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
|