def test_http_method_not_allowed_allowed_methods(self):
        class GetPostView(views.JSONView):
            def get(self, request, *args, **kwargs):
                return 'asdf'

            def post(self, request, *args, **kwargs):
                return 'qwer'
        response = GetPostView().http_method_not_allowed()
        ok_(set(['GET', 'POST']).issubset(set(response['Allow'].split(', '))))

        class GetPostPutDeleteHeadView(views.JSONView):
            def get(self, request, *args, **kwargs):
                return 'asdf'

            def post(self, request, *args, **kwargs):
                return 'qwer'

            def put(self, request, *args, **kwargs):
                return 'qwer'

            def delete(self, request, *args, **kwargs):
                return 'qwer'

            def head(self, request, *args, **kwargs):
                return 'qwer'
        response = GetPostPutDeleteHeadView().http_method_not_allowed()
        expected_methods = set(['GET', 'POST', 'PUT', 'DELETE', 'HEAD'])
        actual_methods = set(response['Allow'].split(', '))
        ok_(expected_methods.issubset(actual_methods))

 def mocked_get(url, **options):
     assert '/crash_data/' in url
     ok_('/datatype/processed/' in url)
     return Response("""
     {
       "product": "Firefox",
       "uuid": "7c44ade2-fdeb-4d6c-830a-07d302120525",
       "version": "13.0",
       "build": "20120501201020",
       "ReleaseChannel": "beta",
       "os_name": "Windows NT",
       "date_processed": "2012-05-25 11:35:57",
       "success": true,
       "signature": "CLocalEndpointEnumerator::OnMediaNotific",
       "addons": [
         [
           "[email protected]",
           "1.2.1"
         ],
         [
           "{972ce4c6-7e08-4474-a285-3208198ce6fd}",
           "13.0"
         ]
       ]
     }
     """)

    def test_current_versions(self, rget):
        model = models.CurrentVersions
        api = model()

        def mocked_get(**options):
            assert '/products/' in options['url']
            return Response("""
                {"hits": {
                   "SeaMonkey": [{
                     "product": "SeaMonkey",
                     "throttle": "100.00",
                     "end_date": "2012-05-10 00:00:00",
                     "start_date": "2012-03-08 00:00:00",
                     "featured": true,
                     "version": "2.1.3pre",
                     "release": "Beta",
                     "id": 922}]
                  },
                  "products": ["SeaMonkey"]
                }
              """)

        rget.side_effect = mocked_get
        info = api.get()
        ok_(isinstance(info, list))
        ok_(isinstance(info[0], dict))
        eq_(info[0]['product'], 'SeaMonkey')

 def test_rename_file(self):
   paths = ['foo', '%s/bar' % (self.client.root.rstrip('/'), )]
   self.client._create(paths[0], data='hello')
   ok_(self.client._rename(paths[0], destination=paths[1]).json()['boolean'])
   ok_(not self._exists(paths[0]))
   eq_(self.client._open(paths[1].rsplit('/', 1)[1]).content, b'hello')
   self.client._delete(paths[1])

    def test_xss_file_attachment_title(self):
        title = '"><img src=x onerror=prompt(navigator.userAgent);>'
        # use view to create new attachment
        file_for_upload = make_test_file()
        post_data = {
            'title': title,
            'description': 'xss',
            'comment': 'xss',
            'file': file_for_upload,
        }
        self.client.login(username='admin', password='testpass')
        resp = self.client.post(reverse('attachments.new_attachment'),
                                data=post_data)
        eq_(302, resp.status_code)

        # now stick it in/on a document
        attachment = Attachment.objects.get(title=title)
        rev = revision(content='<img src="%s" />' % attachment.get_file_url(),
                       save=True)

        # view it and verify markup is escaped
        response = self.client.get(rev.document.get_edit_url())
        eq_(200, response.status_code)
        doc = pq(response.content)
        eq_('%s xss' % title,
            doc('#page-attachments-table .attachment-name-cell').text())
        ok_('&gt;&lt;img src=x onerror=prompt(navigator.userAgent);&gt;' in
            doc('#page-attachments-table .attachment-name-cell').html())

