Django:之中间件、微信接口和单元测试

Django中间件

我们从浏览器发出一个请求 Request,得到一个响应后的内容 HttpResponse ,这个请求传递到 Django的过程如下:

也就是说,每一个请求都是先通过中间件中的 process_request 函数,这个函数返回 None 或者 HttpResponse 对象,如果返回前者,继续处理其它中间件,如果返回一个 HttpResponse,就处理中止,返回到网页上。

中间件不用继承自任何类(可以继承 object ),下面一个中间件大概的样子:

class CommonMiddleware(object):
    def process_request(self, request):
        return None
 
    def process_response(self, request, response):
        return response

还有process_view,process_exception和process_template_response函数。

一、比如我们要做一个拦截器,发生有恶意访问网站的人,就拦截它!

假如我们通过一种技术,比如统计一分钟访问页面数,太多酒把他的ip加入到黑名单BLOCKED_IPS(这部分没有提供代码,主要讲中间件部分)

#项目 wulaoer 文件名 wulaoer/middleware.py
 
class BlockedIpMiddleware(object):
    def process_request(self, request):
        if request.META['REMOTE_ADDR'] in getattr(settings, "BLOCKED_IPS", []):
            return http.HttpResponseForbidden('<h1>Forbidden</h1>')

这里的代码的功能就是获取当前访问者的IP(request.META['REMOTE_ADDR']),如果这个IP在黑名单中就拦截,如果不在就返回None(函数中没有返回值其实就是默认为None),把这个中间件的Python路径写到settings.py中

MIDDLEWARE_CLASSES = (
    'wulaoer.middleware.BlockedIpMiddleware',
    ...其它的中间件
)

Django会从MIDDLEWARE_CLASSES中按照从上到下到顺序一个个执行中间件中的process_request函数,而其中process_response函数则是最前面的最后执行。

二、在比如,我们在网站放到服务器上正式运行后,DEBUG改为了False ,这样更安全,但是有时候发生错误不能显示错误详情页面,有没有办法处理好这两个事情呢?

1、普通访问者看到的是友好的报错信息

2、管理员看到的是错误详情,以便修复BUG

当然可以有,利用中间件就可以做到!代码如下:

import sys
from django.views.debug import technical_500_response
from django.conf import settings
 
class UserBasedExceptionMiddleware(object):
    def process_exception(self, request, exception):
        if request.user.is_superuser or request.META.get('REMOTE_ADDR') in settings.INTERNAL_IPS:
            return technical_500_response(request, *sys.exc_info())

把这个中间件像上面一样,加到你的 settings.py 中的 MIDDLEWARE_CLASSES 中,可以放到最后,这样可以看到其它中间件的 process_request的错误。

当访问者为管理员时,就给出错误详情,比如访问本站的不存在的页面:http://localhost/admin/  

普通人看到的是普通的 404(自己点开看看),而我可以看到:

三,分享一个简单的识别手机的中间件,更详细的可以参考这个:django-mobi 或 django-mobile

MOBILE_USERAGENTS = ("2.0 MMP","240x320","400X240","AvantGo","BlackBerry",
    "Blazer","Cellphone","Danger","DoCoMo","Elaine/3.0","EudoraWeb",
    "Googlebot-Mobile","hiptop","IEMobile","KYOCERA/WX310K","LG/U990",
    "MIDP-2.","MMEF20","MOT-V","NetFront","Newt","Nintendo Wii","Nitro",
    "Nokia","Opera Mini","Palm","PlayStation Portable","portalmmm","Proxinet",
    "ProxiNet","SHARP-TQ-GX10","SHG-i900","Small","SonyEricsson","Symbian OS",
    "SymbianOS","TS21i-10","UP.Browser","UP.Link","webOS","Windows CE",
    "WinWAP","YahooSeeker/M1A1-R2D2","iPhone","iPod","Android",
    "BlackBerry9530","LG-TU915 Obigo","LGE VX","webOS","Nokia5800")
 
class MobileTemplate(object):
    """
    If a mobile user agent is detected, inspect the default args for the view 
    func, and if a template name is found assume it is the template arg and 
    attempt to load a mobile template based on the original template name.
    """
 
    def process_view(self, request, view_func, view_args, view_kwargs):
        if any(ua for ua in MOBILE_USERAGENTS if ua in 
            request.META["HTTP_USER_AGENT"]):
            template = view_kwargs.get("template")
            if template is None:
                for default in view_func.func_defaults:
                    if str(default).endswith(".html"):
                        template = default
            if template is not None:
                template = template.rsplit(".html", 1)[0] + ".mobile.html"
                try:
                    get_template(template)
                except TemplateDoesNotExist:
                    pass
                else:
                    view_kwargs["template"] = template
                    return view_func(request, *view_args, **view_kwargs)
        return None

