2024年4月5日

Python RSA公钥解密 RSA_public_decrypt

我们一般使用 RSA 的时候，都是公钥加密，私钥解密。

因为从设计上来说公钥大家都有，私钥只有一方有，这样更安全，但实际来说也可以私钥加密，公钥解密。openssl有对应实现，但Python没有，项目中需要用到，所以还得尝试实现下。

Python的rsa库虽然没有提供公钥解密但提供了公钥verification方法，里面有使用公钥解密signature的代码，可以依样画葫芦。

keylength = common.byte_size(pub_key.n)

encrypted = transform.bytes2int(signature)

decrypted = core.decrypt_int(encrypted, pub_key.e, pub_key.n)

clearsig = transform.int2bytes(decrypted, keylength)

def decrypt_int(cyphertext: int, dkey: int, n: int) -> int:

"""Decrypts a cypher text using the decryption key 'dkey', working modulo n"""

assert_int(cyphertext, "cyphertext")

assert_int(dkey, "dkey")

assert_int(n, "n")

message = pow(cyphertext, dkey, n)

return message

同时也支持了长文本解密（分段）。

import Crypto.PublicKey.RSA

from rsa import transform, core, common, PublicKey

def rsa_public_decrypt(cls, pub_key: bytes, data: bytes) -> bytes:

pub_key = Crypto.PublicKey.RSA.import_key(pub_key)

block_size = common.byte_size(pub_key.n)

def decrypt(data):

# 解密

cypher_text = transform.bytes2int(data)

decrypted = core.decrypt_int(cypher_text, pub_key.e, pub_key.n)

clear_text = transform.int2bytes(decrypted, block_size)

try:

# Find the 00 separator between the padding and the message

sep_idx = clear_text.index(b'\x00', 2)

except ValueError:

raise RuntimeError('rsa_public_decrypt failed, clear_text=%s' % clear_text)

clear_text = clear_text[sep_idx + 1:]

return clear_text

# 分段解密

total_length = len(data)

default_length = 128

res = []

offset = 0

while offset < total_length:

res.append(decrypt(data[offset:offset + default_length]))

offset += default_length

return b''.join(res)

不过在实际使用过程中，发现rsa_public_decrypt特别慢，起初认为是rsa算法的问题，因为rsa设计上也不是用来加解密较长的文本。一番定位发现是PublicKey.load_pkcs1比较慢，在我Mac上需要3061000 ns，当然一般来说不需要每次都导入公钥，但我实际场景有点特殊，每次公钥都不一样，这个导入操作需要再优化下。

一番定位发现是底层返回public key对象时有个consistency_check参数写死了True，表示会校验n和e是否互为质数，但n和key的位数有关，一般特别大，一般求GCD（最大公约数）使用辗转相除法，即非常耗CPU，这也是Python的弱势，所以比较慢。

但我的场景可以确定rsa key是合法的，因此可以不校验。在我本地测试仅需290000 ns，提高了一个数量级。

def _import_pkcs1_public(cls, pub_key: bytes):

der = DerSequence().decode(pub_key, nr_elements=2, only_ints_expected=True)

return RSA.construct(der, consistency_check=False)

后记：

这个问题是早些年解决的，那时候还没有ChatGPT，不断搜资料，也搞了好久。不过这个问题ChatGPT现在还是不能直接给出答案，会使用rsa.decrypt(encrypted_message, public_key)，但相关库并不支持这样。如果你了解 RSA 原理的话，稍加引导，它最终还是能给出一个非常接近的答案，调一调其实就能用了，不得不感叹科技的进步。

2021年7月7日

Python import机制探究

最近遇到一个需求，需要动态加载一堆 Python 模块，但是这些 .py 是 pb 文件生成的，无法控制命名，导致导入的时候各种报错。

目录结构

├── main.py

└── temp_dir

├── __init__.py

├── basic

│ ├── __init__.py

│ └── package1.py

├── calendar

│ ├── __init__.py

│ └── package2.py

└── math

├── __init__.py

└── package3.py

package1.py 内容

from calendar import package2

from math import package3

print("import package1")

package2.py 内容

1	print("import package2")

package3.py 内容

1	print("import package3")

可以看见在 basic.package1 中导入了 package2、package3。我们在 main 文件中导入 basic 包。

import importlib