def test_on_model_change_delete():
    app, db, admin = setup()
    Model1, _ = create_models(db)
    db.create_all()

    class ModelView(CustomModelView):
        def on_model_change(self, form, model, is_created):
            model.test1 = model.test1.upper()

        def on_model_delete(self, model):
            self.deleted = True

    view = ModelView(Model1, db.session)
    admin.add_view(view)

    client = app.test_client()

    client.post('/admin/model1/new/',
                data=dict(test1='test1large', test2='test2'))

    model = db.session.query(Model1).first()
    eq_(model.test1, 'TEST1LARGE')

    url = '/admin/model1/edit/?id=%s' % model.id
    client.post(url, data=dict(test1='test1small', test2='test2large'))

    model = db.session.query(Model1).first()
    eq_(model.test1, 'TEST1SMALL')

    url = '/admin/model1/delete/?id=%s' % model.id
    client.post(url)
    ok_(view.deleted)

 def test_message_with_url_is_link(self):
     m = Message(message="Go to http://bit.ly/sample-demo", is_global=True,
                 is_active=True, url="/")
     m.save()
     ok_('Go to <a href="http://bit.ly/sample-demo">'
         'http://bit.ly/sample-demo</a>' in
         soapbox_messages(get_soapbox_messages("/")))

 def test_get_jwt(self, client=None, extra_headers=None):
     res = self._post(client=client, extra_headers=extra_headers)
     eq_(res.status_code, 201, res.content)
     contribution = Contribution.objects.get()
     eq_(res.json['contribStatusURL'],
         reverse('webpay-status', kwargs={'uuid': contribution.uuid}))
     ok_(res.json['webpayJWT'])

    def test_saves_promo_img(self, requests_mock, crush_mock):
        img_path = os.path.join(settings.ROOT, 'mkt', 'site', 'tests',
                                'images', 'game_1050.jpg')

        # Mock the image fetch request.
        with open(img_path, 'r') as content:
            requests_mock.return_value = mock.Mock(
                content=content.read(),
                headers={'ok': 'ok'},
                status_code=200)

        result = fetch_promo_imgs(self.website.pk, 'http://mocked_url.ly')
        ok_(result)

        website = Website.objects.all()[0]
        eq_(website.promo_img_hash, '215dd2a2')

        # Check the actual saved image on disk.
        img_dir = website.get_promo_img_dir()
        for size in mkt.PROMO_IMG_SIZES:
            img_path = os.path.join(img_dir, '%s-%s.png' % (str(website.id),
                                                            size))
            with public_storage.open(img_path, 'r') as img:
                checker = ImageCheck(img)
                assert checker.is_image()
                eq_(checker.img.size[0], size)

    def test_title(self):
        event = SuggestedEvent.objects.create(
            user=self.user,
            title='Cool Title',
            slug='cool-title',
        )
        url = reverse('suggest:title', args=(event.pk,))
        response = self.client.get(url)
        eq_(response.status_code, 200)

        data = {
            'title': '',
            'slug': 'contains spaces',
        }
        response = self.client.post(url, data)
        eq_(response.status_code, 200)
        ok_('Form errors' in response.content)

        data = {
            'title': 'New Title',
            'slug': 'new-slug',
        }
        response = self.client.post(url, data)
        eq_(response.status_code, 302)
        next_url = reverse('suggest:description', args=(event.pk,))
        self.assertRedirects(response, next_url)

 def test_details_timezone_formatting(self):
     location = Location.objects.create(
         name='Paris',
         timezone='Europe/Paris'
     )
     start_time = datetime.datetime(
         2013, 5, 6, 11, 0, 0
     ).replace(tzinfo=utc)
     event = SuggestedEvent.objects.create(
         user=self.user,
         title='Cool Title',
         slug='cool-title',
         description='Some long description',
         short_description='',
         location=location,
         privacy=Event.PRIVACY_PUBLIC,
         start_time=start_time,
     )
     url = reverse('suggest:details', args=(event.pk,))
     response = self.client.get(url)
     eq_(response.status_code, 200)
     # the location is `US/Pacific` which means at 13:00 UTC,
     # the time is expected to be 05:00 in US/Pacific
     as_string = '2013-05-06 13:00:00'
     ok_('value="%s"' % as_string in response.content)

 def test_link_to_suggest(self):
     event = Event.objects.get(title='Test event')
     self._attach_file(event, self.placeholder)
     response = self.client.get('/')
     eq_(response.status_code, 200)
     start_url = reverse('suggest:start')
     ok_(start_url in response.content)

    def test_send_mail(self, fake_messages):
        """Test EmailRepsForm email sending functionality."""

        data = {"subject": "Test email subject", "body": "Test email body", "functional_area": self.functional_area.id}

        form = EmailRepsForm(data=data)
        ok_(form.is_valid())

        area = self.functional_area
        UserFactory.create_batch(20, userprofile__functional_areas=[area])

        factory = RequestFactory()
        request = factory.request()
        request.user = UserFactory.create()

        reps = User.objects.filter(userprofile__functional_areas__name=area)

        form.send_email(request, reps)

        eq_(len(mail.outbox), 20)

        address = lambda u: "%s %s <%s>" % (u.first_name, u.last_name, u.email)
        recipients = map(address, reps)

        receivers = []
        for i in range(0, len(mail.outbox)):
            eq_(mail.outbox[i].subject, data["subject"])
            eq_(mail.outbox[i].body, data["body"])
            receivers.append(mail.outbox[i].to[0])

        eq_(set(receivers), set(recipients))
        fake_messages.assert_called_with(ANY, "Email sent successfully.")

 def test_register_without_code_vouched(self):
     user = UserFactory.create()
     with self.login(user) as client:
         response = client.get(reverse('phonebook:register'), follow=True)
     ok_(not self.client.session.get('invite-code'))
     self.assertJinja2TemplateUsed(response, 'phonebook/home.html')
     eq_(response.status_code, 200)

