Added update order line partner management command.

LEARNER-5608

ecommerce/extensions/order/management/commands/prompt.py

@@ -0,0 +1,33 @@
+import sys
+
+
+def query_yes_no(question, default="yes"):
+    """Ask a yes/no question via raw_input() and return their answer.
+
+    "question" is a string that is presented to the user.
+    "default" is the presumed answer if the user just hits <Enter>.
+        It must be "yes" (the default), "no" or None (meaning
+        an answer is required of the user).
+
+    The "answer" return value is one of "yes" or "no".
+    """
+    valid = {
+        "yes": True,
+        "y": True,
+        "no": False,
+        "n": False,
+    }
+    if default is None or default in valid.keys():
+        prompt = " [y/n] "
+    else:
+        raise ValueError("Invalid default answer: '%s'" % default)
+
+    while True:
+        sys.stdout.write(question + prompt)
+        choice = raw_input().lower()
+        if default is not None and choice == '':
+            return valid[default]
+        elif choice in valid:
+            return valid[choice]
+        else:
+            sys.stdout.write("Please respond with one of the following ({}).\n".format(', '.join(valid.keys())))

ecommerce/extensions/order/management/commands/tests/test_prompt.py

@@ -0,0 +1,45 @@
+import __builtin__
+import sys
+from StringIO import StringIO
+
+import ddt
+from mock import patch
+
+from ecommerce.tests.testcases import TestCase
+
+from ..prompt import query_yes_no
+
+
+@ddt.ddt
+class PromptTests(TestCase):
+    """Tests for prompt."""
+
+    CONFIRMATION_PROMPT = u'Do you want to continue?'
+
+    def test_wrong_default(self):
+        """Test that query_yes_no raises ValueError with wrong default."""
+        with self.assertRaises(ValueError):
+            query_yes_no(self.CONFIRMATION_PROMPT, default='wrong')
+
+    def test_wrong_user_input(self):
+        """Test wrong user input."""
+        out = StringIO()
+        sys.stdout = out
+        with patch.object(__builtin__, 'raw_input', side_effect=['wrong', 'no']):
+            query_yes_no(self.CONFIRMATION_PROMPT)
+            output = out.getvalue().strip()
+            self.assertIn("Please respond with one of the following (y, yes, n, no)", output)
+
+    @patch.object(__builtin__, 'raw_input')
+    @ddt.data(
+        ('yes', True, 'no'), ('no', False, 'yes'), ('', True, 'yes'), ('yes', True, None)
+    )
+    @ddt.unpack
+    def test_query_yes_no(self, user_input, return_value, default, mock_raw_input):
+        """Test that query_yes_no works as expected."""
+        mock_raw_input.return_value = user_input
+        expected_value = query_yes_no(self.CONFIRMATION_PROMPT, default=default)
+        if return_value:
+            self.assertTrue(expected_value)
+        else:
+            self.assertFalse(expected_value)

ecommerce/extensions/order/management/commands/tests/test_update_order_lines_partner.py

