Python django.urls 模块，reverse_lazy() 实例源码

def subscriptions(request):        

    data = {}

    for word in r.keys("subscription_*"):
        subscription = re.sub(r'^subscription_', '', str(word.decode('utf-8')))
        try:

            subscription_data = r.lrange(word, 0, -1)
            data[subscription] = subscription_data 

        except:
            raise

    profile_form = ContactForm(instance=Contact.objects.get(user=request.user.id))   

    return render(request, 'isubscribe/subscriptions.html', {'DATA':data, 'profile_form': profile_form})




#@login_required(login_url=reverse_lazy('login'))

def resolve(request):

    mimetype = 'application/json'
    data = {}

    if request.method == 'POST' and 'entity' in request.POST and request.POST['entity'] != '':
        data['entity'] = request.POST['entity']        
        data['status'] = 0
        data['timestamp'] = datetime.datetime.now().timestamp()        
        data['output'] = "resolve request by %s" % (request.user.username)
        data['result'] = 'okay'

        sensu_event_resolve(data)
        Channel('background-alert').send(dict(data))


    return HttpResponse(json.dumps(data), mimetype)



#@login_required(login_url=reverse_lazy('login'))

def rmResult(request):

    mimetype = 'application/json'
    data = {}

    if request.method == 'POST' and 'entity' in request.POST and request.POST['entity'] != '':        
        data['client'], data['check'] = request.POST['entity'].split(':')       
        data['status'] = 0
        data['timestamp'] = datetime.datetime.now().timestamp()

        if sensu_result_delete(data):
            data['result'] = 'okay'
        else:        
            data['result'] = 'failed deleting result using sensu api for: ' + request.POST['entity']

    return HttpResponse(json.dumps(data), mimetype)



#@login_required(login_url=reverse_lazy('login'))

def entity_history(request):            

    data = []
    mimetype = 'application/json'   

    if request.method == 'POST' and 'entity' in request.POST and request.POST['entity'] != '':

        entity = request.POST['entity']
        logger.debug("view entity_history user: %s entity: %s" % (request.user.username, entity))

        for history_data in r.lrange('history_entity_' + entity, 0, 100):
            data.append(pickle.loads(history_data))

    return HttpResponse(json.dumps(data), mimetype)



#@login_required(login_url=reverse_lazy('login'))

def entity_notify_history(request):            

    data = []
    mimetype = 'application/json'   

    if request.method == 'POST' and 'entity' in request.POST and request.POST['entity'] != '':

        entity = request.POST['entity']
        logger.debug("view entity_notify_history user: %s entity: %s" % (request.user.username, entity))

        for history_data in r.lrange('notifyhistory_entity_' + entity, 0, 100):
            data.append(pickle.loads(history_data))

    return HttpResponse(json.dumps(data), mimetype)




#@login_required(login_url=reverse_lazy('login'))

def check_config(request):        

    mimetype = 'application/json'   
    data = {}

    if request.method == 'POST' and 'entity' in request.POST and request.POST['entity'] != '':        
        client_name, check_name = request.POST['entity'].split(':')
        #check_name = 'check_gw_tomcat_errors_1h'
        #data = cache.get('check_' + check_name)
        data = cache.get('check_' + request.POST['entity'])


    return HttpResponse(json.dumps(data), mimetype)



#@login_required(login_url=reverse_lazy('login'))

def user_register(message):

    if settings.DEBUG == True: return

    logger.debug('user_register task start - email: %s' % message['register_user_email'])
    user = User.objects.create_user(message['register_user_name'], message['register_user_email'], is_active = False)
    message['register_password'] = passwd_generator(size=25)
    user.set_password(message['register_password'])
    user.save()    

    if 'slack_user_id' in message:
        user.contact = Contact(email = message['register_user_email'], slack_uid = message['slack_user_id'])
        user.contact.save()     
        registration_link = "%s%s?username=%s&key=%s" % (settings.REGISTRATION_URL_PREFIX, reverse_lazy('register_activate'), message['register_user_name'], message['register_password'])
        SLACK_MESSAGE = "Hello %s! we've detected you are using our team's slack. please take a minute to activate you account in the following <%s|LINK>.\n (please use same email address you used to sign-up with Slack)" % (message['register_user_name'], registration_link)
        logger.debug('user_register sending slack activation message to slack_uid %s' % message['slack_user_id'])
        slack.chat.post_message(message['slack_user_id'], SLACK_MESSAGE, as_user=False, username=settings.SLACK_BOT_NAME, icon_url=settings.SLACK_BOT_ICON)
        message['registration_link'] = registration_link
    else:      
        register_email(message)

    logger.debug('user_register task end - email: %s' % message['register_user_email'])

