conflict-set/test_conflict_set.py

from conflict_set import *


class DebugConflictSet:
    """
    Bisimulates the skip list and radix tree conflict sets for testing purposes
    """

    def __init__(self, version: int = 0) -> None:
        self.skip_list = ConflictSet(version, implementation="skip_list")
        self.radix_tree = ConflictSet(version, implementation="radix_tree")

    def addWrites(self, version: int, *writes: WriteRange):
        self.skip_list.addWrites(version, *writes)
        self.radix_tree.addWrites(version, *writes)

    def check(self, *reads: ReadRange) -> list[Result]:
        expected = self.skip_list.check(*reads)
        actual = self.radix_tree.check(*reads)
        assert expected == actual
        return actual

    def setOldestVersion(self, version: int) -> None:
        self.skip_list.setOldestVersion(version)
        self.radix_tree.setOldestVersion(version)

    def getBytes(self) -> int:
        return self.radix_tree.getBytes()

    def __enter__(self):
        return self

    def close(self) -> None:
        self.skip_list.close()
        self.radix_tree.close()

    def __exit__(self, exception_type, exception_value, exception_traceback):
        self.close()


def test_conflict_set():
    with DebugConflictSet() as cs:
        before = cs.getBytes()
        key = b"a key"
        cs.addWrites(1, write(key))
        assert cs.getBytes() - before > 0
        assert cs.check(read(0, key)) == [Result.CONFLICT]
        cs.setOldestVersion(1)
        assert cs.check(read(0, key)) == [Result.TOO_OLD]


if __name__ == "__main__":
    # budget "pytest" for ctest integration without pulling in a dependency. You can of course still use pytest in local development.
    import argparse
    import inspect
    import sys

    parser = argparse.ArgumentParser()
    subparsers = parser.add_subparsers(dest="command")

    list_parser = subparsers.add_parser("list")
    test_parser = subparsers.add_parser("test")
    test_parser.add_argument("test")

    args = parser.parse_args()

    if args.command == "list":
        sys.stdout.write(
            ";".join(
                name[5:]
                for name in dir()
                if name.startswith("test_")
                and inspect.isfunction(getattr(sys.modules[__name__], name))
            )
        )
    elif args.command == "test":
        globals()["test_" + args.test]()