@@ -0,0 +1,67 @@
+import ddt
+from django.core.management import call_command
+from django.core.management.base import CommandError
+from mock import patch
+from oscar.core.loading import get_model
+
+from ecommerce.extensions.test.factories import create_order
+from ecommerce.tests.factories import PartnerFactory
+from ecommerce.tests.testcases import TestCase
+
+LOGGER_NAME = 'ecommerce.extensions.order.management.commands.update_order_lines_partner'
+OrderLine = get_model('order', 'Line')
+
+
+@ddt.ddt
+class UpdateOrderLinePartnerTests(TestCase):
+    """Tests for update_order_lines_partner management command."""
+
+    PARTNER_CODE = 'testX'
+    YES_NO_PATCH_LOCATION = 'ecommerce.extensions.order.management.commands.update_order_lines_partner.query_yes_no'
+
+    def assert_error_log(self, error_msg, *args):
+        """Helper to call command and assert error log."""
+        with self.assertRaisesRegexp(CommandError, error_msg):
+            call_command('update_order_lines_partner', *args)
+
+    def test_partner_required(self):
+        """Test that command raises partner required error."""
+        self.assert_error_log(
+            'Error: argument --partner is required',
+            'sku12345'
+        )
+
+    def test_partner_does_not_exist(self):
+        """Test that command raises partner does not exist error."""
+        self.assert_error_log(
+            'No Partner exists for code {}.'.format(self.PARTNER_CODE),
+            'sku12345',
+            '--partner={}'.format(self.PARTNER_CODE)
+        )
+
+    def test_one_or_more_sku_required(self):
+        """Test that command raises one or more SKUs required error."""
+        self.assert_error_log(
+            'update_order_lines_partner requires one or more <SKU>s.',
+            '--partner={}'.format(self.PARTNER_CODE)
+        )
+
+    @ddt.data(True, False)
+    def test_update_order_lines_partner(self, yes_no_value):
+        """Test that command works as expected."""
+        new_partner = PartnerFactory(short_code=self.PARTNER_CODE)
+        order = create_order()
+        order_line = order.lines.first()
+        self.assertNotEqual(order_line.partner, new_partner)
+        with patch(self.YES_NO_PATCH_LOCATION) as mocked_yes_no:
+            mocked_yes_no.return_value = yes_no_value
+            call_command('update_order_lines_partner', order_line.partner_sku, '--partner={}'.format(self.PARTNER_CODE))
+            order_line = OrderLine.objects.get(partner_sku=order_line.partner_sku)
+            if yes_no_value:
+                # Verify that partner is updated
+                self.assertEqual(order_line.partner, new_partner)
+                self.assertEqual(order_line.partner_name, new_partner.name)
+            else:
+                # Verify that partner is not updated
+                self.assertNotEqual(order_line.partner, new_partner)
+                self.assertNotEqual(order_line.partner_name, new_partner.name)

ecommerce/extensions/order/management/commands/update_order_lines_partner.py

@@ -0,0 +1,63 @@
+from __future__ import unicode_literals
+
+import logging
+from textwrap import dedent
+
+from django.core.management.base import BaseCommand, CommandError
+from oscar.core.loading import get_model
+
+from .prompt import query_yes_no
+
+logger = logging.getLogger(__name__)
+OrderLine = get_model('order', 'Line')
+Partner = get_model('partner', 'Partner')
+
+
+class Command(BaseCommand):
+    """
+    Command to update order lines partner.
+
+    Example:
+
+        ./manage.py update_order_lines_partner <SKU 1> <SKU 2> ... --partner edX
+    """
+    help = dedent(__doc__)
+    CONFIRMATION_PROMPT = u"You're going to update {count} order lines. Do you want to continue?"
+
+    def add_arguments(self, parser):
+        parser.add_argument('skus',
+                            type=str,
+                            nargs='*',
+                            metavar='SKU',
+                            help='SKUs corresponding to the product for which order lines will be updated.')
+        parser.add_argument('--partner',
+                            action='store',
+                            dest='partner',
+                            type=str,
+                            required=True,
+                            help='Partner code to be updated.')
+
+    def handle(self, *args, **options):
+        skus = options['skus']
+        partner_code = options['partner']
+
+        if not len(skus):
+            msg = 'update_order_lines_partner requires one or more <SKU>s.'
+            logger.exception(msg)
+            raise CommandError(msg)
+
+        try:
+            partner = Partner.objects.get(short_code__iexact=partner_code)
+        except Partner.DoesNotExist:
+            msg = 'No Partner exists for code {}.'.format(partner_code)
+            logger.exception(msg)
+            raise CommandError(msg)
+
+        order_lines = OrderLine.objects.filter(partner_sku__in=skus).exclude(partner=partner)
+        count = len(order_lines)
+        if query_yes_no(self.CONFIRMATION_PROMPT.format(count=count), default="no"):
+            order_lines.update(partner=partner, partner_name=partner.name)
+            logger.info('%d order lines updated.', count)
+        else:
+            logger.info('Operation canceled.')
+            return