def test_call_list_shows_correct_glyphicon_per_call_direction(self):
        """Test that calls list shows correct icon based on call direction."""
        user1 = User()
        user1.save()
        self.client.force_login(user1)
        new_call = CallFactory.create(contact=self.contact)
        response = self.client.get(reverse_lazy('call_list'))
        soup = Soup(response.content, 'html.parser')
        # import pdb;pdb.set_trace()
        if new_call.direction == "incoming":
            span = soup.find_all(
                "span", {"class": "glyphicon glyphicon-log-in"})
            self.assertEqual(
                '<span aria-hidden="true" class="glyphicon glyphicon-log-in"></span>', str(span[0]))
        else:
            span = soup.find_all(
                "span", {"class": "glyphicon glyphicon-log-out"})
            self.assertEqual(
                '<span aria-hidden="true" class="glyphicon glyphicon-log-out"></span>', str(span[0]))

def __init__(self, *args, **kwargs):
        super(LogUpdateForm, self).__init__(*args, **kwargs)
        self.helper = FormHelper(self)

        # set form tag attributes
        self.helper.action = reverse_lazy('logs_edit', kwargs['instance'].id)
        self.helper.form_method = 'POST'
        self.helper.form_class = 'form-horizontal'

        # set form field properties
        self.helper.help_text_inline = True
        self.helper.html5_required = False
        self.helper.attrs = {'novalidate': ''}
        self.helper.label_class = 'col-sm-3 control-label'
        self.helper.field_class = 'col-sm-9'


        # add buttons
        self.helper.layout.append(Layout(
            FormActions(
                Submit('add_button', _(u'Save')),
                Submit('cancel_button', _(u'Cancel'), css_class='btn-link')
            )
        ))

def test_delete_resource_shows_cancel_button(self):
            """Test cancel button shows on delete resource page."""
            self.client.login(username='fred', password='temporary')
            self.assertEqual(Resource.objects.count(), 1)
            idx = self.resource.id
            response = self.client.get(reverse_lazy('delete', kwargs={'pk': idx}))
            html = soup(response.content, 'html.parser')
            cancel = html.findAll('a', {'href': "/resource/1/edit/"})
            self.assertTrue(cancel)

    # def test_homepage_view_links_to_a_single_resource(self):
    #     """Test homepage resource list total."""
    #     response = self.client.get(reverse_lazy('home'))
    #     html = soup(response.content, "html.parser")
    #     link = html.findAll("a", {"href": "/resource/"})
    #     # import pdb; pdb.set_trace()
    #     self.assertTrue(link)

    ##### CSS Element Tests

def __init__(self, query_key, query_url, field_to_update, obj_label=None, *args, **kwargs):

        super(Livesearch, self).__init__(*args, **kwargs)

        self.attrs = {
            'data-key': query_key,
            'data-source': reverse_lazy(query_url),
            'data-field': field_to_update,
        }

        if obj_label:
            self.attrs['data-label'] = obj_label


#
# Form fields
#

def post(self, *args, **kwargs):
        product_delete_pk = kwargs.get('pk', None)
        product_pk = kwargs.get('tpk', None)

        if product_pk and product_delete_pk:
            product_delete = ProductFinal.objects.filter(pk=product_delete_pk).first()
            product = ProductFinal.objects.filter(pk=product_pk).first()
            if product and product_delete:
                product.related.remove(product_delete)

        return redirect(reverse_lazy("status", kwargs={'status': 'accept', 'answer': urlsafe_base64_encode(json.dumps({'__pk__': None, '__str__': 'OK'}))}))