if __name__ == '__main__':

print(importlib.import_module("basic.package1"))

这样肯定不行，报错了：ModuleNotFoundError: No module named 'basic'

sys.path

Google 一下或者稍微了解 Python 的就会知道，Python 导入非系统库需要指定 PYTHONPATH，Python 启动时会读取此环境变量并加入到 sys.path 中。我们可以打印下：

['/mnt/c/Users/PycharmProjects/python-import-test',

'/usr/lib/python38.zip',

'/usr/lib/python3.8',

'/usr/local/lib/python3.8/dist-packages']

可以看到默认加载了当前项目目录、Python系统目录以及 dist-packages（通过 pip 安装的），我们把 temp_dir 目录加进去就行了。代码增加：

1 2	temp_dir = "%s/temp_dir" % os.path.dirname(__file__) sys.path.append(temp_dir)

但还是会报错：ImportError: cannot import name 'package2' from 'calendar' (/usr/lib/python3.8/calendar.py)，不过至少说明，加载路径是设置成功了，basic 包本身导入成功了，只是 from calendar import package2 执行失败了。

这个是一个经典错误，搞 Python 的都会遇到，那就是自己定义的包名不能和系统的重复，一般是改个名字或者加上命名空间，但是我们这个场景中，Python 文件是 pb 生成的，无法修改源文件，修改生成的文件也比较麻烦。也能解决，就是提高我们路径的优先级，改成 sys.path.insert(0, temp_dir) 加载完了再改回来。

BuiltinImporter

不过又有新的报错：ImportError: cannot import name 'package3' from 'math' (unknown location)，有输出 import package2，看着还是导入了系统的 math 库，没有导入自己的，还没有显示路径。这里匪夷所思，卡了很久，继续单步调试代码和查阅资料，发现 Python 其实是支持多种导入方式，甚至可以自定义。importlib._bootstrap._find_spec 中，是循环 sys.meta_path 中的 Importer 哪个找到算哪个。我们可以打印下 sys.meta_path：

1	[<class '_frozen_importlib.BuiltinImporter'>, <class '_frozen_importlib.FrozenImporter'>, <class '_frozen_importlib_external.PathFinder'>]

可以看见最前面有个 BuiltinImporter（具体是在 importlib._bootstrap._install 时注入的），唉，灵光一闪，math 就是一个内建库呀，这里结合源码猜测，C 的库（详见 sys.builtin_module_names）都是通过这个导入的。去掉这个就能导入成功了（记得加载完了还原回来）。

builtin = None

get_class_name = lambda _class: _class.__name__ if hasattr(_class, '__name__') else type(_class).__name__

for importer in sys.meta_path:

if get_class_name(importer) == 'BuiltinImporter':

builtin = importer

break

if builtin:

sys.meta_path.remove(builtin)

终于成功了：

import package2

import package3

import package1

sys.modules

不过，这里只是个 demo。实际情况比这个复杂，还是会报 ImportError: cannot import name 'package2' from 'calendar' (/usr/lib/python3.8/calendar.py)，这里很奇怪，已经保证了加载的优先级，但为什么还不行呢？没办法，扒一下源码，终于找到了一点端倪。importlib._bootstrap._find_and_load 方法中，会先在 sys.modules 里面找，如果有则直接返回，官方文档也有相应说明。

def _find_and_load(name, import_):

"""Find and load the module."""

with _ModuleLockManager(name):

module = sys.modules.get(name, _NEEDS_LOADING)

if module is _NEEDS_LOADING:

return _find_and_load_unlocked(name, import_)

所以还得排除已加载的情况

for exclude in ["calendar", "math"]:

if exclude in sys.modules:

del sys.modules[exclude]

最终到此才彻底解决，完整 demo 代码：https://github.com/iyaozhen/python-import-test

参考资料

https://docs.python.org/3/library/importlib.html

妙用Hook来研究Python的Import机制，https://www.51cto.com/article/527713.html

Python3.7源码剖析之import，https://zhuanlan.zhihu.com/p/361720373

2021年3月29日

Python使用logging为Flask增加logid

我们为了问题定位，常见做法是在日志中加入 logid，用于关联一个请求的上下文。这就涉及两个问题：1. logid 这个“全局”变量如何保存传递。2. 如何让打印日志的时候自动带上 logid（毕竟不能每个打日志的地方都手动传入）

