fix: Fix Vector formatter, add tests

Author: Tycho Hob

xapi_db_load/backends/vector.py

@@ -60,7 +60,9 @@ def __init__(self, config: dict, logger: Logger, event_generator: EventGenerator
         super().__init__(config, logger, event_generator)
 
         stream_handler = logging.StreamHandler(sys.stdout)
-        formatter = logging.Formatter("[%(name)s] %(message)s")
+        # This formatter is different from what the LMS uses, but is the smallest possible
+        # format that passes Vector's regex
+        formatter = logging.Formatter(" [{name}] [] {message}", style="{")
         stream_handler.setFormatter(formatter)
         self.xapi_logger = getLogger("xapi_tracking")
         self.xapi_logger.setLevel(logging.INFO)

xapi_db_load/tests/fixtures/small_vector_config.yaml

@@ -0,0 +1,54 @@
+# Test configuration for Ralph / ClickHouse
+# #########################################
+backend: vector
+db_host: localhost
+db_port: 8123
+db_name: xapi
+db_username: ch_admin
+db_password: foo
+
+# Run options
+log_dir: logs
+num_xapi_batches: 3
+batch_size: 100
+
+# This number is used for each QueueBackend that use workers, so the number of threads if
+# multiplicative. Generally this performs best less than 10, as more threads will cost more
+# in context switching than they save.
+num_workers: 4
+
+# Overall start and end date for the entire run
+start_date: 2014-01-01
+end_date: 2023-11-27
+
+# All courses will be this long, and be fit into the start / end dates
+# This must be less than end_date - start_date days.
+course_length_days: 120
+
+# The size of the test
+num_organizations: 3
+num_actors: 10
+
+# This replicates users updating their profiles several times, creating
+# more rows
+num_actor_profile_changes: 5
+
+# How many of each size course to create. The sum of these is the total number
+# of courses created for the test.
+num_course_sizes:
+  small: 1
+
+# How many times each course will be "published", this creates a more realistic
+# distribution of course blocks where each course can be published dozens or
+# hundreds of times while it is being developed.
+num_course_publishes: 10
+
+course_size_makeup:
+  small:
+    actors: 5
+    problems: 20
+    videos: 10
+    chapters: 3
+    sequences: 10
+    verticals: 20
+    forum_posts: 20

xapi_db_load/tests/test_backends.py

@@ -3,7 +3,9 @@
 """
 
 import gzip
+import json
 import os
+import re
 from contextlib import contextmanager
 from unittest.mock import AsyncMock, MagicMock, patch
 
@@ -120,6 +122,66 @@ def test_clickhouse_backend(_, tmp_path):
     assert "Run duration was" in result.output
 
 
+@patch(
+    "xapi_db_load.backends.base_async_backend.clickhouse_connect",
+    new_callable=AsyncMock,
+)
+@patch(
+    "xapi_db_load.backends.vector.getLogger",
+    new_callable=MagicMock,
+)
+def test_vector_backend(mock_get_logger, _, tmp_path):
+    """
+    Run a test through the Vector backend, currently this just checks that the
+    output indicates success.
+    """
+    test_path = "xapi_db_load/tests/fixtures/small_vector_config.yaml"
+
+    runner = CliRunner()
+
+    with override_config(test_path, tmp_path):
+        result = runner.invoke(
+            load_db,
+            f"--config_file {test_path}",
+            catch_exceptions=False,
+        )
+
+    # This test should create 300 xAPI log statemetns
+    assert mock_get_logger.return_value.info.call_count == 300
+
+    last_logged_statement = mock_get_logger.return_value.info.call_args.args[0]
+
+    # We check to make sure Vector's regex will parse what we're sending. We want it to match both
+    # the LMS and our local logger formatter.
+    # This is how things are generally formatted in the LMS
+    test_str_1 = f"2026-02-24 20:26:13,006 INFO 42 [xapi_tracking] [user None] [ip 172.19.0.1] logger.py:41 - {last_logged_statement}"
+
+    # This returns our message formatted with the abbreviated version we use for size and speed purposes
+    formatter = mock_get_logger.return_value.addHandler.call_args.args[0].formatter
+    test_str_2 = formatter._fmt.format(
+        name="xapi_tracking", message=last_logged_statement
+    )
+
+    # This is a direct copy and paste from Aspects' Vector common-post.toml
+    msg_regex = r"^.* \[xapi_tracking\] [^{}]* (?P<tracking_message>\{.*\})$"
+
+    # Quick test to make sure that what's being stored is at least parseable
+    for s in (test_str_1, test_str_2):
+        try:
+            statement = re.match(msg_regex, s).groups()[0]
+            json.loads(statement)
+        except Exception as e:
+            print(e)
+            print("Exception! Regex testing: ")
+            print(s)
+            raise
+
+    assert "Insert xAPI Events complete." in result.output
+    assert "Insert Initial Enrollments complete." in result.output
+    assert "ALL TASKS DONE!" in result.output
+    assert "Run duration was" in result.output
+
+
 @patch("xapi_db_load.backends.ralph.requests", new_callable=AsyncMock)
 @patch(
     "xapi_db_load.backends.base_async_backend.clickhouse_connect",