software/kisstdlib/./example/sqlite3_ext.py

A set of modules for the Python programming language designed for system programming with conceptually and algebraically simple primitives.

Files

Raw Source

Contents

#!/usr/bin/env python3
#
# Copyright (c) 2025 Jan Malakhovski <oxij@oxij.org>
#
# Permission to use, copy, modify, and/or distribute this software for any
# purpose with or without fee is hereby granted.
#
# THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES WITH
# REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY
# AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY SPECIAL, DIRECT,
# INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING FROM
# LOSS OF USE, DATA OR PROFITS, WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR
# OTHER TORTIOUS ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR
# PERFORMANCE OF THIS SOFTWARE.

"""Some `kisstdlib.sqlite3_ext` examples/tests."""

import logging as _logging
import os as _os
import os.path as _op
import typing as _t

import pytest

from kisstdlib.failure import *
from kisstdlib.sqlite3_ext import *  # pylint: disable=redefined-builtin


class DummyDB(DictSettingsAppDB):
    """A very simple example AppDB."""

    APPLICATION_ID = 1
    MIN_VERSION = 0
    MAX_VERSION = 6

    def upgrade(self, cur: Cursor, target_version: int) -> None:
        if self.settings.get("forbidden", False):
            # this operation is currently forbidden, used for testing below
            raise RuntimeError("forbidden")

        if self.version == 0:
            # init
            cur.execute("CREATE TABLE test (id INTEGER PRIMARY KEY NOT NULL)")

        self.version += 1

        # if at version 6, mark as broken
        if target_version == 6:
            self.settings["broken"] = True

    def check(self, cur: Cursor, final: bool) -> None:
        if self.settings.get("broken", False):
            # this should fail to check, used for testing below
            raise RuntimeError("bad")


class NotDummy(DummyDB):
    """The same thing, but with a different `APPLICATION_ID`."""

    APPLICATION_ID = 2


