Skip to content

23. 练习题:第6版

23.1. 简介

现在,我们回到我们的税费计算应用程序。我们将围绕它构建各种 Web 应用程序。

在应用程序练习的第 5 版中,税务机关的数据存储在数据库中。该第 5 版由两个独立的应用程序组成,它们共享公共层:

  • 一个应用程序,用于以 |批处理| 模式为存储在文本文件中的纳税人计算税款;
  • 一个以 |交互| 模式为通过键盘输入信息的纳税人计算税款的应用程序;

批处理税务计算应用程序的第 5 版具有以下架构:

Image

最终,该应用程序的 Web 版本将采用以下架构:

Image

  • Web客户端 [1] 与Web服务器 [2] 通信,Web服务器 [2] 与数据库管理系统 [3] 通信;
  • Web 服务器 [2] 保留了原应用程序的 [业务] [8] [DAO] [9] 层;
  • 原始应用程序保留了其主脚本[4]及其[业务][15][业务][8][15]是完全相同的;
  • 客户端/服务器通信需要两个额外层:
    • [Web][7],用于实现Web应用程序;
    • [DAO][5],作为Web应用程序[7]的客户端;

在最终版本中,批量税费计算可通过两种方式进行:

  • 税费计算的业务逻辑由服务器的 [业务] 层处理。[main] 脚本将采用此方法;
  • 税费计算的业务逻辑由客户端的 [business] 层处理。[main2] 脚本将采用此方法;

从现在起,我们将开发若干上述类型的客户端/服务器应用程序,每个应用程序都将演示一种或多种新的 Web 开发技术。

23.2. 税费计算 Web 服务器

23.2.1. 版本 1

Image

[server_01]脚本是一个如下所示的Web应用程序:

Image

  • [1] 中,我们使用了一个参数化 URL,其中传递了三个值:
    • [married](是/否)用于表示纳税人是否已婚;
    • [children]:纳税人子女的数量;
    • [salary]:纳税人的年薪;
  • [2]中,Web服务器返回一个JSON字符串,其中包含应缴税额及其各项组成部分;

应用程序架构如下:

Image

  • 浏览器 [1] 向服务器 [2] 发起请求。脚本 [server_01] 实现了服务器的 [Web][2]
  • [3-8] 是税务计算应用程序 |第 5 | 中已使用的层。我们直接复用它们;
    • [业务][3]的定义见|此处|;
    • [DAO][4]的定义见|此处|;

[server_01] Web 应用程序通过三个脚本进行配置:

  • [config],用于配置整个应用程序;
  • [config_database],用于配置数据库访问。我们将使用 MySQL 和 PostgreSQL 数据库管理系统;
  • [config_layers],用于配置应用程序层;

[config] 脚本如下:

def configure(config: dict) -> dict:
    import os

    #  step 1 ------
    #  folder of this file
    script_dir = os.path.dirname(os.path.abspath(__file__))
    #  root path
    root_dir = "C:/Data/st-2020/dev/python/cours-2020/python3-flask-2020"
    #  absolute dependencies
    absolute_dependencies = [
        #  project files
        #  BaseEntity, MyException
        f"{root_dir}/classes/02/entities",
        #  InterfaceImpôtsDao, InterfaceImpôtsMétier, InterfaceImpôtsUi
        f"{root_dir}/impots/v04/interfaces",
        #  AbstractImpôtsdao, ImpôtsConsole, ImpôtsMétier
        f"{root_dir}/impots/v04/services",
        #  ImpotsDaoWithAdminDataInDatabase
        f"{root_dir}/impots/v05/services",
        #  AdminData, ImpôtsError, TaxPayer
        f"{root_dir}/impots/v04/entities",
        #  Constants, slices
        f"{root_dir}/impots/v05/entities",
        #  IndexController
        f"{script_dir}/../controllers",
        #  scripts [config_database, config_layers]
        script_dir,
    ]
    #  set the syspath
    from myutils import set_syspath
    set_syspath(absolute_dependencies)

    #  step 2 ------
    #  application configuration
    #  list of users authorized to use the application
    config['users'] = [
        {
            "login": "admin",
            "password": "admin"
        }
    ]

    #  step 3 ------
    #  database configuration
    import config_database
    config["database"] = config_database.configure(config)

    #  step 4 ------
    #  instantiation of application layers
    import config_layers
    config['layers'] = config_layers.configure(config)

    #  we return the configuration
    return config
  • [configure] 函数将 [config] 字典作为参数(第 1 行),并在丰富其内容后将其作为结果返回(第 54 行)。其实早就应该指出,没有必要 返回结果 [config]。事实上,[config] 是一个字典引用,调用方与被调用方共享该引用。 因此,调用方代码已经拥有这个引用(第 1 行),没有必要再次返回它(第 54 行)。因此,写成:

config=[module].configure(config) (1)

是多余的。只需写:


[module].configure(config) (2)

尽管如此,我还是保留了 (1) 的写法,因为我觉得这样可能更能说明被调用的代码修改了 [config] 字典。

  • 第 1 行:[configure] 函数接收的 [config] 字典包含一个名为 ‘sgbd’ 的键,其值取自列表 [‘mysql’, ‘pgres’]。其中 [‘mysql’] 表示所用数据库由 MySQL 管理,而 [‘pgres’] 表示所用数据库由 PostgreSQL 管理;
  • 第 4–27 行:我们列出了所有包含 Web 应用程序所需元素的目录。这些目录将成为应用程序 Python 路径的一部分(第 30–31 行);
  • 第 33–40 行:仅允许特定用户访问该应用程序。此处提供了一个仅包含单个用户的列表;
  • 第 43–46 行:[config_database] 脚本负责构建所用数据库的配置;
  • 第 46 行:由 [config_database] 脚本生成的配置是一个字典,我们将它存储在与 ‘database’ 键关联的通用配置中;
  • 第 48–51 行:[config_layers] 脚本实例化 Web 应用程序层。它返回一个字典,该字典存储在通用配置的 ‘layers’ 键下;

[config_database] 脚本是 |版本 5| 中已使用的脚本。我们在此将其列出以供参考:

