各种python工具:测试、aio助手等。
tipsi-tools的Python项目详细描述
关于此软件包
下面是一组内部工具,在不同的项目之间在内部共享。最初大多数工具都与测试相关,因此它们为测试中的各种情况提供了一些基类
注意:我们所有的工具实际上只支持3.5+python。
有些可能可以与其他版本一起使用,但我们将从所有这些拐杖中解放出来,将诸如async/await
之类的东西移植到较低版本,因此如果它工作正常(如果不工作),请随意发送PR,但它不会一直被合并。
测试助手
apiurls
定义在tipsi\u tools/testing/\u init\uuu.py
中。定义带格式的嵌套URL时必需。
您可以在设备中使用它,例如:
@pytest.fixture(scope='session')defapi(api_v_base):yieldApiUrls('{}/'.format(api_v_base),{'password_reset_request':'password/request/code/','password_reset':'password/reset/','user_review_list':'user/{user_id}/review/','user_review':'user/{user_id}/review/{review_id}/','wine_review':'wine/{wine_id}/review/','drink_review':'drink/{drink_id}/review/',})deftest_review_list(user,api):resp=user.get_json(api.user_review_list(user_id=user1.id),{'page_size':2})
属性
您可以在tipsi_tools/testing/meta.py
中找到源代码。
目前,它将以prop\ux
开头的方法转换为带缓存的描述符。
classA(metaclass=PropsMeta):defprop__conn(self):conn=SomeConnection()returnconn
变成:
classA:@propertydefconn(self):ifnothasattr(self,'__conn'):setattr(self,'__conn',SomeConnection())returnself.__conn
因此,它允许使用延迟初始化进行相当好的测试风格。喜欢:
classMyTest(TestCase,metaclass=PropsMeta):defprop__conn(self):returnpsycopg2.connect('')defprop__cursor(self):returnself.conn.cursor()deftest_simple_query(self):self.cursor.execute('select 1;')row=self.cursor.fetchone()assertrow[0]==1,'Row: {}'.format(row)
在这里,您只需获取并使用self.cursor
,但会自动获取连接和游标并缓存它们。
这只是一个简单的例子,复杂的测试可以在测试中使用更深层的关系。而且这种方法比复杂的设置方法要简单快捷得多。
其他情况
注意:我们强烈建议将pytest与现有的异步测试插件一起使用
作为异步测试用例的基础,您可以使用它作为预先存在的测试的替换项,以便能够:
- 编写异步测试方法
- 编写异步的
setup
和teardown
方法 - 在
assertraises中使用异步函数
classExampleCase(AIOTestCase):asyncsetUp(self):awaitasync_setup()asynctearDown(self):awaitasync_teardown()asyncdivision(self):1/0asynctest_example(self):awaitself.assertRaises(ZeroDivisionError,self.async_division)
tipsi_tools.unix帮助程序
基本Unix帮助程序
- 在shell中运行-run命令
- suc-wrapper around
run
带有返回代码和stderr检查 - 等待套接字-等待套接字可用(例如,您可以使用
等待套接字('localhost',5432)
- ASUCC-异步版本的
suc
与await
一起使用。支持实时日志记录 - source-类似于bash"source"或"."命令。
- cd-contextmanager对临时更改的目录执行操作
插值系统环境
用系统变量+默认值格式化字符串。
PG_DEFAULTS={'PGDATABASE':'postgres','PGPORT':5432,'PGHOST':'localhost','PGUSER':'postgres','PGPASSWORD':'',}DSN=interpolate_sysenv('postgresql://{PGUSER}:{PGPASSWORD}@{PGHOST}:{PGPORT}/{PGDATABASE}',PG_DEFAULTS)
tipsi_tools.tipsi_logging.jsformatter
使用附加字段启用JSON输出,适合结构化登录ELK或类似解决方案。
接受env_vars
键和应该包含在日志中的环境键。
# this example uses safe_logger as handler (pip install safe_logger)importloggingimportlogging.configLOGGING={'version':1,'disable_existing_loggers':True,'formatters':{'json':{'()':'tipsi_tools.tipsi_logging.JSFormatter','env_vars':['HOME'],},'standard':{'format':'%(asctime)s [%(levelname)s] %(name)s: %(message)s'},},'handlers':{'default':{'level':'DEBUG','class':'safe_logger.TimedRotatingFileHandlerSafe','filename':'test_json.log','when':'midnight','interval':1,'backupCount':30,'formatter':'json',},},'loggers':{'':{'handlers':['default'],'level':'DEBUG',},},}logging.config.dictConfig(LOGGING)log=logging.getLogger('TestLogger')log.debug('test debug')log.info('test info')log.warn('test warn')log.error('test error')
tipsi_tools.mon_server.metricsserver
基于sanic的服务器,以prometheus格式提供度量。
from sanic import Sanic
from tipsi_tools.mon_server import MetricsServer
from tipsi_tools.mon_server.certs import update_certs_loop
app = Sanic()
mserver = MetricsServer(app, status_metric='running{server="localhost"}')
mserver.add_task(update_certs_loop, hosts=['gettipsi.com', 'proofnetwork.io'])
app.run(host='0.0.0.0', port=8000)
tipsi_tools.drf.serializers.enumsserializer
允许您将传入字符串反序列化为enum
值。
您应该手动将枚举序列化程序添加到序列化程序中。
fromenumimportIntEnumfromdjango.dbimportmodelsfromrest_frameworkimportserializersfromtipsi_tools.drf.serializersimportEnumSerializerclassMyEnum(IntEnum):one=1two=2classExampleModel(models.Model):value=models.IntegerField(choices=[(x.name,x.value)forxinMyEnum])classExampleSerializer(serializers.ModelSerializer):value=EnumSerializer(MyEnum)# this allows you to post value as: {'value': 'one'}
由于enum
和integerfield
实现,您可以在querysets中使用enum.value
@pytest.fixture(scope='session')defapi(api_v_base):yieldApiUrls('{}/'.format(api_v_base),{'password_reset_request':'password/request/code/','password_reset':'password/reset/','user_review_list':'user/{user_id}/review/','user_review':'user/{user_id}/review/{review_id}/','wine_review':'wine/{wine_id}/review/','drink_review':'drink/{drink_id}/review/',})deftest_review_list(user,api):resp=user.get_json(api.user_review_list(user_id=user1.id),{'page_size':2})0
tipsi_tools.django.log_requests.loggermiddleware
loggermiddleware将请求meta+原始post数据记录到日志中。
对于django<;1.10,请与我们联系etipsi_tools.django.log_requests.deprecatedLogGermiddleware
tipsi_tools.django.request_uniq
decorator为每个uwsgi请求dict添加一个唯一的作为第一个函数 争论。 对于模拟测试,获取请求唯一缓存
tipsi_tools.django.call_once_on_commit
使函数在事务提交时只调用一次。下面是一些例子
其中函数do_一些有用的
只在之后调用一次
交易已提交。
@pytest.fixture(scope='session')defapi(api_v_base):yieldApiUrls('{}/'.format(api_v_base),{'password_reset_request':'password/request/code/','password_reset':'password/reset/','user_review_list':'user/{user_id}/review/','user_review':'user/{user_id}/review/{review_id}/','wine_review':'wine/{wine_id}/review/','drink_review':'drink/{drink_id}/review/',})deftest_review_list(user,api):resp=user.get_json(api.user_review_list(user_id=user1.id),{'page_size':2})1
对于具有嵌套事务的测试(实际上大多数情况下提交不是
called)覆盖行为在提交时调用一次非常有用
当修饰函数在调用它的地方执行时。
要执行此操作,请在提交函数时模拟
。Pytest夹具示例:
@pytest.fixture(scope='session')defapi(api_v_base):yieldApiUrls('{}/'.format(api_v_base),{'password_reset_request':'password/request/code/','password_reset':'password/reset/','user_review_list':'user/{user_id}/review/','user_review':'user/{user_id}/review/{review_id}/','wine_review':'wine/{wine_id}/review/','drink_review':'drink/{drink_id}/review/',})deftest_review_list(user,api):resp=user.get_json(api.user_review_list(user_id=user1.id),{'page_size':2})2
tipsi_tools.django.fields.choicesEnum
用于模型中字段的选项属性
@pytest.fixture(scope='session')defapi(api_v_base):yieldApiUrls('{}/'.format(api_v_base),{'password_reset_request':'password/request/code/','password_reset':'password/reset/','user_review_list':'user/{user_id}/review/','user_review':'user/{user_id}/review/{review_id}/','wine_review':'wine/{wine_id}/review/','drink_review':'drink/{drink_id}/review/',})deftest_review_list(user,api):resp=user.get_json(api.user_review_list(user_id=user1.id),{'page_size':2})3
tipsi_tools.django.db.utils.set_word_similarity_threshold
允许为默认django数据库连接设置postgres trigram单词相似性阈值
@pytest.fixture(scope='session')defapi(api_v_base):yieldApiUrls('{}/'.format(api_v_base),{'password_reset_request':'password/request/code/','password_reset':'password/reset/','user_review_list':'user/{user_id}/review/','user_review':'user/{user_id}/review/{review_id}/','wine_review':'wine/{wine_id}/review/','drink_review':'drink/{drink_id}/review/',})deftest_review_list(user,api):resp=user.get_json(api.user_review_list(user_id=user1.id),{'page_size':2})4
tipsi_tools.django.contrib.postgres.models.ltreemodel
包含postgres ltree的django模型
@pytest.fixture(scope='session')defapi(api_v_base):yieldApiUrls('{}/'.format(api_v_base),{'password_reset_request':'password/request/code/','password_reset':'password/reset/','user_review_list':'user/{user_id}/review/','user_review':'user/{user_id}/review/{review_id}/','wine_review':'wine/{wine_id}/review/','drink_review':'drink/{drink_id}/review/',})deftest_review_list(user,api):resp=user.get_json(api.user_review_list(user_id=user1.id),{'page_size':2})5
tipsi_tools.django.contrib.postgres.fields.ltreeDescents
查找postgres ltree子体
@pytest.fixture(scope='session')defapi(api_v_base):yieldApiUrls('{}/'.format(api_v_base),{'password_reset_request':'password/request/code/','password_reset':'password/reset/','user_review_list':'user/{user_id}/review/','user_review':'user/{user_id}/review/{review_id}/','wine_review':'wine/{wine_id}/review/','drink_review':'drink/{drink_id}/review/',})deftest_review_list(user,api):resp=user.get_json(api.user_review_list(user_id=user1.id),{'page_size':2})6
tipsi_tools.django.contrib.postgres.fields.ltreelevel
按层深度查找Postgres树
@pytest.fixture(scope='session')defapi(api_v_base):yieldApiUrls('{}/'.format(api_v_base),{'password_reset_request':'password/request/code/','password_reset':'password/reset/','user_review_list':'user/{user_id}/review/','user_review':'user/{user_id}/review/{review_id}/','wine_review':'wine/{wine_id}/review/','drink_review':'drink/{drink_id}/review/',})deftest_review_list(user,api):resp=user.get_json(api.user_review_list(user_id=user1.id),{'page_size':2})7
tipsi_tools.django.db.pgfields.similaritylookup
postgrestext%>;文本
运算符
@pytest.fixture(scope='session')defapi(api_v_base):yieldApiUrls('{}/'.format(api_v_base),{'password_reset_request':'password/request/code/','password_reset':'password/reset/','user_review_list':'user/{user_id}/review/','user_review':'user/{user_id}/review/{review_id}/','wine_review':'wine/{wine_id}/review/','drink_review':'drink/{drink_id}/review/',})deftest_review_list(user,api):resp=user.get_json(api.user_review_list(user_id=user1.id),{'page_size':2})8
tipsi_tools.django.db.pgfields.wordsimilarity
Postgrestext1<;<;>;text2
运算符。它返回1-单词相似性(text1,text2)
@pytest.fixture(scope='session')defapi(api_v_base):yieldApiUrls('{}/'.format(api_v_base),{'password_reset_request':'password/request/code/','password_reset':'password/reset/','user_review_list':'user/{user_id}/review/','user_review':'user/{user_id}/review/{review_id}/','wine_review':'wine/{wine_id}/review/','drink_review':'drink/{drink_id}/review/',})deftest_review_list(user,api):resp=user.get_json(api.user_review_list(user_id=user1.id),{'page_size':2})9
tipsi_tools.drf.filters.numberInfilter
如果整数在用逗号分隔的整数列表中,则匹配的django筛选器
classA(metaclass=PropsMeta):defprop__conn(self):conn=SomeConnection()returnconn0
tipsi_tools.django.mail.mail
使用django模板发送文本和html电子邮件。
classA(metaclass=PropsMeta):defprop__conn(self):conn=SomeConnection()returnconn1
tipsi_tools.django.url.build_absolute_uri
使用django请求对象获取当前页面绝对url的域部分。
classA(metaclass=PropsMeta):defprop__conn(self):conn=SomeConnection()returnconn2
tipsi_tools.drf.forms.use_form
有助于使用序列化程序的强大功能进行简单的API检查。
classA(metaclass=PropsMeta):defprop__conn(self):conn=SomeConnection()returnconn3
tipsi_tools.drf.pagination.apipagenumberpagination
通过指定零页大小,允许关闭分页。
classA(metaclass=PropsMeta):defprop__conn(self):conn=SomeConnection()returnconn4
tipsi_tools.rest_framework.renderers.apirenderer
漂亮的django rest framework api呈现程序,带有错误代码。
classA(metaclass=PropsMeta):defprop__conn(self):conn=SomeConnection()returnconn5
tipsi_tools.rest_framework.handlers.api_exception_处理程序
漂亮的django rest framework api异常处理程序,带有错误代码。
classA(metaclass=PropsMeta):defprop__conn(self):conn=SomeConnection()returnconn6
tipsi_tools.drf.asserts.assert_validation_错误
helper assert函数用于测试以匹配验证错误代码。
classA(metaclass=PropsMeta):defprop__conn(self):conn=SomeConnection()returnconn7
tipsi_tools.aio_utils.dbrecordsProcessorWorker
在postgres db表中等待新记录并处理它们的异步工作进程。
tipsi_tools.aio_utils.dict_query/sql_update
aiopg快捷方式
tipsi_tools.python.execfile
python的2execfile
函数的后台端口。
用法:execfile('path/to/file.py',globals(),locals())
返回:如果文件存在并执行,则为true;如果文件不存在,则为false
tipsi_tools.doc_utils.tipsi_狮身人面像
sphinx扩展,用于生成django restframework序列化程序的文档和http请求的示例。
为了使用它们,请指定包安装的依赖项:
classA(metaclass=PropsMeta):defprop__conn(self):conn=SomeConnection()returnconn8
用法:
classA(metaclass=PropsMeta):defprop__conn(self):conn=SomeConnection()returnconn9
命令
tipsi_env_yaml
将替换为%{env_name}
字符串的模板yaml转换为适当的环境变量。
用法:tipsi_env_yaml src_file dst_file
tipsi_ci_脚本
运行默认CI管道的帮助程序。默认设置为giltab默认值包括阶段:
- 使用临时名称生成Docker映像(默认情况下提交SHA)
- 运行测试(可选)
- 推送分支(默认情况下仅适用于主分支和暂存分支)
- 如果有标签,请按标签
- 使用公用名称缓存图像
- 删除带有临时名称的图像
它针对并行启动进行了优化,因此需要使用唯一的临时名称(--temp name
)。如果可能的话,我们希望保持系统的干净,所以我们最后会删除这个标签。但我们不想一遍又一遍地重复基本步骤,因此我们将使用公共缓存名称(--cache name
)缓存图像,它将删除以前缓存的图像。
请稍候
等待套接字可用/不可用超时。
classA:@propertydefconn(self):ifnothasattr(self,'__conn'):setattr(self,'__conn',SomeConnection())returnself.__conn0
运行文件节拍
- 检查环境变量
-e key=value-e key2=value2
- 转换yaml template
tipsi_env_yaml{template}/tmp/filebeat.yml
- 运行
/usr/bin/filebeat/tmp/filebeat.yml
classA:@propertydefconn(self):ifnothasattr(self,'__conn'):setattr(self,'__conn',SomeConnection())returnself.__conn1
文档序列化程序
- 使用序列化程序列表输出rst
- 为序列化程序生成文档工件
classA:@propertydefconn(self):ifnothasattr(self,'__conn'):setattr(self,'__conn',SomeConnection())returnself.__conn2