def test_DB(tmp_path: str) -> None:
    path = _op.join(tmp_path, "test.db")
    try:
        default_settings: DictSettings = {"name": "value", "default": "yes"}
        non_default_settings: DictSettings = {"default": "not"}

        # for illustrative purposes, always start in WAL mode
        setup = AppDB.SETUP_WAL

        # newly created databases starts with
        dummy = DummyDB(path, 1, default_settings, setup)
        # `version` set to (at least) `default_version`
        assert dummy.version == 1
        # `settings` set to `default_settings`
        assert dummy.settings == default_settings
        # everything written to disk
        assert not dummy.dirty
        # and a transaction open
        assert dummy.db.in_transaction
        # so, a commit is needed to save it all
        dummy.commit()
        dummy.close()

        # doing the same with `commit_intermediate` will leave all changes uncommited
        dummy = DummyDB(path + ".temp", 1, default_settings, setup, commit_intermediate=False)
        assert dummy.dirty
        dummy.close()

        # i.e., in the case of database creation, it will leave the database
        # completely empty
        dummy = DummyDB(path + ".temp", 0, {"forbidden": True}, setup, commit_intermediate=False)
        assert dummy.version == 0
        dummy.close()

        # but `commit_intermediate=False` setting exists to help in developing
        # new `upgrade` steps
        #
        # also, note that openinng with `default_version = 0` and
        # `commit_intermediate=False` will disable all `upgrade`s and `commit`s

        # newly created databases that fail to init (by either `upgrade` or
        # `check` failing) will be cleaned up after
        with pytest.raises(RuntimeError) as exc_info:
            dummy = DummyDB(path + ".trash", 1, {"forbidden": True}, setup)
        _logging.debug("%s", repr(exc_info.value))
        assert not _op.exists(path + ".trash")

        with pytest.raises(RuntimeError) as exc_info:
            dummy = DummyDB(path + ".trash", 1, {"broken": True}, setup)
        _logging.debug("%s", repr(exc_info.value))
        assert not _op.exists(path + ".trash")

        # for existing databases, `default_settings` gets ignored completely
        dummy = DummyDB(path, 1, non_default_settings, setup)
        # old `settings` persist
        assert dummy.settings == default_settings
        # and they usually start clean (unless some upgrades happened and
        # `autocommit_upgrades = False` or `APPLICATION_ID` was just assigned)
        assert not dummy.dirty

        # all changes need to be commited explcitly
        dummy.settings["A"] = 0
        dummy.settings["B"] = None
        # NB: `dirty` sigsifies metadata dirtyness, if you do
        #
        # > dummy.settings = {...}
        #
        # it will be set automatically, but changing `settings` in-place, as
        # above, will not set `dirty`, so it will need to be set explicitly
        assert not dummy.dirty
        # but we are also changing `version`
        dummy.version += 1
        # which sets it
        assert dummy.dirty

        # add some data
        cur = dummy.cursor()
        for i in range(0, 10):
            cur.execute("INSERT INTO test VALUES (?)", (i,))
        del cur

        # and commit
        dummy.commit()
        dummy.close()

        # now, the above changes persist
        dummy = DummyDB(path, 1, non_default_settings, setup)
        assert not dummy.dirty
        assert dummy.version == 2
        assert dummy.settings["default"] == "yes"
        assert dummy.settings["A"] == 0
        assert dummy.settings["B"] is None

        def stored(dummy: DummyDB) -> _t.Iterator[int]:
            return map(lambda x: x[0], dummy.db.execute("SELECT id from test"))

        assert list(stored(dummy)) == list(range(0, 10))

        # but without a `commit`, all changes get rolled back instead
        dummy.version = 3
        dummy.settings = non_default_settings
        assert dummy.dirty
        cur = dummy.cursor()
        for i in range(10, 20):
            cur.execute("INSERT INTO test VALUES (?)", (i,))
        del cur
        assert list(stored(dummy)) == list(range(0, 20))
        dummy.close()

        dummy = DummyDB(path, 1, non_default_settings, setup)
        # NB: old version! previous block did not commit!
        assert dummy.version == 2
        assert dummy.settings["default"] == "yes"
        assert list(stored(dummy)) == list(range(0, 10))
        dummy.close()

        # openning a database with a newer `want_version` (which inherists from
        # `default_version` by default) forces an upgrade, which, by default,
        # also gets commited to disk
        dummy = DummyDB(path, 3, non_default_settings, setup)
        assert not dummy.dirty
        assert dummy.version == 3
        # `settings` stay the same, since even though `upgrade` above
        # recreates the db, it reuses `settings`
        assert dummy.settings["default"] == "yes"
        assert dummy.settings["A"] == 0
        assert dummy.settings["B"] is None
        dummy.close()

        # and another upgrade
        dummy = DummyDB(path, 4, non_default_settings, setup)
        assert not dummy.dirty
        assert dummy.version == 4
        assert dummy.settings["default"] == "yes"
        dummy.close()

        # upgrades do not get rolled back
        dummy = DummyDB(path, 1, non_default_settings, setup)
        assert not dummy.dirty
        assert dummy.version == 4
        assert dummy.settings["default"] == "yes"
        dummy.close()

        # `AppDB` can make backups before upgrades
        dummy = DummyDB(path, 5, non_default_settings, setup, backup_before_upgrades=True)
        assert not dummy.dirty
        assert dummy.version == 5
        assert dummy.settings["default"] == "yes"
        # modify it a bit
        dummy.settings = non_default_settings
        dummy.commit()
        dummy.close()

        # check it exists and is at the old version
        dummy = DummyDB(_op.join(tmp_path, "test.db.v4.bak"), 1, non_default_settings, setup)
        assert not dummy.dirty
        assert dummy.version == 4
        assert dummy.settings["default"] == "yes"
        dummy.close()

        # updates that fail to `check` propagate `check`'s exceptions and get
        # rolled back
        with pytest.raises(RuntimeError) as exc_info:
            _dummy = DummyDB(path, 6, {}, setup)
        _logging.debug("%s", repr(exc_info.value))

        # and the db is left unmodified
        dummy = DummyDB(path, 1, default_settings, setup)
        assert dummy.version == 5
        assert dummy.settings == non_default_settings
        dummy.close()

        # specifying `default_version` outside of supported versions is an error
        with pytest.raises(AssertionFailure) as exc_info:
            _dummy = DummyDB(path, 7, {}, setup)
        _logging.debug("%s", repr(exc_info.value))

        # trying to reuse the database with a different `APPLICATION_ID` is an error
        with pytest.raises(InvalidDB) as exc_info:
            one = NotDummy(path, 2, {}, setup)
            one.close()
        _logging.debug("%s", repr(exc_info.value))
    finally:
        _os.unlink(path)


if __name__ == "__main__":
    _logging.basicConfig(level=_logging.DEBUG)
    _os.makedirs("test-appdb", exist_ok=True)
    test_DB("test-appdb")