# ------- sublista de accesorios productos -------

def post(self, *args, **kwargs):
        product_delete_pk = kwargs.get('pk', None)
        product_pk = kwargs.get('tpk', None)

        if product_pk and product_delete_pk:
            product_delete = ProductFinal.objects.filter(pk=product_delete_pk).first()
            product = ProductFinal.objects.filter(pk=product_pk).first()
            if product and product_delete:
                product.related_accesory.remove(product_delete)

        return redirect(reverse_lazy("status", kwargs={'status': 'accept', 'answer': urlsafe_base64_encode(json.dumps({'__pk__': None, '__str__': 'OK'}))}))


# ###########################################

def index(request):
    return HttpResponseRedirect(reverse_lazy('events'))

def subscribe_toggle(request):

    mimetype = 'application/json'
    data = {}

    if request.method == 'POST' and 'entity' in request.POST and request.POST['entity'] != '' and 'status' in request.POST and request.POST['status'] != '':

        data['entity'] = request.POST['entity']
        data['status'] = request.POST['status']

        if Subscribe.objects.filter(entity=request.POST['entity'], status=int(request.POST['status'])).count() > 0:
            # change existing object
            obj = Subscribe.objects.get(entity=request.POST['entity'], status=int(request.POST['status']))
            if request.user.pk not in obj.friends.values_list('pk', flat=True).all():
                obj.friends.add(request.user.pk)
                data['result'] = "subscription added"
                logger.debug('%s subscribed to %s' % (request.user.username, request.POST['entity']))
            else:
                obj.friends.remove(request.user.pk)
                data['result'] = "subscription removed"
                logger.debug('%s unsubscribed from %s' % (request.user.username, request.POST['entity']))
        else:
            # create new object
            obj = Subscribe(entity=request.POST['entity'], status=int(request.POST['status']))
            obj.save()
            obj.friends.add(request.user.pk)
            data['result'] = "subscription added"
            logger.debug('%s subscribed to new entity %s' % (request.user.username, request.POST['entity']))

        Channel('background-build-entity-rules').send({'entity': request.POST['entity']})    

    return HttpResponse(json.dumps(data), mimetype)




#@login_required(login_url=reverse_lazy('login'))

def ack(request):

    mimetype = 'application/json'
    data = {}

    if request.method == 'POST' and 'entity' in request.POST and request.POST['entity'] != '' and 'ack_interval' in request.POST and request.POST['ack_interval'] != '':

        data['entity'] = request.POST['entity']
        data['ack_interval'] = request.POST['ack_interval']
        data['status'] = request.POST['status']
        data['timestamp'] = datetime.datetime.now().timestamp()
        data['ack_by'] = request.user.username
        data['ack'] = True
        data['output'] = "acknowledged by %s for %s hours" % (request.user.username, request.POST['ack_interval'])

        if 'ack_comment' in request.POST:
            data['ack_comment'] = request.user.username + ': ' + request.POST['ack_comment']

        ack_data = { 'user_id': request.user.pk, 
                    'user_name': request.user.username, 
                    'timestamp': datetime.datetime.now().timestamp(), 
                    'ack_interval': request.POST['ack_interval'],
                    'ack_comment': data['ack_comment']
                    }        

        logger.debug('ack %s' % json.dumps(ack_data))
        cache.set("ack_" + request.POST['entity'], ack_data, timeout=(float(data['ack_interval']) * 3600))                

        Channel('background-ack').send(data)

    return HttpResponse(json.dumps(data), mimetype)




#@login_required(login_url=reverse_lazy('login'))

