addons-server/mkt/submit/tests/test_views.py

import json
import os

from django.conf import settings

import mock
from nose.tools import eq_
from nose import SkipTest
from pyquery import PyQuery as pq
import waffle

import amo
import amo.tests
from amo.tests import formset, initial
from amo.urlresolvers import reverse
from addons.models import (Addon, AddonCategory, AddonDeviceType, AddonUser,
                           Category, DeviceType)
from addons.utils import ReverseNameLookup
from files.tests.test_models import UploadTest as BaseUploadTest
from market.models import Price
import mkt
from mkt.submit.models import AppSubmissionChecklist
from translations.models import Translation
from users.models import UserProfile
from webapps.models import Webapp


class TestSubmit(amo.tests.TestCase):
    fixtures = ['base/users']

    def setUp(self):
        self.user = self.get_user()
        assert self.client.login(username=self.user.email, password='password')

    def get_user(self):
        return UserProfile.objects.get(username='regularuser')

    def _test_anonymous(self):
        self.client.logout()
        r = self.client.get(self.url, follow=True)
        self.assertLoginRedirects(r, self.url)

    def _test_progress_display(self, completed, current):
        """Test that the correct steps are highlighted."""
        r = self.client.get(self.url)
        progress = pq(r.content)('#submission-progress')

        # Check the completed steps.
        completed_found = progress.find('.completed')
        for idx, step in enumerate(completed):
            li = completed_found.eq(idx)
            eq_(li.text(), unicode(mkt.APP_STEPS_TITLE[step]))

        # Check the current step.
        eq_(progress.find('.current').text(),
            unicode(mkt.APP_STEPS_TITLE[current]))


class TestTerms(TestSubmit):
    fixtures = ['base/users']

    def setUp(self):
        super(TestTerms, self).setUp()
        self.url = reverse('submit.app.terms')

    def test_anonymous(self):
        self.client.logout()
        r = self.client.get(self.url, follow=True)
        self.assertLoginRedirects(r, self.url)

    def test_jump_to_step(self):
        r = self.client.get(reverse('submit.app'), follow=True)
        self.assertRedirects(r, self.url)

    def test_page(self):
        r = self.client.get(self.url)
        eq_(r.status_code, 200)
        eq_(pq(r.content)('#submit-terms').length, 1)

    def test_progress_display(self):
        self._test_progress_display([], 'terms')

    def test_agree(self):
        r = self.client.post(self.url, {'read_dev_agreement': True})
        self.assertRedirects(r, reverse('submit.app.manifest'))
        eq_(self.get_user().read_dev_agreement, True)

    def test_disagree(self):
        r = self.client.post(self.url)
        eq_(r.status_code, 200)
        eq_(self.user.read_dev_agreement, False)


class TestManifest(TestSubmit):
    fixtures = ['base/users']

    def setUp(self):
        super(TestManifest, self).setUp()
        self.url = reverse('submit.app.manifest')

    def _step(self):
        self.user.update(read_dev_agreement=True)

    def test_anonymous(self):
        self._test_anonymous()

    def test_cannot_skip_prior_step(self):
        r = self.client.get(self.url, follow=True)
        # And we start back at one...
        self.assertRedirects(r, reverse('submit.app.terms'))

    def test_jump_to_step(self):
        # I already read the Terms.
        self._step()
        # So jump me to the Manifest step.
        r = self.client.get(reverse('submit.app'), follow=True)
        self.assertRedirects(r, reverse('submit.app.manifest'))

    def test_page(self):
        self._step()
        r = self.client.get(self.url)
        eq_(r.status_code, 200)
        eq_(pq(r.content)('#submit-manifest').length, 1)

    def test_progress_display(self):
        self._step()
        self._test_progress_display(['terms'], 'manifest')


class UploadAddon(object):

    def post(self, desktop_platforms=[amo.PLATFORM_ALL], mobile_platforms=[],
             expect_errors=False):
        d = dict(upload=self.upload.pk,
                 desktop_platforms=[p.id for p in desktop_platforms],
                 mobile_platforms=[p.id for p in mobile_platforms])
        r = self.client.post(self.url, d, follow=True)
        eq_(r.status_code, 200)
        if not expect_errors:
            # Show any unexpected form errors.
            if r.context and 'form' in r.context:
                eq_(r.context['form'].errors, {})
        return r