logid保存与传递

传统做法就是讲 logid 保存在 threading.local 里面，一个线程里都是一样的值。在 before_app_request 就生成好，logid并放进去。

import threading

from blueprint.hooks import hooks

thread_local = threading.local()

app = Flask()

app.thread_local = thread_local

import uuid

from flask import Blueprint

from flask import current_app as app

hooks = Blueprint('hooks', __name__)

@hooks.before_app_request

def before_request():

"""

处理请求之前的钩子

:return:

"""

# 生成logid

app.thread_local.logid = uuid.uuid1().time

因为需要一个数字的 logid 所以简单使用 uuid.uuid1().time 一般并发完全够了，不会重复且趋势递增（看logid就能知道请求的早晚）。

打印日志自动带上logid

这个就是 Python 日志库自带的功能了，可以使用 Filter 来实现这个需求。

import logging

# https://docs.python.org/3/library/logging.html#logrecord-attributes

log_format = "%(asctime)s %(levelname)s [%(threadName)s-%(thread)d] %(logid)s %(filename)s:%(lineno)d %(message)s"

file_handler = logging.FileHandler(file_name)

logger = logging.getLogger()

logid_filter = ContextFilter()

file_handler.addFilter(logid_filter)

file_handler.setFormatter(logging.Formatter(log_format))

logger.addHandler(file_handler)

class ContextFilter(logging.Filter):

"""

logging Filter

"""

def filter(self, record):

"""

threading local 获取logid

:param record:

:return:

"""

log_id = thread_local.logid if hasattr(thread_local, 'logid') else '-'

record.logid = log_id

return True

log_format 中我们用了很多系统自带的占位符，但 %(logid)s 默认没有的。每条日志打印输出前都会过 Filter，利用此特征我们就可以把 record.logid 赋值上，最终打印出来的时候就有 logid 了。

虽然最终实现了，但因为是通用化方案，所以有些复杂了。其实官方教程中介绍了一种更加简单的方式：injecting-request-information，看来没事还得多看看官方文档。

2021年3月19日

iTerm2进阶使用技巧

iTerm2 的优点这里不做赘述了，第一次使用的话可以先看看官网的介绍：features、Highlights for New Users。本文主要是结合实际使用场景，介绍一些进阶使用技巧（基本的 Oh My Zsh、rzsz 等配置就不重复说明了）。

跳板机自动登录

现在很多公司登录服务器都需要先登录到一个跳板机然后再登录到目标机器，每次输入密码（一般还是动态的）很麻烦。一般的教程推荐使用 expext 解决这个问题，这里介绍一个更简单、直接的办法。

首先，要先解决登录到跳板机的连接复用问题，这个输入 ssh 本身的范畴，一般教程都是说在 ~/.ssh/config 增加下面的配置：

host *

ControlMaster auto

ControlPath ~/.ssh/master-%r@%h:%p

但这样有个问题，iTerm2 第一次登录的 tab 关闭后，就失效了，再登录就又要密码了。其实再增加一个配置即可。

1 2	ControlPersist yes ServerAliveInterval 60

之前的设置只是实现了连接复用，但 tab 关闭 ssh 进程结束后，连接也被销毁了，ControlPersist 项的意思是进程结束后连接还保持，再有相同的（ControlPath配置）连接，还能继续使用此连接。

再者，每次输入 ssh xxx 很麻烦（而且跳板机登录后还得输入 ssh xxx），当然你可以配置 alias+expext 但也不是太方便，特别是 Windows 转过来的。对此，iTerm2 本身也提供了很强大的 Profiles 功能，能一键登录目标机器（还可以设置快捷键）。

有些自己的常用机器，还可以复制一下这份基础的 relay 配置。

再加一行（Send text at start处）。

还可以把机器打个标签，然后页面上就能分组，方便选择。

还有一个，有时候会发现，有些机器一会儿没操作就会被断开 ssh，一方面可以设置 ServerAliveInterval，但作用不大，因为实际连接服务器的是 relay 跳板机。iTerm2 其实没有可以显式设置这个功能的地方，有些同学可能找到个这个配置： Continue Reading...

2020年3月28日

Python进/线程池使用

