test_box.py

import pytest
import numpy as np
from datetime import datetime, timezone
import time
from math import floor

import objectbox
from objectbox.exceptions import *
from tests.common import *
from tests.model import TestEntity, TestEntityDatetime, TestEntityFlex


def test_box_basics(test_store):
    box = test_store.box(TestEntity)

    assert box.is_empty()
    assert box.count() == 0

    # create
    object = TestEntity()
    id = box.put(object)
    assert id == 1
    assert id == object.id

    # create another one with some data
    object = TestEntity()
    object.bool = True
    object.int64 = 9223372036854775807
    object.int32 = 2147483647
    object.int16 = 32767
    object.int8 = 127
    object.str = "foo"
    object.float64 = 4.2
    object.float32 = 1.5
    object.bytes = bytes([1, 1, 2, 3, 5])
    object.bools = np.array([True, False, True, True, False], dtype=np.bool_)
    object.ints = np.array([1, 2, 3, 555, 120, 222], dtype=np.int32)
    object.shorts = np.array([7, 8, 9, 12, 13, 22], dtype=np.int16)
    object.chars = np.array([311, 426, 852, 927, 1025], dtype=np.uint16)
    object.longs = np.array([4299185519, 155462547, 5019238156195], dtype=np.int64)
    object.floats = np.array([0.1, 1.2, 2.3, 3.4, 4.5], dtype=np.float32)
    object.doubles = np.array([99.99, 88.88, 77.77, 66.66, 55.595425], dtype=np.float64)
    object.bools_list = [True, False, True, True, False]
    object.bytes_list = [-128, -1, 0, 1, 127]
    object.ints_list = [91, 82, 73, 64, 55]
    object.shorts_list = [8, 2, 7, 3, 6]
    object.chars_list = [4, 5, 43, 75, 12]
    object.longs_list = [4568, 8714, 1234, 5678, 9012240941]
    object.floats_list = [0.11, 1.22, 2.33, 3.44, 4.5595]
    object.doubles_list = [99.1999, 88.2888, 77.3777, 66.4666, 55.6597555]
    object.date = time.time()  # seconds since UNIX epoch (float)
    object.date_nano = time.time_ns()  # nanoseconds since UNIX epoch (int)
    object.flex = dict(a=1, b=2, c=3)
    object.transient = "abcd"

    id = box.put(object)
    assert id == 2
    assert id == object.id
    # check the count
    assert not box.is_empty()
    assert box.count() == 2

    # read
    # wrap date so it can be compared (is read as datetime)
    object.date = datetime.fromtimestamp(round(object.date * 1000) / 1000, tz=timezone.utc)
    read = box.get(object.id)
    assert_equal(read, object)
    assert read.transient != object.transient  # !=

    # update
    object.str = "bar"
    object.date = floor(time.time_ns() / 1000000)  # check that date can also be an int
    object.date_nano = time.time()  # check that date_nano can also be a float
    id = box.put(object)
    assert id == 2

    # read again
    read = box.get(object.id)
    assert read.str == "bar"
    assert floor(read.date.timestamp() * 1000) == object.date
    assert read.date_nano == floor(object.date_nano * 1000000000)

    # remove
    success = box.remove(object)
    assert success

    # remove should return success  
    success = box.remove(1)
    assert success
    success = box.remove(1)
    assert success is False

    # check they're gone
    assert box.count() == 0
    assert box.get(object.id) is None
    assert box.get(1) is None


def test_box_bulk(test_store):
    box = test_store.box(TestEntity)

    box.put(TestEntity(str="first"))

    objects = [TestEntity(str="second"), TestEntity(str="third"),
               TestEntity(str="fourth"), box.get(1)]
    box.put(objects)
    assert box.count() == 4
    assert objects[0].id == 2
    assert objects[1].id == 3
    assert objects[2].id == 4
    assert objects[3].id == 1

    read = box.get(objects[0].id)
    assert_equal(read, objects[0])
    assert_equal(box.get(objects[1].id), objects[1])
    assert_equal(box.get(objects[2].id), objects[2])
    assert_equal(box.get(objects[3].id), objects[3])

    objects_read = box.get_all()
    assert len(objects_read) == 4
    assert_equal(objects_read[0], objects[3])
    assert_equal(objects_read[1], objects[0])
    assert_equal(objects_read[2], objects[1])
    assert_equal(objects_read[3], objects[2])

    # remove all
    removed = box.remove_all()
    assert removed == 4
    assert box.count() == 0


