From 6fb1dfd05a3911560c402c6c33e4e4ea4afbb15f Mon Sep 17 00:00:00 2001 From: Felipe Montoya Date: Thu, 29 Oct 2026 10:55:20 -0500 Subject: [PATCH] feat!: load fake Pathways objects into event_sink.catalog_pathway. Requires v2.2.0 from the platform-plugin-aspects --- README.rst | 22 ++ default_config.yaml | 12 ++ xapi_db_load/backends/chdb.py | 121 +++++++++++ xapi_db_load/backends/clickhouse.py | 73 +++++++ xapi_db_load/backends/csv.py | 75 +++++++ xapi_db_load/backends/ralph.py | 4 + xapi_db_load/backends/vector.py | 4 + xapi_db_load/constants.py | 29 +++ xapi_db_load/generate_load_async.py | 145 +++++++++++++ xapi_db_load/tests/test_backends.py | 10 + xapi_db_load/tests/test_event_generator.py | 231 ++++++++++++++++++++- 11 files changed, 725 insertions(+), 1 deletion(-) diff --git a/README.rst b/README.rst index 310a573..966bd0e 100644 --- a/README.rst +++ b/README.rst @@ -308,6 +308,28 @@ test:: # The number of organizations, courses will be evenly spread among these num_organizations: 3 + # The number of Catalog Pathways written to event_sink.catalog_pathway, + # spread across the organizations and the Pathway Categories. That table + # exists only in Aspects installs that have run the tutor-contrib-aspects + # migration 0042_event_sink_pathways (released after v6.1.1). On older + # installs set this to 0, or the pathway insert fails. + num_catalog_pathways: 6 + + # The id given to the first fake Catalog Pathway; the rest count up from + # it. Keep it above the ids of real pathways dumped by the LMS into the + # same table so the two don't collide. + first_catalog_pathway_id: 100001 + + # The number of learners enrolled in each Catalog Pathway, written to + # event_sink.pathway_enrollment (same migration). Learners come from the + # pathway org's courses first, and are capped at num_actors. About 10% of + # the enrollments are unenrolled (is_active false). + num_enrollments_per_pathway: 5 + + # The id given to the first fake Pathway Enrollment, kept above the ids of + # real enrollments dumped by the LMS, as with first_catalog_pathway_id. + first_pathway_enrollment_id: 100001 + # The number of learners to create, random subsets of these will be # "registered" for each course and have statements generated for them # between their registration date and the end of the course diff --git a/default_config.yaml b/default_config.yaml index 77d82ff..c91de0a 100644 --- a/default_config.yaml +++ b/default_config.yaml @@ -50,6 +50,18 @@ course_length_days: 120 num_organizations: 3 num_actors: 10 +# Catalog Pathways (event_sink.catalog_pathway), spread across the organizations +# and the Pathway Categories in constants.py. The table needs the +# tutor-contrib-aspects migration 0042_event_sink_pathways (released after +# v6.1.1); set num_catalog_pathways to 0 on older Aspects installs. +num_catalog_pathways: 6 +first_catalog_pathway_id: 100001 + +# Learners enrolled in each Catalog Pathway (event_sink.pathway_enrollment, same +# migration), taken from the pathway org's courses first and capped at num_actors. +num_enrollments_per_pathway: 5 +first_pathway_enrollment_id: 100001 + # This replicates users updating their profiles several times, creating # more rows num_actor_profile_changes: 5 diff --git a/xapi_db_load/backends/chdb.py b/xapi_db_load/backends/chdb.py index e2ee90e..a4a1114 100644 --- a/xapi_db_load/backends/chdb.py +++ b/xapi_db_load/backends/chdb.py @@ -73,6 +73,18 @@ def get_test_data_tasks(self) -> List[Waiter]: self.event_generator, self.db_loader, ), + WriteCatalogPathways( + self.config, + self.logger, + self.event_generator, + self.db_loader, + ), + WritePathwayEnrollments( + self.config, + self.logger, + self.event_generator, + self.db_loader, + ), WriteBlocks( self.config, self.logger, @@ -545,6 +557,115 @@ async def _run_task(self): self.update_completed_task_count(increment_by=1) +class WriteCatalogPathways(XAPILakeCHDBAsync): # pylint: disable=abstract-method + """ + Write the Catalog Pathways to S3 using CHDB, all under one dump. + """ + + task_name = "Insert Catalog Pathways" + schema = "event_sink" + table = "catalog_pathway" + s3_load_prefix = "catalog_pathway" + table_definition = """ + `id` Int64, + `org` String, + `pathway_code` String, + `title` String, + `category_code` String, + `category_title` String, + `created` String, + `modified` String, + `dump_id` UUID, + `time_last_dumped` String, + `is_deleted` UInt8 + """ + + async def _run_task(self): + """ + Insert the Catalog Pathways into the event sink db. + """ + pathways = self.event_generator.catalog_pathways + self.update_total_task_count(len(pathways)) + + dump_id = str(uuid.uuid4()) + dump_time = datetime.now(UTC) + out_data = [] + + for p in pathways: + # Titles such as "Master's Degree" need their quotes escaped for SQL. + title = p["title"].replace("'", "\\'") + category_title = p["category_title"].replace("'", "\\'") + out_data.append( + f"""( + {p["id"]}, + '{p["org"]}', + '{p["pathway_code"]}', + '{title}', + '{p["category_code"]}', + '{category_title}', + '{p["created"]}', + '{p["modified"]}', + '{dump_id}', + '{dump_time}', + 0 + )""" + ) + self.update_completed_task_count(increment_by=1) + + if out_data: + await self._write_list_sql_retry(out_data, "catalog_pathway") + + +class WritePathwayEnrollments(XAPILakeCHDBAsync): # pylint: disable=abstract-method + """ + Write the Pathway Enrollments to S3 using CHDB, all under one dump. + """ + + task_name = "Insert Pathway Enrollments" + schema = "event_sink" + table = "pathway_enrollment" + s3_load_prefix = "pathway_enrollment" + table_definition = """ + `id` Int64, + `catalog_pathway_id` Int64, + `actor_id` String, + `created` String, + `modified` String, + `is_active` Bool, + `dump_id` UUID, + `time_last_dumped` String + """ + + async def _run_task(self): + """ + Insert the Pathway Enrollments into the event sink db. + """ + enrollments = self.event_generator.pathway_enrollments + self.update_total_task_count(len(enrollments)) + + dump_id = str(uuid.uuid4()) + dump_time = datetime.now(UTC) + out_data = [] + + for e in enrollments: + out_data.append( + f"""( + {e["id"]}, + {e["catalog_pathway_id"]}, + '{e["actor_id"]}', + '{e["created"]}', + '{e["modified"]}', + {str(e["is_active"]).lower()}, + '{dump_id}', + '{dump_time}' + )""" + ) + self.update_completed_task_count(increment_by=1) + + if out_data: + await self._write_list_sql_retry(out_data, "pathway_enrollment") + + class WriteBlocks(XAPILakeCHDBAsync): # pylint: disable=abstract-method """ Write the course block data to S3 using CHDB. diff --git a/xapi_db_load/backends/clickhouse.py b/xapi_db_load/backends/clickhouse.py index edaf320..afac8d5 100644 --- a/xapi_db_load/backends/clickhouse.py +++ b/xapi_db_load/backends/clickhouse.py @@ -43,6 +43,8 @@ def get_test_data_tasks(self) -> List[Waiter]: self.event_generator, InsertInitialEnrollments(self.config, self.logger, self.event_generator), InsertCourses(self.config, self.logger, self.event_generator), + InsertCatalogPathways(self.config, self.logger, self.event_generator), + InsertPathwayEnrollments(self.config, self.logger, self.event_generator), InsertBlocks(self.config, self.logger, self.event_generator), InsertObjectTags(self.config, self.logger, self.event_generator), InsertTaxonomies(self.config, self.logger, self.event_generator), @@ -278,6 +280,77 @@ async def _run_task(self): self.update_completed_task_count(increment_by=1) +class InsertCatalogPathways(XAPILakeClickhouseAsync): # pylint: disable=abstract-method + """Inserts Catalog Pathway data to ClickHouse.""" + + task_name = "Insert Catalog Pathways" + + async def _run_task(self): + """ + Insert the Catalog Pathways to ClickHouse, all under one dump. + """ + pathways = self.event_generator.catalog_pathways + self.update_total_task_count(1) + + dump_id = str(uuid.uuid4()) + dump_time = datetime.now(UTC) + out_data = [ + ( + p["id"], + p["org"], + p["pathway_code"], + p["title"], + p["category_code"], + p["category_title"], + p["created"], + p["modified"], + dump_id, + str(dump_time), + 0, # is_deleted: fake pathways are never tombstones + ) + for p in pathways + ] + + if out_data: + await self._insert_rows("catalog_pathway", out_data) + self.update_completed_task_count(increment_by=1) + + +class InsertPathwayEnrollments(XAPILakeClickhouseAsync): # pylint: disable=abstract-method + """Inserts Pathway Enrollment data to ClickHouse.""" + + task_name = "Insert Pathway Enrollments" + + async def _run_task(self): + """ + Insert the Pathway Enrollments to ClickHouse, all under one dump, in batches. + """ + enrollments = self.event_generator.pathway_enrollments + self.update_total_task_count(len(enrollments)) + + dump_id = str(uuid.uuid4()) + dump_time = str(datetime.now(UTC)) + batch_size = self.config["batch_size"] + + for i in range(0, len(enrollments), batch_size): + batch = enrollments[i:i + batch_size] + out_data = [ + ( + e["id"], + e["catalog_pathway_id"], + e["actor_id"], + e["created"], + e["modified"], + e["is_active"], + dump_id, + dump_time, + ) + for e in batch + ] + await self._insert_rows("pathway_enrollment", out_data) + self.update_completed_task_count(increment_by=len(batch)) + + class InsertBlocks(XAPILakeClickhouseAsync): # pylint: disable=abstract-method """Inserts course block data to ClickHouse.""" diff --git a/xapi_db_load/backends/csv.py b/xapi_db_load/backends/csv.py index addfefb..4152948 100644 --- a/xapi_db_load/backends/csv.py +++ b/xapi_db_load/backends/csv.py @@ -48,6 +48,8 @@ def get_test_data_tasks(self) -> List[Waiter]: return [ self.event_generator, WriteCSVCourses(self.config, self.logger, self.event_generator), + WriteCSVCatalogPathways(self.config, self.logger, self.event_generator), + WriteCSVPathwayEnrollments(self.config, self.logger, self.event_generator), WriteCSVBlocks(self.config, self.logger, self.event_generator), WriteCSVObjectTags(self.config, self.logger, self.event_generator), WriteCSVTaxonomies(self.config, self.logger, self.event_generator), @@ -224,6 +226,79 @@ async def _run_task(self): self.update_completed_task_count(increment_by=1) +class WriteCSVCatalogPathways(XAPILakeCSVAsync): + """Writes Catalog Pathway data to a gzipped CSV file.""" + + task_name = "Write CSV Catalog Pathways" + schema = "event_sink" + table = "catalog_pathway" + filename = "catalog_pathways.csv.gz" + + async def _run_task(self): + """ + Write out the Catalog Pathways data file, all under one dump. + """ + async with self.get_csv_handle("catalog_pathways") as pathway_csv_writer: + pathways = self.event_generator.catalog_pathways + self.update_total_task_count(len(pathways)) + + dump_id = str(uuid.uuid4()) + dump_time = datetime.now(UTC) + + for p in pathways: + pathway_csv_writer.writerow( + ( + p["id"], + p["org"], + p["pathway_code"], + p["title"], + p["category_code"], + p["category_title"], + p["created"], + p["modified"], + dump_id, + dump_time, + 0, # is_deleted: fake pathways are never tombstones + ) + ) + self.update_completed_task_count(increment_by=1) + + +class WriteCSVPathwayEnrollments(XAPILakeCSVAsync): + """Writes Pathway Enrollment data to a gzipped CSV file.""" + + task_name = "Write CSV Pathway Enrollments" + schema = "event_sink" + table = "pathway_enrollment" + filename = "pathway_enrollments.csv.gz" + + async def _run_task(self): + """ + Write out the Pathway Enrollments data file, all under one dump. + """ + async with self.get_csv_handle("pathway_enrollments") as enrollment_csv_writer: + enrollments = self.event_generator.pathway_enrollments + self.update_total_task_count(len(enrollments)) + + dump_id = str(uuid.uuid4()) + dump_time = datetime.now(UTC) + + for e in enrollments: + enrollment_csv_writer.writerow( + ( + e["id"], + e["catalog_pathway_id"], + e["actor_id"], + e["created"], + e["modified"], + str(e["is_active"]).lower(), + dump_id, + dump_time, + ) + ) + self.update_completed_task_count(increment_by=1) + + class WriteCSVBlocks(XAPILakeCSVAsync): """Writes course block data to a gzipped CSV file.""" diff --git a/xapi_db_load/backends/ralph.py b/xapi_db_load/backends/ralph.py index 6428cf0..d9d292f 100644 --- a/xapi_db_load/backends/ralph.py +++ b/xapi_db_load/backends/ralph.py @@ -15,10 +15,12 @@ from xapi_db_load.backends.base_async_backend import BaseBackendTasks from xapi_db_load.backends.clickhouse import ( InsertBlocks, + InsertCatalogPathways, InsertCourses, InsertExternalIDs, InsertInitialEnrollments, InsertObjectTags, + InsertPathwayEnrollments, InsertProfiles, InsertTags, InsertTaxonomies, @@ -41,6 +43,8 @@ def get_test_data_tasks(self): self.event_generator, InsertInitialEnrollments(self.config, self.logger, self.event_generator), InsertCourses(self.config, self.logger, self.event_generator), + InsertCatalogPathways(self.config, self.logger, self.event_generator), + InsertPathwayEnrollments(self.config, self.logger, self.event_generator), InsertBlocks(self.config, self.logger, self.event_generator), InsertObjectTags(self.config, self.logger, self.event_generator), InsertTaxonomies(self.config, self.logger, self.event_generator), diff --git a/xapi_db_load/backends/vector.py b/xapi_db_load/backends/vector.py index 572cf81..ca59c41 100644 --- a/xapi_db_load/backends/vector.py +++ b/xapi_db_load/backends/vector.py @@ -16,10 +16,12 @@ from xapi_db_load.backends.base_async_backend import BaseBackendTasks from xapi_db_load.backends.clickhouse import ( InsertBlocks, + InsertCatalogPathways, InsertCourses, InsertExternalIDs, InsertInitialEnrollments, InsertObjectTags, + InsertPathwayEnrollments, InsertProfiles, InsertTags, InsertTaxonomies, @@ -42,6 +44,8 @@ def get_test_data_tasks(self): self.event_generator, InsertInitialEnrollments(self.config, self.logger, self.event_generator), InsertCourses(self.config, self.logger, self.event_generator), + InsertCatalogPathways(self.config, self.logger, self.event_generator), + InsertPathwayEnrollments(self.config, self.logger, self.event_generator), InsertBlocks(self.config, self.logger, self.event_generator), InsertObjectTags(self.config, self.logger, self.event_generator), InsertTaxonomies(self.config, self.logger, self.event_generator), diff --git a/xapi_db_load/constants.py b/xapi_db_load/constants.py index 0e5d36c..a884968 100644 --- a/xapi_db_load/constants.py +++ b/xapi_db_load/constants.py @@ -29,6 +29,35 @@ UUID_SHORT_LENGTH = 8 COURSE_ID_SHORT_LENGTH = 6 +# Catalog Pathways (openedx-core ``openedx_catalog``) written to the +# ``event_sink.catalog_pathway`` table when ``num_catalog_pathways`` is not set. +DEFAULT_NUM_CATALOG_PATHWAYS = 6 + +# First id given to a fake Catalog Pathway when ``first_catalog_pathway_id`` is +# not set. High enough not to collide with real pathways dumped by the LMS +# into the same table, the way ``first_user_id`` keeps fake users apart. +DEFAULT_FIRST_CATALOG_PATHWAY_ID = 100001 + +# Pathway Categories the fake Catalog Pathways rotate through, as +# (category_code, title) pairs. +PATHWAY_CATEGORIES = ( + ("masters", "Master's Degree"), + ("annual-training", "Annual Training"), + ("professional-certificate", "Professional Certificate"), +) + +# Learners enrolled in each fake Catalog Pathway (``event_sink.pathway_enrollment``) +# when ``num_enrollments_per_pathway`` is not set. Capped at ``num_actors``. +DEFAULT_NUM_ENROLLMENTS_PER_PATHWAY = 5 + +# First id given to a fake Pathway Enrollment when ``first_pathway_enrollment_id`` +# is not set, kept above the ids of real enrollments for the same reason as +# ``DEFAULT_FIRST_CATALOG_PATHWAY_ID``. +DEFAULT_FIRST_PATHWAY_ENROLLMENT_ID = 100001 + +# Share of fake Pathway Enrollments that end up unenrolled (``is_active`` False). +PATHWAY_UNENROLLED_RATIO = 0.1 + # Range of "course runs" generated per logical course (inclusive of low, # exclusive of high — passed straight to ``random.randrange``). MIN_COURSE_RUNS = 1 diff --git a/xapi_db_load/generate_load_async.py b/xapi_db_load/generate_load_async.py index 67f38c1..415b0a4 100644 --- a/xapi_db_load/generate_load_async.py +++ b/xapi_db_load/generate_load_async.py @@ -3,6 +3,7 @@ """ import asyncio +import datetime import json import os import random @@ -20,9 +21,15 @@ from xapi_db_load.constants import ( COURSE_ID_SHORT_LENGTH, + DEFAULT_FIRST_CATALOG_PATHWAY_ID, + DEFAULT_FIRST_PATHWAY_ENROLLMENT_ID, DEFAULT_LMS_URL, + DEFAULT_NUM_CATALOG_PATHWAYS, + DEFAULT_NUM_ENROLLMENTS_PER_PATHWAY, MAX_COURSE_RUNS, MIN_COURSE_RUNS, + PATHWAY_CATEGORIES, + PATHWAY_UNENROLLED_RATIO, ) from xapi_db_load.course_configs import ( Actor, @@ -100,6 +107,29 @@ def _get_uuid() -> str: return str(uuid.uuid4()) +def _random_datetime( + start: datetime.datetime, end: datetime.datetime +) -> datetime.datetime: + """ + Return a random UTC datetime between start and end, with microseconds like + the timestamps Django stores. + """ + span = int((end - start).total_seconds() * 1_000_000) + return start + datetime.timedelta(microseconds=random.randrange(max(span, 1))) + + +def to_sink_datetime(value: datetime.datetime) -> str: + """ + Format a UTC datetime the way the platform-plugin-aspects model sinks write + created/modified: DRF's DateTimeField, which is isoformat() with "+00:00" + turned into "Z", e.g. 2026-10-29T21:18:23.286693Z. + """ + formatted = value.astimezone(datetime.UTC).isoformat() + if formatted.endswith("+00:00"): + formatted = formatted[:-6] + "Z" + return formatted + + class EventGenerator(Waiter): # pylint: disable=abstract-method """ Generates a batch of random xAPI events based on the EVENT_WEIGHTS proportions. @@ -118,6 +148,8 @@ def __init__( self.actors: List[Actor] = [] self.courses: List[RandomCourse] = [] self.orgs: List[str] = [] + self.catalog_pathways: List[Dict] = [] + self.pathway_enrollments: List[Dict] = [] self.taxonomies: Dict = {} self.tags: List = [] self.setup_complete = False @@ -153,12 +185,15 @@ async def run_task(self) -> None: """ self.complete_pct = 0.0 self.setup_orgs() + self.setup_catalog_pathways() self.complete_pct = 0.1 self.setup_taxonomies_tags() self.complete_pct = 0.2 self.setup_actors() self.complete_pct = 0.3 await self.setup_courses() + # Needs the courses, to enroll learners of the pathway's org. + self.setup_pathway_enrollments() # Force this in case of a rounding error self.complete_pct = 1.0 @@ -180,6 +215,116 @@ def setup_orgs(self) -> None: for i in range(self.config["num_organizations"]): self.orgs.append(f"Org{i}") + def setup_catalog_pathways(self) -> None: + """ + Create the Catalog Pathways, rotating through the orgs and categories so + that each org gets pathways of different categories. + + Rows match the ``event_sink.catalog_pathway`` table written by the + platform-plugin-aspects ``catalog_pathway`` sink, including the + created/modified string format (see ``to_sink_datetime``). + """ + num_pathways = self.config.get( + "num_catalog_pathways", DEFAULT_NUM_CATALOG_PATHWAYS + ) + first_id = self.config.get( + "first_catalog_pathway_id", DEFAULT_FIRST_CATALOG_PATHWAY_ID + ) + num_orgs = len(self.orgs) + start, end = self._run_window() + + for i in range(num_pathways): + # Shift each org's category by one on every full pass over the orgs. + category_code, category_title = PATHWAY_CATEGORIES[ + (i % num_orgs + i // num_orgs) % len(PATHWAY_CATEGORIES) + ] + created = _random_datetime(start, end) + modified = _random_datetime( + created, min(created + datetime.timedelta(days=30), end) + ) + self.catalog_pathways.append( + { + "id": first_id + i, + "org": self.orgs[i % num_orgs], + "pathway_code": f"Pathway{i}", + "title": f"{category_title} {i}", + "category_code": category_code, + "category_title": category_title, + "created": to_sink_datetime(created), + "modified": to_sink_datetime(modified), + } + ) + + def setup_pathway_enrollments(self) -> None: + """ + Enroll learners in each Catalog Pathway. + + Each pathway gets ``num_enrollments_per_pathway`` distinct learners (at + most ``num_actors``), taken first from learners of the pathway org's + courses so that their xAPI activity is in the same org, then from the + remaining actors. ``actor_id`` is the actor's xAPI id, as written by the + platform-plugin-aspects ``pathway_enrollment`` sink, so it joins the + xAPI tables. About ``PATHWAY_UNENROLLED_RATIO`` of the enrollments end + up unenrolled (``is_active`` False, modified after created); the rest + keep ``modified`` equal to ``created``. + + Rows match the ``event_sink.pathway_enrollment`` table. + """ + per_pathway = min( + self.config.get( + "num_enrollments_per_pathway", DEFAULT_NUM_ENROLLMENTS_PER_PATHWAY + ), + len(self.actors), + ) + next_id = self.config.get( + "first_pathway_enrollment_id", DEFAULT_FIRST_PATHWAY_ENROLLMENT_ID + ) + _, end = self._run_window() + + org_learners: Dict[str, Dict[int, Actor]] = {} + for course in self.courses: + learners = org_learners.setdefault(course.org, {}) + for enrolled in course.actors: + learners[enrolled.actor.user_id] = enrolled.actor + + for pathway in self.catalog_pathways: + in_org = list(org_learners.get(pathway["org"], {}).values()) + random.shuffle(in_org) + chosen = in_org[:per_pathway] + if len(chosen) < per_pathway: + chosen_ids = {a.user_id for a in chosen} + others = [a for a in self.actors if a.user_id not in chosen_ids] + chosen += random.sample(others, per_pathway - len(chosen)) + + pathway_created = datetime.datetime.fromisoformat(pathway["created"]) + for actor in chosen: + created = _random_datetime(pathway_created, end) + is_active = random.random() >= PATHWAY_UNENROLLED_RATIO + modified = created if is_active else _random_datetime(created, end) + self.pathway_enrollments.append( + { + "id": next_id, + "catalog_pathway_id": pathway["id"], + "actor_id": actor.id, + "created": to_sink_datetime(created), + "modified": to_sink_datetime(modified), + "is_active": is_active, + } + ) + next_id += 1 + + def _run_window(self) -> tuple[datetime.datetime, datetime.datetime]: + """ + Return the configured start and end dates as UTC datetimes. + """ + start = datetime.datetime.combine( + self.start_date, datetime.time(), tzinfo=datetime.UTC + ) + end = datetime.datetime.combine( + self.end_date, datetime.time(), tzinfo=datetime.UTC + ) + return start, end + async def setup_courses(self) -> None: """ Pre-create a number of courses based on the config. diff --git a/xapi_db_load/tests/test_backends.py b/xapi_db_load/tests/test_backends.py index 2db5939..9a84f3c 100644 --- a/xapi_db_load/tests/test_backends.py +++ b/xapi_db_load/tests/test_backends.py @@ -16,6 +16,10 @@ import yaml from click.testing import CliRunner +from xapi_db_load.constants import ( + DEFAULT_NUM_CATALOG_PATHWAYS, + DEFAULT_NUM_ENROLLMENTS_PER_PATHWAY, +) from xapi_db_load.main import load_db @@ -90,6 +94,12 @@ def test_csv_backend(tmp_path): ("blocks", expected_blocks), ("external_ids", expected_external_ids), ("user_profiles", expected_profiles), + ("catalog_pathways", DEFAULT_NUM_CATALOG_PATHWAYS), + ( + "pathway_enrollments", + DEFAULT_NUM_CATALOG_PATHWAYS + * min(DEFAULT_NUM_ENROLLMENTS_PER_PATHWAY, test_config["num_actors"]), + ), ): with gzip.open( os.path.join(test_config["log_dir"], f"{prefix}.csv.gz"), "r" diff --git a/xapi_db_load/tests/test_event_generator.py b/xapi_db_load/tests/test_event_generator.py index 40271cb..b65dc2a 100644 --- a/xapi_db_load/tests/test_event_generator.py +++ b/xapi_db_load/tests/test_event_generator.py @@ -2,12 +2,24 @@ Validation tests for ``EventGenerator._validate_config`` and basic setup. """ +import asyncio import datetime import logging +import re import pytest -from xapi_db_load.generate_load_async import EventGenerator +from xapi_db_load.constants import ( + DEFAULT_FIRST_CATALOG_PATHWAY_ID, + DEFAULT_FIRST_PATHWAY_ENROLLMENT_ID, + DEFAULT_NUM_CATALOG_PATHWAYS, + DEFAULT_NUM_ENROLLMENTS_PER_PATHWAY, + PATHWAY_CATEGORIES, +) +from xapi_db_load.generate_load_async import ( + EventGenerator, + to_sink_datetime, +) def _base_config(**overrides) -> dict: @@ -113,3 +125,220 @@ def test_instances_have_isolated_state(): assert not b.actors assert not b.orgs + + +def test_catalog_pathways_defaults(): + """ + Without pathway settings the default number of pathways is created, starting + at the default id, and every category is used. + """ + gen = _make_generator(_base_config(num_organizations=3)) + gen.setup_orgs() + gen.setup_catalog_pathways() + pathways = gen.catalog_pathways + + assert len(pathways) == DEFAULT_NUM_CATALOG_PATHWAYS + assert pathways[0]["id"] == DEFAULT_FIRST_CATALOG_PATHWAY_ID + assert {p["category_code"] for p in pathways} == { + code for code, _ in PATHWAY_CATEGORIES + } + + +def test_catalog_pathways_first_id(): + """first_catalog_pathway_id moves the id range.""" + gen = _make_generator( + _base_config(num_catalog_pathways=2, first_catalog_pathway_id=500) + ) + gen.setup_orgs() + gen.setup_catalog_pathways() + + assert [p["id"] for p in gen.catalog_pathways] == [500, 501] + + +@pytest.mark.parametrize("num_orgs", [1, 2, 3, 5]) +def test_catalog_pathways_spread_across_orgs_and_categories(num_orgs): + """ + Pathways rotate through every org, every org gets every category once it has + as many pathways as there are categories, and ids and codes are unique. + """ + num_categories = len(PATHWAY_CATEGORIES) + num_pathways = num_orgs * num_categories + gen = _make_generator( + _base_config(num_organizations=num_orgs, num_catalog_pathways=num_pathways) + ) + gen.setup_orgs() + gen.setup_catalog_pathways() + pathways = gen.catalog_pathways + + assert len(pathways) == num_pathways + assert {p["org"] for p in pathways} == set(gen.orgs) + assert len({p["id"] for p in pathways}) == num_pathways + assert len({(p["org"], p["pathway_code"]) for p in pathways}) == num_pathways + for org in gen.orgs: + org_categories = {p["category_code"] for p in pathways if p["org"] == org} + assert len(org_categories) == num_categories + for p in pathways: + created = datetime.datetime.fromisoformat(p["created"]) + modified = datetime.datetime.fromisoformat(p["modified"]) + assert created <= modified + assert gen.start_date <= created.date() + assert modified.date() <= gen.end_date + + +def test_catalog_pathways_can_be_disabled(): + """num_catalog_pathways: 0 creates no pathways.""" + gen = _make_generator(_base_config(num_catalog_pathways=0)) + gen.setup_orgs() + gen.setup_catalog_pathways() + + assert not gen.catalog_pathways + + +# The format DRF's DateTimeField writes, which the platform-plugin-aspects +# sinks send to ClickHouse, e.g. 2026-10-29T21:18:23.286693Z. +SINK_DATETIME_RE = re.compile(r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(\.\d{6})?Z$") + + +@pytest.mark.parametrize( + "value,expected", + [ + ( + datetime.datetime(2026, 10, 29, 21, 18, 23, 286693, tzinfo=datetime.UTC), + "2026-10-29T21:18:23.286693Z", + ), + # DRF drops zero microseconds, because isoformat() does. + ( + datetime.datetime(2026, 1, 1, 12, 34, 56, tzinfo=datetime.UTC), + "2026-01-01T12:34:56Z", + ), + # Other offsets are converted to UTC first. + ( + datetime.datetime( + 2026, + 1, + 1, + 7, + 34, + 56, + tzinfo=datetime.timezone(datetime.timedelta(hours=-5)), + ), + "2026-01-01T12:34:56Z", + ), + ], +) +def test_to_sink_datetime(value, expected): + """Datetimes are formatted the way the LMS sinks write them.""" + assert to_sink_datetime(value) == expected + + +def test_catalog_pathways_use_sink_datetime_format(): + """created and modified are strings in the format the LMS sink writes.""" + gen = _make_generator(_base_config()) + gen.setup_orgs() + gen.setup_catalog_pathways() + + for p in gen.catalog_pathways: + assert SINK_DATETIME_RE.match(p["created"]), p["created"] + assert SINK_DATETIME_RE.match(p["modified"]), p["modified"] + + +async def _make_enrolled_generator(**overrides) -> EventGenerator: + """Run the whole setup, which ends by enrolling learners in the pathways.""" + gen = _make_generator(_base_config(**overrides)) + await gen.run_task() + return gen + + +def test_pathway_enrollments_defaults(): + """ + Each pathway gets the default number of distinct learners, ids count up + from the default first id, and every actor_id is a known actor's xAPI id. + """ + gen = asyncio.run(_make_enrolled_generator(num_organizations=3)) + enrollments = gen.pathway_enrollments + per_pathway = min(DEFAULT_NUM_ENROLLMENTS_PER_PATHWAY, len(gen.actors)) + + assert len(enrollments) == DEFAULT_NUM_CATALOG_PATHWAYS * per_pathway + assert [e["id"] for e in enrollments] == list( + range( + DEFAULT_FIRST_PATHWAY_ENROLLMENT_ID, + DEFAULT_FIRST_PATHWAY_ENROLLMENT_ID + len(enrollments), + ) + ) + actor_ids = {a.id for a in gen.actors} + pathways = {p["id"]: p for p in gen.catalog_pathways} + for pathway_id in pathways: + learners = [ + e["actor_id"] for e in enrollments if e["catalog_pathway_id"] == pathway_id + ] + assert len(learners) == len(set(learners)) == per_pathway + assert set(learners) <= actor_ids + + for e in enrollments: + assert SINK_DATETIME_RE.match(e["created"]), e["created"] + assert SINK_DATETIME_RE.match(e["modified"]), e["modified"] + created = datetime.datetime.fromisoformat(e["created"]) + modified = datetime.datetime.fromisoformat(e["modified"]) + pathway_created = datetime.datetime.fromisoformat( + pathways[e["catalog_pathway_id"]]["created"] + ) + assert pathway_created <= created <= modified + assert modified.date() <= gen.end_date + if e["is_active"]: + assert created == modified + + +def test_pathway_enrollments_prefer_learners_of_the_pathway_org(): + """ + Learners come from the pathway org's courses first, so their xAPI activity + is in the same org. + """ + gen = asyncio.run( + _make_enrolled_generator( + num_organizations=1, + num_actors=5, + num_catalog_pathways=1, + # Courses pick their actors with replacement, so only one distinct + # course learner is guaranteed. + num_enrollments_per_pathway=1, + ) + ) + course_learners = {ea.actor.id for c in gen.courses for ea in c.actors} + + assert len(gen.pathway_enrollments) == 1 + assert gen.pathway_enrollments[0]["actor_id"] in course_learners + + +def test_pathway_enrollments_capped_at_num_actors(): + """A pathway can't have more distinct learners than there are actors.""" + gen = asyncio.run( + _make_enrolled_generator( + num_actors=3, num_catalog_pathways=2, num_enrollments_per_pathway=10 + ) + ) + + assert len(gen.pathway_enrollments) == 2 * 3 + + +def test_pathway_enrollments_first_id_and_inactive(): + """first_pathway_enrollment_id moves the ids, and some learners unenroll.""" + gen = asyncio.run( + _make_enrolled_generator( + num_actors=50, + num_catalog_pathways=10, + num_enrollments_per_pathway=50, + first_pathway_enrollment_id=7, + ) + ) + enrollments = gen.pathway_enrollments + + assert enrollments[0]["id"] == 7 + # 500 enrollments at a 10% unenroll ratio; both kinds are all but certain. + assert {e["is_active"] for e in enrollments} == {True, False} + + +def test_pathway_enrollments_none_without_pathways(): + """num_catalog_pathways: 0 also means no enrollments.""" + gen = asyncio.run(_make_enrolled_generator(num_catalog_pathways=0)) + + assert not gen.pathway_enrollments