在 Python 中，进程/线程是个非常重要的概念，特别是 Python 还有 GIL（同一时刻只有一个线程在执行 Python bytecode）限制，使得 Python 线程并不那么好用。但 GIL 更多的是影响 CPU 密集型任务，实际业务场景更多的是 IO 密集型任务，多线程还是适用绝大多数场景。不过话又说回来，很多时候不太好判断是 IO 密集型多还是 CPU 密集型多，需要在多进程、多线程环境下分别验证。

但多线程和多进程手写起来还是有点差别，好在 multiprocessing.Pool 提供了统一的接口，可以无缝切换：

# 进程池

from multiprocessing import Pool

# dummy（假）的进程，线程池

from multiprocessing.dummy import Pool as ThreadPool

下面介绍下 Pool 类如何使用。

首先是构造函数：

class multiprocessing.Pool([processes[, initializer[, initargs[, maxtasksperchild]]]])

接收 processes initializer initargs maxtasksperchild 4个参数：

processes，就是池里有多少个进程，可以不传，默认 CPU 个数，可以按需多设置几倍

initializer、initargs，如果设置了参数，则会在每个进程初始化的时候调用 initializer(*initargs)。这个非常有用，比如可以在初始化的时候建立连接，连接重用。

maxtasksperchild，用于设置每个子进程执行多少个任务后重启。虽然很简单粗暴，但这是防止内存溢出、资源未释放等问题的常见手段，类似 PHP-FPM 的 pm.max_requests 参数。线程池无此参数。

def _send_request(sql):

"""

发送单个请求

:param request:

:return:

"""

# 取出绑定的数据库连接

db_conn = _send_request.db_conn

db_conn.execute(sql)

def _initializer(func):

"""

线程初始化工作

:param func:

:return:

"""

# 绑定建立的数据库连接到请求函数的属性上

func.db_conn = mysql.connector.connect()

pool = ThreadPool(10, initializer=_initializer, initargs=(_send_request))

result = pool.map(func=_send_request, request_list)

pool.close()

pool.join()

这里有一个 hack 技巧，将初始化的数据库连接，绑定在了请求函数上，这样调用请求函数发送请求时，就不用重新建立连接，直接使用即可。

我们一般使用线程池有两个场景，一是关注执行结果，比如我们并行去 redis mysql 各个地方请求数据，然后整合这些数据，二是不关注执行结果，比如新开一个线程打印一个请求的审计日志，不阻塞主进程返回数据。这里主要是介绍下 apply 和 apply_async 的区别，其它的都类似。

apply(func[, args[, kwds]])，主要是传一个执行函数和参数，阻塞并得到返回结果。

apply_async(func[, args[, kwds[, callback]]])，也是主要传执行函数和参数，但返回的是一个 multiprocessing.pool.AsyncResult 对象，AsyncResult 主要有 get([timeout])、wait([timeout])、ready()、successful() 4 个方法，都很好理解，用的比较多的是 get 方法，给定超时时间内获取执行的结果，如果超时抛出 multiprocessing.TimeoutError 异常。如果 timeout 是 None，则一直就等待，行为就和 apply 一致了，实际上 apply 也是调用的 apply_async get：

apply_async 还有一个有用的参数 callback，相对于异步回调了，一般用于上述的场景二。下面是一些示例：

#!/usr/bin/env python

# -*- coding: utf-8 -*

from multiprocessing.dummy import Pool

import time

callback_start_time = 0

def f_sleep(x):

time.sleep(3)

return x*x

def callback_func(result):

print 'callback result:%s use:%s' % (result, current_time_millis() - callback_start_time)

def current_time_millis():

"""

当前时间戳 ms

:return:

"""

return int(round(time.time() * 1000))

if __name__ == '__main__':

pool = Pool(processes=10)

t1 = current_time_millis()

result = pool.apply(f_sleep, (10,))

t2 = current_time_millis()

print 'apply result:%s use:%s' % (result, t2 - t1)

result = pool.apply_async(f_sleep, (10,))

t3 = current_time_millis()

print 'apply_async result:%s use:%s' % (result, t3 - t2)

result = result.get(timeout=4)

t4 = current_time_millis()

print 'apply_async result2:%s use:%s' % (result, t4 - t3)

callback_start_time = current_time_millis()

result = pool.apply_async(f_sleep, (10,), callback=callback_func)