def configure(config: dict) -> dict:
    #  sqlalchemy configuration
    from sqlalchemy import create_engine, Table, Column, Integer, MetaData, Float
    from sqlalchemy.orm import mapper, sessionmaker

    #  connection chains to the databases used
    connection_strings = {
        'mysql': "mysql+mysqlconnector://admimpots:mdpimpots@localhost/dbimpots-2019",
        'pgres': "postgresql+psycopg2://admimpots:mdpimpots@localhost/dbimpots-2019"
    }
    #  connection chain to the database used
    engine = create_engine(connection_strings[config['sgbd']])

    #  metadata
    metadata = MetaData()

    #  the constants table
    constantes_table = Table("tbconstantes", metadata,
                             Column('id', Integer, primary_key=True),
                             Column('plafond_qf_demi_part', Float, nullable=False),
                             Column('plafond_revenus_celibataire_pour_reduction', Float, nullable=False),
                             Column('plafond_revenus_couple_pour_reduction', Float, nullable=False),
                             Column('valeur_reduc_demi_part', Float, nullable=False),
                             Column('plafond_decote_celibataire', Float, nullable=False),
                             Column('plafond_decote_couple', Float, nullable=False),
                             Column('plafond_impot_celibataire_pour_decote', Float, nullable=False),
                             Column('plafond_impot_couple_pour_decote', Float, nullable=False),
                             Column('abattement_dixpourcent_max', Float, nullable=False),
                             Column('abattement_dixpourcent_min', Float, nullable=False)
                             )

    #  tax bracket table
    tranches_table = Table("tbtranches", metadata,
                           Column('id', Integer, primary_key=True),
                           Column('limite', Float, nullable=False),
                           Column('coeffr', Float, nullable=False),
                           Column('coeffn', Float, nullable=False)
                           )
    #  mappings
    from Tranche import Tranche
    mapper(Tranche, tranches_table)

    from Constantes import Constantes
    mapper(Constantes, constantes_table)

    #  the factory session
    session_factory = sessionmaker()
    session_factory.configure(bind=engine)

    #  a session
    session = session_factory()

    #  certain information is recorded and rendered in a dictionary
    return {"engine": engine, "metadata": metadata, "tranches_table": tranches_table,
            "constantes_table": constantes_table, "session": session}

[config_layers] 脚本用于配置 Web 服务器层。我们复用之前见过的 |script|:

def configure(config: dict) -> dict:
    #  instantiation of application layers

    #  dao
    from ImpotsDaoWithAdminDataInDatabase import ImpotsDaoWithAdminDataInDatabase
    dao = ImpotsDaoWithAdminDataInDatabase(config)

    #  business
    from ImpôtsMétier import ImpôtsMétier
    métier = ImpôtsMétier()

    #  put layer instances in a dictionary and return them to the calling code
    return {
        "dao": dao,
        "métier": métier
    }
  • 第 6 行:[dao] 层通过数据库实现;
  • [ImpotsDaoWithAdminDataInDatabase] 已在此处定义;
  • [BusinessTaxes] 已在此处定义;

主脚本 [server_01] 如下:

#  a mysql or pgres parameter is expected
import sys
syntaxe = f"{sys.argv[0]} mysql / pgres"
erreur = len(sys.argv) != 2
if not erreur:
    sgbd = sys.argv[1].lower()
    erreur = sgbd != "mysql" and sgbd != "pgres"
if erreur:
    print(f"syntaxe : {syntaxe}")
    sys.exit()

#  configure the application
import config
config = config.configure({'sgbd': sgbd})

#  dependencies
from ImpôtsError import ImpôtsError
from TaxPayer import TaxPayer
import re
from flask import request
from myutils import json_response
from flask import Flask
from flask_api import status

#  data recovery from tax authorities
try:
    #  admindata will be read-only application data
    admindata = config["layers"]["dao"].get_admindata()
except ImpôtsError as erreur:
    print(f"L'erreur suivante s'est produite : {erreur}")
    sys.exit(1)

#  flask application
app = Flask(__name__)


#  Home URL : /?married=xx&children=yy&salary=zz
@app.route('/', methods=['GET'])
def index():
    #  initially no errors
    erreurs = []
    #  the query must have three parameters in the URL
    if len(request.args) != 3:
        erreurs.append("Méthode GET requise avec les seuls paramètres [marié, enfants, salaire]")

    #  retrieve marital status in URL
    marié = request.args.get('marié')
    if marié is None:
        erreurs.append("paramètre [marié] manquant")
    else:
        marié = marié.strip().lower()
        erreur = marié != "oui" and marié != "non"
        if erreur:
            erreurs.append(f"paramétre marié [{marié}] invalide")

    #  retrieve the number of children in the URL
    enfants = request.args.get('enfants')
    if enfants is None:
        erreurs.append("paramètre [enfants] manquant")
    else:
        enfants = enfants.strip()
        match = re.match(r"^\d+", enfants)
        if not match:
            erreurs.append(f"paramétre enfants [{enfants}] invalide")
        else:
            enfants = int(enfants)

    #  the salary is retrieved from the URL
    salaire = request.args.get('salaire')
    if salaire is None:
        erreurs.append("paramètre [salaire] manquant")
    else:
        salaire = salaire.strip()
        match = re.match(r"^\d+", salaire)
        if not match:
            erreurs.append(f"paramétre salaire [{salaire}] invalide")
        else:
            salaire = int(salaire)

    #  invalid parameters in the URL?
    for key in request.args.keys():
        if key not in ['marié', 'enfants', 'salaire']:
            erreurs.append(f"paramètre [{key}] invalide")

    #  mistakes?
    if erreurs:
        #  an error response is sent to the client
        résultats = {"réponse": {"erreurs": erreurs}}
        return json_response(résultats, status.HTTP_400_BAD_REQUEST)

    #  no mistakes, we can work
    #  tAX CALCULATION
    taxpayer = TaxPayer().fromdict({'marié': marié, 'enfants': enfants, 'salaire': salaire})
    config["layers"]["métier"].calculate_tax(taxpayer, admindata)
    #  we send the response to the customer
    return json_response({"réponse": {"result": taxpayer.asdict()}}, status.HTTP_200_OK)