def test_non_int_pk():
    app, db, admin = setup()

    class Model(db.Model):
        id = db.Column(db.String, primary_key=True)
        test = db.Column(db.String)

    db.create_all()

    view = CustomModelView(Model, db.session, form_columns=['id', 'test'])
    admin.add_view(view)

    client = app.test_client()

    rv = client.get('/admin/model/')
    eq_(rv.status_code, 200)

    rv = client.post('/admin/model/new/',
                     data=dict(id='test1', test='test2'))
    eq_(rv.status_code, 302)

    rv = client.get('/admin/model/')
    eq_(rv.status_code, 200)
    data = rv.data.decode('utf-8')
    ok_('test1' in data)

    rv = client.get('/admin/model/edit/?id=test1')
    eq_(rv.status_code, 200)
    data = rv.data.decode('utf-8')
    ok_('test2' in data)

def test_model():
    app, db, admin = setup()

    view = TestView(db.test, 'Test')
    admin.add_view(view)

    # Drop existing data (if any)
    db.test.remove()

    eq_(view.name, 'Test')
    eq_(view.endpoint, 'testview')

    ok_('test1' in view._sortable_columns)
    ok_('test2' in view._sortable_columns)

    ok_(view._create_form_class is not None)
    ok_(view._edit_form_class is not None)
    eq_(view._search_supported, False)
    eq_(view._filters, None)

    # Make some test clients
    client = app.test_client()

    rv = client.get('/admin/testview/')
    eq_(rv.status_code, 200)

    rv = client.get('/admin/testview/new/')
    eq_(rv.status_code, 200)

    rv = client.post('/admin/testview/new/',
                     data=dict(test1='test1large', test2='test2'))
    eq_(rv.status_code, 302)

    model = db.test.find()[0]
    print(model)
    eq_(model['test1'], 'test1large')
    eq_(model['test2'], 'test2')

    rv = client.get('/admin/testview/')
    eq_(rv.status_code, 200)
    ok_('test1large' in rv.data.decode('utf-8'))

    url = '/admin/testview/edit/?id=%s' % model['_id']
    rv = client.get(url)
    eq_(rv.status_code, 200)

    rv = client.post(url,
                     data=dict(test1='test1small', test2='test2large'))
    eq_(rv.status_code, 302)

    print(db.test.find()[0])

    model = db.test.find()[0]
    eq_(model['test1'], 'test1small')
    eq_(model['test2'], 'test2large')

    url = '/admin/testview/delete/?id=%s' % model['_id']
    rv = client.post(url)
    eq_(rv.status_code, 302)
    eq_(db.test.count(), 0)