def test_datetime(test_store):
    box = test_store.box(TestEntityDatetime)

    assert box.is_empty()
    assert box.count() == 0

    # creat - deferred for now, as there is an issue with 0 timestamp on Windows
    # object = TestEntityDatetime()
    # id = box.put(object)
    # assert id == 1
    # assert id == object.id

    # create with some data
    object = TestEntityDatetime()
    object.date = datetime.now(timezone.utc)  # milliseconds since UNIX epoch
    object.date_nano = datetime.now(timezone.utc)  # nanoseconds since UNIX epoch

    id = box.put(object)
    assert id == 1
    assert id == object.id
    # check the count
    assert not box.is_empty()
    assert box.count() == 1

    # read
    read = box.get(object.id)
    assert type(read.date) == float
    assert type(read.date_nano) == datetime
    assert pytest.approx(read.date) == object.date.timestamp()

    # update
    object.date = datetime.now(timezone.utc)
    object.date_nano = datetime.now(timezone.utc)
    id = box.put(object)
    assert id == 1

    # read again
    read = box.get(object.id)
    assert pytest.approx(read.date) == object.date.timestamp()
    assert pytest.approx(read.date_nano.timestamp()) == object.date_nano.timestamp()

    # remove
    success = box.remove(object)
    assert success

    # check they're gone
    assert box.count() == 0
    assert box.get(object.id) is None
    assert box.get(1) is None


def test_datetime_special_values(test_store):
    box = test_store.box(TestEntityDatetime)
    assert box.is_empty()

    object = TestEntityDatetime()
    object.date = 0
    object.date_nano = 0.0
    id = box.put(object)
    assert object.id == id

    read = box.get(id)
    assert isinstance(read.date, float)
    assert read.date == 0.0
    assert isinstance(read.date_nano, datetime)
    assert read.date_nano == datetime.fromtimestamp(0, timezone.utc)

    object.date = datetime.fromtimestamp(1.0, timezone.utc)
    object.date_nano = datetime.fromtimestamp(1.0, timezone.utc)
    id = box.put(object)

    read = box.get(id)
    assert isinstance(read.date, float)
    assert read.date == 1.0
    assert isinstance(read.date_nano, datetime)
    assert read.date_nano == datetime.fromtimestamp(1.0, timezone.utc)


def test_flex(test_store):
    def test_put_get(object: TestEntity, box: objectbox.Box, property):
        object.flex = property
        id = box.put(object)
        assert id == object.id
        read = box.get(object.id)
        assert read.flex == object.flex

    box = test_store.box(TestEntity)
    object = TestEntity()

    # Put an empty object
    id = box.put(object)
    assert id == object.id

    # Put a None type object
    test_put_get(object, box, None)

    # Update to int
    test_put_get(object, box, 1)

    # Update to float
    test_put_get(object, box, 1.2)

    # Update to string
    test_put_get(object, box, "foo")

    # Update to int list
    test_put_get(object, box, [1, 2, 3])

    # Update to float list
    test_put_get(object, box, [1.1, 2.2, 3.3])

    # Update to dict
    test_put_get(object, box, {"a": 1, "b": 2})

    # Update to bool
    test_put_get(object, box, True)

    # Update to dict inside dict
    test_put_get(object, box, {"a": 1, "b": {"c": 2}})

    # Update to list inside dict
    test_put_get(object, box, {"a": 1, "b": [1, 2, 3]})


def test_flex_values(test_store):
    box = test_store.box(TestEntityFlex)

    # Test empty object
    obj_id = box.put(TestEntityFlex())
    read_obj = box.get(obj_id)
    assert read_obj.flex is None

    # Test int
    obj_id = box.put(TestEntityFlex(flex=23))
    read_obj = box.get(obj_id)
    assert read_obj.flex == 23

    # Test string
    obj_id = box.put(TestEntityFlex(flex="hello"))
    read_obj = box.get(obj_id)
    assert read_obj.flex == "hello"

    # Test mixed list
    obj_id = box.put(TestEntityFlex(flex=[4, 5, 1, "foo", 23, "bar"]))
    read_obj = box.get(obj_id)
    assert read_obj.flex == [4, 5, 1, "foo", 23, "bar"]

    # Test dictionary
    dict_ = {"a": 1, "b": {"list": [1, 2, 3], "int": 5}}
    obj_id = box.put(TestEntityFlex(flex=dict_))
    read_obj = box.get(obj_id)
    assert read_obj.flex == dict_


def test_contains(test_store):
    box = test_store.box(TestEntity)

    box.put([TestEntity(str=f"Object {i}") for i in range(1, 4)])
    assert box.count() == 3

    assert box.contains(1)
    assert box.contains(2)
    assert box.contains(3)
    assert not box.contains(4)
    assert not box.contains(3248)

    with pytest.raises(IllegalArgumentError, match="Illegal ID value: 0"):
        box.contains(0)


