gremlin-python/src/main/python/tests/unit/structure/io/test_graphbinaryv4model.py

"""
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

"""
Round trip testing of GraphBinaryV4 compared to a correct "model".

Set the IO_TEST_DIRECTORY environment variable to the directory where
the .gbin files that represent the serialized "model" are located. 
"""

from gremlin_python.structure.io.graphbinaryV4 import GraphBinaryWriter, GraphBinaryReader
import math
import os

from .model import model

gremlin_test_dir = "gremlin-test/src/main/resources/org/apache/tinkerpop/gremlin/structure/io/graphbinary/"
directory_search_pattern = "gremlin-python" + os.sep + "src" + os.sep + "main"
default_dir = __file__[:__file__.find(directory_search_pattern)]
test_resource_directory = os.environ.get('IO_TEST_DIRECTORY', default_dir + gremlin_test_dir)
writer = GraphBinaryWriter()
reader = GraphBinaryReader()

def decimal_cmp(x, y):
    return x.scale == y.scale and x.unscaled_value == y.unscaled_value and x.value == y.value

def vertex_label_cmp(x, y):
    return x == y and x.labels == y.labels

def property_cmp(x, y):
    return x.key == y.key and x.value == y.value

def vertex_property_cmp(x, y):
    return x == y and x.label == y.label and x.value == y.value and \
           len(x.properties) == len(y.properties) and \
           all(property_cmp(xp, yp) for xp, yp in zip(x.properties, y.properties))

def vertex_cmp(x, y):
    return x == y and x.label == y.label and x.labels == y.labels and \
           len(x.properties) == len(y.properties) and \
           all(vertex_property_cmp(xp, yp) for xp, yp in zip(x.properties, y.properties))

def edge_cmp(x, y):
    return x == y and x.label == y.label and x.labels == y.labels and \
           x.outV == y.outV and x.inV == y.inV and \
           len(x.properties) == len(y.properties) and \
           all(property_cmp(xp, yp) for xp, yp in zip(x.properties, y.properties))

def graph_cmp(x, y):
    return x.vertices.keys() == y.vertices.keys() and \
           x.edges.keys() == y.edges.keys() and \
           all(vertex_cmp(x.vertices[k], y.vertices[k]) for k in x.vertices.keys()) and \
           all(edge_cmp(x.edges[k], y.edges[k]) for k in x.edges.keys())

def get_entry(title):
    return model[title]

def read_file_by_name(resource_name):
    full_name = test_resource_directory + resource_name + "-" + "v4" + ".gbin"
    with open(full_name, 'rb') as resource_file:
        return bytearray(resource_file.read())

def test_pos_bigdecimal():
    # BigDecimal writes through Python BigInteger, which may not preserve byte-exact sign padding.
    run_writeread("pos-bigdecimal", decimal_cmp)

def test_neg_bigdecimal():
    # BigDecimal writes through Python BigInteger, which may not preserve byte-exact sign padding.
    run_writeread("neg-bigdecimal", decimal_cmp)

def test_zero_bigdecimal():
    # Use the same reduced mode as other Python BigDecimal model cases.
    run_writeread("zero-bigdecimal", decimal_cmp)

def test_scale_zero_bigdecimal():
    # Use the same reduced mode as other Python BigDecimal model cases.
    run_writeread("scale-zero-bigdecimal", decimal_cmp)

def test_negative_scale_bigdecimal():
    # Use the same reduced mode as other Python BigDecimal model cases.
    run_writeread("negative-scale-bigdecimal", decimal_cmp)

def test_small_decimal_bigdecimal():
    # Use the same reduced mode as other Python BigDecimal model cases.
    run_writeread("small-decimal-bigdecimal", decimal_cmp)

def test_pos_biginteger():
    # gremlin-python adds an extra 0 byte to the value.
    run_writeread("pos-biginteger")

def test_neg_biginteger():
    # gremlin-python adds an extra 0 byte to the value.
    run_writeread("neg-biginteger")

def test_zero_biginteger():
    # gremlin-python may add an extra sign-padding byte to the value.
    run_writeread("zero-biginteger")

def test_sign_boundary_pos_biginteger():
    # gremlin-python may add an extra sign-padding byte to the value.
    run_writeread("sign-boundary-pos-biginteger")

def test_sign_boundary_neg_biginteger():
    # gremlin-python can't serialize -129 because BigIntIO computes a byte length
    # that is too small for signed two's-complement encoding.
    run_read("sign-boundary-neg-biginteger")

def test_uint8_primitive_pdt():
    run("uint8-primitive-pdt")

def test_point_composite_pdt():
    run("point-composite-pdt")

def test_min_byte():
    run("min-byte")
    
def test_max_byte():
    run("max-byte")

def test_empty_binary():
    run("empty-binary")

def test_str_binary():
    run("str-binary")

def test_max_double():
    run("max-double")

def test_min_double():
    run("min-double")

def test_neg_max_double():
    run("neg-max-double")

def test_neg_min_double():
    run("neg-min-double")

def test_nan_double():
    def nan_cmp(x, y):
        if math.isnan(x) and math.isnan(y):
            return True
        else:
            return False
    run("nan-double", nan_cmp)

def test_pos_inf_double():
    run("pos-inf-double")

def test_neg_inf_double():
    run("neg-inf-double")

def test_neg_zero_double():
    run("neg-zero-double")

def test_zero_duration():
    run("zero-duration")

def test_positive_duration():
    run("positive-duration")

def test_negative_duration():
    # timedelta.seconds normalizes negative values, so the writer can't preserve the wire value.
    run_read("negative-duration")

def test_nanos_duration():
    # Python timedelta only preserves microseconds, not arbitrary nanoseconds.
    run_read("nanos-duration")

def test_traversal_edge():
    # properties aren't serialized in gremlin-python
    run_writeread("traversal-edge")

def test_no_prop_edge():
    # gremlin-python serializes/deserializes "null" for props not empty list
    run_writeread("no-prop-edge")

def test_max_int():
    run("max-int")

def test_min_int():
    run("min-int")

def test_max_long():
    # attempts to serialize a long as an int
    run_writeread("max-long")

def test_min_long():
    # attempts to serialize a long as an int
    run_writeread("min-long")

def test_var_type_list():
    run("var-type-list")

def test_empty_list():
    run("empty-list")

def test_var_type_map():
    # can't write tuple
    run_read("var-type-map")

def test_empty_map():
    run("empty-map")

def test_ordered_string_int_map():
    run("ordered-string-int-map")

def test_traversal_path():
    # gremlin-python serializes/deserializes "null" for props not empty list
    run_writeread("traversal-path")

def test_empty_path():
    run("empty-path")

def test_path_zero_labels():
    run("path-zero-labels")

def test_path_multiple_labels():
    # label set ordering isn't guaranteed after a Python object round trip.
    run_writeread("path-multiple-labels")

def test_traversal_tree():
    # gremlin-python doesn't serialize vertex properties, so the read tree's
    # vertices have no properties; vertex equality is by id so the model matches.
    run_writeread("traversal-tree")

def test_empty_tree():
    run("empty-tree")

def test_tree_null_key():
    run("tree-null-key")

def test_tree_mixed_key_types():
    run("tree-mixed-key-types")

def test_tree_deep_nesting():
    run("tree-deep-nesting")

def test_prop_path():
    # gremlin-python doesn't serialize properties
    run_writeread("prop-path")

def test_edge_property():
    run("edge-property")

def test_null_property():
    run("null-property")

def test_var_type_set():
    # ordering within set might be different
    run_writeread("var-type-set")

def test_empty_set():
    run("empty-set")

def test_max_short():
    # short deserialized to int
    run_writeread("max-short")

def test_min_short():
    # short deserialized to int
    run_writeread("min-short")

def test_specified_uuid():
    run("specified-uuid")

def test_nil_uuid():
    run("nil-uuid")

def test_no_prop_vertex():
    # gremlin-python serializes/deserializes "null" for props not empty list
    run_writeread("no-prop-vertex")

def test_traversal_vertexproperty():
    # properties aren't serialized in gremlin-python
    run_writeread("traversal-vertexproperty")

def test_meta_vertexproperty():
    # properties aren't serialized in gremlin-python
    run_writeread("meta-vertexproperty")

def test_set_cardinality_vertexproperty():
    # properties aren't serialized in gremlin-python
    run_writeread("set-cardinality-vertexproperty")

def test_id_t():
    run("id-t")

def test_out_direction():
    run("out-direction")

def test_var_bulklist():
    # BulkList is deserialized as a plain list, so the writer emits a regular List.
    run_read("var-bulklist")

def test_empty_bulklist():
    # BulkList is deserialized as a plain list, so the writer emits a regular List.
    run_read("empty-bulklist")

def test_single_byte_char():
    # char is serialized as string
    run_writeread("single-byte-char")

def test_two_byte_char():
    # char is serialized as string
    run_writeread("two-byte-char")

def test_three_byte_char():
    # char is serialized as string
    run_writeread("three-byte-char")

def test_four_byte_char():
    # char is serialized as string
    run_writeread("four-byte-char")

def test_unspecified_null():
    # no serializer for plain null
    run_writeread("unspecified-null")

def test_null_int():
    # typed nulls deserialize to plain None, so the original type can't be re-emitted.
    run_read("null-int")

def test_null_long():
    # typed nulls deserialize to plain None, so the original type can't be re-emitted.
    run_read("null-long")

def test_null_string():
    # typed nulls deserialize to plain None, so the original type can't be re-emitted.
    run_read("null-string")

def test_null_list():
    # typed nulls deserialize to plain None, so the original type can't be re-emitted.
    run_read("null-list")

def test_null_map():
    # typed nulls deserialize to plain None, so the original type can't be re-emitted.
    run_read("null-map")

def test_null_set():
    # typed nulls deserialize to plain None, so the original type can't be re-emitted.
    run_read("null-set")

def test_true_boolean():
    run("true-boolean")

def test_false_boolean():
    run("false-boolean")

def test_single_byte_string():
    run("single-byte-string")

def test_mixed_string():
    run("mixed-string")

def test_empty_string():
    run("empty-string")

def test_traversal_vertex():
    # properties aren't serialized in gremlin-python
    run_writeread("traversal-vertex")

def test_multi_label_vertex():
    # Vertex equality is id-only, so compare labels explicitly.
    run_writeread("multi-label-vertex", vertex_label_cmp)

def test_empty_label_vertex():
    # Vertex equality is id-only, so compare labels explicitly.
    run_writeread("empty-label-vertex", vertex_label_cmp)

def test_tinker_graph():
    # The hand-built graph is structurally equivalent, but it doesn't preserve
    # every wire-level numeric type tag needed for byte-exact writes.
    run_writeread("tinker-graph", graph_cmp)

def test_merge_on_create():
    run("merge-on-create")

def test_merge_on_match():
    run("merge-on-match")

def test_merge_out_v():
    run("merge-out-v")

def test_merge_in_v():
    run("merge-in-v")

def run(resource_name, comparator = None):
    """
    Runs the regular set of tests for the type which is
        1. model to deserialized object
        2. written bytes to read bytes
        3. round tripped (read then written) bytes to read bytes
    
    Optionally takes in a comparator that used for comparison 1.
    """
    resource_bytes = read_file_by_name(resource_name)

    model = get_entry(resource_name)
    read = reader.read_object(resource_bytes)
    if comparator is not None:
        assert comparator(model, read)
    else:
        assert model == read

    written = writer.write_object(model)
    assert resource_bytes == written

    round_tripped = writer.write_object(reader.read_object(resource_bytes))
    assert resource_bytes == round_tripped

def run_read(resource_name):
    """
    Runs the read test which compares the model to deserialized object

    This should only be used in cases where there is only a deserializer
    but no serializer for the same type.
    """
    resource_bytes = read_file_by_name(resource_name)

    model = get_entry(resource_name)
    read = reader.read_object(resource_bytes)
    assert model == read

def run_writeread(resource_name, comparator = None):
    """
    Runs a reduced set of tests for the type which is
        1. model to deserialized object
        2. model to round tripped (written then read) object
    
    Optionally takes in a comparator that used for comparison 1.

    Use this in cases where the regular run() function in too stringent.
    E.g. when ordering doesn't matter like for sets. Ideally, a type
    should be tested with run() instead when possible.
    """
    resource_bytes = read_file_by_name(resource_name)

    model = get_entry(resource_name)
    read = reader.read_object(resource_bytes)
    round_tripped = reader.read_object(writer.write_object(model))

    if comparator is not None:
        assert comparator(model, read)
        assert comparator(model, round_tripped)
    else:
        assert model == read
        assert model == round_tripped