blob: 22bcd1b1b4ac127f92b1d9733f91d367bc681d29 [file]
"""
Licensed to the Apache Software Foundation (ASF) under one
or more contributor license agreements. See the NOTICE file
distributed with this work for additional information
regarding copyright ownership. The ASF licenses this file
to you under the Apache License, Version 2.0 (the
"License"); you may not use this file except in compliance
with the License. You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing,
software distributed under the License is distributed on an
"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
KIND, either express or implied. See the License for the
specific language governing permissions and limitations
under the License.
"""
import pytest
from gremlin_python.structure.graph import Tree
from gremlin_python.structure.io.graphbinaryV4 import GraphBinaryWriter, GraphBinaryReader
writer = GraphBinaryWriter()
reader = GraphBinaryReader()
def build_tree():
"""
Builds the following tree in-memory (mirrors Go's TestTreeReadAPI fixture):
a
|-- b
| |-- d
| |-- e
|-- c
| |-- f
"""
tree = Tree()
a = tree.get_or_create_child("a")
b = a.get_or_create_child("b")
b.get_or_create_child("d")
b.get_or_create_child("e")
c = a.get_or_create_child("c")
c.get_or_create_child("f")
return tree
def test_root_nodes():
tree = build_tree()
assert tree.root_nodes() == ["a"]
assert tree.child_at("a").root_nodes() == ["b", "c"]
def test_child_at():
tree = build_tree()
a = tree.child_at("a")
assert isinstance(a, Tree)
assert a.root_nodes() == ["b", "c"]
def test_child_at_missing_key_raises():
tree = build_tree()
with pytest.raises(KeyError):
tree.child_at("does-not-exist")
def test_has_child():
tree = build_tree()
assert tree.has_child("a") is True
assert tree.has_child("b") is False
assert tree.child_at("a").has_child("b") is True
assert tree.child_at("a").has_child("z") is False
def test_contains():
tree = build_tree()
assert tree.contains("a") is True
assert tree.contains("d") is True
assert tree.contains("f") is True
assert tree.contains("z") is False
def test_find_subtree():
tree = build_tree()
b = tree.find_subtree("b")
assert b is not None
assert b.root_nodes() == ["d", "e"]
assert tree.find_subtree("f").is_leaf()
assert tree.find_subtree("z") is None
def test_get_or_create_child():
tree = Tree()
first = tree.get_or_create_child("k")
assert isinstance(first, Tree)
# returns the existing child on re-request rather than replacing it
first.get_or_create_child("nested")
second = tree.get_or_create_child("k")
assert second is first
assert second.has_child("nested")
assert tree.root_nodes() == ["k"]
def test_is_leaf():
tree = build_tree()
assert tree.is_leaf() is False
assert tree.find_subtree("d").is_leaf() is True
assert Tree().is_leaf() is True
def test_node_count():
tree = build_tree()
# a, b, c, d, e, f
assert tree.node_count() == 6
assert Tree().node_count() == 0
def test_get_nodes_at_depth():
tree = build_tree()
assert tree.get_nodes_at_depth(0) == ["a"]
assert tree.get_nodes_at_depth(1) == ["b", "c"]
assert tree.get_nodes_at_depth(2) == ["d", "e", "f"]
# out of range
assert tree.get_nodes_at_depth(3) == []
# negative
assert tree.get_nodes_at_depth(-1) == []
def test_get_trees_at_depth():
tree = build_tree()
depth0 = tree.get_trees_at_depth(0)
assert depth0 == [tree]
depth1 = tree.get_trees_at_depth(1)
assert len(depth1) == 1
assert depth1[0].root_nodes() == ["b", "c"]
depth2 = tree.get_trees_at_depth(2)
roots = []
for t in depth2:
roots.extend(t.root_nodes())
assert roots == ["d", "e", "f"]
# depth 3 is the (empty) subtrees hanging off the leaf nodes d, e, f: they
# exist as empty trees, so no nodes but three empty trees are returned.
depth3 = tree.get_trees_at_depth(3)
assert len(depth3) == 3
assert all(t.is_leaf() for t in depth3)
# depth 4 is genuinely beyond the tree's height -> empty list
assert tree.get_trees_at_depth(4) == []
assert tree.get_trees_at_depth(-1) == []
def test_get_leaf_nodes():
tree = build_tree()
assert tree.get_leaf_nodes() == ["d", "e", "f"]
def test_get_leaf_trees():
tree = build_tree()
leaf_trees = tree.get_leaf_trees()
roots = [t.root_nodes()[0] for t in leaf_trees]
assert roots == ["d", "e", "f"]
for t in leaf_trees:
assert len(t.root_nodes()) == 1
assert t.child_at(t.root_nodes()[0]).is_leaf()
def test_add_tree_merge():
base = Tree()
a = base.get_or_create_child("a")
a.get_or_create_child("b")
other = Tree()
oa = other.get_or_create_child("a")
oa.get_or_create_child("c")
other.get_or_create_child("x")
base.add_tree(other)
assert base.root_nodes() == ["a", "x"]
# overlapping key 'a' has both 'b' (original) and 'c' (merged in)
assert base.child_at("a").root_nodes() == ["b", "c"]
# key present only in other adopted directly
assert base.has_child("x")
def test_split_parents_multiple():
tree = Tree()
tree.get_or_create_child("a").get_or_create_child("a1")
tree.get_or_create_child("b").get_or_create_child("b1")
parts = tree.split_parents()
assert len(parts) == 2
assert parts[0].root_nodes() == ["a"]
assert parts[0].child_at("a").root_nodes() == ["a1"]
assert parts[1].root_nodes() == ["b"]
def test_split_parents_single():
tree = build_tree()
parts = tree.split_parents()
assert parts == [tree]
def test_add_tree_none_raises():
# Merging a None tree is invalid: add_tree dereferences other._entries,
# which raises on a None argument (AttributeError/TypeError).
tree = build_tree()
with pytest.raises((AttributeError, TypeError)):
tree.add_tree(None)
def test_element_keys_dedup_by_id_and_type():
from gremlin_python.structure.graph import Vertex, Edge
# Same id, differing label/properties -> same Element (Element.__eq__ is
# isinstance(self.__class__) and id == id), so the key collapses.
same_id = Tree()
same_id.get_or_create_child(Vertex(1, "person"))
same_id.get_or_create_child(Vertex(1, "software"))
assert len(same_id.root_nodes()) == 1
# Differing ids -> two distinct keys.
diff_id = Tree()
diff_id.get_or_create_child(Vertex(1))
diff_id.get_or_create_child(Vertex(2))
assert len(diff_id.root_nodes()) == 2
# Same id but different concrete type (Vertex vs Edge) -> two distinct keys.
diff_type = Tree()
diff_type.get_or_create_child(Vertex(1))
diff_type.get_or_create_child(Edge(1, Vertex(2), "knows", Vertex(3)))
assert len(diff_type.root_nodes()) == 2
def test_hash_consistent_with_eq():
# Equal trees must have equal hashes. Python's Tree.__hash__ is based on the
# root entry count, which structurally-equal trees share.
t1 = build_tree()
t2 = build_tree()
assert t1 == t2
assert hash(t1) == hash(t2)
def test_pretty_print():
tree = build_tree()
expected = "\n".join([
"|--a",
" |--b",
" |--d",
" |--e",
" |--c",
" |--f",
])
assert tree.pretty_print() == expected
# no trailing newline
assert not tree.pretty_print().endswith("\n")
def test_eq_order_insensitivity():
t1 = Tree()
r1 = t1.get_or_create_child("root")
r1.get_or_create_child("a")
r1.get_or_create_child("b")
t2 = Tree()
r2 = t2.get_or_create_child("root")
# insert in reversed sibling order
r2.get_or_create_child("b")
r2.get_or_create_child("a")
assert t1 == t2
assert not (t1 != t2)
def test_eq_differing_structure():
t1 = build_tree()
t2 = build_tree()
assert t1 == t2
t2.child_at("a").get_or_create_child("extra")
assert t1 != t2
def test_null_key_round_trip():
tree = Tree()
tree.get_or_create_child(None).get_or_create_child("child")
round_tripped = reader.read_object(writer.write_object(tree))
assert isinstance(round_tripped, Tree)
assert round_tripped.has_child(None)
assert round_tripped.child_at(None).has_child("child")
assert round_tripped == tree
# ---- Deserialization tests (duplicate-sibling-key dedup) ----
import io
from gremlin_python.structure.io.graphbinaryV4 import DataType, TreeIO, int32_pack
def _fq_string(s):
"""Fully-qualified GraphBinary string: {type-id}{null flag}{int32 len}{bytes}."""
b = bytearray([DataType.string.value, 0x00])
encoded = s.encode("utf-8")
b.extend(int32_pack(len(encoded)))
b.extend(encoded)
return b
def test_deserializer_duplicate_sibling_key():
# length=2; key='dup' appears twice with distinct children -> one merged entry.
b = bytearray(int32_pack(2))
# entry 1: key='dup', child={ 'a' -> {} }
b.extend(_fq_string("dup"))
b.extend(int32_pack(1))
b.extend(_fq_string("a"))
b.extend(int32_pack(0))
# entry 2: key='dup', child={ 'b' -> {} }
b.extend(_fq_string("dup"))
b.extend(int32_pack(1))
b.extend(_fq_string("b"))
b.extend(int32_pack(0))
tree = TreeIO._read_tree(io.BytesIO(bytes(b)), reader)
assert tree.root_nodes() == ["dup"]
merged = tree.child_at("dup")
assert merged.has_child("a")
assert merged.has_child("b")