def register_activate(request):

    if 'key' not in request.GET or request.GET['key'] == '':
        return HttpResponse('Unauthorized', status=401)

    data = {}
    data['username'] = request.GET['username']
    data['key'] = request.GET['key']
    data['errors'] = ''

    if 'email' in request.POST and request.POST['key'] != '' and request.POST['password'] == request.POST['password_repeat']:

        try:

            validate_password(request.POST['password'])            
            # check for key against current user password
            logger.debug("validating registration key for user %s" % request.POST['username'])

            try:
                u = User.objects.get(username=request.POST['username'])
                if u.contact.email == request.POST['email']:
                    if u.check_password(request.POST['key']):
                        #user activate
                        logger.debug("activating user %s" % request.POST['username'])
                        u.set_password(request.POST['password'])
                        u.is_active = True
                        u.save()
                        login(request, u)
                        return HttpResponseRedirect(reverse_lazy('entities'))
            except:
                data['errors'] = 'An exception flew by!'

        except ValidationError as err:
            logger.debug("validating registration new password for user %s FAILED - %s" % (request.POST['username'], err))
            data['errors'] = err

    return render(request, 'registration/activate.html', {'DATA':data})

def get_success_url(self):
        return reverse_lazy(
            'game', 
            kwargs={
                'group_pk': self.kwargs.get('pk', None),
                'game_pk': self.object.pk,
            },
        )

def get_success_url(self):
        """Redirect back to the game."""

        return reverse_lazy(
            'game', 
            kwargs={
                'group_pk': self.object.group_set.first().pk,
                'game_pk': self.object.pk,
            },
        )

def get(self, request, *args, **kwargs):

        group = get_object_or_404(Group, id=kwargs.get('pk', None))

        player = request.user.player

        if player not in group.players.all():
            group.players.add(player)
            group.save()

            return HttpResponseRedirect(
                reverse_lazy('group', kwargs={'pk': group.pk}))

        return HttpResponseRedirect(reverse_lazy('groups'))

def get_success_url(self):
        return reverse_lazy('group', kwargs={'pk': self.object.pk})

def get_success_url(self):
        return reverse_lazy('proposals:detail', kwargs={'pk': self.get_object().proposal.id})

def __init__(self, *args, **kwargs):
        self.user = kwargs.pop('user', None)
        super().__init__(*args, **kwargs)
        if 'group' in self.fields:
            self.fields['group'].help_text = _(
                'Group projects under the same namespace, just like a folder. '
                '<a href="{url}">Create group</a>.'
            ).format(url=reverse_lazy('groups:group-create'))
        self.fields['repo'].widget.attrs['placeholder'] = \
            'ex: https://github.com/srtab/alexandriadocs'
        self.fields['tags'].widget.attrs['placeholder'] = 'ex: django, python'
        self.form_helper()

def get_success_url(self):
        success_url = reverse_lazy(
            "home",
        )

        return success_url

def post(self, request, *args, **kwargs):
        if "save" in request.POST:
            return super().post(request, *args, **kwargs)
        else:
            return HttpResponseRedirect(reverse_lazy('home'))

def get_absolute_url(self):
        return reverse_lazy('matches:control', kwargs={'pk': self.id})

def get_overlay_url(self):
        return reverse_lazy('matches:overlay', kwargs={'pk': self.id})

def test_new_call_instance_created_on_outgoing_call(self):
        """When outgoing call initiated, new call instance is created."""
        self.assertEqual(Call.objects.count(), 0)
        self.client.post(reverse_lazy('call'), {'phoneNumber': '+12345678910'})
        self.assertEqual(Call.objects.count(), 1)

def test_new_call_instance_created_on_incoming_call(self):
        """When incoming call initiated, new call instance is created."""
        self.assertEqual(Call.objects.count(), 0)
        self.client.get(reverse_lazy('call'), {'From': '+12345678910'})
        self.assertEqual(Call.objects.count(), 1)

def test_new_contact_created_on_call_if_new_number(self):
        """When call made or received, new contact is made if new number."""
        self.assertEqual(Contact.objects.count(), 1)
        self.client.post(reverse_lazy('call'), {'phoneNumber': '+12345678910'})
        self.assertEqual(Contact.objects.count(), 2)

def test_new_contact_not_created_on_call_if_already_contact(self):
        """When call made, received to old number, new contact is not made."""
        self.assertEqual(Contact.objects.count(), 1)
        existing_number = str(self.contact.number).strip('+')
        self.client.post(
            reverse_lazy('call'), {'phoneNumber': existing_number})
        self.assertEqual(Contact.objects.count(), 1)

