Pytest测试框架一键动态切换测试环境实现思路及方案

defpytest_addoption(parser):
"""    添加命令行参数    parser.addoption为固定写法    default 设置一个默认值，此处设置默认值为test    choices 参数范围，传入其他值无效    help 帮助信息    """parser.addoption(
"--env", default="test", choices=["dev", "test", "pre"], help="enviroment parameter"    )

@pytest.fixture(scope="session")
defget_env(request):
returnrequest.config.getoption("--env")

deftest_env(get_env):
print(f"The current environment is: {get_env}")

pytest-s-v--envdevtest_env.py::test_env

importpytestfromconfig.configimportURLConf, PasswordConf, UsernameConf, ProductIDConffromapi.jc_commonimportJCCommonfromapi.jc_resourceimportJCResourcefromconfig.db_configimportDBConfigfromcommon.mysql_handlerimportMySQL@pytest.fixture(scope="session")
defget_url(get_env):
"""解析URL"""globalurlifget_env=="test":
print("当前环境为测试环境")
url=URLConf.RS_TEST_URL.valueelifget_env=="dev":
print("当前环境为开发环境")
url=URLConf.RS_DEV_URL.valueelifget_env=="pre":
print("当前环境为预发布环境")
url=URLConf.RS_PRE_URL.valuereturnurl@pytest.fixture(scope="session")
defget_user(get_env):
"""解析登录用户"""globalusername_admin, username_boss# 若get_env获取到的是test，则读取配置文件中测试环境的用户名ifget_env=="test":
username_admin=UsernameConf.RS_TEST_ADMIN.valueusername_boss=UsernameConf.RS_TEST_BOSS.value# 若get_env获取到的是dev，则读取配置文件中开发环境的用户名elifget_env=="dev":
username_admin=UsernameConf.RS_TEST_ADMIN.valueusername_boss=UsernameConf.RS_TEST_BOSS.value# 若get_env获取到的是pre，则读取配置文件中预发布环境的用户名elifget_env=="pre":
username_admin=UsernameConf.RS_TEST_ADMIN.valueusername_boss=UsernameConf.RS_TEST_BOSS.value@pytest.fixture(scope="session")
defget_db(get_env):
"""解析数据库配置"""globaldb_host, db_pwd, db_ssh_host, db_ssh_pwd, db_nameifget_env=="test":
db_host=DBConfig.db_test.get('host')
db_pwd=DBConfig.db_test.get('pwd')
db_ssh_host=DBConfig.db_test.get('ssh_host')
db_ssh_pwd=DBConfig.db_test.get('ssh_pwd')
db_name=DBConfig.db_test.get('dbname_jc')
elifget_env=="dev":
db_host=DBConfig.db_test.get('host')
db_pwd=DBConfig.db_test.get('pwd')
db_ssh_host=DBConfig.db_test.get('ssh_host')
db_ssh_pwd=DBConfig.db_test.get('ssh_pwd')
db_name=DBConfig.db_test.get('dbname_jc')
elifget_env=="pre":
db_host=DBConfig.db_test.get('host')
db_pwd=DBConfig.db_test.get('pwd')
db_ssh_host=DBConfig.db_test.get('ssh_host')
db_ssh_pwd=DBConfig.db_test.get('ssh_pwd')
db_name=DBConfig.db_test.get('dbname_jc')
@pytest.fixture(scope="session")
defjc_common(get_env, get_url):
"""传入解析到的URL、实例化jc项目公共接口类"""product_id=ProductIDConf.JC_PRODUCT_ID.valuejc_common=JCCommon(product_id=product_id, url=get_url)
returnjc_common@pytest.fixture(scope="session")
defjc_resource(get_env, get_url):
"""传入解析到的URL、实例化jc项目测试接口类"""product_id=ProductIDConf.JC_PRODUCT_ID.valuejc_resource=JCResource(product_id=product_id, url=get_url)
returnjc_resource@pytest.fixture(scope="class")
defrs_admin_login(get_user, jc_common):
"""登录的fixture函数"""password=PasswordConf.PASSWORD_MD5.valuelogin=jc_common.login(username=username_shipper, password=password)
shipper_user_id=login["b"]
returnadmin_user_id@pytest.fixture(scope="class")
defjc_get_admin_user_info(jc_common, jc_admin_login):
"""获取用户信息的fixture函数"""user_info=jc_common.get_user_info(user_id=rs_shipper_login)
shipper_cpy_id=user_info["d"]["b"]
returnadmin_cpy_id@pytest.fixture(scope="class")
defuse_db(get_db):
"""链接数据库的fixture函数"""mysql=MySQL(host=db_host, pwd=db_pwd, ssh_host=db_ssh_host, ssh_pwd=db_ssh_pwd, dbname=db_name)
yieldmysqlmysql.disconnect()