time.sleep(4) # 看回调结果

运行结果：

apply result:100 use:3001

apply_async result:<multiprocessing.pool.ApplyResult object at 0x10876dc50> use:0

apply_async result2:100 use:3041

callback result:100 use:3002

注意代码的最后一行，这里不仅仅是为了看回调结果，还因为回调是回调到主进程执行，如果主进（线）程已经退出了，那就处理不到回调了，实际使用需要注意运行环境。

最后，multiprocessing 模块的 pool 功能只是其中很小的一部分，但比较实用，后面有新的心得再介绍其它功能。

参考资料：

https://docs.python.org/2.7/library/multiprocessing.html#module-multiprocessing.pool

2019年11月24日

Python Requests Session set-cookie不生效的坑

我们知道 Python Requests库中的 Session 模块有连接池和会话管理的功能，比如请求一个登录接口后，会自动处理 response 中的 set-cookie，下次再请求时会自动把 cookie 带上。但最近出现了一个诡异的事情，cookie 没有自动带上，导致请求 403。

一开始怀疑是登录接口错误了，没有 set-cookie，但抓包发现 response header 中有 set-cookie，打印请求的 response.cookies 也有需要的 cookie。又怀疑是 set-cookie 的格式不对或者其它问题，但用浏览器实际跑了下流程，发现系统一切正常，那基本就是 requests 库的问题了。

没办法，只能 debug 了，单步调试了几轮，基本了解了 requests 的处理方式，首先把请求参数转变为 Request 对象，然后对使用 prepare_request 对 Request 进行预处理，其中有一步 merge_cookies 的操作（还有各种其它处理），把传入的 cookies 和 self.cookies merge 到 RequestsCookieJar 对象上去，这一步也没啥问题，merged_cookies 变量也是对的。后续将预处理过的请求，通过内置的 http adapter 发送出去。http adapter 底层是通过 urllib3.poolmanager 获取到 urllib3.connectionpool 连接（这里是连接池的核心部分），再通过 conn.urlopen 实际发送请求。虽然跟踪了解到了整个请求逻辑，但最终发出的请求还是没有带上需要的 cookie。

问题定位一度陷入僵局，只能再回顾上面的流程，cookie 肯定就是在 merged_cookies 和 conn.urlopen 之间没的，再仔细观察发现，conn.urlopen 请求参数里面压根没有 cookie 字段。

# https://github.com/psf/requests/blob/master/requests/adapters.py#L439

resp = conn.urlopen(

method=request.method,

url=url,

body=request.body,

headers=request.headers,

redirect=False,

assert_same_host=False,

preload_content=False,

decode_content=False,

retries=self.max_retries,

timeout=timeout

)

查阅资料发现，urllib3 的作者说，连接池只处理底层连接，cookie 跟踪等事情应该上层来做。大胆猜测，那 cookie 应该是放在 header 里了，往前捣看看 request.headers 是怎么变动的（此时里面的 Cookie 字段确实不正确）。

再走查代码发现 prepare_request 里面是调用了 PreparedRequest.prepare，其中有一步 prepare_cookies，主要是调用了 cookielib.CookieJar.add_cookie_header 最终将 cookie 放到了 self.headers['Cookie']。但里面有个 request.has_header("Cookie") 的判断，header 中没有 Cookie 字段才会放，不知道为什么这么考虑（最新版 3.8 还是这样），估计是 merge cookie 比较麻烦，但问题确实就出在这里，这次请求之前，Requests Session 已经直接通过 headers['Cookie'] 设置了 cookie。复现代码（修改自官方示例）：

import requests

s = requests.Session()

s.headers.update({

'Cookie': 'k=v'

})

s.get('https://httpbin.org/cookies/set/sessioncookie/123456789')

r = s.get('https://httpbin.org/cookies')

print(r.text)

# 预期：{"cookies":{"k":"v"},"cookies":{"sessioncookie":"123456789"}}

# 实际：{"cookies":{"k":"v"}}，未包含sessioncookie

虽然找到了问题的原因，但又不好解决，总不能不让直接操作 headers['Cookie'] 吧，先不说无法限制使用者，而且之前的代码已经这样做了，改动量非常之大。不过好在，现在自己用的框架是在 Requests Session 上封装了一层，操作 header 都是调用的统一的 update_headers 方法：