class BaseWebAppTest(BaseUploadTest, UploadAddon, amo.tests.TestCase):
    fixtures = ['base/apps', 'base/users', 'base/platforms']

    def setUp(self):
        super(BaseWebAppTest, self).setUp()
        waffle.models.Flag.objects.create(name='accept-webapps', everyone=True)
        self.manifest = os.path.join(settings.ROOT, 'mkt', 'submit', 'tests',
                                     'webapps', 'mozball.webapp')
        self.upload = self.get_upload(abspath=self.manifest)
        self.url = reverse('submit.app.manifest')
        assert self.client.login(username='regular@mozilla.com',
                                 password='password')
        # Complete first step.
        self.client.post(reverse('submit.app.terms'),
                         {'read_dev_agreement': True})

    def post_addon(self):
        eq_(Addon.objects.count(), 0)
        self.post()
        return Addon.objects.get()


class TestCreateWebApp(BaseWebAppTest):

    def test_page_title(self):
        eq_(pq(self.client.get(self.url).content)('title').text(),
            'App Manifest | Developer Hub | Mozilla Marketplace')

    def test_post_app_redirect(self):
        r = self.post()
        webapp = Webapp.objects.get()
        self.assertRedirects(r,
            reverse('submit.app.details', args=[webapp.app_slug]))

    def test_app_from_uploaded_manifest(self):
        addon = self.post_addon()
        eq_(addon.type, amo.ADDON_WEBAPP)
        eq_(addon.guid, None)
        eq_(unicode(addon.name), 'MozillaBall')
        eq_(addon.slug, 'app-%s' % addon.id)
        eq_(addon.app_slug, 'mozillaball')
        eq_(addon.summary, u'Exciting Open Web development action!')
        eq_(Translation.objects.get(id=addon.summary.id, locale='it'),
            u'Azione aperta emozionante di sviluppo di fotoricettore!')

    def test_version_from_uploaded_manifest(self):
        addon = self.post_addon()
        eq_(addon.current_version.version, '1.0')

    def test_file_from_uploaded_manifest(self):
        addon = self.post_addon()
        files = addon.current_version.files.all()
        eq_(len(files), 1)
        eq_(files[0].status, amo.STATUS_PUBLIC)


class TestCreateWebAppFromManifest(BaseWebAppTest):

    def setUp(self):
        super(TestCreateWebAppFromManifest, self).setUp()
        Webapp.objects.create(app_slug='xxx', app_domain='existing-app.com')

    def upload_webapp(self, manifest_url, **post_kw):
        self.upload.update(name=manifest_url)  # Simulate JS upload.
        return self.post(**post_kw)

    def post_manifest(self, manifest_url):
        rs = self.client.post(reverse('mkt.developers.upload_manifest'),
                              dict(manifest=manifest_url))
        if 'json' in rs['content-type']:
            rs = json.loads(rs.content)
        return rs

    @mock.patch.object(settings, 'WEBAPPS_UNIQUE_BY_DOMAIN', True)
    def test_duplicate_domain(self):
        rs = self.upload_webapp('http://existing-app.com/my.webapp',
                                expect_errors=True)
        eq_(rs.context['form'].errors,
            {'upload':
             ['An app already exists on this domain; only one '
              'app per domain is allowed.']})

    @mock.patch.object(settings, 'WEBAPPS_UNIQUE_BY_DOMAIN', False)
    def test_allow_duplicate_domains(self):
        self.upload_webapp('http://existing-app.com/my.webapp')  # No errors.

    @mock.patch.object(settings, 'WEBAPPS_UNIQUE_BY_DOMAIN', True)
    def test_duplicate_domain_from_js(self):
        data = self.post_manifest('http://existing-app.com/my.webapp')
        eq_(data['validation']['errors'], 1)
        eq_(data['validation']['messages'][0]['message'],
            'An app already exists on this domain; '
            'only one app per domain is allowed.')

    @mock.patch.object(settings, 'WEBAPPS_UNIQUE_BY_DOMAIN', False)
    def test_allow_duplicate_domains_from_js(self):
        rs = self.post_manifest('http://existing-app.com/my.webapp')
        eq_(rs.status_code, 302)