fromcommon.http_requestsimportHttpRequestsclassJcCommon(HttpRequests):
def__init__(self, url, product_id):
super(JcCommon, self).__init__(url)
self.product_id=product_iddeflogin(self, username, password):
'''用户登录'''headers= {"product_id": str(self.product_id)}
params= {"a": int(username), "b": str(password)}
response=self.post(uri="/userlogin", headers=headers, params=params)
returnresponsedefget_user_info(self, uid, token):
'''获取用户信息'''headers= {"user_id": str(uid), "product_id": str(self.product_id), "token": token}
response=self.post(uri="/user/login/info", headers=headers)
returnresponse

importrandomfromcommon.http_requestsimportHttpRequestsfromfakerimportFakerclassRSResource(HttpRequests):
def__init__(self, url, product_id):
super(RSResource, self).__init__(url)
self.product_id=product_idself.faker=Faker(locale="zh_CN")
defadd_goods(self, cpy_id, user_id, goods_name, goos_desc='', goods_type='', goos_price=''):
"""新增商品"""headers= {"product_id": str(self.product_id), "cpy_id": str(cpy_id), "user_id": str(user_id)}
params= {"a": goods_name, "b": goos_desc, "c": goods_type, "d": goos_price}
r=self.post(uri="/add/goods", params=params, headers=headers)
returnrdefmodify_goods(self, cpy_id, user_id, goods_name, goos_desc='', goods_type='', goos_price=''):
"""修改商品信息"""headers= {"product_id": str(self.product_id), "cpy_id": str(cpy_id), "user_id": str(user_id)}
params= {"a": car_name, "ab": car_id, "b": company_id, "c": car_or_gua}
r=self.post(uri="/risun/res/car/add/blacklist?md=065&cmd=006", params=params, headers=headers)
returnr

importosimportsyssys.path.append(os.path.dirname(os.path.dirname(os.path.dirname(os.path.realpath(__file__)))))
importallurefromfixture.jc_fixtureimport*fromcommon.parse_excelimportParseExcellogger=LogGen("JC接口Smoke测试").getLog()
@allure.feature("JC项目接口冒烟测试")
classTestJcSmoke:
defsetup_class(self):
self.fake=Faker("zh_CN")
# 将fixture中的jc_resource实例、数据库实例、登录等fixture函数传递给测试用例进行调用@pytest.mark.jc_smoke@allure.story("商品管理")
deftest_01_goods_flow(self, jc_resource, jc_admin_login, jc_get_admin_user_info, use_db):
"""测试商品增删改查接口"""user_id=jc_admin_logincpy_id=jc_get_admin_user_infogoods_name="iphone 14pro max 512G"try:
logger.info(f"新增'{goods_name}'商品")
withallure.step("调用添加商品接口"):
add_goods=jc_resource.add_goods(cpy_id, user_id, goods_name, goods_type=1)
assertadd_goods["a"] ==200self.goods_id=add_goods["d"]
select_db=use_db.execute_sql(
f"SELECT * FROM goods_info WHERE company_id = {cpy_id} AND id = {self.goods_id}")  # 查询数据库是否存在新增的数据assertgoods_nameinstr(select_db)
logger.info(f"商品'{goods_name}'新增成功")
logger.info(f"修改'{goods_name}'的商品信息")
withallure.step("调用修改商品接口"):
modify_goods=jc_resource.modify_goods(cpy_id, user_id, goods_id=self.goods_id, goods_name=goods_name, goods_type=2)
assertmodify_goods["a"] ==200select_db=use_db.execute_sql(
f"SELECT goodsType FROM goods_info WHERE company_id = {cpy_id} AND id = {self.goods_id}")
assertstr(select_db[0]) =='2'logger.info(f"修改'{goods_name}'的商品信息成功")
logger.info(f"开始删除商品'{goods_name}'")
withallure.step("调用删除商品接口"):
del_goods=jc_resource.delete_goods(cpy_id, user_id, goods_id=self.goods_id)
assertdel_goods["a"] ==200select_db=use_db.execute_sql(
f"SELECT * FROM goods_info WHERE id = {self.goods_id}")
print(select_db)
logger.info(f"删除商品'{goods_name}'成功")
exceptAssertionErrorase:
logger.info(f"商品流程测试失败")
raisee