From 6770b0fd8e49620ffe4edcd588177e2213a368c5 Mon Sep 17 00:00:00 2001 From: James Kachel Date: Thu, 29 Feb 2024 07:43:56 -0600 Subject: [PATCH] Adds the auth variegated viewset and adds system slug to the integrated system --- .../management/commands/generate_test_data.py | 218 ++++++++++++++++++ .../migrations/0002_add_system_slug.py | 17 ++ unified_ecommerce/permissions.py | 20 ++ unified_ecommerce/viewsets.py | 39 ++++ 4 files changed, 294 insertions(+) create mode 100644 system_meta/management/commands/generate_test_data.py create mode 100644 system_meta/migrations/0002_add_system_slug.py create mode 100644 unified_ecommerce/permissions.py create mode 100644 unified_ecommerce/viewsets.py diff --git a/system_meta/management/commands/generate_test_data.py b/system_meta/management/commands/generate_test_data.py new file mode 100644 index 00000000..ea4fa59b --- /dev/null +++ b/system_meta/management/commands/generate_test_data.py @@ -0,0 +1,218 @@ +""" +Adds some test data to the system. This includes three IntegratedSystems with three +products each. + +Ignoring A003 because "help" is valid for argparse. +Ignoring S311 because it's complaining about the faker package. +""" +# ruff: noqa: A003, S311 + +import random +import uuid +from decimal import Decimal + +import faker +from django.core.management import BaseCommand +from django.core.management.base import CommandParser +from django.db import transaction + +from system_meta.models import IntegratedSystem, Product + + +def fake_courseware_id(courseware_type: str, **kwargs) -> str: + """ + Generate a fake courseware id. + + Courseware IDs generally are in the format: + -v1:+(+) + + Type is either "course" or "program", depending on what you specify. School ID is + one of "MITx", "MITxT", "edX", "xPRO", or "Sample". Courseware ID is a set of + numbers: a number < 100, a number < 1000 with a leading zero, and an optional + number < 10, separated by periods. Courseware ID is followed by an "x". This + should be pretty like the IDs that are on MITx Online now (but pretty unlike the + xPRO ones, which usually use a text courseware ID, but that's fine since these + are fake). + + Arguments: + - courseware_type (str): "course" or "program"; the type of + courseware id to generate. + + Keyword Arguments: + - include_run_tag (bool): include the run tag. Defaults to False. + + Returns: + - str: The generated courseware id, in the normal format. + """ + fake = faker.Faker() + + school_id = random.choice(["MITx", "MITxT", "edX", "xPRO", "Sample"]) + courseware_id = f"{random.randint(0, 99)}.{random.randint(0, 999):03d}" + courseware_type = courseware_type.lower() + optional_third_digit = random.randint(0, 9) if fake.boolean() else "" + optional_run_tag = ( + f"+{random.randint(1,3)}T{fake.date_this_decade().year}" + if kwargs["include_run_tag"] + else "" + ) + + return ( + f"{courseware_type}-v1:{school_id}+{courseware_id}" + f"{optional_third_digit}x{optional_run_tag}" + ) + + +class Command(BaseCommand): + """Adds some test data to the system.""" + + def add_arguments(self, parser: CommandParser) -> None: + """Add arguments to the command parser.""" + parser.add_argument( + "--remove", + action="store_true", + help="Remove the test data. This is potentially dangerous.", + ) + + parser.add_argument( + "--only-systems", + action="store_true", + help="Only add test systems.", + ) + + parser.add_argument( + "--only-products", + action="store_true", + help="Only add test products.", + ) + + parser.add_argument( + "--system", + type=str, + help=( + "The name of the system to add products to." + " Only used with --only-products." + ), + nargs="?", + ) + + def add_test_systems(self) -> None: + """Add the test systems.""" + max_systems = 3 + for i in range(1, max_systems + 1): + IntegratedSystem.objects.create( + name=f"Test System {i}", + description=f"Test System {i} description.", + is_active=True, + api_key=uuid.uuid4(), + ) + + def add_test_products(self, system: str) -> None: + """Add the test products to the specified system.""" + + if not IntegratedSystem.objects.filter(name=system).exists(): + self.stdout.write( + self.style.ERROR(f"Integrated system {system} does not exist.") + ) + return + + system = IntegratedSystem.objects.get(name=system) + + for i in range(1, 4): + product_sku = fake_courseware_id("course", include_run_tag=True) + Product.objects.create( + name=f"Test Product {i}", + description=f"Test Product {i} description.", + sku=product_sku, + system=system, + is_active=True, + price=Decimal(random.random() * 10000).quantize(Decimal("0.01")), + system_data={ + "courserun": product_sku, + "program": fake_courseware_id("program"), + }, + ) + + def remove_test_data(self) -> None: + """Remove the test data.""" + + test_systems = ( + IntegratedSystem.all_objects.prefetch_related("products") + .filter(name__startswith="Test System") + .all() + ) + + self.stdout.write( + self.style.WARNING("This command will remove these systems and products:") + ) + + for system in test_systems: + self.stdout.write( + self.style.WARNING(f"System: {system.name} ({system.id})") + ) + + for product in system.products.all(): + self.stdout.write( + self.style.WARNING(f"\tProduct: {product.name} ({product.id})") + ) + + self.stdout.write( + self.style.WARNING( + "This will ACTUALLY DELETE these records." + " Are you sure you want to do this?" + ) + ) + + if input("Type 'yes' to continue: ") != "yes": + self.stdout.write(self.style.ERROR("Aborting.")) + return + + for system in test_systems: + Product.all_objects.filter( + pk__in=[product.id for product in system.products.all()] + ).delete() + IntegratedSystem.all_objects.filter(pk=system.id).delete() + + self.stdout.write(self.style.SUCCESS("Test data removed.")) + + def handle(self, *args, **options) -> None: # noqa: ARG002 + """Handle the command.""" + remove = options["remove"] + only_systems = options["only_systems"] + only_products = options["only_products"] + systems = [options["system"]] if options["system"] else [] + + with transaction.atomic(): + if remove: + self.remove_test_data() + return + + if not only_products: + self.add_test_systems() + + if not only_systems: + if only_products and len(systems) == 0: + self.stdout.write( + self.style.ERROR( + "You must specify a system when using --only-products." + ) + ) + return + else: + systems = [ + system.name + for system in ( + IntegratedSystem.all_objects.filter( + name__startswith="Test System" + ).all() + ) + ] + + [self.add_test_products(system) for system in systems] + return + + if not only_products: + third_test_system = IntegratedSystem.all_objects.filter( + name__startswith="Test System" + ).get() + third_test_system.is_active = False + third_test_system.save(update_fields=("is_active",)) diff --git a/system_meta/migrations/0002_add_system_slug.py b/system_meta/migrations/0002_add_system_slug.py new file mode 100644 index 00000000..779b7331 --- /dev/null +++ b/system_meta/migrations/0002_add_system_slug.py @@ -0,0 +1,17 @@ +# Generated by Django 4.2.8 on 2024-01-11 16:49 + +from django.db import migrations, models + + +class Migration(migrations.Migration): + dependencies = [ + ("system_meta", "0001_add_integrated_system_and_product_models"), + ] + + operations = [ + migrations.AddField( + model_name="integratedsystem", + name="slug", + field=models.CharField(blank=True, max_length=80, null=True, unique=True), + ), + ] diff --git a/unified_ecommerce/permissions.py b/unified_ecommerce/permissions.py new file mode 100644 index 00000000..1c30574b --- /dev/null +++ b/unified_ecommerce/permissions.py @@ -0,0 +1,20 @@ +"""Custom DRF permissions.""" + +from rest_framework import permissions + + +class IsAdminUserOrReadOnly(permissions.BasePermission): + """Determines if the user owns the object""" + + def has_permission(self, request, view): # noqa: ARG002 + """ + Return True if the user is an admin user requesting a write operation, + or if the user is logged in. Otherwise, return False. + """ + + if request.method in permissions.SAFE_METHODS or ( + request.user.is_authenticated and request.user.is_staff + ): + return True + + return False diff --git a/unified_ecommerce/viewsets.py b/unified_ecommerce/viewsets.py new file mode 100644 index 00000000..32de394e --- /dev/null +++ b/unified_ecommerce/viewsets.py @@ -0,0 +1,39 @@ +"""Common viewsets for Unified Ecommerce.""" + +import logging + +from rest_framework import viewsets + +log = logging.getLogger(__name__) + + +class AuthVariegatedModelViewSet(viewsets.ModelViewSet): + """ + Viewset with customizable serializer based on user authentication. + + This bifurcates the ModelViewSet so that if the user is a read-only user (i.e. + not a staff or superuser, or not logged in), they get a separate "read-only" + serializer. Otherwise, we use a regular serializer. The read-only serializer can + then have different fields so you can hide irrelevant data from anonymous users. + + You will need to enforce the read-onlyness of the API yourself; use something like + the IsAuthenticatedOrReadOnly permission class or do something in the serializer. + + Set read_write_serializer_class to the serializer you want to use for admins and + set read_only_serializer_class to the one for regular users. + """ + + read_write_serializer_class = None + read_only_serializer_class = None + + def get_serializer_class(self): + """Get the serializer class for the route.""" + + if hasattr(self, "request") and ( + self.request.user.is_staff or self.request.user.is_superuser + ): + log.debug("get_serializer_class returning the Admin one") + return self.read_write_serializer_class + + log.debug("get_serializer_class returning the regular one") + return self.read_only_serializer_class