class TestDetails(TestSubmit):
    fixtures = ['base/apps', 'base/users', 'webapps/337141-steamcube']

    def setUp(self):
        super(TestDetails, self).setUp()
        self.webapp = self.get_webapp()
        self.url = reverse('submit.app.details', args=[self.webapp.app_slug])
        waffle.models.Flag.objects.create(name='accept-webapps', everyone=True)

    def get_webapp(self):
        return Webapp.objects.get(id=337141)

    def _step(self):
        self.user.update(read_dev_agreement=True)
        self.cl = AppSubmissionChecklist.objects.create(addon=self.webapp,
            terms=True, manifest=True)

        # Associate app with user.
        AddonUser.objects.create(addon=self.webapp, user=self.user)

        # Associate device type with app.
        self.dtype = DeviceType.objects.create(name='fligphone')
        AddonDeviceType.objects.create(addon=self.webapp,
                                       device_type=self.dtype)

        # Associate category with app.
        self.cat1 = Category.objects.create(type=amo.ADDON_WEBAPP, name='Fun')
        AddonCategory.objects.create(addon=self.webapp, category=self.cat1)

        # Create initial formset for categories.
        ctx = self.client.get(self.url).context['form_cats']
        self.cat_initial = amo.tests.initial(ctx.initial_forms[0])

    def test_anonymous(self):
        self._test_anonymous()

    def test_resume_step(self):
        self._step()
        payments_url = reverse('submit.app.payments',
                               args=[self.webapp.app_slug])
        r = self.client.get(payments_url, follow=True)
        self.assertRedirects(r, reverse('submit.app.details',
                                        args=[self.webapp.app_slug]))

    def test_not_owner(self):
        self._step()
        assert self.client.login(username='clouserw@gmail.com',
                                 password='password')
        eq_(self.client.get(self.url).status_code, 403)

    def test_page(self):
        self._step()
        r = self.client.get(self.url)
        eq_(r.status_code, 200)
        eq_(pq(r.content)('#submit-details').length, 1)

    def test_progress_display(self):
        self._step()
        self._test_progress_display(['terms', 'manifest'], 'details')

    def new_preview_formset(self, *args, **kw):
        ctx = self.client.get(self.url).context

        blank = initial(ctx['form_previews'].forms[-1])
        blank.update(**kw)
        return blank

    def preview_formset(self, *args, **kw):
        kw.setdefault('initial_count', 0)
        kw.setdefault('prefix', 'files')

        fs = formset(*[a for a in args] + [self.new_preview_formset()], **kw)
        return dict([(k, '' if v is None else v) for k, v in fs.items()])

    def get_dict(self, **kw):
        data = {
            'name': 'Test name',
            'slug': 'testname',
            'summary': 'Hello!',
            'description': 'desc',
            'privacy_policy': 'XXX <script>alert("xss")</script>',
            'device_types': [self.dtype.id],
        }
        # Add the required screenshot.
        data.update(self.preview_formset({
            'upload_hash': '<hash>',
            'position': 0
        }))
        data.update(**kw)
        # Build formset for categories.
        data.update(amo.tests.formset(self.cat_initial, initial_count=1))
        # Remove fields without values.
        data = dict((k, v) for k, v in data.iteritems() if v is not None)
        return data

    def check_dict(self, data=None, expected=None):
        if data is None:
            data = self.get_dict()
        addon = self.get_webapp()

        # Build a dictionary of expected results.
        expected = {
            'name': 'Test name',
            'app_slug': 'testname',
            'summary': 'Hello!',
            'description': 'desc',
            'privacy_policy': 'XXX &lt;script&gt;alert("xss")&lt;/script&gt;',
        }
        expected.update(expected)

        for field, expected in expected.iteritems():
            got = unicode(getattr(addon, field))
            eq_(got, expected,
                'Expected %r for %r. Got %r.' % (expected, field, got))
        eq_(list(addon.device_types), [self.dtype])

    def test_success(self):
        self._step()
        data = self.get_dict()
        # Post and be redirected.
        r = self.client.post(self.url, data)
        self.assertNoFormErrors(r)
        # TODO: Assert redirects when we go to next step.
        self.check_dict(data=data)

    def _setup_other_webapp(self):
        self._step()
        # Generate another webapp to test name uniqueness.
        app = amo.tests.addon_factory(type=amo.ADDON_WEBAPP, name='Cool App')
        eq_(ReverseNameLookup(webapp=True).get(app.name), app.id)

    def test_name_unique(self):
        self._setup_other_webapp()
        r = self.client.post(self.url, self.get_dict(name='Cool App'))
        error = 'This name is already in use. Please choose another.'
        self.assertFormError(r, 'form_basic', 'name', error)

    def test_name_unique_strip(self):
        # Make sure we can't sneak in a name by adding a space or two.
        self._setup_other_webapp()
        r = self.client.post(self.url, self.get_dict(name='  Cool App  '))
        error = 'This name is already in use. Please choose another.'
        self.assertFormError(r, 'form_basic', 'name', error)

    def test_name_unique_case(self):
        # Make sure unique names aren't case sensitive.
        self._setup_other_webapp()
        r = self.client.post(self.url, self.get_dict(name='cool app'))
        error = 'This name is already in use. Please choose another.'
        self.assertFormError(r, 'form_basic', 'name', error)

    def test_name_required(self):
        self._step()
        r = self.client.post(self.url, self.get_dict(name=''))
        eq_(r.status_code, 200)
        self.assertFormError(r, 'form_basic', 'name',
                             'This field is required.')

    def test_screenshot_required(self):
        self._step()
        data = self.get_dict()
        for k in data:
            if k.startswith('files') and k.endswith('upload_hash'):
                data[k] = ''
        rp = self.client.post(self.url, data)
        eq_(rp.context['form_previews'].non_form_errors(),
            ['You must upload at least one screen shot.'])

    def test_name_length(self):
        self._step()
        r = self.client.post(self.url, self.get_dict(name='a' * 129))
        eq_(r.status_code, 200)
        self.assertFormError(r, 'form_basic', 'name',
            'Ensure this value has at most 128 characters (it has 129).')

    def test_slug_invalid(self):
        self._step()
        # Submit an invalid slug.
        d = self.get_dict(slug='slug!!! aksl23%%')
        r = self.client.post(self.url, d)
        eq_(r.status_code, 200)
        self.assertFormError(r, 'form_basic', 'slug',
            "Enter a valid 'slug' consisting of letters, numbers, underscores "
            "or hyphens.")

    def test_slug_required(self):
        self._step()
        r = self.client.post(self.url, self.get_dict(slug=''))
        eq_(r.status_code, 200)
        self.assertFormError(r, 'form_basic', 'slug',
                             'This field is required.')

    def test_summary_required(self):
        self._step()
        r = self.client.post(self.url, self.get_dict(summary=''))
        eq_(r.status_code, 200)
        self.assertFormError(r, 'form_basic', 'summary',
                             'This field is required.')

    def test_summary_length(self):
        self._step()
        r = self.client.post(self.url, self.get_dict(summary='a' * 251))
        eq_(r.status_code, 200)
        self.assertFormError(r, 'form_basic', 'summary',
            'Ensure this value has at most 250 characters (it has 251).')

    def test_description_optional(self):
        self._step()
        r = self.client.post(self.url, self.get_dict(description=None))
        self.assertNoFormErrors(r)

    def test_privacy_policy_required(self):
        self._step()
        r = self.client.post(self.url, self.get_dict(privacy_policy=None))
        self.assertFormError(r, 'form_basic', 'privacy_policy',
                             'This field is required.')

    def test_device_types_required(self):
        self._step()
        r = self.client.post(self.url, self.get_dict(device_types=None))
        self.assertFormError(r, 'form_devices', 'device_types',
                          'This field is required.')

    def test_device_types_invalid(self):
        self._step()
        r = self.client.post(self.url, self.get_dict(device_types='999'))
        self.assertFormError(r, 'form_devices', 'device_types',
            'Select a valid choice. 999 is not one of the available choices.')

    def test_categories_required(self):
        self._step()
        del self.cat_initial['categories']
        r = self.client.post(self.url, self.get_dict())
        eq_(r.context['form_cats'].errors[0]['categories'],
            ['This field is required.'])

    def test_categories_max(self):
        self._step()
        eq_(amo.MAX_CATEGORIES, 2)
        cat2 = Category.objects.create(type=amo.ADDON_WEBAPP, name='bling')
        cat3 = Category.objects.create(type=amo.ADDON_WEBAPP, name='blang')
        self.cat_initial['categories'] = [self.cat1.id, cat2.id, cat3.id]
        r = self.client.post(self.url, self.get_dict())
        eq_(r.context['form_cats'].errors[0]['categories'],
            ['You can have only 2 categories.'])

    def _post_cats(self, cats):
        self.cat_initial['categories'] = [c.id for c in cats]
        self.client.post(self.url, self.get_dict())
        eq_(sorted(self.get_webapp().categories.values_list('id', flat=True)),
            sorted(c.id for c in cats))

    def test_categories_add(self):
        self._step()
        cat2 = Category.objects.create(type=amo.ADDON_WEBAPP, name='bling')
        self._post_cats([self.cat1, cat2])

    def test_categories_add_and_remove(self):
        self._step()
        cat2 = Category.objects.create(type=amo.ADDON_WEBAPP, name='bling')
        self._post_cats([cat2])

    def test_categories_remove(self):
        # Add another category here so it gets added to the initial formset.
        cat2 = Category.objects.create(type=amo.ADDON_WEBAPP, name='bling')
        AddonCategory.objects.create(addon=self.webapp, category=cat2)
        self._step()

        # `cat2` should get removed.
        self._post_cats([self.cat1])