#  hand only
if __name__ == '__main__':
    #  start the Flask server
    app.config.update(ENV="development", DEBUG=True)
    app.run()
  • 第 1–10 行:获取指示应使用哪个数据库管理系统 (DBMS) 的参数;
  • 第 12–14 行:利用此信息,我们可以配置应用程序。特别是,构建 Python 路径;
  • 第 16–23 行:利用新的 Python 路径,导入必要的模块;
  • 第 25–31 行:从税务机关获取数据以计算税款;
  • 第 33–34 行:实例化 Flask 应用程序;
  • 第 38 行:Flask 应用程序仅处理 URL [/]。它期望 URL 采用以下格式:[/ ?married=xx&children=yy&salary=zz],其中:
    • xx:是 / 否;
    • yy:子女数量;
    • zz:年薪;
  • 第 40–89 行:我们检查 URL 参数的有效性;
  • 第 41 行:我们将错误消息累积到 [errors] 列表中;
  • 第 43 行:您可能还记得,URL 的参数位于 [request.args] 中(参见 |此处|):
    • [request] 对象即第 20 行导入的 Flask 对象;
    • [request.args] 对象的行为类似于字典;
  • 第 43–44 行:我们验证参数数量恰好为三个(不多不少);
  • 第 46–49 行:我们检查 URL 中是否包含 [married] 参数;
  • 第 50–54 行:如果存在,则检查其小写值(去除首尾空格)是否为“yes”或“no”;
  • 第 56–59 行:检查 URL 中是否包含 [children] 参数;
  • 第 60–66 行:如果存在,则检查其值是否为正整数;
  • 第 66 行:请注意,URL 参数及其值均为字符串。[children] 参数的值被转换为 'int' 类型;
  • 第 68–78 行:对于 [salary] 参数,我们执行与 [children] 参数相同的检查;
  • 第 81–83 行:检查 URL 中是否仅包含 [‘married’, ‘children’, ‘salary’] 这三个参数;
  • 第 85–89 行:如果经过所有这些检查后,[errors] 列表不为空,我们将该错误列表作为 JSON 字符串连同状态码 [400 Bad Request] 一起发送给客户端;

由于后续我们经常需要向客户端发送 JSON 字符串作为响应,因此实现这一功能的几行代码已被提取到我们之前用过的 [myutils.py] 模块中:

Image

[myutils.py] 脚本内容如下:

#  imports
import json
import os
import sys

from flask import make_response


def set_syspath(absolute_dependencies: list):
    #  absolute_dependencies: a list of absolute folder names

    .


#  response generation HTTP jSON
def json_response(réponse: dict, status_code: int) -> tuple:
    #  response body HTTP
    response = make_response(json.dumps(réponse, ensure_ascii=False))
    #  response body HTTP is jSON
    response.headers['Content-Type'] = 'application/json; charset=utf-8'
    #  we send the HTTP response
    return response, status_code
  • 第 16 行:[json_response] 函数需要两个参数:
    • [response]:包含要发送给 Web 客户端的 JSON 字符串的字典;
    • [status_code]:响应的 HTTP 状态码;
  • 第 18 行:我们设置响应的 JSON 正文;
  • 第 20 行:我们添加 HTTP 头部,告知 Web 客户端将接收 JSON 数据;
  • 第 22 行:我们将 HTTP 响应发送给调用代码。是否将其发送给 Web 客户端由调用代码决定;

[__init__.py] 文件的修改如下:


from .myutils import set_syspath, json_response

通过在 PyCharm 终端中执行 [pip install .] 命令,将 [myutils] 的新版本安装到全局模块中:


(venv) C:\Data\st-2020\dev\python\cours-2020\python3-flask-2020\packages>pip install .
Processing c:\data\st-2020\dev\python\cours-2020\python3-flask-2020\packages
Using legacy setup.py install for myutils, since package 'wheel' is not installed.
Installing collected packages: myutils
  Attempting uninstall: myutils
    Found existing installation: myutils 0.1
    Uninstalling myutils-0.1:
      Successfully uninstalled myutils-0.1
    Running setup.py install for myutils ... done
Successfully installed myutils-0.1
  • 第 1 行:您必须位于 [packages] 文件夹中才能输入此命令;

[server_01] 脚本的代码继续如下:

    
    #  mistakes?
    if erreurs:
        #  an error response is sent to the client
        résultats = {"réponse": {"erreurs": erreurs}}
        return json_response(résultats, status.HTTP_400_BAD_REQUEST)

    #  no mistakes, we can work
    #  tAX CALCULATION
    taxpayer = TaxPayer().fromdict({'id': 0, 'marié': marié, 'enfants': enfants, 'salaire': salaire})
    config["layers"]["métier"].calculate_tax(taxpayer, admindata)
    #  we send the response to the client
    return json_response({"réponse": {"result": taxpayer.asdict()}}, status.HTTP_200_OK)
  • 第 10 行:此时,URL 中所需的参数已正确存在;
  • 第 10 行:我们创建了表示纳税人的 [TaxPayer] 对象;
  • 第 11 行:我们请求 [business] 层计算税款。请注意,[business] 层计算出的结果会被插入到作为参数传递的 [taxpayer] 对象中;
  • 第 13 行:将响应作为 JSON 字符串发送给 Web 客户端。这是一个字典形式的 JSON 字符串。我们通过 [result] 键关联,将 [taxpayer] 对象的字典放入其中。我们无法直接放入 [taxpayer] 对象本身,因为它无法在 JSON 中序列化;

我们创建两个执行配置,一个用于 MySQL,另一个用于 PostgreSQL:

Image

