trailbase/client/python/tests/test_client.py

from trailbase import (
    EVENT_ERROR_STATUS_FORBIDDEN,
    ErrorEvent,
    parseEvent,
    Client,
    CompareOp,
    FetchException,
    Filter,
    InsertEvent,
    UpdateEvent,
    DeleteEvent,
    RecordId,
    JSON,
    JSON_OBJECT,
    EVENT,
)

import httpx
import json
import logging
import mintotp  # type: ignore
import os
import pytest
import subprocess

from time import time, sleep
from typing import List, cast

logging.basicConfig(level=logging.DEBUG)

port = 4007
address = f"127.0.0.1:{port}"
site = f"http://{address}"


class TrailBaseFixture:
    process: None | subprocess.Popen[bytes]

    def __init__(self) -> None:
        cwd = os.getcwd()
        traildepot = "../testfixture" if cwd.endswith("python") else "client/testfixture"

        logger.info("Building TrailBase")
        build = subprocess.run(["cargo", "build"])
        assert build.returncode == 0, f"{build.stderr}"

        logger.info("Starting TrailBase")
        self.process = subprocess.Popen(
            [
                "cargo",
                "run",
                "--",
                "--data-dir",
                traildepot,
                "run",
                "-a",
                address,
                "--runtime-threads",
                "1",
            ]
        )

        client = httpx.Client()
        for _ in range(100):
            try:
                response = client.get(f"http://{address}/api/healthcheck")
                if response.status_code == 200:
                    return
            except Exception:
                pass

            sleep(0.5)

        logger.error("Failed ot start TrailBase")

    def isUp(self) -> bool:
        p = self.process
        return p is not None and p.returncode is None

    def shutdown(self) -> None:
        p = self.process
        if p is not None:
            p.send_signal(9)
            p.wait()
            assert isinstance(p.returncode, int)


@pytest.fixture(scope="session")
def trailbase():
    fixture = TrailBaseFixture()
    yield fixture
    fixture.shutdown()


def connect() -> Client:
    client = Client(site, tokens=None)
    client.login("admin@localhost", "secret")
    return client


def test_authentication(trailbase: TrailBaseFixture):
    assert trailbase.isUp()

    client = connect()
    assert client.site() == site

    tokens = client.tokens()
    assert tokens is not None and tokens.valid()

    client.refresh_auth_tokens()

    user = client.user()
    assert user is not None and user.id != ""
    assert user is not None and user.email == "admin@localhost"

    client.logout()
    assert client.tokens() is None

    client.refresh_auth_tokens()


def test_second_factor_authentication(trailbase: TrailBaseFixture):
    assert trailbase.isUp()

    client = Client(site, tokens=None)
    mfaToken = client.login("alice@trailbase.io", "secret")
    assert mfaToken is not None

    secret = "YCUTAYEZ346ZUEI7FLCG57BOMZQHHRA5"
    code: str = mintotp.totp(secret)  # pyright: ignore [reportUnknownMemberType]

    client.login_second(mfaToken, code)
    user = client.user()
    assert user is not None and user.email == "alice@trailbase.io"

    client.logout()
    assert client.tokens() is None


def test_otp_auth(trailbase: TrailBaseFixture):
    assert trailbase.isUp()

    client = Client(site, tokens=None)
    client.request_otp("fake0@trailbase.io")
    client.request_otp("fake1@trailbase.io", redirect_uri="/target")

    with pytest.raises(FetchException) as exec:
        client.login_otp("fake0@trailbase.io", "invalid")

    assert exec.value.status == 401


def test_records(trailbase: TrailBaseFixture):
    assert trailbase.isUp()

    client = connect()
    api = client.records("simple_strict_table")

    now = int(time())
    messages = [
        f"python client test 0: =?&{now}",
        f"python client test 1: =?&{now}",
    ]
    ids: List[RecordId] = []
    for msg in messages:
        ids.append(api.create({"text_not_null": msg}))

    if True:
        bulk_ids = api.create_bulk(
            [
                {"text_not_null": "python bulk test 0"},
                {"text_not_null": "python bulk test 1"},
            ]
        )
        assert len(bulk_ids) == 2

    if True:
        response = api.list(
            filters=[Filter("text_not_null", messages[0])],
        )
        records = response.records
        assert len(records) == 1
        assert records[0]["text_not_null"] == messages[0]

    if True:
        recordsAsc = api.list(
            order=["+text_not_null"],
            filters=[Filter(column="text_not_null", value=f"% =?&{now}", op=CompareOp.LIKE)],
            count=True,
        )

        assert recordsAsc.total_count == 2
        assert [el["text_not_null"] for el in recordsAsc.records] == messages

        recordsDesc = api.list(
            order=["-text_not_null"],
            filters=[Filter(column="text_not_null", value=f"%{now}", op=CompareOp.LIKE)],
        )

        assert [el["text_not_null"] for el in recordsDesc.records] == list(reversed(messages))

    if True:
        record = api.read(ids[0])
        assert record["text_not_null"] == messages[0]

        record = api.read(ids[1])
        assert record["text_not_null"] == messages[1]

    if True:
        updatedMessage = f"python client updated test 0: {now}"
        api.update(ids[0], {"text_not_null": updatedMessage})
        record = api.read(ids[0])
        assert record["text_not_null"] == updatedMessage

    if True:
        api.delete(ids[0])

        with pytest.raises(FetchException):
            api.read(ids[0])