class TestPayments(TestSubmit):
    fixtures = ['base/users', 'webapps/337141-steamcube']

    def setUp(self):
        super(TestPayments, self).setUp()
        self.webapp = self.get_webapp()
        self.url = reverse('submit.app.payments', args=[self.webapp.app_slug])

    def get_webapp(self):
        return Webapp.objects.get(id=337141)

    def _step(self):
        self.user.update(read_dev_agreement=True)
        self.cl = AppSubmissionChecklist.objects.create(addon=self.webapp,
            terms=True, manifest=True, details=True)
        AddonUser.objects.create(addon=self.webapp, user=self.user)

    def test_anonymous(self):
        self._test_anonymous()

    def test_required(self):
        self._step()
        res = self.client.post(self.url, {'premium_type': ''})
        eq_(res.status_code, 200)
        self.assertFormError(res, 'form', 'premium_type',
                             'This field is required.')

    def test_not_valid(self):
        self._step()
        res = self.client.post(self.url, {'premium_type': 124})
        eq_(res.status_code, 200)
        self.assertFormError(res, 'form', 'premium_type',
            'Select a valid choice. 124 is not one of the available choices.')

    def test_valid(self):
        self._step()
        res = self.client.post(self.url, {'premium_type': amo.ADDON_PREMIUM})
        eq_(res.status_code, 302)
        eq_(self.get_webapp().premium_type, amo.ADDON_PREMIUM)