以下是一些执行示例(您已启动 [server_01] 应用程序和数据库管理系统,随后通过浏览器请求了 URLhttp://localhost:5000/):

Image

Image

以下是在 Postman 控制台中发送请求的示例:

Image


GET /?mari%C3%A9=xx&enfants=yy&salaire=zz HTTP/1.1
User-Agent: PostmanRuntime/7.26.1
Accept: */*
Cache-Control: no-cache
Postman-Token: e4c5df8c-4bd6-4250-b789-b7b164db4eff
Host: localhost:5000
Accept-Encoding: gzip, deflate, br
Connection: keep-alive
 
HTTP/1.0 400 BAD REQUEST
Content-Type: application/json; charset=utf-8
Content-Length: 134
Server: Werkzeug/1.0.1 Python/3.8.1
Date: Fri, 17 Jul 2020 06:15:44 GMT
 
{"réponse": {"erreurs": ["paramètre marié [xx] invalide", "paramètre enfants [yy] invalide", "paramètre salaire [zz] invalide"]}}
  • 第 1 行:请求的 URL 不正确;
  • 第 10 行:服务器返回 400 BAD REQUEST 状态码;

23.2.2. 版本 2

Image

服务器第 2 版将 URL 处理功能独立到 [index_controller] 模块中 [5]

#  import dependencies
import re

from flask_api import status
from werkzeug.local import LocalProxy


#  URL set: /?married=xx&children=yy&salary=zz
def execute(request: LocalProxy, config: dict) -> tuple:
    #  dependencies
    from TaxPayer import TaxPayer

    #  initially no errors
    erreurs = []
    #  the query must have three parameters
    if len(request.args) != 3:
        erreurs.append("Méthode GET requise avec les seuls paramètres [marié, enfants, salaire]")

    #  we retrieve the marital status of the URL
    marié = request.args.get('marié')
    if marié is None:
        erreurs.append("paramètre [marié] manquant")
    else:
        marié = marié.strip().lower()
        erreur = marié != "oui" and marié != "non"
        if erreur:
            erreurs.append(f"paramétre marié [{marié}] invalide")

    #  we retrieve the number of children in URL
    enfants = request.args.get('enfants')
    if enfants is None:
        erreurs.append("paramètre [enfants] manquant")
    else:
        enfants = enfants.strip()
        match = re.match(r"^\d+", enfants)
        if not match:
            erreurs.append(f"paramétre enfants {enfants} invalide")
        else:
            enfants = int(enfants)

    #  we recover the URL salary
    salaire = request.args.get('salaire')
    if salaire is None:
        erreurs.append("paramètre [salaire] manquant")
    else:
        salaire = salaire.strip()
        match = re.match(r"^\d+", salaire)
        if not match:
            erreurs.append(f"paramétre salaire {salaire} invalide")
        else:
            salaire = int(salaire)

    #  other parameters in the URL?
    for key in request.args.keys():
        if not key in ['marié', 'enfants', 'salaire']:
            erreurs.append(f"paramètre [{key}] invalide")

    #  mistakes?
    if erreurs:
        #  an error response is sent to the client
        résultats = {"réponse": {"erreurs": erreurs}}
        return résultats, status.HTTP_400_BAD_REQUEST

    #  no mistakes, we can work
    #  tAX CALCULATION
    taxpayer = TaxPayer().fromdict({'marié': marié, 'enfants': enfants, 'salaire': salaire})
    config["layers"]["métier"].calculate_tax(taxpayer, config["admindata"])
    #  we send the response to the customer
    return {"réponse": {"result": taxpayer.asdict()}}, status.HTTP_200_OK
  • 第 9 行:[execute] 函数接收两个参数:
    • [request]:客户端的 HTTP 请求;
    • [config]:应用程序的配置字典;

[server_02] 脚本如下:

#  a mysql or pgres parameter is expected
import sys
syntaxe = f"{sys.argv[0]} mysql / pgres"
erreur = len(sys.argv) != 2
if not erreur:
    sgbd = sys.argv[1].lower()
    erreur = sgbd != "mysql" and sgbd != "pgres"
if erreur:
    print(f"syntaxe : {syntaxe}")
    sys.exit()

#  configure the application
import config
config = config.configure({'sgbd': sgbd})

#  dependencies
from ImpôtsError import ImpôtsError
from flask import request
from myutils import json_response
from flask import Flask
import index_controller

#  data recovery from tax authorities
try:
    #  admindata will be read-only application data
    config['admindata'] = config["layers"]["dao"].get_admindata()
except ImpôtsError as erreur:
    print(f"L'erreur suivante s'est produite : {erreur}")
    sys.exit(1)

#  flask application
app = Flask(__name__)


#  Home URL : /?married=xx&child=yy&salary=zz
@app.route('/', methods=['GET'])
def index():
    #  execute the query
    résultat, statusCode = index_controller.execute(request, config)
    #  we send the answer
    return json_response(résultat, statusCode)


#  hand only
if __name__ == '__main__':
    #  start the server
    app.config.update(ENV="development", DEBUG=True)
    app.run()
  • 第 36–41 行:处理 / 路由;
  • 第 39 行:使用 [IndexController.execute] 函数;

接下来我们将采用这种方法:每个路由将由其专属的模块进行处理。

执行结果与版本 1 相同。

23.2.3. 版本 3

第 3 版引入了身份验证的概念。

[server_03] 脚本变为如下所示:

#  a mysql or pgres parameter is expected
import sys
syntaxe = f"{sys.argv[0]} mysql / pgres"
erreur = len(sys.argv) != 2
if not erreur:
    sgbd = sys.argv[1].lower()
    erreur = sgbd != "mysql" and sgbd != "pgres"
if erreur:
    print(f"syntaxe : {syntaxe}")
    sys.exit()

#  configure the application
import config
config = config.configure({'sgbd': sgbd})

#  dependencies
from ImpôtsError import ImpôtsError
from flask import request
from myutils import json_response
from flask import Flask
from flask_httpauth import HTTPBasicAuth
import index_controller

#  data recovery from tax authorities
try:
    #  config['admindata'] will be read-only application data
    config["admindata"] = config["layers"]["dao"].get_admindata()
except ImpôtsError as erreur:
    print(f"L'erreur suivante s'est produite : {erreur}")
    sys.exit(1)

#  authentication manager
auth = HTTPBasicAuth()


#  authentication method
@auth.verify_password
def verify_credentials(login: str, password: str) -> bool:
    #  user list
    users = config['users']
    #  browse this list
    for user in users:
        if user['login'] == login and user['password'] == password:
            return True
    #  we didn't find
    return False


#  flask application
app = Flask(__name__)


#  Home URL : /?married=xx&child=yy&salary=zz
@app.route('/', methods=['GET'])
@auth.login_required
def index():
    #  execute the query
    résultat, statusCode = index_controller.execute(request, config)
    #  we send the answer
    return json_response(résultat, statusCode)


#  hand only
if __name__ == '__main__':
    #  start the server
    app.config.update(ENV="development", DEBUG=True)
    app.run()
  • 第 21 行:导入身份验证处理程序。Web 服务器有多种身份验证类型。这里使用的是 [HTTP Basic]。每种身份验证类型都遵循特定的客户端/服务器对话;
  • 第 33 行:创建身份验证处理程序的实例;
  • 第 37 行:[@auth.verify_password] 注解标记了当身份验证处理程序需要根据 [HTTP Basic] 协议验证客户端发送的用户名和密码时应执行的函数;
  • 第 55 行:[@auth.login_required] 注解标记了一条必须对 Web 客户端进行身份验证的路由。如果 Web 客户端尚未发送其凭据,Web 服务器将自动使用 HTTP Basic 协议请求这些凭据;

必须安装 [flask_httpauth] 模块:


(venv) C:\Data\st-2020\dev\python\cours-2020\python3-flask-2020\impots\http-servers\01\flask>pip install flask_httpauth
Collecting flask_httpauth
  Downloading Flask_HTTPAuth-4.1.0-py2.py3-none-any.whl (5.8 kB)
Requirement already satisfied: Flask in c:\data\st-2020\dev\python\cours-2020\python3-flask-2020\venv\lib\site-packages (from flask_httpauth) (1.1.2)
Requirement already satisfied: itsdangerous>=0.24 in c:\data\st-2020\dev\python\cours-2020\python3-flask-2020\venv\lib\site-packages (from Flask->flask_httpauth) (1.1.0)
Requirement already satisfied: click>=5.1 in c:\data\st-2020\dev\python\cours-2020\python3-flask-2020\venv\lib\site-packages (from Flask->flask_httpauth) (7.1.2)
Requirement already satisfied: Jinja2>=2.10.1 in c:\data\st-2020\dev\python\cours-2020\python3-flask-2020\venv\lib\site-packages (from Flask->flask_httpauth) (2.11.2)
Requirement already satisfied: Werkzeug>=0.15 in c:\data\st-2020\dev\python\cours-2020\python3-flask-2020\venv\lib\site-packages (from Flask->flask_httpauth) (1.0.1)
Requirement already satisfied: MarkupSafe>=0.23 in c:\data\st-2020\dev\python\cours-2020\python3-flask-2020\venv\lib\site-packages (from Jinja2>=2.10.1->Flask->flask_httpauth) (1.1.1
)
Installing collected packages: flask-httpauth
Successfully installed flask-httpauth-4.1.0

让我们看看 Postman 控制台中发生了什么。您:

  • 创建一个运行配置;
  • 启动 Web 应用程序;
  • 启动您选择的数据库;
  • 使用 Postman 请求 URL [/]

Postman 控制台中的客户端/服务器对话如下:

GET / HTTP/1.1
User-Agent: PostmanRuntime/7.26.1
Accept: */*
Cache-Control: no-cache
Postman-Token: e65e2a28-4fe3-423b-88b3-b3e5a83092b1
Host: localhost:5000
Accept-Encoding: gzip, deflate, br
Connection: keep-alive

HTTP/1.0 401 UNAUTHORIZED
Content-Type: text/html; charset=utf-8
Content-Length: 19
WWW-Authenticate: Basic realm="Authentication Required"
Server: Werkzeug/1.0.1 Python/3.8.1
Date: Fri, 17 Jul 2020 07:05:37 GMT

Unauthorized Access
  • 第 10 行:服务器响应称我们无权访问 URL [/]
  • 第13行:它告知我们应使用哪种身份验证协议,本例中为基本身份验证协议;

可以配置 Postman 根据基本认证协议发送用户凭据:

Image

  • [6-7] 处,输入 [config] 脚本中的凭据: Image

    config['users'] = [
        {
            "login": "admin",
            "password": "admin"
        }
    ]

Postman 控制台中的客户端/服务器对话如下所示:


GET / HTTP/1.1
Authorization: Basic YWRtaW46YWRtaW4=
User-Agent: PostmanRuntime/7.26.1
Accept: */*
Cache-Control: no-cache
Postman-Token: 5ce20822-e87c-4eef-a2f4-b9eaec38d881
Host: localhost:5000
Accept-Encoding: gzip, deflate, br
Connection: keep-alive
 
HTTP/1.0 400 BAD REQUEST
Content-Type: application/json; charset=utf-8
Content-Length: 203
Server: Werkzeug/1.0.1 Python/3.8.1
Date: Fri, 17 Jul 2020 07:20:01 GMT
 
{"réponse": {"erreurs": ["Méthode GET requise avec les seuls paramètres [marié, enfants, salaire]", "paramètre [marié] manquant", "paramètre [enfants] manquant", "paramètre [salaire] manquant"]}}
  • 第2行:Postman客户端以加密形式发送用户凭据 [admin / admin]
  • 第 17 行:服务器响应正确。它报告了错误,因为未发送参数 [married, children, salary](第 1 行),但未报告身份验证错误;

现在让我们使用浏览器(下图为 Firefox)请求该 URL:

Image

  • 与 Postman 一样,Firefox 收到了来自服务器的 HTTP 响应,其中包含以下 HTTP 头部:
1
2
3
4
HTTP/1.0 401 UNAUTHORIZED
WWW-Authenticate: Basic realm="Authentication Required"

Firefox 与其他浏览器一样,在接收到这些标头时不会停止对话框。它会要求用户输入服务器所请求的凭据。在上例中,只需输入 admin / admin 即可收到服务器的响应:

Image

23.3. 税务计算服务器的 Web 客户端

23.3.1. 简介

在前一节中,税费计算服务器的 Web 客户端是一个浏览器。在本节中,Web 客户端将是一个控制台脚本。架构如下所示:

Image

  • Web客户端由层[1-2]组成;
  • Web 服务器由层 [3-9] 组成。如上一节所述;

因此,我们需要编写第 [1-2] 层。

第 [dao] [2] 层必须能够与 Web 服务器 [3] 进行通信。现在我们已经理解了 HTTP 协议,可以利用我们之前学过的 [pycurl] 模块等工具,编写一个与 Web 服务器 [3] 进行通信的脚本。不过,也有专门用于 HTTP 客户端/服务器通信的模块。我们将使用其中之一,即 [requests] 模块:


(venv) C:\Data\st-2020\dev\python\cours-2020\python3-flask-2020\impots\http-servers\01\flask>pip install requests
Collecting requests
  Downloading requests-2.24.0-py2.py3-none-any.whl (61 kB)
     || 61 kB 137 kB/s
Collecting idna<3,>=2.5
  Downloading idna-2.10-py2.py3-none-any.whl (58 kB)
     || 58 kB 692 kB/s
Collecting chardet<4,>=3.0.2
  Downloading chardet-3.0.4-py2.py3-none-any.whl (133 kB)
     || 133 kB 1.3 MB/s
Collecting urllib3!=1.25.0,!=1.25.1,<1.26,>=1.21.1
  Downloading urllib3-1.25.9-py2.py3-none-any.whl (126 kB)
     || 126 kB 1.1 MB/s
Collecting certifi>=2017.4.17
  Downloading certifi-2020.6.20-py2.py3-none-any.whl (156 kB)
     || 156 kB 1.1 MB/s
Installing collected packages: idna, chardet, urllib3, certifi, requests
Successfully installed certifi-2020.6.20 chardet-3.0.4 idna-2.10 requests-2.24.0 urllib3-1.25.9

Web 客户端脚本的目录结构如下:

Image

该脚本将实现 |版本 1| 中描述的批处理模式税费计算应用程序。该应用程序的最新版本为 |版本 5|。以下是其工作原理的简要说明:

  • 需要计算税款的纳税人列在文本文件 [taxpayersdata.txt] 中:
# données valides : id, marié, enfants, salaire
1,oui,2,55555
2,oui,2,50000
3,oui,3,50000
4,non,2,100000
5,non,3,100000
6,oui,3,100000
7,oui,5,100000
8,non,0,100000
9,oui,2,30000
10,non,0,200000
11,oui,3,200000
# on crée des lignes erronées
# pas assez de valeurs
11,12
# des valeurs erronées
x,x,x,x
  • 结果保存在两个文件中:
  • 文本文件 [errors.txt] 列出了在纳税人文件中检测到的错误:

Analyse du fichier C:\Data\st-2020\dev\python\cours-2020\python3-flask-2020\impots\http-clients\01\main/../data/input/taxpayersdata.txt
 
Ligne 15, not enough values to unpack (expected 4, got 2)
Ligne 17, MyException[1, L'identifiant d'une entité <class 'TaxPayer.TaxPayer'> doit être un entier >=0]
  • (续)
    • JSON 文件 [results.json] 包含各纳税人的税款计算结果:

[
  {
    "id": 0,
    "marié": "oui",
    "enfants": 2,
    "salaire": 55555,
    "impôt": 2814,
    "surcôte": 0,
    "taux": 0.14,
    "décôte": 0,
    "réduction": 0
  },
  {
    "id": 1,
    "marié": "oui",
    "enfants": 2,
    "salaire": 50000,
    "impôt": 1384,
    "surcôte": 0,
    "taux": 0.14,
    "décôte": 384,
    "réduction": 347
  },

]

23.3.2. Web 客户端配置

Image

配置通过以下两个脚本进行:

  • [config],用于处理架构层之外的所有配置;
  • [config_layers],用于处理架构层的配置;

[config] 脚本如下:

def configure(config: dict) -> dict:
    import os

    #  step 1 ------

    #  folder of this file
    script_dir = os.path.dirname(os.path.abspath(__file__))

    #  root path
    root_dir = "C:/Data/st-2020/dev/python/cours-2020/python3-flask-2020"

    #  absolute dependencies
    absolute_dependencies = [
        #  project files
        #  BaseEntity, MyException
        f"{root_dir}/classes/02/entities",
        #  InterfaceImpôtsDao, InterfaceImpôtsMétier, InterfaceImpôtsUi
        f"{root_dir}/impots/v04/interfaces",
        #  AbstractImpôtsdao, ImpôtsConsole, ImpôtsMétier
        f"{root_dir}/impots/v04/services",
        #  ImpotsDaoWithAdminDataInDatabase
        f"{root_dir}/impots/v05/services",
        #  AdminData, ImpôtsError, TaxPayer
        f"{root_dir}/impots/v04/entities",
        #  Constants, slices
        f"{root_dir}/impots/v05/entities",
        #  ImpôtsDaoWithHttpClient
        f"{script_dir}/../services",
        #  configuration scripts
        script_dir,
    ]

    #  set the syspath
    from myutils import set_syspath
    set_syspath(absolute_dependencies)

    #  step 2 ------
    #  application configuration with constants
    config.update({
        "taxpayersFilename": f"{script_dir}/../data/input/taxpayersdata.txt",
        "resultsFilename": f"{script_dir}/../data/output/résultats.json",
        "errorsFilename": f"{script_dir}/../data/output/errors.txt",
        "server": {
            "urlServer": "http://127.0.0.1:5000/",
            "authBasic": True,
            "user": {
                "login": "admin",
                "password": "admin"
            }
        }
    }
    )

    #  step 3 ------
    #  layer instantiation
    import config_layers
    config['layers'] = config_layers.configure(config)

    #  we return the configuration
    return config
  • 第 1 行:[configure] 函数将作为参数接收一个用于填充配置信息的字典。该字典可能已被预先填充,也可能为空。在此处,它将为空;
  • 第 40–42 行:由 [dao] 层管理的三个文本文件的绝对路径;
  • 第 43–50 行:与 [server] 键相关联,[dao] 层需要了解的、用于与其进行通信的 Web 服务器的信息:
    • 第 44 行:Web 服务的 URL;
    • 第 45 行:若访问该 URL 需要基本认证,则将 [authBasic] 键设置为 True;
    • 第 46–49 行:若需认证,此处为将进行认证的用户凭据;
  • 第 56–57 行:我们实例化各层(本例中为单个 [dao] 层),并将层引用放置在 [config] 中的 [layers] 键下;

[config_layers] 脚本如下:

def configure(config: dict) -> dict:
    #  instantiation of applicatuon layers

    #  dao layer
    from ImpôtsDaoWithHttpClient import ImpôtsDaoWithHttpClient
    dao = ImpôtsDaoWithHttpClient(config)

    #  make the layer configuration
    return {
        "dao": dao
    }
  • 第 1 行:[configure] 函数接收用于配置应用程序的字典;
  • 第 4–6 行:实例化 [dao] 层。在第 6 行,我们将应用程序配置传递给它,它将从中获取所需的信息;
  • 第 8–11 行:返回一个包含 [dao] 层引用信息的字典;

23.3.3. 主脚本 [main]

主脚本 [main]|版本 5| 中脚本的一个变体:

#  configure the application
import config
config = config.configure({})

#  dependencies
from ImpôtsError import ImpôtsError

#  code
try:
    #  we recover the [dao] layer
    dao = config["layers"]["dao"]
    #  reading taxpayer data
    taxpayers = dao.get_taxpayers_data()["taxpayers"]
    #  taxpayers?
    if not taxpayers:
        raise ImpôtsError(f"Pas de contribuables valides dans le fichier {config['taxpayersFilename']}")
    #  tax calculation
    for taxpayer in taxpayers:
        #  taxpayer is both an input and output parameter
        #  taxpayer will be modified
        dao.calculate_tax(taxpayer)
    #  writing results to a text file
    dao.write_taxpayers_results(taxpayers)
except ImpôtsError  as erreur:
    #  error display
    print(f"L'erreur suivante s'est produite : {erreur}")
finally:
    #  completed
    print("Travail terminé...")
  • 第2-3行:应用程序配置完成;
  • 第 13 行:[dao] 层提供需要计算税款的纳税人列表;
  • 第 21 行:[dao] 层为每位纳税人计算税款;
  • 第 23 行:将结果保存到 JSON 文件中;

23.3.4. [dao]层的实现

Image

让我们回顾一下所使用的客户端/服务器架构:

Image

  • [2, 6]中,我们可以看到[dao]层有两个作用:
    • 它既用于读取纳税人数据,也用于将税务计算结果写入文件系统。我们已经有一个 |AbstractImpôtsDao| 类可以完成这些任务。该类自 |第 4 版| 起便已投入使用;
    • 它与 Web 服务器进行通信 [3]

|版本 5| 中,主脚本 [main] [1] 直接与 [业务][4] 进行通信。 我们希望尽量不修改该脚本。为此,我们将确保 [DAO][2] 实现 [business][4] 的接口。这样,主脚本 [main] 看起来就好像直接与 [business][4] 通信,并且可以完全忽略它位于另一台机器上的事实。

实现 [DAO][2] 的类定义可以如下所示:


class ImpôtsDaoWithHttpClient(AbstractImpôtsDao, InterfaceImpôtsMétier):
  • [TaxDaoWithHttpClient] 类:
    • 继承自 [AbstractTaxDao] 类,从而能够处理与文件系统的通信 [6]
    • 实现了 [InterfaceImpôtsMétier] 接口,从而无需修改 |版本 5| 的主脚本 [main]

[TaxDaoWithHttpClient] 类的完整代码如下:

#  imports
import requests
from flask_api import status

from AbstractImpôtsDao import AbstractImpôtsDao
from AdminData import AdminData
from ImpôtsError import ImpôtsError
from InterfaceImpôtsMétier import InterfaceImpôtsMétier
from TaxPayer import TaxPayer


class ImpôtsDaoWithHttpClient(AbstractImpôtsDao, InterfaceImpôtsMétier):

    #  manufacturer
    def __init__(self, config: dict):
        #  parent initialization
        AbstractImpôtsDao.__init__(self, config)
        #  parameter memory
        self.__config_server = config["server"]

    #  unused [AbstractImpôtsDao] method
    def get_admindata(self) -> AdminData:
        pass

    #  tAX CALCULATION
    def calculate_tax(self: object, taxpayer: TaxPayer, admindata: AdminData = None):
        #  we let the exceptions rise
        #  get parameters
        params = {"marié": taxpayer.marié, "enfants": taxpayer.enfants, "salaire": taxpayer.salaire}
        #  connection with Auth Basic authentication?
        if self.__config_server['authBasic']:
            response = requests.get(
                #  URL of the queried server
                self.__config_server['urlServer'],
                #  URL parameters
                params=params,
                #  basic authentication
                auth=(
                    self.__config_server["user"]["login"],
                    self.__config_server["user"]["password"]))
        else:
            #  connection without Auth Basic authentication
            response = requests.get(self.__config_server['urlServer'], params=params)
        #  check
        print(response.text)
        #  response status code HTTP
        status_code = response.status_code
        #  we put the response jSON in a dictionary
        résultat = response.json()
        #  error if status code other than 200 OK
        if status_code != status.HTTP_200_OK:
            #  we know that the errors have been associated with the [errors] key in the response
            raise ImpôtsError(87, résultat['réponse']['erreurs'])
        #  we know that the result has been associated with the [result] key in the response
        #  modify the input parameter with this result
        taxpayer.fromdict(résultat["réponse"]["result"])
  • 第 21–23 行:[AbstractTaxDao] 类(第 12 行)有一个抽象方法 [get_admindata]。即使我们不使用它,也必须实现该方法(admindata 由服务器管理,而非客户端);
  • 第 26 行:方法 [calculate_tax] 属于接口 [InterfaceImpôtsMétier](第 12 行)。我们必须实现它;
  • 第 15 行:构造函数接收应用程序配置字典作为其唯一参数;
  • 第 16–17 行:父类 [AbstractTaxDao] 通过传递应用程序配置进行初始化。它将在此处找到需要管理的三个文本文件的名称;
  • 第 18–19 行:与税费计算 Web 服务器相关的信息被存储在类内部;
  • 第 26 行:[calculate_tax] 方法接收一个类型为 |Taxpayer| 的对象作为参数。为了符合 [InterfaceImpôtsMétier.calculate_tax] 方法的签名,它还接收一个 [admindata] 参数,该参数本应封装税务管理数据。 在客户端,我们并不拥有这些数据。该参数将始终保持为 [None]。这一变通方案表明 [ImpôtsMétier] 类最初的设计存在缺陷:
  • [calculate_tax] 的签名本应仅为:

def calculate_tax(self, taxpayer: TaxPayer)

并且 [admindata: AdminData] 参数本应传递给类构造函数;

  • 第 27 行:[calculate_tax] 方法的代码未封装在 try / catch / finally 代码块中。这意味着任何异常都不会被处理,并将传播到调用代码中,在本例中即 [main] 脚本。该脚本确实捕获了从 [dao] 层传播过来的所有异常;
  • 第 28 行:税费计算在服务器端进行。因此,我们需要与服务器进行通信。我们使用第 2 行导入的 [requests] 模块来实现这一点;
  • 第 31–43 行:为了向 Web 服务器发送 GET 请求,我们使用 [requests.get] 方法:
    • 第 33–34 行:该方法的第一个参数是要连接的 URL;
    • 第 35–40 行:其余两个参数是命名参数,其顺序无关紧要;
    • 第 35–36 行:命名参数 [params] 的值必须是一个字典,其中包含以 [/url?param1=value1&param2=value2&…] 形式写入 URL 的信息;
    • 第 29 行:包含 Web 服务器所期望的三个参数 [married, children, salary] 的字典。我们无需担心这些参数必须经过的编码(称为 urlencoded)。[requests] 会自动处理这一点;
    • 第 37–40 行:名为 [auth] 的参数是一个包含两个元素(login, password)的元组。它代表 Basic 认证的凭据;
  • 第 44–45 行:这两行仅用于教学目的(调试完成后我们将将其注释掉):
    • [response] 代表服务器的 HTTP 响应;
    • [response.text] 表示该响应中包含的文档正文。在调试过程中,它有助于验证服务器发送给我们的内容;
  • 第 47 行:[response.status_code] 是接收到的响应的 HTTP 状态码。我们的服务器仅发送三种状态码:
    • 200 OK
    • 400 请求错误
    • 500 服务器内部错误
  • 第 49 行:我们的服务器始终返回 JSON 格式,即使发生错误也是如此。[response.json()] 函数会根据接收到的 JSON 字符串创建一个字典。让我们回顾一下 JSON 字符串的两种可能形式:

{"réponse": {"erreurs": ["Méthode GET requise avec les seuls paramètres [marié, enfants, salaire]", "paramètre [marié] manquant", "paramètre [enfants] manquant", "paramètre [salaire] manquant"]}}
{"réponse": {"result": {"id": 0, "marié": "oui", "enfants": 3, "salaire": 200000, "impôt": 42842, "surcôte": 17283, "taux": 0.41, "décôte": 0, "réduction": 0}}}
  • 第 51–53 行:如果状态码不是 200,则抛出异常,并包含响应中的错误消息;
  • 第 56 行:检索税费计算生成的字典,并用其更新 [taxpayer] 输入参数;

23.3.5. 执行

运行客户端:

  • 使用您选择的数据库管理系统启动服务器 [server_03]
  • 运行客户端的 [main] 脚本;

结果将保存在 [data/output] 文件夹中。这些结果与第 5 版相同。

23.4. [dao] 层的测试

让我们回到客户端/服务器应用程序架构:

Image

  • 在客户端代码中,我们已确保 [dao][1] 提供的接口与 [business][3] 完全一致。因此,我们将使用之前学习过的测试类 |TestDaoMétier| 来测试 [business][3]

该测试类将在以下环境中执行:

Image

  • 配置 [2] 与我们刚刚分析过的配置 [1] 完全相同;

测试类 [TestHttpClientDao] 如下所示:

import unittest


class TestHttpClientDao(unittest.TestCase):

    def test_1(self) -> None:
        from TaxPayer import TaxPayer

        #  { 'married': 'yes', 'children': 2, 'salary': 55555,
        #  tax': 2814, 'surcôte': 0, 'décôte': 0, 'réduction': 0, 'taux': 0.14}
        taxpayer = TaxPayer().fromdict({"marié": "oui", "enfants": 2, "salaire": 55555})
        dao.calculate_tax(taxpayer)
        #  check
        self.assertAlmostEqual(taxpayer.impôt, 2815, delta=1)
        self.assertEqual(taxpayer.décôte, 0)
        self.assertEqual(taxpayer.réduction, 0)
        self.assertAlmostEqual(taxpayer.taux, 0.14, delta=0.01)
        self.assertEqual(taxpayer.surcôte, 0)

    

    def test_11(self) -> None:
        from TaxPayer import TaxPayer

        #  { 'married': 'yes', 'children': 3, 'salary': 200000,
        #  tax': 42842, 'surcôte': 17283, 'décôte': 0, 'réduction': 0, 'taux': 0.41}
        taxpayer = TaxPayer().fromdict({'marié': 'oui', 'enfants': 3, 'salaire': 200000})
        dao.calculate_tax(taxpayer)
        #  checks
        self.assertAlmostEqual(taxpayer.impôt, 42842, 1)
        self.assertEqual(taxpayer.décôte, 0)
        self.assertEqual(taxpayer.réduction, 0)
        self.assertAlmostEqual(taxpayer.taux, 0.41, delta=0.01)
        self.assertAlmostEqual(taxpayer.surcôte, 17283, delta=1)


if __name__ == '__main__':

    #  configure the application
    import config
    config = config.configure({})

    #  dao layer
    dao = config['layers']['dao']

    #  test methods are executed
    print("tests en cours...")
    unittest.main()

该类与应用程序第 4 版中已学过的类类似。

  • 第 40-41 行:配置测试环境;
  • 第 44 行:获取 [DAO] 层的引用;
  • 第47-48行:运行测试;

要运行测试,我们需要创建一个 |运行配置|:

Image

  • 我们为控制台脚本创建运行配置,而非为 UnitTest 创建;

运行此配置时,将得到以下结果:

C:\Data\st-2020\dev\python\cours-2020\python3-flask-2020\venv\Scripts\python.exe C:/Data/st-2020/dev/python/cours-2020/python3-flask-2020/impots/http-clients/01/tests/TestHttpClientDao.py
tests en cours...
{"réponse": {"result": {"marié": "oui", "enfants": 2, "salaire": 55555, "impôt": 2814, "surcôte": 0, "taux": 0.14, "décôte": 0, "réduction": 0}}}
....{"réponse": {"result": {"marié": "non", "enfants": 0, "salaire": 200000, "impôt": 64210, "surcôte": 7498, "taux": 0.45, "décôte": 0, "réduction": 0}}}
{"réponse": {"result": {"marié": "oui", "enfants": 3, "salaire": 200000, "impôt": 42842, "surcôte": 17283, "taux": 0.41, "décôte": 0, "réduction": 0}}}
{"réponse": {"result": {"marié": "oui", "enfants": 2, "salaire": 50000, "impôt": 1384, "surcôte": 0, "taux": 0.14, "décôte": 384, "réduction": 347}}}
{"réponse": {"result": {"marié": "oui", "enfants": 3, "salaire": 50000, "impôt": 0, "surcôte": 0, "taux": 0.14, "décôte": 720, "réduction": 0}}}
...{"réponse": {"result": {"marié": "non", "enfants": 2, "salaire": 100000, "impôt": 19884, "surcôte": 4480, "taux": 0.41, "décôte": 0, "réduction": 0}}}
{"réponse": {"result": {"marié": "non", "enfants": 3, "salaire": 100000, "impôt": 16782, "surcôte": 7176, "taux": 0.41, "décôte": 0, "réduction": 0}}}
{"réponse": {"result": {"marié": "oui", "enfants": 3, "salaire": 100000, "impôt": 9200, "surcôte": 2180, "taux": 0.3, "décôte": 0, "réduction": 0}}}
{"réponse": {"result": {"marié": "oui", "enfants": 5, "salaire": 100000, "impôt": 4230, "surcôte": 0, "taux": 0.14, "décôte": 0, "réduction": 0}}}
{"réponse": {"result": {"marié": "non", "enfants": 0, "salaire": 100000, "impôt": 22986, "surcôte": 0, "taux": 0.41, "décôte": 0, "réduction": 0}}}
....
{"réponse": {"result": {"marié": "oui", "enfants": 2, "salaire": 30000, "impôt": 0, "surcôte": 0, "taux": 0.0, "décôte": 0, "réduction": 0}}}
----------------------------------------------------------------------
Ran 11 tests in 0.130s

OK

Process finished with exit code 0

全部 11 个测试均通过。