def test_update(test_store):
    box = test_store.box(TestEntity)

    box.put([TestEntity(str=f"Object {i}") for i in range(1, 4)])
    assert box.count() == 3

    with test_store.write_tx():
        for obj in box.get_all():
            obj.str += " (updated)"
            box.update(obj)

    assert box.get(1).str == "Object 1 (updated)"
    assert box.get(2).str == "Object 2 (updated)"
    assert box.get(3).str == "Object 3 (updated)"
    assert box.count() == 3

    with pytest.raises(ValueError, match="without an ID"):
        box.update(TestEntity(str="New object"))

    obj = box.get(1)
    box.remove(1)
    with pytest.raises(IdNotFoundError, match="Update failed, given ID doesn't exist: 1"):
        box.update(obj)
    assert not box.contains(1)  # Failed update must not re-create the object


def test_get_many(test_store):
    box = test_store.box(TestEntity)

    box.put([TestEntity(str=f"Object {i}") for i in range(1, 6)])
    assert box.count() == 5

    objects = box.get_many([1, 3, 4])
    assert len(objects) == 3
    assert objects[0].str == "Object 1"
    assert objects[1].str == "Object 3"
    assert objects[2].str == "Object 4"

    objects = box.get_many([5, 2])
    assert len(objects) == 2
    assert objects[0].str == "Object 5"
    assert objects[1].str == "Object 2"

    objects = box.get_many([5, 5, 2])
    assert len(objects) == 3
    assert objects[0].str == "Object 5"
    assert objects[1].str == "Object 5"
    assert objects[2].str == "Object 2"

    objects = box.get_many([2])
    assert len(objects) == 1
    assert objects[0].str == "Object 2"

    objects = box.get_many([])
    assert len(objects) == 0

    objects = box.get_many([1, 789, 2, 10])
    assert len(objects) == 4
    assert objects[0].str == "Object 1"
    assert objects[1] is None
    assert objects[2].str == "Object 2"
    assert objects[3] is None


def test_put_new_object_ids(test_store):
    """New objects get their ID while putting; on failure, the object keeps ID 0."""
    box = test_store.box(TestEntity)
    first, second = TestEntity(str="first"), TestEntity(str="second")
    assert box.put(first) == first.id == 1
    with test_store.write_tx():
        assert box.put(second) == second.id == 2
    assert box.get(2).id == 2 and box.get(2).str == "second"  # the ID is also in the stored data
    first.str = "first updated"
    assert box.put(first) == 1  # update
    assert box.count() == 2 and box.get(1).str == "first updated"

    failing = TestEntity(str="fails")
    with pytest.raises(DbError):
        with test_store.read_tx():
            box.put(failing)  # no write in a read transaction
    assert box._entity._get_object_id(failing) == 0  # not set
    assert box.count() == 2


def _id_entity(assignable: bool):
    @Entity(model="ids")
    class Item:
        id = Id(assignable=assignable)
        name = String

    return Item


def _open_id_store(tmp_path, entity) -> objectbox.Store:
    model = objectbox.Model()
    model.entity(entity)
    return objectbox.Store(model=model, directory=str(tmp_path / "db"),
                           model_json_file=str(tmp_path / "model.json"))


def test_put_unassigned_id(tmp_path):
    """ Putting an object with an ID that ObjectBox did not assign raises; this used to put it, and a later new
    object got the same ID and replaced it. """
    Item = _id_entity(assignable=False)
    store = _open_id_store(tmp_path, Item)
    box = store.box(Item)
    message = "ID is higher or equal to internal ID sequence"
    with pytest.raises(IllegalArgumentError, match=message):
        box.put(Item(id=3, name="preset"))
    assert box.count() == 0

    # put(list): nothing is put if one ID was not assigned, and new objects keep ID 0
    new = Item(name="new")
    with pytest.raises(IllegalArgumentError, match=message):
        box.put([new, Item(id=9, name="preset")])
    assert box.count() == 0 and box._entity._get_object_id(new) == 0

    # New objects, then updates by their (assigned) IDs; new objects do not reuse an ID
    ids = [box.put(Item(name="a%d" % i)) for i in range(3)]
    assert ids == [1, 2, 3]
    assert box.put(Item(id=2, name="b2")) == 2
    box.put([Item(id=1, name="b1"), Item(id=3, name="b3"), Item(name="c")])
    assert [o.name for o in box.get_all()] == ["b1", "b2", "b3", "c"]
    assert box.put(Item(name="d")) == 5

    # IDs of removed objects were assigned, so can be put again
    box.remove(5)
    assert box.put(Item(id=5, name="d again")) == 5
    with pytest.raises(IllegalArgumentError, match=message):
        box.put(Item(id=6, name="preset"))
    store.close()

    # The JSON model file has the ID flag only
    assert _json_id_flags(tmp_path) == 1  # OBXPropertyFlags_ID


