python/tests/test_basic.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 numpy as np
import pytest
from tsfile import schema, Field
from tsfile import Tablet
from tsfile.constants import *
from tsfile.schema import *


def test_tablet():
    column_names = ["temp1", "temp2", "value1", "value2", "string1"]
    data_types = [TSDataType.INT32, TSDataType.INT64, TSDataType.FLOAT, TSDataType.DOUBLE, TSDataType.STRING]
    tablet = Tablet(column_names, data_types)
    tablet.set_table_name("test")

    assert "test" == tablet.get_target_name()
    assert 5 == len(tablet.get_column_name_list())
    assert TSDataType.INT32 == tablet.get_data_type_list()[0]

    tablet.add_timestamp(0, 10)
    tablet.add_value_by_name("temp1", 0, 10)
    tablet.add_value_by_name("temp2", 0, 100)
    tablet.add_value_by_index(2, 0, 0.1)
    tablet.add_value_by_index(3, 0, 0.1)
    tablet.add_value_by_index(4, 0, "test")
    tablet.add_value_by_name("string1", 0, "test")
    # Illegal column name
    with pytest.raises(ValueError):
        tablet.add_value_by_name("temp3", 0, 10)
    # Illegal exists column index
    with pytest.raises(IndexError):
        tablet.add_value_by_index(5, 0, 10)
    # Illegal row index
    with pytest.raises(IndexError):
        tablet.add_value_by_name("temp1", 2048, 10)
    # Illegal data type
    with pytest.raises(TypeError):
        tablet.add_value_by_name("temp1", 2, 10.0)

    # Illegal data scope
    with pytest.raises(OverflowError):
        tablet.add_value_by_index(0, 20, np.iinfo(np.int64).max)

    tablet.add_value_by_index(0, 30, np.iinfo(np.int32).max)

    assert 0.1 == tablet.get_value_list_by_name("value1")[0]
    assert np.iinfo(np.int32).max == tablet.get_value_list_by_name("temp1")[30]

def test_field():
    field_int32 = Field("int32",10, TSDataType.INT32)
    field_int64 = Field("int64", np.iinfo(np.int32).max + 1, TSDataType.INT64)
    field_float = Field("float",10.0, TSDataType.FLOAT)
    field_double = Field("double",10.0, TSDataType.DOUBLE)
    field_bool = Field("bool",True, TSDataType.BOOLEAN)
    field = Field("t",100)

    assert 100 == field.get_value()
    assert np.int64(10) == field_int32.get_long_value()
    assert True == field_bool.get_bool_value()
    assert np.float64(10) == field_double.get_double_value()
    assert np.float32(10) == field_float.get_float_value()
    assert TSDataType.INT64 == field_int64.get_data_type()
    assert False == field_int32.is_null()
    field.set_value(200)
    field.set_data_type(TSDataType.DOUBLE)
    assert np.float32(200) == field.get_float_value()
    assert np.float64(200) == field.get_double_value()
    assert field_int64.get_value() == field_int64.get_timestamp_value()

    field = Field("t", "t1", TSDataType.STRING)
    assert "t1" == field.get_string_value()
    assert "10" == field_int32.get_string_value()

    with pytest.raises(OverflowError):
        field_int64.get_int_value()

def test_schema():
    column1 = ColumnSchema("device", TSDataType.STRING, ColumnCategory.TAG)
    column2 = ColumnSchema("sensor", TSDataType.STRING, ColumnCategory.TAG)
    # Default by FIELD.
    column3 = ColumnSchema("value1", TSDataType.DOUBLE)
    column4 = ColumnSchema("value2", TSDataType.INT32, ColumnCategory.FIELD)
    table = TableSchema("test_table", [column1, column2, column3, column4])

    assert column3.get_category() == ColumnCategory.FIELD
    assert column4.__str__() == "ColumnSchema(value2, INT32, FIELD)"

    with pytest.raises(ValueError):
        tablet = TableSchema("", [column1, column2, column3, column4])

    with pytest.raises(ValueError):
        tablet = TableSchema("test_table", [])

    with pytest.raises(ValueError):
        column = ColumnSchema("test_column",None, ColumnCategory.TAG)

    with pytest.raises(ValueError):
        tablet = TableSchema("test_table", [ColumnSchema("", TSDataType.DOUBLE)])