import sys
import os
sys.path.insert(0, os.path.abspath('../dzh_pb2/'))
sys.path.append('../../')
import base64
from _datetime import datetime
import json
import requests
import pytest
from loguru import logger
from ycyj_hangqing.app_hangqing import PORT
IP = 'http://localhost:{}'.format(PORT)
token = 'test_app'
class Dict(dict):
__setattr__ = dict.__setitem__
__getattr__ = dict.__getitem__
def dict_to_object(dictObj):
"""字典转换为对象,方便调用"""
if not isinstance(dictObj, dict):
return dictObj
inst = Dict()
for k, v in dictObj.items():
inst[k] = dict_to_object(v)
return inst
def dict_to_parm_str(dict):
"""字典 转 parm 拼接"""
parma = set()
for k, v in dict.items():
parma.add('{}={}'.format(k, v))
return '&'.join(parma)
@pytest.fixture(scope="module")
def servers():
"""模块初始化代码 servers"""
return dict_to_object(dict(url=IP))
def assert_getreq(urlstr):
logger.debug(urlstr)
resp = requests.get(urlstr)
logger.debug(resp.status_code)
if resp.status_code != 200:
raise Exception(
'status_code error, status_code={}'.format(resp.status_code))
respjson = resp.content.decode('utf-8')
logger.debug(respjson)
respData = json.loads(respjson)
if respData['State'] != 1:
raise Exception(
'respData State error,respjson={}'.format(respjson))
def get_ts_time():
ts = str(int(datetime.utcnow().timestamp())).encode()
ts = base64.b64encode(ts).hex()
return ts
class Test_GetFenleiCodeNameList2():
"""
测试一个接口, 每个测试类里包含多个方法,测试不同情况
PC 分类接口
"""
@pytest.fixture(scope="class")
def paths(self):
return "/api/GetFenleiCodeNameList2"
@pytest.fixture(scope="class")
def parms(self):
return dict(ts=get_ts_time(),
fenlei=0,
start=0,
count=20,
texttype=0,
orderby='ZhangFu',
field='Obj,ZhongWenJianCheng,ZhangFu,ZhangDie',
desc='True',
token=token)
def test_0(self, servers, paths, parms): # 分类0 沪深市场
urlstr = "{}{}?{}".format(servers.url, paths, dict_to_parm_str(parms))
assert_getreq(urlstr)
def test_201(self, servers, paths, parms): # 分类201 港股
parms['fenlei'] = 201
urlstr = "{}{}?{}".format(servers.url, paths, dict_to_parm_str(parms))
assert_getreq(urlstr)
def test_900(self, servers, paths, parms): # 分类900 全球指数
parms['fenlei'] = 900
urlstr = "{}{}?{}".format(servers.url, paths, dict_to_parm_str(parms))
assert_getreq(urlstr)
class Test_GetFenleiCodeNameList2App():
"""
测试一个接口, 每个测试类里包含多个方法,测试不同情况
APP 分类接口
"""
@pytest.fixture(scope="class")
def paths(self):
return "/api/GetFenleiCodeNameList2App"
@pytest.fixture(scope="class")
def parms(self):
return dict(ts=get_ts_time(),
fenlei=0,
page=1,
prePageCount=100,
texttype=0,
paixu=4,
field='',
token=token)
def test_0(self, servers, paths, parms): # 分类0 沪深市场
urlstr = "{}{}?{}".format(servers.url, paths, dict_to_parm_str(parms))
assert_getreq(urlstr)
def test_201(self, servers, paths, parms): # 分类201 港股
parms['fenlei'] = 201
urlstr = "{}{}?{}".format(servers.url, paths, dict_to_parm_str(parms))
assert_getreq(urlstr)
def test_900(self, servers, paths, parms): # 分类900 全球指数
parms['fenlei'] = 900
urlstr = "{}{}?{}".format(servers.url, paths, dict_to_parm_str(parms))
assert_getreq(urlstr)
class Test_GetPankouDataList2():
"""
测试一个接口, 每个测试类里包含多个方法,测试不同情况
PC 盘口数据
"""
@pytest.fixture(scope="class")
def paths(self):
return "/api/GetPankouDataList2"
@pytest.fixture(scope="class")
def parms(self):
return dict(ts=get_ts_time(),
codes="000001.SH,000002.SZ,00636.HK,B$99375",
token=token)
def test_01(self, servers, paths, parms):
urlstr = "{}{}?{}".format(servers.url, paths, dict_to_parm_str(parms))
assert_getreq(urlstr)
class Test_GetPankouDataList2App():
"""
测试一个接口, 每个测试类里包含多个方法,测试不同情况
APP 盘口数据
"""
@pytest.fixture(scope="class")
def paths(self):
return "/api/GetPankouDataList2App"
@pytest.fixture(scope="class")
def parms(self):
return dict(ts=get_ts_time(),
codes="000001.SH,000002.SZ,00636.HK,B$99375",
token=token)
def test_01(self, servers, paths, parms):
urlstr = "{}{}?{}".format(servers.url, paths, dict_to_parm_str(parms))
assert_getreq(urlstr)
class Test_GetShuoZhaiBanKuai():
"""
测试一个接口, 每个测试类里包含多个方法,测试不同情况
PC 股票所在板块列表 接口
"""
@pytest.fixture(scope="class")
def paths(self):
return "/api/GetShuoZhaiBanKuai"
@pytest.fixture(scope="class")
def parms(self):
return dict(ts=get_ts_time(),
code='600000.SH',
token=token)
def test_0(self, servers, paths, parms):
urlstr = "{}{}?{}".format(servers.url, paths, dict_to_parm_str(parms))
assert_getreq(urlstr)
class Test_GetSuoZhaiBanKuaiApp():
"""
测试一个接口, 每个测试类里包含多个方法,测试不同情况
APP 股票所在板块列表 接口
"""
@pytest.fixture(scope="class")
def paths(self):
return "/api/GetSuoZhaiBanKuaiApp"
@pytest.fixture(scope="class")
def parms(self):
return dict(ts=get_ts_time(),
code='600000.SH',
token=token)
def test_0(self, servers, paths, parms):
urlstr = "{}{}?{}".format(servers.url, paths, dict_to_parm_str(parms))
assert_getreq(urlstr)
class Test_GetShiChangFenXi():
"""
沪深 市场分析
"""
@pytest.fixture(scope="class")
def paths(self):
return "/api/GetShiChangFenXi"
@pytest.fixture(scope="class")
def parms(self):
return dict(ts=get_ts_time(),token=token)
def test_0(self, servers, paths, parms):
urlstr = "{}{}?{}".format(servers.url, paths, dict_to_parm_str(parms))
assert_getreq(urlstr)