def test_call_has_contact_outgoing_call(self):
        """When outgoing call initiated, new contact should have number."""
        self.client.post(reverse_lazy('call'), {'phoneNumber': '2345678910'})
        contact = Call.objects.first().contact
        self.assertEqual(contact.number, '+12345678910')

def test_call_list_shows_all_previous_calls(self):
        """Call history should show up in order on the call list view."""
        user1 = User()
        user1.save()
        self.client.force_login(user1)
        response = self.client.get(reverse_lazy('call_list'))
        soup = Soup(response.content, 'html.parser')
        trs = soup.find_all('tr')
        call_length = len(trs)
        [CallFactory.create(contact=self.contact) for i in range(20)]
        response = self.client.get(reverse_lazy('call_list'))
        soup = Soup(response.content, 'html.parser')
        trs = soup.find_all('tr')
        self.assertEqual(len(trs), call_length + 20)

def test_api_contacts_detail_view_status_ok(self):
        """Test api contacts detail view is status ok."""
        user1 = User()
        user1.save()
        self.client.force_login(user1)
        jabba = Contact(name="Jabba", number="+12068675309")
        jabba.save()
        request = self.request.get(reverse_lazy('api_contacts_retrieve',
                                   kwargs={"pk": jabba.pk}))
        request.user = user1
        view = ContactViewSet.as_view({'get': 'retrieve'})
        response = view(request, pk=jabba.id)
        self.assertEqual(response.status_code, 200)

def test_text_view_status_200(self):
        """Test that text view returns ok status."""
        user1 = User()
        user1.save()
        self.client.force_login(user1)
        text1 = Text(body="Jabba no watta.", sender="them", contact=self.contacts[0])
        text1.save()
        response = self.client.get(reverse_lazy('texts', kwargs={"pk": self.contacts[0].id}))
        self.assertTrue(response.status_code == 200)

def test_text_view_template(self):
        """Test that text view uses texts template."""
        user1 = User()
        user1.save()
        self.client.force_login(user1)
        text1 = Text(body="Jabba no watta.", sender="them", contact=self.contacts[0])
        text1.save()
        response = self.client.get(reverse_lazy('texts', kwargs={"pk": self.contacts[0].id}))
        self.assertTemplateUsed(response, 'texts/texting.html')

# Needs tests that use self.client and bs4 to count texts on page

def test_get_request_on_hook_view_not_allowed(self):
        """Get request is not allowed on text_hook url."""
        request = self.client.get(reverse_lazy('text_hook'))
        self.assertEqual(request.status_code, 405)

def test_post_request_to_text_hook_status_code(self):
        """A post request with correct kwargs returns a status code of 200."""
        req = self.client.post(reverse_lazy('text_hook'), {
            'Body': 'ToCountry=US&ToState=&FromCity=SEATTLE&Body=Test&FromCountry=US&To=%2B11111111111&From=%2B12064190136&ApiVersion=2010-04-01'})
        self.assertEqual(req.status_code, 200)

def test_post_request_to_text_hook_create_new_contact(self):
        """Test a new contact is added when a text from unknown number is received."""
        old_contacts = Contact.objects.count()
        self.client.post(reverse_lazy('text_hook'), {
            'Body': 'ToCountry=US&ToState=&FromCity=SEATTLE&Body=Test&FromCountry=US&To=%2B1222222222&From=%2B11111111111&ApiVersion=2010-04-01'})
        contacts = Contact.objects.count()
        self.assertGreater(contacts, old_contacts)

def test_no_contact_is_added_when_text_from_contact_received(self):
        """Test that no new contact is added when receiving text from known number."""
        new_contact = Contact()
        new_contact.number = "+11111111111"
        new_contact.name = "test"
        new_contact.save()
        contact_count = Contact.objects.count()
        self.client.post(reverse_lazy('text_hook'), {
            'Body': 'ToCountry=US&ToState=&FromCity=SEATTLE&Body=Test&FromCountry=US&To=%2B1222222222&From=%2B11111111111&ApiVersion=2010-04-01'})
        self.assertEqual(contact_count, Contact.objects.count())

