Source code for examples.ftscontentmanagement.test_application

from __future__ import annotations

import os
from typing import ClassVar
from unittest import TestCase
from uuid import uuid4

from eventsourcing.postgres import PostgresDatastore
from eventsourcing.tests.postgres_utils import drop_postgres_table
from eventsourcing.utils import get_topic
from examples.contentmanagement.domainmodel import user_id_cvar
from examples.ftscontentmanagement.application import FtsContentManagement
from examples.ftscontentmanagement.postgres import PostgresFtsApplicationRecorder
from examples.ftscontentmanagement.sqlite import SQLiteFtsApplicationRecorder


class SearchableContentApplicationTestCase(TestCase):
    env: ClassVar[dict[str, str]] = {}

    def test_app(self) -> None:
        app = FtsContentManagement(env=self.env)

        # Set user_id context variable.
        user_id = uuid4()
        user_id_cvar.set(user_id)

        # Create empty pages.
        app.create_page(title="Animals", slug="animals")
        app.create_page(title="Plants", slug="plants")
        app.create_page(title="Minerals", slug="minerals")

        # Search, expect no results.
        self.assertEqual(0, len(app.search("dog")))
        self.assertEqual(0, len(app.search("rose")))
        self.assertEqual(0, len(app.search("zinc")))

        # Update the pages.
        app.update_body(slug="animals", body="cat dog zebra")
        app.update_body(slug="plants", body="bluebell rose jasmine")
        app.update_body(slug="minerals", body="iron zinc calcium")

        # Search for single words, expect results.
        pages = app.search("dog")
        self.assertEqual(1, len(pages))
        self.assertEqual(pages[0]["slug"], "animals")
        self.assertEqual(pages[0]["body"], "cat dog zebra")

        pages = app.search("rose")
        self.assertEqual(1, len(pages))
        self.assertEqual(pages[0]["slug"], "plants")
        self.assertEqual(pages[0]["body"], "bluebell rose jasmine")

        pages = app.search("zinc")
        self.assertEqual(1, len(pages))
        self.assertEqual(pages[0]["slug"], "minerals")
        self.assertEqual(pages[0]["body"], "iron zinc calcium")

        # Search for multiple words in same page.
        pages = app.search("dog cat")
        self.assertEqual(1, len(pages))
        self.assertEqual(pages[0]["slug"], "animals")
        self.assertEqual(pages[0]["body"], "cat dog zebra")

        # Search for multiple words in same page, expect no results.
        pages = app.search("rose zebra")
        self.assertEqual(0, len(pages))

        # Search for alternative words, expect two results.
        pages = app.search("rose OR zebra")
        self.assertEqual(2, len(pages))
        self.assertEqual(["animals", "plants"], sorted(p["slug"] for p in pages))




class TestWithSQLite(SearchableContentApplicationTestCase):
    env: ClassVar[dict[str, str]] = {
        "PERSISTENCE_MODULE": "eventsourcing.sqlite",
        "APPLICATION_RECORDER_TOPIC": get_topic(SQLiteFtsApplicationRecorder),
        "SQLITE_DBNAME": ":memory:",
    }





class TestWithPostgres(SearchableContentApplicationTestCase):
    env: ClassVar[dict[str, str]] = {
        "PERSISTENCE_MODULE": "eventsourcing.postgres",
        "APPLICATION_RECORDER_TOPIC": get_topic(PostgresFtsApplicationRecorder),
    }



    def setUp(self) -> None:
        super().setUp()
        os.environ["POSTGRES_DBNAME"] = "eventsourcing"
        os.environ["POSTGRES_HOST"] = "127.0.0.1"
        os.environ["POSTGRES_PORT"] = "5432"
        os.environ["POSTGRES_USER"] = "eventsourcing"
        os.environ["POSTGRES_PASSWORD"] = "eventsourcing"  # noqa: S105
        self.drop_tables()




    def tearDown(self) -> None:
        self.drop_tables()
        super().tearDown()




    def drop_tables(self) -> None:
        with PostgresDatastore(
            os.environ["POSTGRES_DBNAME"],
            os.environ["POSTGRES_HOST"],
            os.environ["POSTGRES_PORT"],
            os.environ["POSTGRES_USER"],
            os.environ["POSTGRES_PASSWORD"],
        ) as datastore:
            drop_postgres_table(datastore, "ftscontentmanagement_events")
            drop_postgres_table(datastore, "ftsprojection")




del SearchableContentApplicationTestCase