def test_put_assignable_id(tmp_path):
    Item = _id_entity(assignable=True)
    store = _open_id_store(tmp_path, Item)
    box = store.box(Item)
    assert box.put(Item(id=10, name="ten")) == 10
    new = Item(name="new")
    assert box.put(new) == new.id == 11  # Above the highest ID put before
    box.put([Item(id=100, name="hundred"), Item(id=5, name="five"), Item(name="new 2")])
    assert box.put(Item(name="new 3")) == 102
    assert [(o.id, o.name) for o in box.get_all()] == [(5, "five"), (10, "ten"), (11, "new"), (100, "hundred"),
                                                         (101, "new 2"), (102, "new 3")]
    store.close()
    assert _json_id_flags(tmp_path) == 129  # OBXPropertyFlags_ID | OBXPropertyFlags_ID_SELF_ASSIGNABLE


def test_put_assignable_id_changed(tmp_path):
    """ An existing ID property can become assignable and back. """
    Item = _id_entity(assignable=False)
    store = _open_id_store(tmp_path, Item)
    store.box(Item).put(Item(name="a"))
    store.close()

    Item = _id_entity(assignable=True)
    store = _open_id_store(tmp_path, Item)
    box = store.box(Item)
    assert box.put(Item(id=50, name="b")) == 50
    store.close()

    Item = _id_entity(assignable=False)
    store = _open_id_store(tmp_path, Item)
    box = store.box(Item)
    assert box.count() == 2
    assert box.put(Item(name="c")) == 51
    with pytest.raises(IllegalArgumentError):
        box.put(Item(id=60, name="d"))
    store.close()


def _json_id_flags(tmp_path) -> int:
    import json
    with open(str(tmp_path / "model.json")) as file:
        doc = json.load(file)
    return [p["flags"] for p in doc["entities"][0]["properties"] if p["name"] == "id"][0]


def test_remove_many(test_store):
    box = test_store.box(TestEntity)
    objects = [TestEntity(str="o%d" % i) for i in range(5)]
    box.put(objects)
    ids = [o.id for o in objects]

    assert box.remove_many([ids[0], ids[1], 9999]) == 2  # Missing IDs are skipped
    assert box.remove_many((objects[2], ids[3])) == 2  # Objects and IDs, any sequence
    assert box.remove_many(iter([ids[4]])) == 1
    assert box.remove_many([]) == 0
    assert box.count() == 0
    assert box.remove_many(ids) == 0

    with pytest.raises(TypeError, match=r"use remove_many\(\) for several"):
        box.remove([1, 2])  # Failed with a ctypes.ArgumentError


def test_put_many_returns_ids(test_store):
    box = test_store.box(TestEntity)
    first = TestEntity(str="first")
    assert box.put(first) == 1
    objects = [TestEntity(str="a"), first, TestEntity(str="b")]  # New and existing objects
    assert box.put(objects) == [2, 1, 3]
    assert [o.id for o in objects] == [2, 1, 3]
    assert box.put((TestEntity(str="c"), TestEntity(str="d"))) == [4, 5]  # A tuple, too
    assert box.put(TestEntity(str="e"), TestEntity(str="f")) == [6, 7]  # Several arguments
    assert box.put([]) == []
    assert box.put() == []
    assert box.count() == 7


def test_date_float_seconds(tmp_path):
    """ With py_type=float, Date and DateNano read back seconds (DateNano read nanoseconds, as an int, before). """

    @objectbox.Entity(model="date-float")
    class Event:
        id = objectbox.Id
        date = objectbox.Date(float)
        date_nano = objectbox.DateNano(float)
        date_nano_int = objectbox.DateNano(int)

    model = objectbox.Model()
    model.entity(Event)
    store = objectbox.Store(model=model, directory=str(tmp_path / "db"), model_json_file=str(tmp_path / "m.json"))
    box = store.box(Event)
    seconds = 1767268800.123456  # 2026-01-01 12:00:00.123456 UTC
    read = box.get(box.put(Event(date=seconds, date_nano=seconds, date_nano_int=seconds)))
    assert type(read.date) == float and read.date == pytest.approx(seconds, abs=0.001)
    assert type(read.date_nano) == float and read.date_nano == pytest.approx(seconds, abs=1e-6)
    assert type(read.date_nano_int) == int and read.date_nano_int == pytest.approx(seconds * 1e9, abs=1000)
    read = box.get(box.put(Event()))  # Not set: 0
    assert (read.date, read.date_nano, read.date_nano_int) == (0.0, 0.0, 0)
    store.close()