def update_headers(self, headers):

"""

更新当前会话的header

:param headers: header字典

"""

self.headers.update(headers)

对 headers['Cookie'] 的操作拦截一下，变成对 cookies 的操作：

def update_headers(self, headers):

"""

更新当前会话的header

:param headers: header字典

"""

for header_key, header_value in headers.items:

if header_key == 'Cookie' or header_key == 'cookie':

c = Cookie.SimpleCookie()

c.load(header_value)

cookies = {}

for key, morsel in c.items():

cookies[key] = morsel.value

requests.utils.add_dict_to_cookiejar(self.cookies, cookies)

del headers[header_key]

self.headers.update(headers)

这样即不用改之前的调用方代码，也防止了后人掉坑。

参考资料

https://requests.readthedocs.io/en/master/user/advanced/#session-objects

https://stackoverflow.com/questions/2422922/python-urllib3-and-how-to-handle-cookie-support

https://urllib3.readthedocs.io/en/latest/reference/index.html#module-urllib3.poolmanager

https://urllib3.readthedocs.io/en/latest/reference/index.html#urllib3.connectionpool.HTTPConnectionPool.urlopen

https://docs.python.org/2/library/cookie.html

2017年10月15日

使用 Python nose 组织 HTTP 接口测试

现在前端 Web、移动端 APP 开发基本上是面向接口编程，后端各种 HTTP 接口已经提供好了，大家只要实现逻辑交互和展示就行。那么问题来了，这么多接口如何方便的进行测试呢？根据个人经验，我认为需要解决三个问题：1. HTTP 接口多种多样，测试程序如何统一？2. 测试程序如何规范组织？3. 接口间有上下文依赖，如何解决？

HTTP 接口多种多样，测试程序如何统一？

后端接口可能来自各个系统，GET/POST 协议的、遵循 RESTful 规范的，使用 session 认证、token 认证的，各式各样的接口都存在。但无论怎么变都无法脱离 HTTP 协议。因为组内的技术栈是 Python，这就很容易想到使用 Python 的 requests 库。首先我们使用requests.Session()会话对象，来进行会话保持和 Header、Cookie 等处理。这里我们可以简单封装一个 HttpSession 类，把一些常用操作和设置封装一下：

#!/usr/bin/env python

# -*- coding: utf-8 -*-

import requests

# 屏蔽https安全警告

# https://urllib3.readthedocs.io/en/latest/advanced-usage.html#ssl-warnings

from requests.packages.urllib3.exceptions import InsecurePlatformWarning

from requests.packages.urllib3.exceptions import InsecureRequestWarning

from requests.packages.urllib3.exceptions import SNIMissingWarning

requests.packages.urllib3.disable_warnings(InsecurePlatformWarning)

requests.packages.urllib3.disable_warnings(InsecureRequestWarning)

requests.packages.urllib3.disable_warnings(SNIMissingWarning)

class HttpSession(requests.Session):

"""

http请求会话

"""

def __init__(self):

super(HttpSession, self).__init__()

# 不校验SSL证书

self.verify = False

# 超时时间

self.timeout = (60, 600)

def update_cookies(self, cookies):

"""

更新当前会话cookie

:param cookies: cookie字典

"""

requests.utils.add_dict_to_cookiejar(self.cookies, cookies)

def update_headers(self, headers):

"""

更新当前会话的header

:param headers: header字典

"""

self.headers.update(headers)

通过HttpSession()来获取一个requests session对象，后续的 HTTP 请求都通过它发起。requests 提供 get()、post() 等各种方法调用，但这会让测试代码显得不统一，这里我们直接调用底层的 request 方法，该方法几乎提供了发起一个 HTTP 请求需要的所有参数，非常强大。

http_session = HttpSession()

def test_get_access_token(self):

"""

获取access_token

"""

params = {

"corpid": 'corp_id',

"corpsecret": 'corp_secret',

}

response = http_session.request(

'GET',

config['dingtalk_oapi'] + '/gettoken',

params=params

).json()

data = response['access_token']

print data is not None

这样每一个 HTTP 接口的测试都可以通过准备参数、发起请求、断言结果三板斧来解决。

测试程序如何规范组织？

