Skip to content

Commit dca9107

Browse files
BryanttVBryanttV
committed
test: add unit tests for course authoring migration functionality
1 parent 33f1bc3 commit dca9107

1 file changed

Lines changed: 347 additions & 0 deletions

File tree

‎openedx_authz/tests/test_migrations.py‎

Lines changed: 347 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,7 @@
66
from django.contrib.auth.models import Group
77
from django.core.management import CommandError, call_command
88
from django.test import TestCase
9+
from opaque_keys.edx.django.models import CourseKeyField
910

1011
from openedx_authz.api.data import OrgCourseOverviewGlobData
1112
from openedx_authz.api.users import (
@@ -16,6 +17,7 @@
1617
from openedx_authz.constants.roles import (
1718
COURSE_ADMIN,
1819
COURSE_DATA_RESEARCHER,
20+
COURSE_EDITOR,
1921
COURSE_LIMITED_STAFF,
2022
COURSE_STAFF,
2123
LEGACY_COURSE_ROLE_EQUIVALENCES,
@@ -24,9 +26,18 @@
2426
)
2527
from openedx_authz.engine.enforcer import AuthzEnforcer
2628
from openedx_authz.engine.utils import (
29+
MigrationErrorReason,
30+
MigrationMetadata,
2731
migrate_authz_to_legacy_course_roles,
2832
migrate_legacy_course_roles_to_authz,
2933
migrate_legacy_permissions,
34+
run_course_authoring_migration,
35+
)
36+
from openedx_authz.models.authz_migration import (
37+
AuthzCourseAuthoringMigrationRun,
38+
MigrationType,
39+
ScopeType,
40+
Status,
3041
)
3142
from openedx_authz.models.subjects import UserSubject
3243
from openedx_authz.tests.stubs.models import (
@@ -1413,3 +1424,339 @@ def test_course_id_list_filter_excludes_glob_and_other_courses(self):
14131424
# library assignment in self.org is excluded — library scopes are not course scopes
14141425
self.assertNotIn(user_lib.username, migrated_users)
14151426
self.assertEqual(len(errors), 0)
1427+
1428+
1429+
class TestRunCourseAuthoringMigration(TestCase):
1430+
"""Exercise ``run_course_authoring_migration`` lifecycle and outcomes using stub ``CourseAccessRole``."""
1431+
1432+
def setUp(self):
1433+
self.course_id = "course-v1:TestOrg+TestCourse+2024"
1434+
self.org_id = "TestOrg"
1435+
1436+
def test_skipped_when_another_run_is_already_running_for_scope(self):
1437+
"""Second call for the same scope creates a SKIPPED run and returns without migrating."""
1438+
AuthzCourseAuthoringMigrationRun.create_running(MigrationType.FORWARD, ScopeType.COURSE, self.course_id)
1439+
1440+
run_course_authoring_migration(
1441+
migration_type=MigrationType.FORWARD,
1442+
scope_type=ScopeType.COURSE,
1443+
scope_key=self.course_id,
1444+
course_access_role_model=CourseAccessRole,
1445+
user_subject_model=UserSubject,
1446+
course_id_list=[self.course_id],
1447+
org_id=None,
1448+
delete_after_migration=False,
1449+
)
1450+
1451+
self.assertTrue(
1452+
AuthzCourseAuthoringMigrationRun.objects.filter(scope_key=self.course_id, status=Status.SKIPPED).exists()
1453+
)
1454+
self.assertEqual(AuthzCourseAuthoringMigrationRun.objects.filter(scope_key=self.course_id).count(), 2)
1455+
1456+
def test_forward_completes_when_no_legacy_rows_match(self):
1457+
"""No matching ``CourseAccessRole`` rows yields a completed run with zero successes."""
1458+
run_course_authoring_migration(
1459+
migration_type=MigrationType.FORWARD,
1460+
scope_type=ScopeType.COURSE,
1461+
scope_key=self.course_id,
1462+
course_access_role_model=CourseAccessRole,
1463+
user_subject_model=UserSubject,
1464+
course_id_list=[self.course_id],
1465+
org_id=None,
1466+
delete_after_migration=False,
1467+
)
1468+
1469+
run = AuthzCourseAuthoringMigrationRun.objects.get(scope_key=self.course_id, status=Status.COMPLETED)
1470+
self.assertEqual(run.metadata.get("success_count"), 0)
1471+
self.assertEqual(run.metadata.get("error_count"), 0)
1472+
1473+
def test_marks_failed_when_migration_input_invalid(self):
1474+
"""``_validate_migration_input`` raises (e.g. non-course-v1 keys); orchestration marks FAILED."""
1475+
invalid_scope_key = "ccx-v1:Org+Course+Run"
1476+
1477+
run_course_authoring_migration(
1478+
migration_type=MigrationType.FORWARD,
1479+
scope_type=ScopeType.COURSE,
1480+
scope_key=invalid_scope_key,
1481+
course_access_role_model=CourseAccessRole,
1482+
user_subject_model=UserSubject,
1483+
course_id_list=[invalid_scope_key],
1484+
org_id=None,
1485+
delete_after_migration=False,
1486+
)
1487+
1488+
run = AuthzCourseAuthoringMigrationRun.objects.get(scope_key=invalid_scope_key, status=Status.FAILED)
1489+
self.assertIn("error", run.metadata)
1490+
1491+
def test_forward_partial_success_when_legacy_role_unknown(self):
1492+
"""Unknown legacy role strings are reported as errors; run ends PARTIAL_SUCCESS."""
1493+
user = User.objects.create_user(username="legacy_unknown_user", email="[email protected]")
1494+
CourseAccessRole.objects.create(
1495+
user=user, org=self.org_id, course_id=self.course_id, role="not_a_defined_legacy_role"
1496+
)
1497+
1498+
run_course_authoring_migration(
1499+
migration_type=MigrationType.FORWARD,
1500+
scope_type=ScopeType.COURSE,
1501+
scope_key=self.course_id,
1502+
course_access_role_model=CourseAccessRole,
1503+
user_subject_model=UserSubject,
1504+
course_id_list=[self.course_id],
1505+
org_id=None,
1506+
delete_after_migration=False,
1507+
)
1508+
1509+
run = AuthzCourseAuthoringMigrationRun.objects.get(scope_key=self.course_id, status=Status.PARTIAL_SUCCESS)
1510+
self.assertGreaterEqual(run.metadata.get("error_count"), 1)
1511+
self.assertEqual(run.metadata.get("success_count"), 0)
1512+
self.assertIn(MigrationErrorReason.UNKNOWN_ROLE, run.metadata.get("errors"))
1513+
1514+
def test_rollback_completes_when_no_role_assignments(self):
1515+
"""Rollback with no matching authz assignments finishes as COMPLETED with empty tallies."""
1516+
run_course_authoring_migration(
1517+
migration_type=MigrationType.ROLLBACK,
1518+
scope_type=ScopeType.COURSE,
1519+
scope_key=self.course_id,
1520+
course_access_role_model=CourseAccessRole,
1521+
user_subject_model=UserSubject,
1522+
course_id_list=[self.course_id],
1523+
org_id=None,
1524+
delete_after_migration=False,
1525+
)
1526+
1527+
run = AuthzCourseAuthoringMigrationRun.objects.get(scope_key=self.course_id, status=Status.COMPLETED)
1528+
self.assertEqual(run.metadata.get("error_count"), 0)
1529+
self.assertEqual(run.metadata.get("success_count"), 0)
1530+
1531+
def test_rollback_partial_success_when_authz_role_unknown(self):
1532+
"""Roles with no legacy ``CourseAccessRole`` mapping (e.g. ``course_editor``) → PARTIAL_SUCCESS."""
1533+
user = User.objects.create_user(username="rb_no_legacy_user", email="[email protected]")
1534+
assign_role_to_user_in_scope(user.username, COURSE_EDITOR.external_key, self.course_id)
1535+
1536+
run_course_authoring_migration(
1537+
migration_type=MigrationType.ROLLBACK,
1538+
scope_type=ScopeType.COURSE,
1539+
scope_key=self.course_id,
1540+
course_access_role_model=CourseAccessRole,
1541+
user_subject_model=UserSubject,
1542+
course_id_list=[self.course_id],
1543+
org_id=None,
1544+
delete_after_migration=False,
1545+
)
1546+
1547+
run = AuthzCourseAuthoringMigrationRun.objects.get(scope_key=self.course_id, status=Status.PARTIAL_SUCCESS)
1548+
self.assertGreaterEqual(run.metadata.get("error_count", 0), 1)
1549+
self.assertEqual(run.metadata.get("success_count"), 0)
1550+
self.assertIn(MigrationErrorReason.NO_LEGACY_EQUIVALENT, run.metadata.get("errors", {}))
1551+
1552+
def test_forward_partial_success_when_assignment_failed_on_duplicate(self):
1553+
"""Second FORWARD over the same legacy row hits duplicate AuthZ assignment → ``ASSIGNMENT_FAILED``."""
1554+
user = User.objects.create_user(username="dup_assign_user", email="[email protected]")
1555+
CourseAccessRole.objects.create(user=user, org=self.org_id, course_id=self.course_id, role="staff")
1556+
common = {
1557+
"migration_type": MigrationType.FORWARD,
1558+
"scope_type": ScopeType.COURSE,
1559+
"scope_key": self.course_id,
1560+
"course_access_role_model": CourseAccessRole,
1561+
"user_subject_model": UserSubject,
1562+
"course_id_list": [self.course_id],
1563+
"org_id": None,
1564+
"delete_after_migration": False,
1565+
}
1566+
1567+
run_course_authoring_migration(**common)
1568+
run_obj = AuthzCourseAuthoringMigrationRun.objects.get(scope_key=self.course_id, status=Status.COMPLETED)
1569+
self.assertEqual(run_obj.metadata.get("success_count"), 1)
1570+
1571+
run_course_authoring_migration(**common)
1572+
run = AuthzCourseAuthoringMigrationRun.objects.get(scope_key=self.course_id, status=Status.PARTIAL_SUCCESS)
1573+
self.assertEqual(run.metadata.get("success_count"), 0)
1574+
self.assertGreaterEqual(run.metadata.get("error_count"), 1)
1575+
self.assertIn(MigrationErrorReason.ASSIGNMENT_FAILED, run.metadata.get("errors", {}))
1576+
1577+
@patch("openedx_authz.engine.utils.migrate_legacy_course_roles_to_authz")
1578+
def test_forward_partial_success_when_no_scope(self, mock_migrate):
1579+
"""``NO_SCOPE`` is surfaced when the migration body reports neither course nor org (orchestration path)."""
1580+
mock_migrate.return_value = (
1581+
[
1582+
MigrationMetadata(
1583+
subject="no_scope_user",
1584+
role="staff",
1585+
reason=MigrationErrorReason.NO_SCOPE,
1586+
details="neither course_id nor org",
1587+
)
1588+
],
1589+
[],
1590+
)
1591+
1592+
run_course_authoring_migration(
1593+
migration_type=MigrationType.FORWARD,
1594+
scope_type=ScopeType.COURSE,
1595+
scope_key=self.course_id,
1596+
course_access_role_model=CourseAccessRole,
1597+
user_subject_model=UserSubject,
1598+
course_id_list=[self.course_id],
1599+
org_id=None,
1600+
delete_after_migration=False,
1601+
)
1602+
1603+
run = AuthzCourseAuthoringMigrationRun.objects.get(scope_key=self.course_id, status=Status.PARTIAL_SUCCESS)
1604+
self.assertIn(MigrationErrorReason.NO_SCOPE, run.metadata.get("errors", {}))
1605+
1606+
@patch("openedx_authz.engine.utils.migrate_authz_to_legacy_course_roles")
1607+
def test_rollback_partial_success_when_unexpected_scope_type(self, mock_migrate):
1608+
"""``UNEXPECTED_SCOPE_TYPE`` from rollback migration is grouped in run metadata."""
1609+
mock_migrate.return_value = (
1610+
[
1611+
MigrationMetadata(
1612+
subject="u",
1613+
role="course_staff",
1614+
scope=self.course_id,
1615+
reason=MigrationErrorReason.UNEXPECTED_SCOPE_TYPE,
1616+
details="UnexpectedScope",
1617+
)
1618+
],
1619+
[],
1620+
)
1621+
1622+
run_course_authoring_migration(
1623+
migration_type=MigrationType.ROLLBACK,
1624+
scope_type=ScopeType.COURSE,
1625+
scope_key=self.course_id,
1626+
course_access_role_model=CourseAccessRole,
1627+
user_subject_model=UserSubject,
1628+
course_id_list=[self.course_id],
1629+
org_id=None,
1630+
delete_after_migration=False,
1631+
)
1632+
1633+
run = AuthzCourseAuthoringMigrationRun.objects.get(scope_key=self.course_id, status=Status.PARTIAL_SUCCESS)
1634+
self.assertIn(MigrationErrorReason.UNEXPECTED_SCOPE_TYPE, run.metadata.get("errors", {}))
1635+
1636+
@patch("openedx_authz.engine.utils.migrate_authz_to_legacy_course_roles")
1637+
def test_rollback_partial_success_when_unexpected_error(self, mock_migrate):
1638+
"""Exceptions inside the rollback loop become ``UNEXPECTED_ERROR`` entries in metadata."""
1639+
mock_migrate.return_value = (
1640+
[
1641+
MigrationMetadata(
1642+
subject="u",
1643+
role=COURSE_STAFF.external_key,
1644+
scope=self.course_id,
1645+
reason=MigrationErrorReason.UNEXPECTED_ERROR,
1646+
details="KeyError: 'missing'",
1647+
)
1648+
],
1649+
[],
1650+
)
1651+
1652+
run_course_authoring_migration(
1653+
migration_type=MigrationType.ROLLBACK,
1654+
scope_type=ScopeType.COURSE,
1655+
scope_key=self.course_id,
1656+
course_access_role_model=CourseAccessRole,
1657+
user_subject_model=UserSubject,
1658+
course_id_list=[self.course_id],
1659+
org_id=None,
1660+
delete_after_migration=False,
1661+
)
1662+
1663+
run = AuthzCourseAuthoringMigrationRun.objects.get(scope_key=self.course_id, status=Status.PARTIAL_SUCCESS)
1664+
self.assertIn(MigrationErrorReason.UNEXPECTED_ERROR, run.metadata.get("errors", {}))
1665+
1666+
def test_forward_completes_org_wide_legacy_when_org_id(self):
1667+
"""FORWARD with ``org_id`` migrates org-level (no ``course_id``) legacy roles for that org."""
1668+
user = User.objects.create_user(username="rcam_org_fwd_user", email="[email protected]")
1669+
CourseAccessRole.objects.create(
1670+
user=user,
1671+
org=self.org_id,
1672+
course_id=CourseKeyField.Empty,
1673+
role="instructor",
1674+
)
1675+
1676+
run_course_authoring_migration(
1677+
migration_type=MigrationType.FORWARD,
1678+
scope_type=ScopeType.ORG,
1679+
scope_key=self.org_id,
1680+
course_access_role_model=CourseAccessRole,
1681+
user_subject_model=UserSubject,
1682+
course_id_list=None,
1683+
org_id=self.org_id,
1684+
delete_after_migration=False,
1685+
)
1686+
1687+
run = AuthzCourseAuthoringMigrationRun.objects.get(scope_key=self.org_id, status=Status.COMPLETED)
1688+
self.assertEqual(run.metadata.get("success_count"), 1)
1689+
self.assertEqual(run.metadata.get("error_count"), 0)
1690+
self.assertListEqual(
1691+
run.metadata.get("successes"),
1692+
[
1693+
{
1694+
"subject": user.username,
1695+
"role": "instructor",
1696+
"scope": OrgCourseOverviewGlobData.build_external_key(self.org_id),
1697+
}
1698+
],
1699+
)
1700+
1701+
def test_forward_completes_when_no_legacy_rows_for_org_id(self):
1702+
"""FORWARD with ``org_id`` and no matching ``CourseAccessRole`` rows finishes with zero tallies."""
1703+
run_course_authoring_migration(
1704+
migration_type=MigrationType.FORWARD,
1705+
scope_type=ScopeType.ORG,
1706+
scope_key=self.org_id,
1707+
course_access_role_model=CourseAccessRole,
1708+
user_subject_model=UserSubject,
1709+
course_id_list=None,
1710+
org_id=self.org_id,
1711+
delete_after_migration=False,
1712+
)
1713+
1714+
run = AuthzCourseAuthoringMigrationRun.objects.get(scope_key=self.org_id, status=Status.COMPLETED)
1715+
self.assertEqual(run.metadata.get("success_count"), 0)
1716+
self.assertEqual(run.metadata.get("error_count"), 0)
1717+
1718+
def test_rollback_completes_org_glob_assignment_when_org_id(self):
1719+
"""ROLLBACK with ``org_id`` processes org glob / course scopes for that organization."""
1720+
user = User.objects.create_user(username="rcam_org_rb_user", email="[email protected]")
1721+
glob_scope = OrgCourseOverviewGlobData.build_external_key(self.org_id)
1722+
assign_role_to_user_in_scope(user.username, COURSE_STAFF.external_key, glob_scope)
1723+
AuthzEnforcer.get_enforcer().load_policy()
1724+
1725+
run_course_authoring_migration(
1726+
migration_type=MigrationType.ROLLBACK,
1727+
scope_type=ScopeType.ORG,
1728+
scope_key=self.org_id,
1729+
course_access_role_model=CourseAccessRole,
1730+
user_subject_model=UserSubject,
1731+
course_id_list=None,
1732+
org_id=self.org_id,
1733+
delete_after_migration=False,
1734+
)
1735+
1736+
run = AuthzCourseAuthoringMigrationRun.objects.get(scope_key=self.org_id, status=Status.COMPLETED)
1737+
self.assertGreaterEqual(run.metadata.get("success_count"), 1)
1738+
self.assertEqual(run.metadata.get("error_count"), 0)
1739+
self.assertListEqual(
1740+
run.metadata.get("successes"),
1741+
[{"subject": user.username, "role": COURSE_STAFF.external_key, "scope": glob_scope}],
1742+
)
1743+
1744+
def test_skipped_when_org_scope_run_already_running(self):
1745+
"""Concurrent guard applies when ``scope_type`` is ORG and ``scope_key`` is the org id."""
1746+
AuthzCourseAuthoringMigrationRun.create_running(MigrationType.FORWARD, ScopeType.ORG, self.org_id)
1747+
1748+
run_course_authoring_migration(
1749+
migration_type=MigrationType.FORWARD,
1750+
scope_type=ScopeType.ORG,
1751+
scope_key=self.org_id,
1752+
course_access_role_model=CourseAccessRole,
1753+
user_subject_model=UserSubject,
1754+
course_id_list=None,
1755+
org_id=self.org_id,
1756+
delete_after_migration=False,
1757+
)
1758+
1759+
self.assertTrue(
1760+
AuthzCourseAuthoringMigrationRun.objects.filter(scope_key=self.org_id, status=Status.SKIPPED).exists()
1761+
)
1762+
self.assertEqual(AuthzCourseAuthoringMigrationRun.objects.filter(scope_key=self.org_id).count(), 2)

0 commit comments

Comments
 (0)