参考文档:https://docs.djangoproject.com/en/1.8/topics/http/middleware/

Python/Django微信接口

填写相应的网址,Token(令牌) 是随便写的,你自己想写什么就写什么,微信验证时检验是否写的和服务器上的TOKEN一样,一样则通过。

关注一下吴老二的微信号吧,可以随时随地查阅教程哦,体验一下自强学堂的微信的各种功能再阅读效果更佳!

自己动手写微信的验证: views.py

#coding=utf-8
import hashlib
import json
from lxml import etree
from django.utils.encoding import smart_str
from django.views.decorators.csrf import csrf_exempt
from django.http import HttpResponse
from auto_reply.views import auto_reply_main # 修改这里
 
WEIXIN_TOKEN = 'write-a-value'
 
@csrf_exempt
def weixin_main(request):
    """
    所有的消息都会先进入这个函数进行处理,函数包含两个功能,
    微信接入验证是GET方法,
    微信正常的收发消息是用POST方法。
    """
    if request.method == "GET":
        signature = request.GET.get("signature", None)
        timestamp = request.GET.get("timestamp", None)
        nonce = request.GET.get("nonce", None)
        echostr = request.GET.get("echostr", None)
        token = WEIXIN_TOKEN
        tmp_list = [token, timestamp, nonce]
        tmp_list.sort()
        tmp_str = "%s%s%s" % tuple(tmp_list)
        tmp_str = hashlib.sha1(tmp_str).hexdigest()
        if tmp_str == signature:
            return HttpResponse(echostr)
        else:
            return HttpResponse("weixin  index")
    else:
        xml_str = smart_str(request.body)
        request_xml = etree.fromstring(xml_str)
        response_xml = auto_reply_main(request_xml)# 修改这里
        return HttpResponse(response_xml)

auto_reply_main 是用来处理消息,回复消息的,需要自己进一步完善。

使用第三方包实现:

关于Django开发微信,有已经做好的现在的包可以使用 wechat_sdk 这个包,使用文档 也比较完善,但是在处理加密一部分没有做,在微信公众平台上,需要用明文验证,如果要加密,自己参照微信官网的加密算法。

使用 wechat_sdk 的例子(吴老二微信号简化后的例子):

# -*- coding: utf-8 -*-
from __future__ import unicode_literals
 
from django.http.response import HttpResponse, HttpResponseBadRequest
from django.views.decorators.csrf import csrf_exempt
 
from wechat_sdk import WechatBasic
from wechat_sdk.exceptions import ParseError
from wechat_sdk.messages import TextMessage
 
 
WECHAT_TOKEN = 'zqxt'
AppID = ''
AppSecret = ''
 
# 实例化 WechatBasic
wechat_instance = WechatBasic(
    token=WECHAT_TOKEN,
    appid=AppID,
    appsecret=AppSecret
)
 
@csrf_exempt
def index(request):
    if request.method == 'GET':
        # 检验合法性
        # 从 request 中提取基本信息 (signature, timestamp, nonce, xml)
        signature = request.GET.get('signature')
        timestamp = request.GET.get('timestamp')
        nonce = request.GET.get('nonce')
 
        if not wechat_instance.check_signature(
                signature=signature, timestamp=timestamp, nonce=nonce):
            return HttpResponseBadRequest('Verify Failed')
 
        return HttpResponse(
            request.GET.get('echostr', ''), content_type="text/plain")
 
 
    # 解析本次请求的 XML 数据
    try:
        wechat_instance.parse_data(data=request.body)
    except ParseError:
        return HttpResponseBadRequest('Invalid XML Data')
 
    # 获取解析好的微信请求信息
    message = wechat_instance.get_message()
 
    # 关注事件以及不匹配时的默认回复
    response = wechat_instance.response_text(
        content = (
            '感谢您的关注!
回复【功能】两个字查看支持的功能,还可以回复任意内容开始聊天'
            '
【<a href="http://www.ziqiangxuetang.com">自强学堂手机版</a>】'
            ))
    if isinstance(message, TextMessage):
        # 当前会话内容
        content = message.content.strip()
        if content == '功能':
            reply_text = (
                    '目前支持的功能:
1. 关键词后面加上【教程】两个字可以搜索教程,'
                    '比如回复 "Django 后台教程"
'
                    '2. 回复任意词语,查天气,陪聊天,讲故事,无所不能!
'
                    '还有更多功能正在开发中哦 ^_^
'
                    '【<a href="http://www.ziqiangxuetang.com">自强学堂手机版</a>】'
                )
        elif content.endswith('教程'):
            reply_text = '您要找的教程如下:'
 
        response = wechat_instance.response_text(content=reply_text)
 
    return HttpResponse(response, content_type="application/xml")