class TestPaymentsAdvanced(TestSubmit):
    fixtures = ['base/users', 'webapps/337141-steamcube']

    def setUp(self):
        super(TestPaymentsAdvanced, self).setUp()
        AppSubmissionChecklist.objects.all().delete()  # TODO fix this.
        self.webapp = self.get_webapp()
        self.url = self.get_url('payments')
        self.price = Price.objects.create(price='1.00')
        flag = (waffle.models.Flag
                      .objects.create(name='advanced-payments-submission'))
        flag.everyone = True
        flag.save()
        self._step()

    def get_url(self, url):
        return reverse('submit.app.%s' % url, args=[self.webapp.app_slug])

    def get_webapp(self):
        return Webapp.objects.get(id=337141)

    def _step(self):
        self.user.update(read_dev_agreement=True)
        self.cl = AppSubmissionChecklist.objects.create(addon=self.webapp,
            terms=True, manifest=True, details=True)
        AddonUser.objects.create(addon=self.webapp, user=self.user)

    def test_anonymous(self):
        self._test_anonymous()

    def test_valid(self):
        res = self.client.post(self.get_url('payments'),
                               {'premium_type': amo.ADDON_FREE})
        eq_(res.status_code, 302)
        self.assertRedirects(res, self.get_url('done'))

    def test_premium(self):
        for type_ in [amo.ADDON_PREMIUM, amo.ADDON_PREMIUM_INAPP]:
            res = self.client.post(self.get_url('payments'),
                                  {'premium_type': type_})
            eq_(res.status_code, 302)
            self.assertRedirects(res, self.get_url('payments.upsell'))

    def test_free_inapp(self):
        res = self.client.post(self.get_url('payments'),
                               {'premium_type': amo.ADDON_FREE_INAPP})
        eq_(res.status_code, 302)
        self.assertRedirects(res, self.get_url('payments.paypal'))

    def test_price(self):
        self.webapp.update(premium_type=amo.ADDON_PREMIUM)
        res = self.client.post(self.get_url('payments.upsell'),
                               {'price': self.price.pk})
        eq_(res.status_code, 302)
        eq_(self.get_webapp().premium.price.pk, self.price.pk)
        self.assertRedirects(res, self.get_url('payments.paypal'))

    def test_upsell(self):
        free = Addon.objects.create(type=amo.ADDON_WEBAPP)
        AddonUser.objects.create(addon=free,
                                 user=self.webapp.authors.all()[0])
        self.webapp.update(premium_type=amo.ADDON_PREMIUM)
        res = self.client.post(self.get_url('payments.upsell'),
                               {'price': self.price.pk,
                                'do_upsell': 1,
                                'free': free.pk,
                                'text': 'some upsell',
                                })
        eq_(self.get_webapp().premium.price.pk, self.price.pk)
        eq_(self.get_webapp().upsold.free.pk, free.pk)
        eq_(self.get_webapp().upsold.premium.pk, self.get_webapp().pk)
        eq_(res.status_code, 302)
        self.assertRedirects(res, self.get_url('payments.paypal'))

    def test_no_upsell(self):
        self.webapp.update(premium_type=amo.ADDON_PREMIUM)
        res = self.client.get(self.get_url('payments.upsell'),
                               {'price': self.price.pk})
        eq_(res.status_code, 200)
        eq_(len(pq(res.content)('div.brform')), 2)

    def test_upsell_missing(self):
        free = Addon.objects.create(type=amo.ADDON_WEBAPP)
        AddonUser.objects.create(addon=free,
                                 user=self.webapp.authors.all()[0])
        self.webapp.update(premium_type=amo.ADDON_PREMIUM)
        res = self.client.post(self.get_url('payments.upsell'),
                               {'price': self.price.pk,
                                'do_upsell': 1,
                                })
        eq_(res.status_code, 200)

    def test_bad_upsell(self):
        self.webapp.update(premium_type=amo.ADDON_PREMIUM)
        res = self.client.post(self.get_url('payments.upsell'), {'price': ''})
        eq_(res.status_code, 200)
        self.assertFormError(res, 'form', 'price', 'This field is required.')

    def test_paypal(self):
        self.webapp.update(premium_type=amo.ADDON_PREMIUM)
        res = self.client.post(self.get_url('payments.paypal'),
                               {'business_account': 'yes',
                                'email': 'foo@bar.com'})
        eq_(self.get_webapp().paypal_id, 'foo@bar.com')
        eq_(res.status_code, 302)
        self.assertRedirects(res, self.get_url('payments.bounce'))

    def test_bad_paypal(self):
        # some tests for when it goes wrong
        raise SkipTest

    def test_acquire(self):
        raise SkipTest

    def test_confirm(self):
        raise SkipTest
        # and this is where it all breaks down, need to figure out
        # how to test this new flow.


class TestDone(TestSubmit):
    fixtures = ['base/users', 'webapps/337141-steamcube']

    def setUp(self):
        super(TestDone, self).setUp()
        self.webapp = self.get_webapp()
        self.url = reverse('submit.app.done', args=[self.webapp.app_slug])

    def get_webapp(self):
        return Webapp.objects.get(id=337141)

    def _step(self):
        self.user.update(read_dev_agreement=True)
        self.cl = AppSubmissionChecklist.objects.create(addon=self.webapp,
            terms=True, manifest=True, details=True, payments=True)
        AddonUser.objects.create(addon=self.webapp, user=self.user)

    def test_anonymous(self):
        self._test_anonymous()

    def test_done(self):
        self._step()
        res = self.client.get(self.url)
        eq_(res.status_code, 200)