前面我们已经解决了如何快捷的发起一个 HTTP 请求的问题，这让我们几行代码就可以测试一个接口的返回值。但你会发现新的问题又来了，各种脚本散乱在一堆，返回值解析各种中间变量，各种配置硬编码在代码中，测试结果全靠 print，这时 nose 框架就派上用场了。nose 是 Python 最流行的一个单测框架，提供测试 case 设置标签、测试 case 查找、强大的断言函数，格式化/可视化报告输出等功能。这样就可以像写单测一样写 HTTP 接口测试了。我们可以把一类接口的测试放在一个类里面，甚至可以把基础变量的定义放在基础测试类里面，其它测试类继承这个基类。

#!/usr/bin/env python

# -*- coding: utf-8 -*-

try:

import ConfigParser

except ImportError:

import configparser as ConfigParser

import os

import sys

import unittest

import nose.config

import ruamel.yaml

from py_http_api_test.http_session import HttpSession

class HttpTest(unittest.TestCase):

"""

http接口测试类

"""

# case配置信息(执行环境)

config = []

# http会话对象

http_session = HttpSession()

@classmethod

def setUpClass(cls):

"""

初始化配置信息和http_session对象

:return:

"""

env = None

nose_cfg = None

argvs = sys.argv[1:]

for idx, arg in enumerate(argvs):

# 直接获取case配置文件路径

if '-env=' in arg:

env = arg.split('=')[-1]

break

# 获取nose的配置文件

if '--config=' in arg:

nose_cfg = arg.split('=')[-1]

if '-c' == arg:

nose_cfg = argvs[idx + 1]

# 未获取到case配置文件路径，尝试从nose配置文件中获取

nose_config_files = nose.config.all_config_files()

if env is None and (nose_cfg is not None or len(nose_config_files) > 0):

if nose_cfg is None:

# 用户未指定nose配置文件，则从系统全局配置文件中获取

nose_cfg = nose_config_files[-1]

if not os.path.isabs(nose_cfg):

nose_cfg = os.getcwd() + '/' + nose_cfg

cf = ConfigParser.ConfigParser()

cf.read(nose_cfg)

try:

env = cf.get('others', 'env')

except ConfigParser.Error:

env = None

if env is not None:

# 参数不是绝对路径

if not os.path.isabs(env):

# 根据当前工作路径获取到绝对路径

env = os.getcwd() + '/' + env

with open(env) as f:

inp = f.read()

cls.config = ruamel.yaml.safe_load(inp)

@classmethod

def tearDownClass(cls):

cls.http_session.close()

HttpTest基类只做了两个工作：1. 创建http会话对象；2. 读取配置文件到类变量config中。配置文件是一个很好的编程实践，这让我们测试程序和数据能够分离，可以通过调用不同的配置文件来测试不同环境的接口，让我们测试程序不光能做线下测试，还能做线上回归和监控，切换成本非常低。

我这里选择 yaml 语法来组织配置信息，yaml 语法风格类似 json，自带基础数据结构，但更加易于书写和阅读，非常适合做配置文件。通过-env=xxx指定配置文件路径（或者 nose.cfg 配置文件中指定），使用 ruamel.yaml 来解析 yaml，转换为 Python 中能直接操作的数据结构。

%YAML 1.2

---

# http://open-dev.dingtalk.com/#/corpAuthInfo

dingtalk_oapi: 'https://oapi.dingtalk.com'

corpid: 'dingc2ac01025c48fb7635c2f4657eb6378f'

corpsecret: 'RTw5Vbsh6uKSOu2G25wOmQjQK6SA2NOHNyfHtlEgzQ-CWmvjoGH3c4a-MWTzaJ6Q'

department_parentid: 52711158

这一切都放在setUpClass方法中，在具体测试 case 运行之前就已经准备好这些工作了。 Continue Reading...

我的编程人生

yaozhen's blog

Tag Archives: python

Python RSA公钥解密 RSA_public_decrypt

Python import机制探究

目录结构

sys.path

BuiltinImporter

sys.modules

参考资料

Python使用logging为Flask增加logid

logid保存与传递

打印日志自动带上logid

iTerm2进阶使用技巧

跳板机自动登录

Python进/线程池使用

使用 Python nose 组织 HTTP 接口测试

HTTP 接口多种多样，测试程序如何统一？

测试程序如何规范组织？