test_store_options.py

from objectbox import Store
from objectbox.c import *  # TODO ideally we wouldn't have to import c.py
from objectbox.store_options import StoreOptions
from objectbox.exceptions import SchemaError
from tests.common import *

def test_set_options():
    """ Test setting dummy values for each option.
    Checks that Python types are correctly forwarded to C API. """

    options = StoreOptions()
    options.directory("test-db")
    options.max_db_size_in_kb(8192)
    options.max_data_size_in_kb(4096)
    options.file_mode(755)
    options.max_readers(10)
    options.no_reader_thread_locals(False)
    # options.model
    # options.model_bytes
    # options.model_bytes_direct
    # options.validate_on_open_pages
    # options.validate_on_open_kv
    options.put_padding_mode(OBXPutPaddingMode_PaddingAutomatic)
    options.read_schema(False)
    options.use_previous_commit(False)
    options.read_only(True)
    options.debug_flags(OBXDebugFlags_LOG_TRANSACTIONS_READ)
    options.add_debug_flags(OBXDebugFlags_LOG_CACHE_HITS)
    options.async_max_queue_length(100)
    options.async_throttle_at_queue_length(1024)
    options.async_throttle_micros(1000)
    options.async_max_in_tx_duration(1000)
    options.async_max_in_tx_operations(20)
    options.async_pre_txn_delay(500)
    options.async_pre_txn_delay4(500, 700, 100)
    options.async_post_txn_delay(500)
    options.async_post_txn_delay5(500, 700, 100, False)
    options.async_minor_refill_threshold(100)
    options.async_minor_refill_max_count(500)
    options.async_max_tx_pool_size(100)
    options.async_object_bytes_max_cache_size(4096)
    options.async_object_bytes_max_size_to_cache(4096)
    # options.log_callback
    # options.backup_restore

    assert options.get_directory() == "test-db"
    assert options.get_max_db_size_in_kb() == 8192
    assert options.get_max_data_size_in_kb() == 4096
    assert options.get_debug_flags() == (OBXDebugFlags_LOG_TRANSACTIONS_READ | OBXDebugFlags_LOG_CACHE_HITS)

    del options

@pytest.mark.parametrize("option", ["model_bytes", "model_bytes_direct"])
def test_model_bytes(option):
    """ The bytes must reach the C library, which rejects these invalid model bytes. """
    Store.remove_db_files("testdata")
    invalid_model = b"invalid model"
    with pytest.raises(SchemaError, match=f"Verification of the model object \\({len(invalid_model)} bytes\\) failed"):
        Store(model=None, directory="testdata", **{option: invalid_model})


def test_store_with_options():
    Store.remove_db_files("testdata")
    remove_json_model_file()

    store = Store(
        model=create_default_model(),
        directory="testdata",
        max_db_size_in_kb=1<<20,
        max_data_size_in_kb=(1<<20)-(1<<10),
        file_mode=int('664',8),
        max_readers=126,
        no_reader_thread_locals=True,
        read_schema=True,
        use_previous_commit=False,
        read_only=False,
        debug_flags=DebugFlags.LOG_TRANSACTIONS_READ|DebugFlags.LOG_TRANSACTIONS_WRITE,
        async_max_queue_length=100,
        async_throttle_at_queue_length=100,
        async_throttle_micros=50000,
        async_max_in_tx_duration=50000,
        async_max_in_tx_operations=1000,
        async_pre_txn_delay=100000,
        async_post_txn_delay=100000,
        async_minor_refill_threshold=10,
        async_minor_refill_max_count=100,
        async_object_bytes_max_cache_size=1<<20,
        async_object_bytes_max_size_to_cache=100<<10
    )
    store.close()

def test_log_callback():
    Store.remove_db_files("testdata")
    remove_json_model_file()
    
    log_entries = []

    def mylog(level: OBXLogLevel, message: str):
        levelText = "?"
        if level == LogLevel.Debug:
            levelText = "DEBUG"
        print(f"MYLOG: {levelText} {message}")
        log_entries.append( (level, message) )        

    store = Store(
        model=create_default_model(),
        directory="testdata",
        log_callback=mylog
    )
    
    box = store.box(TestEntity)
    assert len(log_entries) == 2
    assert log_entries[0][0] == LogLevel.Debug
    assert log_entries[0][1].startswith("Opening store:")
    assert log_entries[1][0] == LogLevel.Debug
    assert log_entries[1][1].startswith("Opening store:")
    store.close()
    


def test_debug_flags_values():
    """ DebugFlags has the values of the C API (LOG_QUERIES was 3, i.e. read and write transactions). """
    import objectbox.c
    c_flags = {name[len("OBXDebugFlags_"):]: value for name, value in vars(objectbox.c).items()
               if name.startswith("OBXDebugFlags_")}
    assert len(c_flags) == 10
    assert {flag.name: flag.value for flag in DebugFlags if flag != DebugFlags.NONE} == c_flags
    assert DebugFlags.LOG_QUERIES == 4


def test_debug_flags_header():
    """ The OBXDebugFlags_ constants match the C header (if downloaded, e.g. by `make depend`). """
    import os
    import re
    import objectbox.c
    header = os.path.join(os.path.dirname(objectbox.c.__file__), "lib", "objectbox.h")
    if not os.path.exists(header):
        pytest.skip("No C header")
    with open(header) as file:
        header_flags = dict((name, int(value)) for name, value in
                            re.findall(r"^\s*(OBXDebugFlags_\w+) = (\d+),", file.read(), re.MULTILINE))
    assert header_flags == {name: value for name, value in vars(objectbox.c).items()
                            if name.startswith("OBXDebugFlags_")}