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

时间:2023-06-05 23:03:14

Django中间件

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

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 = (
'感谢您的关注!\n回复【功能】两个字查看支持的功能,还可以回复任意内容开始聊天'
'\n【<a href="http://www.ziqiangxuetang.com">自强学堂手机版</a>】'
))
if isinstance(message, TextMessage):
# 当前会话内容
content = message.content.strip()
if content == '功能':
reply_text = (
'目前支持的功能:\n1. 关键词后面加上【教程】两个字可以搜索教程,'
'比如回复 "Django 后台教程"\n'
'2. 回复任意词语,查天气,陪聊天,讲故事,无所不能!\n'
'还有更多功能正在开发中哦 ^_^\n'
'【<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-(好委屈,数据库翻个遍也没找到你输的关键词!\n'
'试试下面这些关键词吧:\nKEYWORD_LIST\n'
'<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', '\n'.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 += '\n来源:扫描二维码关注'
else:
reply_text += '\n来源:搜索名称关注'
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)