下面是一个更详细复杂的使用例子:

models.py

# -*- coding: utf-8 -*-
from __future__ import unicode_literals
 
from django.db import models
 
 
class KeyWord(models.Model):
    keyword = models.CharField(
        '关键词', max_length=256, primary_key=True, help_text='用户发出的关键词')
    content = models.TextField(
        '内容', null=True, blank=True, help_text='回复给用户的内容')
 
    pub_date = models.DateTimeField('发表时间', auto_now_add=True)
    update_time = models.DateTimeField('更新时间', auto_now=True, null=True)
    published = models.BooleanField('发布状态', default=True)
 
    def __unicode__(self):
        return self.keyword
 
    class Meta:
        verbose_name='关键词'
        verbose_name_plural=verbose_name

views.py

# -*- coding: utf-8 -*-
from __future__ import unicode_literals
 
from django.http.response import HttpResponse, HttpResponseBadRequest
from django.views.decorators.csrf import csrf_exempt
 
from wechat_sdk import WechatBasic
from wechat_sdk.exceptions import ParseError
from wechat_sdk.messages import (TextMessage, VoiceMessage, ImageMessage,
    VideoMessage, LinkMessage, LocationMessage, EventMessage
)
 
from wechat_sdk.context.framework.django import DatabaseContextStore
from .models import KeyWord as KeyWordModel
 
 
# 实例化 WechatBasic
wechat_instance = WechatBasic(
    token='zqxt',
    appid='xx',
    appsecret='xx'
)
 
 
@csrf_exempt
def index(request):
    if request.method == 'GET':
        # 检验合法性
        # 从 request 中提取基本信息 (signature, timestamp, nonce, xml)
        signature = request.GET.get('signature')
        timestamp = request.GET.get('timestamp')
        nonce = request.GET.get('nonce')
 
        if not wechat_instance.check_signature(
                signature=signature, timestamp=timestamp, nonce=nonce):
            return HttpResponseBadRequest('Verify Failed')
 
        return HttpResponse(
            request.GET.get('echostr', ''), content_type="text/plain")
 
    # POST
    # 解析本次请求的 XML 数据
    try:
        wechat_instance.parse_data(data=request.body)
    except ParseError:
        return HttpResponseBadRequest('Invalid XML Data')
 
    # 获取解析好的微信请求信息
    message = wechat_instance.get_message()
    # 利用本次请求中的用户OpenID来初始化上下文对话
    context = DatabaseContextStore(openid=message.source)
 
    response = None
 
    if isinstance(message, TextMessage):
        step = context.get('step', 1)  # 当前对话次数,如果没有则返回 1
        # last_text = context.get('last_text')  # 上次对话内容
        content = message.content.strip()  # 当前会话内容
 
        if message.content == '新闻':
            response = wechat_instance.response_news([
                {
                    'title': '自强学堂',
                    'picurl': 'http://www.ziqiangxuetang.com/static/images/newlogo.png',
                    'description': '自强学堂致力于提供优质的IT技术教程, 网页制作,服务器后台编写,以及编程语言,如HTML,JS,Bootstrap,Python,Django。同时也提供大量在线实例,通过实例,学习更容易,更轻松。',
                    'url': 'http://www.ziqiangxuetang.com',
                }, {
                    'title': '百度',
                    'picurl': 'http://doraemonext.oss-cn-hangzhou.aliyuncs.com/test/wechat-test.jpg',
                    'url': 'http://www.baidu.com',
                }, {
                    'title': 'Django 教程',
                    'picurl': 'http://www.ziqiangxuetang.com/media/uploads/images/django_logo_20140508_061519_35.jpg',
                    'url': 'http://www.ziqiangxuetang.com/django/django-tutorial.html',
                }
            ])
            return HttpResponse(response, content_type="application/xml")
 
        else:
            try:
                keyword_object = KeyWordModel.objects.get(keyword=content)
                reply_text = keyword_object.content
            except KeyWordModel.DoesNotExist:
                try:
                    reply_text = KeyWordModel.objects.get(keyword='提示').content
                except KeyWordModel.DoesNotExist:
                    reply_text = ('/:P-(好委屈,数据库翻个遍也没找到你输的关键词!
'
                        '试试下面这些关键词吧:
KEYWORD_LIST
'
                        '<a href="http://www.rxnfinder.org">RxnFinder</a>'
                        '感谢您的支持!/:rose'
                        )
 
        # 将新的数据存入上下文对话中
        context['step'] = step + 1
        context['last_text'] = content
        context.save()  # 非常重要!请勿忘记!临时数据保存入数据库!
 
        if 'KEYWORD_LIST' in reply_text:
            keyword_objects = KeyWordModel.objects.exclude(keyword__in=[
                '关注事件', '测试', 'test', '提示']).filter(published=True)
            keywords = ('{}. {}'.format(str(i), k.keyword)
                        for i, k in enumerate(keyword_objects, 1))
            reply_text = reply_text.replace(
                'KEYWORD_LIST', '
'.join(keywords))
 
        # 文本消息结束
 
    elif isinstance(message, VoiceMessage):
        reply_text = '语音信息我听不懂/:P-(/:P-(/:P-('
    elif isinstance(message, ImageMessage):
        reply_text = '图片信息我也看不懂/:P-(/:P-(/:P-('
    elif isinstance(message, VideoMessage):
        reply_text = '视频我不会看/:P-('
    elif isinstance(message, LinkMessage):
        reply_text = '链接信息'
    elif isinstance(message, LocationMessage):
        reply_text = '地理位置信息'
    elif isinstance(message, EventMessage):  # 事件信息
        if message.type == 'subscribe':  # 关注事件(包括普通关注事件和扫描二维码造成的关注事件)
            follow_event = KeyWordModel.objects.get(keyword='关注事件')
            reply_text = follow_event.content
 
            # 如果 key 和 ticket 均不为空,则是扫描二维码造成的关注事件
            if message.key and message.ticket:
                reply_text += '
来源:扫描二维码关注'
            else:
                reply_text += '
来源:搜索名称关注'
        elif message.type == 'unsubscribe':
            reply_text = '取消关注事件'
        elif message.type == 'scan':
            reply_text = '已关注用户扫描二维码!'
        elif message.type == 'location':
            reply_text = '上报地理位置'
        elif message.type == 'click':
            reply_text = '自定义菜单点击'
        elif message.type == 'view':
            reply_text = '自定义菜单跳转链接'
        elif message.type == 'templatesendjobfinish':
            reply_text = '模板消息'
 
    response = wechat_instance.response_text(content=reply_text)
    return HttpResponse(response, content_type="application/xml")

Django单元测试

Django一系列教程,前面的例子都是我们写好代码后,运行开发服务器,在浏览器上自己点击测试,看写的代码是否正常,但是这样做很麻烦,因为以后如果有改动,可能会影响以前本来正常的功能,这样以前的功能又得测试一遍,非常不方便,Django中有完善的单元测试,我们可以对开发的每一个功能进行单元测试,这样只要运行一个命令 python manage.py test,就可以测试功能是否正常。  

一言以蔽之,测试就是检查代码是否按照自己的预期那样运行。

测试驱动开发: 有时候,我们知道自己需要的功能(结果),并不知道代码如何书写,这时候就可以利用测试驱动开发(Test Driven Development),先写出我们期待得到的结果(把测试代码先写出来),再去完善代码,直到不报错,我们就完成了。

《改善Python的91个建议》一书中说:单元测试绝不是浪费时间的无用功,它是高质量代码的保障之一,在软件开发的一节中值得投入精力和时间去把好这一关。 

1. Python 中 单元测试简介:

下面是一个 Python的单元测试简单的例子: 

假如我们开发一个除法的功能,有的同学可能觉得很简单,代码是这样的:

def division_funtion(x, y):
    return x / y

但是这样写究竟对还是不对呢,有些同学可以在代码下面这样测试:

def division_funtion(x, y):
    return x / y
 
 
if __name__ == '__main__':
    print division_funtion(2, 1)
    print division_funtion(2, 4)
    print division_funtion(8, 3)

但是这样运行后得到的结果,自己每次都得算一下去核对一遍,很不方便,Python中有 unittest 模块,可以很方便地进行测试,详情可以文章最后的链接,看官网文档的详细介绍。

下面是一个简单的示例:

import unittest
 
 
def division_funtion(x, y):
    return x / y
 
 
class TestDivision(unittest.TestCase):
    def test_int(self):
        self.assertEqual(division_funtion(9, 3), 3)
 
    def test_int2(self):
        self.assertEqual(division_funtion(9, 4), 2.25)
 
    def test_float(self):
        self.assertEqual(division_funtion(4.2, 3), 1.4)
 
 
if __name__ == '__main__':
    unittest.main()

我简单地写了三个测试示例(不一定全面,只是示范,比如没有考虑除数是0的情况),运行后发现:

F.F
======================================================================
FAIL: test_float (__main__.TestDivision)
----------------------------------------------------------------------
Traceback (most recent call last):
  File "/Users/tu/YunPan/mydivision.py", line 16, in test_float
    self.assertEqual(division_funtion(4.2, 3), 1.4)
AssertionError: 1.4000000000000001 != 1.4
 
======================================================================
FAIL: test_int2 (__main__.TestDivision)
----------------------------------------------------------------------
Traceback (most recent call last):
  File "/Users/tu/YunPan/1.py", line 13, in test_int2
    self.assertEqual(division_funtion(9, 4), 2.25)
AssertionError: 2 != 2.25
 
----------------------------------------------------------------------
Ran 3 tests in 0.001s
 
FAILED (failures=2)

汗!发现了没,竟然两个都失败了,测试发现:

4.2除以3 等于 1.4000000000000001 不等于期望值 1.4

9除以4等于2,不等于期望的 2.25

下面我们就是要修复这些问题,再次运行测试,直到运行不报错为止。

譬如根据实际情况,假设我们只需要保留到小数点后6位,可以这样改:

def division_funtion(x, y):
    return round(float(x) / y, 6)

再次运行就不报错了:

...
----------------------------------------------------------------------
Ran 3 tests in 0.000s
 
OK

Python 单元测试 官方文档:

Python 2 (https://docs.python.org/2/library/unittest.html

Python 3 (https://docs.python.org/3/library/unittest.html)

2. Django 中 单元测试:(不断完善中,后期会增加对前面讲解的内容的测试)

2.1 简单测试例子:

from django.test import TestCase
from myapp.models import Animal
 
 
class AnimalTestCase(TestCase):
    def setUp(self):
        Animal.objects.create(name="lion", sound="roar")
        Animal.objects.create(name="cat", sound="meow")
 
    def test_animals_can_speak(self):
        """Animals that can speak are correctly identified"""
        lion = Animal.objects.get(name="lion")
        cat = Animal.objects.get(name="cat")
        self.assertEqual(lion.speak(), 'The lion says "roar"')
        self.assertEqual(cat.speak(), 'The cat says "meow"')

这个例子是测试myapp.models 中的 Animal 类相关的方法功能。

2.2 用代码访问网址的方法:

>>> from django.test import Client
>>> c = Client()
>>> response = c.post('/login/', {'username': 'john', 'password': 'smith'})
>>> response.status_code
200
>>> response = c.get('/customer/details/')
>>> response.content
'<!DOCTYPE html...'

我们可以用 django.test.Client 的实例来实现 get 或 post 内容,检查一个网址返回的网页源代码。

默认情况下CSRF检查是被禁用的,如果测试需要,可以用下面的方法:

>>> from django.test import Client
>>> csrf_client = Client(enforce_csrf_checks=True)

使用 csrf_client 这个实例进行请求即可。

指定浏览USER-AGENT:

>>> c = Client(HTTP_USER_AGENT='Mozilla/5.0')

模拟post上传附件:

from django.test import Client
c = Client()
 
with open('wishlist.doc') as fp:
    c.post('/customers/wishes/', {'name': 'fred', 'attachment': fp})

测试网页返回状态: 

from django.test import TestCase
 
class SimpleTest(TestCase):
    def test_details(self):
        response = self.client.get('/customer/details/')
        self.assertEqual(response.status_code, 200)
 
    def test_index(self):
        response = self.client.get('/customer/index/')
        self.assertEqual(response.status_code, 200)

我们用 self.client 即可,不用 client = Client() 这样实例化,更方便,我们还可以继承 Client,添加一些其它方法:  

from django.test import TestCase, Client
 
class MyTestClient(Client):
    # Specialized methods for your environment
    ...
     
class MyTest(TestCase):
    client_class = MyTestClient
 
    def test_my_stuff(self):
        # Here self.client is an instance of MyTestClient...
        call_some_test_code()

定制 self.client 的方法:

from django.test import Client, TestCase
 
 
class MyAppTests(TestCase):
    def setUp(self):
        super(MyAppTests, self).setUp()
        self.client = Client(enforce_csrf_checks=True)
 
    def test_home(self):
        response = self.client.get('/')
        self.assertEqual(response.status_code, 200)

  

原文地址:https://www.cnblogs.com/wulaoer/p/5397346.html