Add mgmt command for creating catalog integrations

openedx/core/djangoapps/catalog/management/commands/create_catalog_integrations.py

@@ -0,0 +1,75 @@
+'''CatalogIntegration management command'''
+
+from django.core.management import BaseCommand, CommandError
+from openedx.core.djangoapps.catalog.models import CatalogIntegration
+
+
+class Command(BaseCommand):
+    """Management command used to create catalog integrations."""
+    help = "Create catalog integration record in LMS"
+
+    def add_arguments(self, parser):
+        parser.add_argument(
+            '--enabled',
+            dest='enabled',
+            action='store_true',
+            help='Enable the catalog integration'
+        )
+
+        parser.add_argument(
+            '--internal_api_url',
+            help='The url of the internal api',
+            required=True
+        )
+
+        parser.add_argument(
+            '--service_username',
+            help='The username for the service',
+            required=True
+        )
+        parser.add_argument(
+            '--cache_ttl',
+            type=int,
+            default=0,
+            help='Enable caching of API responses by setting this to a value greater than 0 (seconds)'
+        )
+        parser.add_argument(
+            '--long_term_cache_ttl',
+            type=int,
+            default=86400,
+            help='Enable long term caching of API responses by setting this to a value greater than 0 (seconds)'
+        )
+        parser.add_argument(
+            '--page_size',
+            type=int,
+            default=100,
+            help='Maximum number of records in paginated response of a single request to catalog service'
+        )
+
+    def handle(self, *args, **options):
+
+        enabled = options.get('enabled')
+        internal_api_url = options.get('internal_api_url')
+        service_username = options.get('service_username')
+        cache_ttl = options.get('cache_ttl')
+        long_term_cache_ttl = options.get('long_term_cache_ttl')
+        page_size = options.get('page_size')
+
+        try:
+            catalog_integration = CatalogIntegration.objects.create(
+                enabled=enabled,
+                internal_api_url=internal_api_url,
+                service_username=service_username,
+                cache_ttl=cache_ttl,
+                long_term_cache_ttl=long_term_cache_ttl,
+                page_size=page_size
+            )
+        except Exception as err:
+            raise CommandError('Error creating CatalogIntegration: {}'.format(err))
+
+        self.stdout.write(self.style.SUCCESS(
+            'Successfully created CatalogIntegration enabled={} url={} service_username={}').format(
+                catalog_integration.enabled,
+                catalog_integration.internal_api_url,
+                catalog_integration.service_username
+        ))

openedx/core/djangoapps/catalog/management/commands/tests/test_create_catalog_integrations.py

@@ -0,0 +1,144 @@
+"""
+Test cases for catalog_integrations command.
+"""
+from django.test import TestCase
+from django.core.management import call_command, CommandError
+
+from openedx.core.djangoapps.catalog.models import CatalogIntegration
+from openedx.core.djangoapps.catalog.tests.mixins import CatalogIntegrationMixin
+
+
+class TestCreateCatalogIntegrations(CatalogIntegrationMixin, TestCase):
+    """ Test the create_catalog_integrations command """
+
+    def test_without_required(self):
+        ''' Test that required values are supplied '''
+
+        # test without service_username
+        with self.assertRaises(CommandError):
+            call_command(
+                "create_catalog_integrations",
+                "--internal_api_url", self.catalog_integration_defaults['internal_api_url'],
+            )
+
+        # test without internal_api_url
+        with self.assertRaises(CommandError):
+            call_command(
+                "create_catalog_integrations",
+                "--service_username", self.catalog_integration_defaults['service_username'],
+            )
+
+    def test_with_required(self):
+        ''' Test with required arguments supplied'''
+
+        initial = CatalogIntegration.current()
+
+        # test with both required args
+        call_command(
+            "create_catalog_integrations",
+            "--internal_api_url", self.catalog_integration_defaults['internal_api_url'],
+            "--service_username", self.catalog_integration_defaults['service_username']
+        )
+
+        current = CatalogIntegration.current()
+
+        # assert current has changed
+        self.assertNotEqual(
+            initial,
+            current
+        )
+
+        self.assertEqual(
+            current.enabled,
+            False
+        )
+        self.assertEqual(
+            current.internal_api_url,
+            self.catalog_integration_defaults['internal_api_url']
+        )
+
+        self.assertEqual(
+            current.service_username,
+            self.catalog_integration_defaults['service_username']
+        )
+
+    def test_with_optional(self):
+        ''' Test with optionals arguments supplied'''
+        initial = CatalogIntegration.current()
+
+        # test --enabled
+        call_command(
+            "create_catalog_integrations",
+            "--internal_api_url", self.catalog_integration_defaults['internal_api_url'],
+            "--service_username", self.catalog_integration_defaults['service_username'],
+            "--enabled"
+        )
+
+        current = CatalogIntegration.current()
+
+        # assert current has changed
+        self.assertNotEqual(
+            initial,
+            current
+        )
+
+        self.assertEqual(
+            current.enabled,
+            True
+        )
+        self.assertEqual(
+            current.internal_api_url,
+            self.catalog_integration_defaults['internal_api_url']
+        )
+
+        self.assertEqual(
+            current.service_username,
+            self.catalog_integration_defaults['service_username']
+        )
+
+        # test with all args
+        call_command(
+            "create_catalog_integrations",
+            "--internal_api_url", self.catalog_integration_defaults['internal_api_url'],
+            "--service_username", self.catalog_integration_defaults['service_username'],
+            "--enabled",
+            "--cache_ttl", 500,
+            "--long_term_cache_ttl", 500,
+            "--page_size", 500
+        )
+
+        current = CatalogIntegration.current()
+
+        # assert current has changed
+        self.assertNotEqual(
+            initial,
+            current
+        )
+
+        self.assertEqual(
+            current.enabled,
+            True
+        )
+        self.assertEqual(
+            current.internal_api_url,
+            self.catalog_integration_defaults['internal_api_url']
+        )
+
+        self.assertEqual(
+            current.service_username,
+            self.catalog_integration_defaults['service_username']
+        )
+
+        self.assertEqual(
+            current.cache_ttl,
+            500
+        )
+
+        self.assertEqual(
+            current.long_term_cache_ttl,
+            500
+        )
+        self.assertEqual(
+            current.page_size,
+            500
+        )