gmakstutis/graphiti
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

make unit tests optional

Browse source
This commit is contained in:
prestonrasmussen 2025-06-30 11:18:12 -04:00
parent 0a7fadf62c
commit 885d14c9f8
1 changed files with 14 additions and 3 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

17
tests/test_node_falkordb_int.py
View file

@ -15,12 +15,12 @@ limitations under the License.
"""
import os
import unittest
from datetime import datetime, timezone
from uuid import uuid4
import pytest
from graphiti_core.driver.falkordb_driver import FalkorDriver
from graphiti_core.nodes import (
CommunityNode,
EntityNode,
@ -33,6 +33,14 @@ FALKORDB_PORT = os.getenv('FALKORDB_PORT', '6379')
FALKORDB_USER = os.getenv('FALKORDB_USER', None)
FALKORDB_PASSWORD = os.getenv('FALKORDB_PASSWORD', None)
try:
from graphiti_core.driver.falkordb_driver import FalkorDriver
HAS_FALKORDB = True
except ImportError:
FalkorDriver = None
HAS_FALKORDB = False
@pytest.fixture
def sample_entity_node():
@ -72,6 +80,7 @@ def sample_community_node():
@pytest.mark.asyncio
@pytest.mark.integration
@unittest.skipIf(not HAS_FALKORDB, "FalkorDB is not installed")
async def test_entity_node_save_get_and_delete(sample_entity_node):
falkor_driver = FalkorDriver(
host=FALKORDB_HOST,
@ -79,9 +88,9 @@ async def test_entity_node_save_get_and_delete(sample_entity_node):
username=FALKORDB_USER,
password=FALKORDB_PASSWORD
)
await sample_entity_node.save(falkor_driver)
retrieved = await EntityNode.get_by_uuid(falkor_driver, sample_entity_node.uuid)
assert retrieved.uuid == sample_entity_node.uuid
assert retrieved.name == 'Test Entity'
@ -93,6 +102,7 @@ async def test_entity_node_save_get_and_delete(sample_entity_node):
@pytest.mark.asyncio
@pytest.mark.integration
@unittest.skipIf(not HAS_FALKORDB, "FalkorDB is not installed")
async def test_community_node_save_get_and_delete(sample_community_node):
falkor_driver = FalkorDriver(
host=FALKORDB_HOST,
@ -115,6 +125,7 @@ async def test_community_node_save_get_and_delete(sample_community_node):
@pytest.mark.asyncio
@pytest.mark.integration
@unittest.skipIf(not HAS_FALKORDB, "FalkorDB is not installed")
async def test_episodic_node_save_get_and_delete(sample_episodic_node):
falkor_driver = FalkorDriver(
host=FALKORDB_HOST,