def test_sender_set_as_you_when_text_from_yourself(self):
        self.client.post(reverse_lazy('text_hook'), {
            'Body': 'ToCountry=US&ToState=&FromCity=SEATTLE&Body=Test&FromCountry=US&To=%2B1222222222&From=%2B18555345517&ApiVersion=2010-04-01'})
        text = Text.objects.first()
        self.assertEqual(text.sender, "you")

def test_new_text_in_db_has_correct_info(self):
        """Test that a new text is added in the database with correct info."""
        self.client.post(reverse_lazy('text_hook'), {
            'Body': 'ToCountry=US&ToState=&FromCity=SEATTLE&Body=Test&FromCountry=US&To=%2B1222222222&From=%2B11111111111&ApiVersion=2010-04-01'})
        text = Text.objects.first()
        self.assertTrue(text.body == "Test")

def test_new_text_has_correct_contact_when_contact_exists(self):
        """Test that incoming test gets matched to correct contact if contact already exists."""
        new_contact = Contact()
        new_contact.number = "+11111111111"
        new_contact.name = "test"
        new_contact.save()
        self.client.post(reverse_lazy('text_hook'), {
            'Body': 'ToCountry=US&ToState=&FromCity=SEATTLE&Body=Test&FromCountry=US&To=%2B1222222222&From=%2B11111111111&ApiVersion=2010-04-01'})
        text = Text.objects.first()
        self.assertEqual(text.contact, new_contact)

def test_new_contact_has_empty_string_when_new_text(self):
        """Test that a new text from a new contact will create a contact with empty string as name."""
        self.client.post(reverse_lazy('text_hook'), {
            'Body': 'ToCountry=US&ToState=&FromCity=SEATTLE&Body=Test&FromCountry=US&To=%2B1222222222&From=%2B11111111111&ApiVersion=2010-04-01'})
        contact = Contact.objects.last()
        self.assertEqual(contact.name, "")

def test_message_list_view_client(self):
        """Test that contact list view returns a response from the same client."""
        ContactFactory.create(name="Bob Barker", number="+15555555555")
        response = self.client.get(reverse_lazy("message_list"))
        self.assertEqual(response.client, self.client)

def test_message_list_view_status(self):
        """Test that contact list view returns 200 OK response."""
        user1 = User()
        user1.save()
        view = MessageListView.as_view()
        req = self.request.get(reverse_lazy("message_list"))
        req.user = user1
        response = view(req)
        self.assertEqual(response.status_code, 200)

def test_message_list_view_content_title(self):
        """Test that contact list view returns 'Message List' as the title of the body."""
        user1 = User()
        user1.save()
        self.client.force_login(user1)
        ContactFactory.create(name="Bob Barker", number="+15555555555")
        response = self.client.get(reverse_lazy("message_list"))
        self.assertTemplateUsed(response, "texts/message_list.html")

def test_message_list_view_returns_first_contact(self):
        """Test that contact list view returns name of first contact."""
        user1 = User()
        user1.save()
        self.client.force_login(user1)
        self.add_text_to_contact(self.contacts[0])
        response = self.client.get(reverse_lazy("message_list"))
        self.assertIn(self.contacts[0].name, response.content.decode("utf-8"))

def test_message_list_view_returns_middle_contact(self):
        """Test that contact list view returns name of 10th contact."""
        user1 = User()
        user1.save()
        self.client.force_login(user1)
        self.add_text_to_contact(self.contacts[10])
        response = self.client.get(reverse_lazy("message_list"))
        self.assertIn(self.contacts[10].name, response.content.decode("utf-8"))

def test_message_list_view_returns_last_contact(self):
        """Test that contact list view returns name of last contact."""
        user1 = User()
        user1.save()
        self.client.force_login(user1)
        self.add_text_to_contact(self.contacts[-1])
        response = self.client.get(reverse_lazy("message_list"))
        self.assertIn(self.contacts[-1].name, response.content.decode("utf-8"))