diff --git a/openedx_authz/management/commands/data/sandbox_seed_data.json b/openedx_authz/management/commands/data/sandbox_seed_data.json new file mode 100644 index 00000000..dc7833d1 --- /dev/null +++ b/openedx_authz/management/commands/data/sandbox_seed_data.json @@ -0,0 +1,37 @@ +{ + "organizations": [ + {"name": "OpenedX Sandbox Org", "short_name": "OpenedX"} + ], + "users": [ + { + "username": "authz_platform_admin", + "email": "authz_platform_admin@example.com", + "roles": [{"role": "course_admin", "scope": "course-v1:*"}] + }, + { + "username": "authz_org_admin", + "email": "authz_org_admin@example.com", + "roles": [{"role": "course_admin", "scope": "course-v1:OpenedX+*"}] + }, + { + "username": "authz_course_staff", + "email": "authz_course_staff@example.com", + "roles": [{"role": "course_staff", "scope": "course-v1:OpenedX+DemoX+DemoCourse"}] + }, + { + "username": "authz_course_editor", + "email": "authz_course_editor@example.com", + "roles": [{"role": "course_editor", "scope": "course-v1:OpenedX+DemoX+DemoCourse"}] + }, + { + "username": "authz_course_auditor", + "email": "authz_course_auditor@example.com", + "roles": [{"role": "course_auditor", "scope": "course-v1:OpenedX+DemoX+DemoCourse"}] + }, + { + "username": "authz_library_admin", + "email": "authz_library_admin@example.com", + "roles": [{"role": "library_admin", "scope": "lib:OpenedX:*"}] + } + ] +} diff --git a/openedx_authz/management/commands/seed_sandbox_data.py b/openedx_authz/management/commands/seed_sandbox_data.py new file mode 100644 index 00000000..d063f47c --- /dev/null +++ b/openedx_authz/management/commands/seed_sandbox_data.py @@ -0,0 +1,116 @@ +"""Django management command to seed the Sandbox environment with deterministic test data. + +Reads a JSON fixture describing organizations, users, and authz role assignments and +inserts it idempotently (get_or_create), so the command is safe to run repeatedly +against a Sandbox database without creating duplicates. + +Example usage: + python manage.py lms seed_sandbox_data + python manage.py lms seed_sandbox_data --data-file /path/to/custom.json + python manage.py lms seed_sandbox_data --reset +""" + +import json +import logging +import os + +from django.contrib.auth import get_user_model +from django.core.management.base import BaseCommand, CommandError +from organizations.api import add_organization, get_organizations + +from openedx_authz import ROOT_DIRECTORY +from openedx_authz.api.users import assign_role_to_user_in_scope +from openedx_authz.engine.enforcer import AuthzEnforcer + +log = logging.getLogger(__name__) + +DEFAULT_DATA_FILE = os.path.join(ROOT_DIRECTORY, "management", "commands", "data", "sandbox_seed_data.json") +DEFAULT_PASSWORD = "edx" + + +class Command(BaseCommand): + """Seed the Sandbox environment with a base set of orgs, users, and authz role assignments.""" + + help = "Seed the Sandbox environment with orgs, users, and authz role assignments from a JSON fixture." + + def add_arguments(self, parser) -> None: + parser.add_argument( + "--data-file", + type=str, + default=None, + help="Path to the JSON seed data file (default: bundled sandbox_seed_data.json).", + ) + parser.add_argument( + "--reset", + action="store_true", + help="Delete previously seeded users (identified by username) before seeding.", + ) + + def handle(self, *args, **options): + data_file = options["data_file"] or DEFAULT_DATA_FILE + with open(data_file, encoding="utf-8") as fh: + seed_data = json.load(fh) + + user_model = get_user_model() + usernames = [user["username"] for user in seed_data.get("users", [])] + + if options["reset"]: + deleted, _ = user_model.objects.filter(username__in=usernames).delete() + self.stdout.write(self.style.WARNING(f"Removed {deleted} previously seeded record(s).")) + + counts = {"created": 0, "skipped": 0, "failed": 0} + self._seed_organizations(seed_data.get("organizations", []), counts) + self._seed_users(seed_data.get("users", []), user_model, counts) + + AuthzEnforcer.get_enforcer().load_policy() + + summary = "Seeding complete: {created} created, {skipped} skipped, {failed} failed.".format(**counts) + if counts["failed"]: + self.stdout.write(self.style.ERROR(summary)) + raise CommandError(f"{counts['failed']} seed item(s) failed, see logs above for details.") + self.stdout.write(self.style.SUCCESS(summary)) + + def _seed_organizations(self, organizations, counts): + """Create any organization from the fixture that doesn't already exist, tallying counts.""" + existing = {org["short_name"] for org in get_organizations()} + for org in organizations: + if org["short_name"] in existing: + counts["skipped"] += 1 + continue + try: + add_organization(org) + counts["created"] += 1 + # One bad organization shouldn't stop the rest of the fixture from seeding. + except Exception: # pylint: disable=broad-exception-caught + log.exception("Failed to create organization %s", org.get("short_name")) + counts["failed"] += 1 + + def _seed_users(self, users, user_model, counts): + """Create/update each user from the fixture and assign their roles, tallying counts.""" + for user_data in users: + username = user_data["username"] + user, was_created = user_model.objects.get_or_create( + username=username, + defaults={"email": user_data.get("email", f"{username}@example.com")}, + ) + if was_created: + user.set_password(user_data.get("password", DEFAULT_PASSWORD)) + user.is_staff = user_data.get("is_staff", False) + user.is_superuser = user_data.get("is_superuser", False) + user.save() + counts["created"] += 1 + else: + counts["skipped"] += 1 + + for assignment in user_data.get("roles", []): + try: + assign_role_to_user_in_scope(username, assignment["role"], assignment["scope"]) + # One bad role assignment shouldn't stop the rest of the fixture from seeding. + except Exception: # pylint: disable=broad-exception-caught + log.exception( + "Failed to assign role %s to %s in scope %s", + assignment["role"], + username, + assignment["scope"], + ) + counts["failed"] += 1 diff --git a/openedx_authz/tests/test_commands.py b/openedx_authz/tests/test_commands.py index 8d270571..b0bc2917 100644 --- a/openedx_authz/tests/test_commands.py +++ b/openedx_authz/tests/test_commands.py @@ -3,17 +3,23 @@ """ import io +import json +import os from tempfile import NamedTemporaryFile from unittest import TestCase from unittest.mock import Mock, patch from ddt import data, ddt +from django.contrib.auth import get_user_model from django.core.management import call_command from django.core.management.base import CommandError +from django.test import TestCase as DjangoTestCase +from organizations.models import Organization from openedx_authz import ROOT_DIRECTORY from openedx_authz import api as authz_api from openedx_authz.api.data import ContentLibraryData +from openedx_authz.api.users import get_user_role_assignments_filtered from openedx_authz.constants.permissions import ( DELETE_LIBRARY, MANAGE_LIBRARY_TEAM, @@ -509,3 +515,97 @@ def test_handle_clear_existing_both_denied(self, mock_confirm, mock_casbin_enfor command._delete_existing_roles.assert_not_called() command._delete_permissions_inheritance.assert_not_called() command.migrate_policies.assert_called_once_with(mock_source_enforcer, mock_target_enforcer) + + +class SeedSandboxDataCommandTests(DjangoTestCase): + """ + Tests for the `seed_sandbox_data` Django management command. + + Uses a real database (organizations, users, and the Casbin adapter) since the + command's whole point is idempotent creation of that data, not something a mock + could verify. + """ + + def setUp(self): + super().setUp() + self.buffer = io.StringIO() + self.username = "seed_test_user" + seed_data = { + "organizations": [{"name": "Seed Test Org", "short_name": "SeedTestOrg"}], + "users": [ + { + "username": self.username, + "email": "seed_test_user@example.com", + "roles": [{"role": LIBRARY_ADMIN.external_key, "scope": "lib:SeedTestOrg:*"}], + } + ], + } + self.data_file = NamedTemporaryFile(mode="w", suffix=".json", delete=False) + json.dump(seed_data, self.data_file) + self.data_file.close() + self.addCleanup(os.remove, self.data_file.name) + + def _call(self, **options): + call_command("seed_sandbox_data", data_file=self.data_file.name, stdout=self.buffer, **options) + + def test_seed_creates_org_user_and_role_assignment(self): + """Seeding creates the organization, user, and role assignment from the fixture.""" + self._call() + + user = get_user_model().objects.get(username=self.username) + assert user.check_password("edx") + assert Organization.objects.filter(short_name="SeedTestOrg").exists() + + assignments = get_user_role_assignments_filtered(user_external_key=self.username) + assert any(role.external_key == LIBRARY_ADMIN.external_key for a in assignments for role in a.roles) + assert "Seeding complete: 2 created, 0 skipped, 0 failed." in self.buffer.getvalue() + + def test_seed_is_idempotent(self): + """Running the command twice does not duplicate the org, user, or role assignment.""" + self._call() + self.buffer = io.StringIO() + self._call() + + assert get_user_model().objects.filter(username=self.username).count() == 1 + assert Organization.objects.filter(short_name="SeedTestOrg").count() == 1 + assignments = get_user_role_assignments_filtered(user_external_key=self.username) + matches = [role for a in assignments for role in a.roles if role.external_key == LIBRARY_ADMIN.external_key] + assert len(matches) == 1 + assert "Seeding complete: 0 created, 2 skipped, 0 failed." in self.buffer.getvalue() + + def test_reset_removes_previously_seeded_user(self): + """--reset deletes users from the fixture before re-seeding them.""" + self._call() + user_id = get_user_model().objects.get(username=self.username).id + + self.buffer = io.StringIO() + self._call(reset=True) + + new_user = get_user_model().objects.get(username=self.username) + assert new_user.id != user_id + + @patch("openedx_authz.management.commands.seed_sandbox_data.add_organization") + def test_seed_reports_and_raises_on_organization_failure(self, mock_add_organization): + """A failed organization creation is counted, logged, and turns into a CommandError.""" + mock_add_organization.side_effect = ValueError("boom") + + with self.assertRaises(CommandError) as ctx: + self._call() + + assert "1 seed item(s) failed" in str(ctx.exception) + assert "Seeding complete: 1 created, 0 skipped, 1 failed." in self.buffer.getvalue() + assert not Organization.objects.filter(short_name="SeedTestOrg").exists() + + @patch("openedx_authz.management.commands.seed_sandbox_data.assign_role_to_user_in_scope") + def test_seed_reports_and_raises_on_role_assignment_failure(self, mock_assign_role): + """A failed role assignment is counted, logged, and turns into a CommandError.""" + mock_assign_role.side_effect = ValueError("boom") + + with self.assertRaises(CommandError) as ctx: + self._call() + + assert "1 seed item(s) failed" in str(ctx.exception) + assert "Seeding complete: 2 created, 0 skipped, 1 failed." in self.buffer.getvalue() + assert get_user_model().objects.filter(username=self.username).exists() + assignments = get_user_role_assignments_filtered(user_external_key=self.username) + assert not any(role.external_key == LIBRARY_ADMIN.external_key for a in assignments for role in a.roles)