Django:之中間件、微信接口和單元測試

時間 2019-12-08

原文原文鏈接

Django中間件css

咱們從瀏覽器發出一個請求 Request，獲得一個響應後的內容 HttpResponse ，這個請求傳遞到 Django的過程以下：html

也就是說，每個請求都是先經過中間件中的 process_request 函數，這個函數返回 None 或者 HttpResponse 對象，若是返回前者，繼續處理其它中間件，若是返回一個 HttpResponse，就處理停止，返回到網頁上。python

中間件不用繼承自任何類（能夠繼承 object ），下面一箇中間件大概的樣子：web

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函數。算法

1、好比咱們要作一個攔截器，發生有惡意訪問網站的人，就攔截它！數據庫

假如咱們經過一種技術，好比統計一分鐘訪問頁面數，太多酒把他的ip加入到黑名單BLOCKED_IPS(這部分沒有提供代碼，主要講中間件部分)django

#項目 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函數則是最前面的最後執行。json

2、在好比，咱們在網站放到服務器上正式運行後，DEBUG改成了False ,這樣更安全，可是有時候發生錯誤不能顯示錯誤詳情頁面，有沒有辦法處理好這兩個事情呢？瀏覽器

一、普通訪問者看到的是友好的報錯信息

二、管理員看到的是錯誤詳情，以便修復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)