def test_expand_foreign_records(trailbase: TrailBaseFixture):
    assert trailbase.isUp()

    client = connect()
    api = client.records("comment")

    def get_nested(obj: JSON_OBJECT, k0: str, k1: str) -> JSON | None:
        x = obj[k0]
        assert type(x) is dict
        return x.get(k1)

    if True:
        comment = api.read(1)

        assert comment.get("id") == 1
        assert comment.get("body") == "first comment"
        assert get_nested(comment, "author", "id") != ""
        assert get_nested(comment, "author", "data") is None
        assert get_nested(comment, "post", "id") != ""

    if True:
        comment = api.read(1, expand=["post"])

        assert comment.get("id") == 1
        assert comment.get("body") == "first comment"
        assert get_nested(comment, "author", "data") is None

        x = get_nested(comment, "post", "data")
        assert type(x) is dict
        assert x.get("title") == "first post"

    if True:
        comments = api.list(
            expand=["author", "post"],
            order=["-id"],
            limit=1,
            count=True,
        )

        assert comments.total_count == 2
        assert len(comments.records) == 1

        comment = comments.records[0]

        assert comment.get("id") == 2
        assert comment.get("body") == "second comment"

        x = get_nested(comment, "post", "data")
        assert type(x) is dict
        assert x.get("title") == "first post"

        y = get_nested(comment, "author", "data")
        assert type(y) is dict
        assert y.get("name") == "SecondUser"

    if True:
        comments = api.list(
            expand=["author", "post"],
            order=["-id"],
            limit=2,
        )

        assert len(comments.records) == 2

        first = comments.records[0]
        assert first.get("id") == 2
        second = comments.records[1]
        assert second.get("id") == 1

        offset_comments = api.list(
            expand=["author", "post"],
            order=["-id"],
            limit=1,
            offset=1,
        )

        assert len(offset_comments.records) == 1
        assert second == offset_comments.records[0]


def test_parse_event():
    err_json: str = """
    {
      "Error": {
        "status": 1,
        "message": "test"
      },
      "seq": 3
    }
    """

    err_event = cast(ErrorEvent | None, parseEvent(json.loads(err_json)))
    assert err_event is not None
    assert err_event.seq == 3
    assert err_event.status == EVENT_ERROR_STATUS_FORBIDDEN
    assert err_event.message == "test"

    update_json: str = """
    {
      "Update": {
        "col0": "val0",
        "col1": 4
      },
      "seq": 4
    }
    """

    update_event = cast(UpdateEvent | None, parseEvent(json.loads(update_json)))
    assert update_event is not None
    assert update_event.seq == 4


def test_subscriptions(trailbase: TrailBaseFixture):
    assert trailbase.isUp()

    client = connect()
    api = client.records("simple_strict_table")

    table_subscription = api.subscribe_all()

    now = int(time())
    create_message = f"python client subscription test 0: =?&{now}"
    id = api.create({"text_not_null": create_message})

    update_message = f"python client subscription test 1: =?&{now}"
    api.update(id, {"text_not_null": update_message})

    api.delete(id)

    events: List[EVENT] = []
    for ev in table_subscription:
        events.append(ev)
        if len(events) == 3:
            break

    table_subscription.close()

    ev0 = events[0]
    assert type(ev0) is InsertEvent
    assert ev0.seq == 1
    assert ev0.value["text_not_null"] == create_message

    ev1 = events[1]
    assert type(ev1) is UpdateEvent
    assert ev1.seq == 2
    assert ev1.value["text_not_null"] == update_message

    ev2 = events[2]
    assert type(ev2) is DeleteEvent
    assert ev2.seq == 3
    assert ev2.value["text_not_null"] == update_message


logger = logging.getLogger(__name__)