feat: add seed_sandbox_data management command

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:*"}]
+    }
+  ]
+}

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

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)