wehub-resource-sync
c889a57b6b
Test Suites / Build CI Environment (push) Has been cancelled
Test Suites / Basic Tests (push) Has been cancelled
Test Suites / End-to-End Tests (push) Has been cancelled
Test Suites / CLI Tests (push) Has been cancelled
Test Suites / Slow End-to-End Tests (push) Has been cancelled
Test Suites / Graph Database Tests (push) Has been cancelled
Test Suites / Vector DB Tests (push) Has been cancelled
Test Suites / Temporal Graph Test (push) Has been cancelled
Test Suites / Search Test on Different DBs (push) Has been cancelled
Test Suites / Example Tests (push) Has been cancelled
Test Suites / Notebook Tests (push) Has been cancelled
Test Suites / OS and Python Tests Ubuntu (push) Has been cancelled
Test Suites / OS and Python Tests Extended (push) Has been cancelled
Test Suites / LLM Test Suite (push) Has been cancelled
Test Suites / S3 File Storage Test (push) Has been cancelled
Test Suites / Run Integration Tests (push) Has been cancelled
Test Suites / MCP Tests (push) Has been cancelled
Test Suites / Docker Compose Test (push) Has been cancelled
Test Suites / Docker CI test (push) Has been cancelled
Test Suites / Relational DB Migration Tests (push) Has been cancelled
Test Suites / Distributed Cognee Test (push) Has been cancelled
Test Suites / DB Examples Tests (push) Has been cancelled
Test Suites / Test Completion Status (push) Has been cancelled
Test Suites / Claude Code Review (push) Has been cancelled
Test Suites / basic checks (push) Has been cancelled
build | Build and Push Cognee MCP Docker Image to dockerhub / docker-build-and-push (push) Has been cancelled
Scorecard supply-chain security / Scorecard analysis (push) Has been cancelled
build | Build and Push Docker Image to dockerhub / docker-build-and-push (push) Has been cancelled
Weighted Edges Tests / Test Weighted Edges Core Functionality (3.11) (push) Has been cancelled
Weighted Edges Tests / Test Weighted Edges Core Functionality (3.12) (push) Has been cancelled
Weighted Edges Tests / Test Weighted Edges with Different Graph Databases (kuzu, kuzu) (push) Has been cancelled
Weighted Edges Tests / Test Weighted Edges with Different Graph Databases (neo4j, neo4j) (push) Has been cancelled
Weighted Edges Tests / Test Weighted Edges Examples (push) Has been cancelled
Weighted Edges Tests / Code Quality for Weighted Edges (push) Has been cancelled
83 lines
2.4 KiB
Python
83 lines
2.4 KiB
Python
import pytest
|
|
|
|
from distributed.graph_write_batch import apply_grouped_graph_writes, group_graph_writes
|
|
|
|
|
|
def test_group_graph_writes_groups_by_provenance_key():
|
|
items = [
|
|
(["n1"], [], "srA", "run1"),
|
|
([], ["e1"], "srA", "run1"),
|
|
(["n2"], ["e2"], "srB", "run2"),
|
|
(["n3"], [], "srA", "run1"),
|
|
]
|
|
|
|
groups = group_graph_writes(items)
|
|
|
|
assert groups[("srA", "run1")] == (["n1", "n3"], ["e1"])
|
|
assert groups[("srB", "run2")] == (["n2"], ["e2"])
|
|
# First-seen key order is preserved.
|
|
assert list(groups.keys()) == [("srA", "run1"), ("srB", "run2")]
|
|
|
|
|
|
def test_group_graph_writes_keeps_unprovenanced_items_separate():
|
|
items = [
|
|
(["n1"], [], None, None),
|
|
(["n2"], [], "srA", "run1"),
|
|
]
|
|
|
|
groups = group_graph_writes(items)
|
|
|
|
assert groups[(None, None)] == (["n1"], [])
|
|
assert groups[("srA", "run1")] == (["n2"], [])
|
|
|
|
|
|
def test_group_graph_writes_empty():
|
|
assert group_graph_writes([]) == {}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_apply_grouped_graph_writes_writes_all_nodes_before_any_edge():
|
|
calls = []
|
|
|
|
async def add_nodes(nodes, source_ref_key, pipeline_run_id):
|
|
calls.append(("nodes", tuple(nodes), source_ref_key, pipeline_run_id))
|
|
|
|
async def add_edges(edges, source_ref_key, pipeline_run_id):
|
|
calls.append(("edges", tuple(edges), source_ref_key, pipeline_run_id))
|
|
|
|
groups = {
|
|
("srA", "run1"): (["n1"], ["e1"]),
|
|
("srB", "run2"): (["n2"], ["e2"]),
|
|
}
|
|
|
|
await apply_grouped_graph_writes(groups, add_nodes, add_edges)
|
|
|
|
# Every group's nodes are written before ANY edge, and each write carries its
|
|
# own group's provenance.
|
|
assert calls == [
|
|
("nodes", ("n1",), "srA", "run1"),
|
|
("nodes", ("n2",), "srB", "run2"),
|
|
("edges", ("e1",), "srA", "run1"),
|
|
("edges", ("e2",), "srB", "run2"),
|
|
]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_apply_grouped_graph_writes_skips_empty_batches():
|
|
calls = []
|
|
|
|
async def add_nodes(nodes, source_ref_key, pipeline_run_id):
|
|
calls.append(("nodes", source_ref_key))
|
|
|
|
async def add_edges(edges, source_ref_key, pipeline_run_id):
|
|
calls.append(("edges", source_ref_key))
|
|
|
|
groups = {
|
|
("srA", "run1"): (["n1"], []), # nodes only
|
|
("srB", "run2"): ([], ["e2"]), # edges only
|
|
}
|
|
|
|
await apply_grouped_graph_writes(groups, add_nodes, add_edges)
|
|
|
|
assert calls == [("nodes", "srA"), ("edges", "srB")]
|