def test_form_columns():
    app, db, admin = setup()

    class Model(db.Model):
        id = db.Column(db.String, primary_key=True)
        int_field = db.Column(db.Integer)
        datetime_field = db.Column(db.DateTime)
        text_field = db.Column(db.UnicodeText)
        excluded_column = db.Column(db.String)

    class ChildModel(db.Model):
        id = db.Column(db.String, primary_key=True)
        model_id = db.Column(db.Integer, db.ForeignKey(Model.id))
        model = db.relationship(Model, backref='backref')

    db.create_all()

    view1 = CustomModelView(Model, db.session, endpoint='view1',
                            form_columns=('int_field', 'text_field'))
    view2 = CustomModelView(Model, db.session, endpoint='view2',
                            form_excluded_columns=('excluded_column',))
    view3 = CustomModelView(ChildModel, db.session, endpoint='view3')

    form1 = view1.create_form()
    form2 = view2.create_form()
    form3 = view3.create_form()

    ok_('int_field' in form1._fields)
    ok_('text_field' in form1._fields)
    ok_('datetime_field' not in form1._fields)

    ok_('excluded_column' not in form2._fields)

    ok_(type(form3.model).__name__ == 'QuerySelectField')

    def test_index(self):
        """Integration-test ``index()`` with some decorator-handled arg."""
        def valid_responder(*args, **kwargs):
            """Return an arbitrary successful Response."""
            response = requests.Response()
            response._content = six.b('{"some": "json"}')
            response.status_code = 200
            return response

        conn = ElasticSearch('http://example.com:9200/')
        with patch.object(conn.session, 'put') as put:
            put.side_effect = valid_responder
            conn.index('some_index',
                       'some_type',
                       {'some': 'doc'},
                       id=3,
                       routing='boogie',
                       es_snorkfest=True,
                       es_borkfest='gerbils:great')

        # Make sure all the query string params got into the URL:
        url = put.call_args[0][0]
        ok_(
            url.startswith('http://example.com:9200/some_index/some_type/3?'))
        ok_('routing=boogie' in url)
        ok_('snorkfest=true' in url)
        ok_('borkfest=gerbils%3Agreat' in url)
        ok_('es_' not in url)  # We stripped the "es_" prefixes.

    def test_amara_callback_successful(self):
        url = reverse('subtitles:amara_callback')
        event = Event.objects.get(title='Test event')
        amara_video = AmaraVideo.objects.create(
            event=event,
            video_id='abc123',
            video_url='http://example.com/foo.mp4',
        )

        response = self.post_json(url, {
            'event': 'new-language',
            'video_id': amara_video.video_id,
            'api_url': 'http://example.com/api/url',
            'team': 'myteam',
            'project': 'myproject',
            'language_code': 'sv',
        })
        eq_(response.status_code, 200)

        amara_callback = AmaraCallback.objects.get(
            api_url='http://example.com/api/url'
        )
        eq_(amara_callback.amara_video, amara_video)
        ok_(amara_callback.payload)
        eq_(amara_callback.api_url, 'http://example.com/api/url')
        eq_(amara_callback.team, 'myteam')
        eq_(amara_callback.project, 'myproject')
        eq_(amara_callback.language_code, 'sv')

    def test_webpush_data_delivery_to_disconnected_client(self):
        tests = {
            "d248d4e0-0ef4-41d9-8db5-2533ad8e4041": dict(data=b"\xe2\x82\x28\xf0\x28\x8c\xbc", result="4oIo8CiMvA=="),
            "df2363be-4d55-49c5-a1e3-aeae9450692e": dict(
                data=b"\xf0\x90\x28\xbc\xf0\x28\x8c\x28", result="8JAovPAojCg="
            ),
            "6c33e055-5762-47e5-b90c-90ad9bfe3f53": dict(data=b"\xc3\x28\xa0\xa1\xe2\x28\xa1", result="wyigoeIooQ=="),
        }

        client = Client("ws://localhost:9010/", use_webpush=True)
        yield client.connect()
        yield client.hello()
        for chan, test in tests.items():
            yield client.register(chid=chan)

        yield client.disconnect()
        for chan, test in tests.items():
            yield client.send_notification(channel=chan, data=test["data"])

        yield client.connect()
        yield client.hello()

        for chan in tests:
            result = yield client.get_notification()
            ok_(result is not None)
            chan = result["channelID"]
            test = tests[chan]
            eq_(result["data"], test["result"])
            yield client.ack(chan, result["version"])

        yield self.shut_down(client)