方便在数据管道内测试数据的库。结果被推送到某种类型的消息队列中，供应用程序、持久性等使用。
dtest-framework的Python项目详细描述
dtest
方便在数据管道内测试数据的库。结果被推送到某种类型的消息队列中，供应用程序、持久性等使用。
支持的消息队列/流媒体平台
- [X]拉比特MQ
- []MQTT
- []redis
- []卡夫卡
- []动觉
安装
pip3 install dtest-framework
单元测试
使用pytest设置测试
使用pip3 install -U pytest
使用根目录中的pytest运行测试。
圆ci
有一个.circleci/config.yml文件将针对python 3.6执行构建和单元测试。
快速启动
from dtest.dtest import Dtest
from hamcrest import *
# If publishing to a RabbitMQ queue, specify 'queue' \
# If publishing to a key-value store, specify 'kv-store' \
# Or specify both
connectionConfig = {
"queue": {
"host": "localhost",
"username": "guest",
"password": "guest",
"exchange": "test.dtest",
"exchange_type": "fanout"
},
"kv-store": {
"api_url": "localhost:8080/api/",
"retrieve_path": "getKeyValue/",
"publish_path": "postKeyValue/"
}
}
metadata = {
"description": "This is a test of the assertCondition",
"topic": "test.dtest",
"ruleSet": "Testing some random data",
"dataSet": "random_data_set_123912731.csv"
}
dt = Dtest(connectionConfig, metadata)
dsQubert = [0,1]
dt.assert_that(dsQubert, has_length(2))
// True
dt.publish()
// Publishes test suite to MQ server
////////////////////////////////////////
// Store value in KV store for later use
dt.publishKeyValue('some-descriptor-dsQubert-length', len(dqQubert))
// Retrieve value from KV store to compare other files against
avg_count = dt.retrieveKeyValue('some-descriptor-dsQubert-length')
dt.assert_that(dsQubert, has_length(avg_count))
自定义处理程序
可以创建自定义消息队列和键值存储处理程序。根据需要实现从dtest.handler.MqHandler或dtest.handler.KvHandler继承的类。
class MqHandler:
@classmethod
def version(self): return "1.0"
@abstractmethod
def connect(self): raise NotImplementedError
@abstractmethod
def publishResults(self): raise NotImplementedError
@abstractmethod
def closeConnection(self): raise NotImplementedError
class KvHandler:
@classmethod
def version(self): return "1.0"
@abstractmethod
def retrieve(self): raise NotImplementedError
@abstractmethod
def publish(self): raise NotImplementedError
欢迎加入QQ群-->： 979659372
