dev-practice/2023-04-10/python-eventsourcing_notes/test_notes_app_es.py

from dataclasses import replace
from datetime import datetime
import os


from typing import cast
import unittest
from unittest import TestCase
from uuid import uuid4

from notes_app_es import (
    NotesApplication,
    NoteNotFound,
    SlugConflictError,
    
)
from notes_app_es import (
    Index,
    Note,
    user_id_cvar,
)

from notes_app_es import (
    NoteAnalytics,
    Counter,
)

from eventsourcing.system import NotificationLogReader
from eventsourcing.domain import create_utc_datetime_now

from eventsourcing.system import System, SingleThreadedRunner


from eventsourcing.application import ProcessingEvent

class TestContentManagement(TestCase):
    def test(self) -> None:
        # Set user_id context variable.
        user_id = uuid4()
        user_id_cvar.set(user_id)

        # Construct application.
        #app = NotesApplication()
        #analytics = NoteAnalytics()
        #system = System(pipes=[[app, analytics]])
        
        system = System(pipes=[[NotesApplication, NoteAnalytics]])
        
        runner = SingleThreadedRunner(system)
        runner.start()
        
        app = runner.get(NotesApplication)
        analytics = runner.get(NoteAnalytics)

        # Check the note doesn't exist.
        with self.assertRaises(NoteNotFound):
            app.get_note_details(slug="welcome")

        # Check the list of notes is empty.
        notes = list(app.get_notes())
        self.assertEqual(len(notes), 0)

        # Create a note.
        app.create_note(title="Welcome", slug="welcome")

        # Present note identified by the given slug.
        note = app.get_note_details(slug="welcome")
        
        #print(note)

        # Check we got a dict that has the given title and slug.
        self.assertEqual(note["title"], "Welcome")
        self.assertEqual(note["slug"], "welcome")
        self.assertEqual(note["body"], "")
        self.assertEqual(note["modified_by"], user_id)

        # Update the title.
        app.update_title(slug="welcome", title="Welcome Visitors")

        # Check the title was updated.
        note = app.get_note_details(slug="welcome")
        self.assertEqual(note["title"], "Welcome Visitors")
        self.assertEqual(note["modified_by"], user_id)

        # Update the slug.
        app.update_slug(old_slug="welcome", new_slug="welcome-visitors")

        # Check the index was updated.
        with self.assertRaises(NoteNotFound):
            app.get_note_details(slug="welcome")

        # Check we can get the note by the new slug.
        note = app.get_note_details(slug="welcome-visitors")
        self.assertEqual(note["title"], "Welcome Visitors")
        self.assertEqual(note["slug"], "welcome-visitors")

        # Update the body.
        app.update_body(slug="welcome-visitors", body="Welcome to my wiki")

        # Check the body was updated.
        note = app.get_note_details(slug="welcome-visitors")
        self.assertEqual(note["body"], "Welcome to my wiki")

        # Update the body.
        app.update_body(slug="welcome-visitors", body="Welcome to this wiki")

        # Check the body was updated.
        note = app.get_note_details(slug="welcome-visitors")
        self.assertEqual(note["body"], "Welcome to this wiki")

        # Update the body.
        app.update_body(
            slug="welcome-visitors",
            body="""
Welcome to this wiki!

This is a wiki about...
""",
        )

        # Check the body was updated.
        note = app.get_note_details(slug="welcome-visitors")
        self.assertEqual(
            note["body"],
            """
Welcome to this wiki!

This is a wiki about...
""",
        )

        # Check all the Note events have the user_id.
        for notification in NotificationLogReader(app.notification_log).read(start=1):
            domain_event = app.mapper.to_domain_event(notification)
            if isinstance(domain_event, Note.Event):
                self.assertEqual(domain_event.user_id, user_id)

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

        # Update the body.
        app.update_body(
            slug="welcome-visitors",
            body="""
Welcome to this wiki!

This is a wiki about us!
""",
        )

        # Check 'modified_by' changed.
        note = app.get_note_details(slug="welcome-visitors")
        self.assertEqual(note["title"], "Welcome Visitors")
        self.assertEqual(note["modified_by"], user_id)

        # Check a snapshot was created by now.
        assert app.snapshots
        index = cast(Index, app.repository.get(Index.create_id("welcome-visitors")))
        assert index.ref
        self.assertTrue(len(list(app.snapshots.get(index.ref))))

        # Create some more pages and list all the pages.
        app.create_note("Note 2", "note-2")
        app.create_note("Note 3", "note-3")
        app.create_note("Note 4", "note-4")
        app.create_note("Note 5", "note-5")
        
        app.create_note(None, "note-no-title")
        
        notes = list(app.get_notes(desc=True))
        self.assertEqual(notes[1]["title"], "Note 5")
        self.assertEqual(notes[1]["slug"], "note-5")
        self.assertEqual(notes[2]["title"], "Note 4")
        self.assertEqual(notes[2]["slug"], "note-4")
        self.assertEqual(notes[3]["title"], "Note 3")
        self.assertEqual(notes[3]["slug"], "note-3")
        self.assertEqual(notes[4]["title"], "Note 2")
        self.assertEqual(notes[4]["slug"], "note-2")
        self.assertEqual(notes[5]["title"], "Welcome Visitors")
        self.assertEqual(notes[5]["slug"], "welcome-visitors")
        
        self.assertEqual(notes[0]["title"], None)
        self.assertEqual(notes[0]["slug"], "note-no-title")
        
        notes = list(app.get_notes(desc=True, limit=3))
        self.assertEqual(len(notes), 3)
        self.assertEqual(notes[0]["slug"], "note-no-title")
        self.assertEqual(notes[1]["slug"], "note-5")
        self.assertEqual(notes[2]["slug"], "note-4")

        notes = list(app.get_notes(desc=True, limit=3, lte=2))
        self.assertEqual(len(notes), 2)
        self.assertEqual(notes[0]["slug"], "note-2")
        self.assertEqual(notes[1]["slug"], "welcome-visitors")

        notes = list(app.get_notes(desc=True, lte=2))
        self.assertEqual(len(notes), 2)
        self.assertEqual(notes[0]["slug"], "note-2")
        self.assertEqual(notes[1]["slug"], "welcome-visitors")

        # Check we can't change the slug of a note to one
        # that is being used by another note.
        with self.assertRaises(SlugConflictError):
            app.update_slug("note-2", "note-3")

        # Check we can change the slug of a note to one
        # that was previously being used.
        app.update_slug("welcome-visitors", "welcome")

        note = app.get_note_details(slug="welcome")
        self.assertEqual(note["title"], "Welcome Visitors")
        self.assertEqual(note["modified_by"], user_id)
        
        now_dt = create_utc_datetime_now()
        app.create_note(title='imported-note.md', slug='imported-note.md', created_at=now_dt, body='new note')
        
        note = app.get_note_details(slug="imported-note.md")
        
        self.assertEqual(note["created_at"], now_dt)
        self.assertEqual(note["modified_at"], now_dt)
        self.assertEqual(note["body"], 'new note')
        
        app.create_note(title='imported-note-no-ts.md', slug='imported-note-no-ts.md', body='new note no ts')
        
        note = app.get_note_details(slug="imported-note-no-ts.md")
        
        # can we get to the underlying event?
        self.assertEqual(note["created_at"], note["modified_at"])
        self.assertEqual(note["body"], 'new note no ts')
        
        # ----
        # Import a note with a retro dated creation event
        #
        # Re-implement the Application.save machinery and the create_note logic.
        
        created_at = datetime(2020,1,2)
        import_events = Note(
                slug='abc.md', 
                body='abc note', 
                created_at= created_at
                ).collect_events()
            
        import_events[0] = replace(import_events[0], 
                                timestamp = created_at)
        
        print(import_events)
        
        processing_event = ProcessingEvent()
        processing_event.collect_events(*import_events)
        
        recordings = app._record(processing_event)
        app._take_snapshots(processing_event)
        app._notify(recordings)
        app.notify(processing_event.events)
        
        note_logged = app.note_log.trigger_event(
            note_id=import_events[0].originator_id
            )
        index_entry = Index(import_events[0].slug, 
            ref=import_events[0].originator_id
            )
        
        app.save(note_logged, index_entry)
        
        imported_note = app.get_note_details(slug = 'abc.md')
        
        print(imported_note)
        
        # ---
        
        
        # Get a stream of all events
        #print(analytics.recorder.select_notifications(0, 1000)[0])
        #print(app.recorder.protocol)
        
        runner.stop()
        
        
if __name__ == '__main__':
    db_path = "./test_notes_app_es.db"
    
    os.environ["PERSISTENCE_MODULE"] = "eventsourcing.sqlite"
    os.environ["SQLITE_DBNAME"] = db_path
    os.environ["SQLITE_LOCK_TIMEOUT"] = "10"
    #os.environ["COMPRESSOR_TOPIC"] = "gzip"
    
    print(f'Using {db_path}')
    
    if os.path.exists(db_path):
        os.remove(db_path)
    
    unittest.main()