feat: 新增JoyHub C端测试用例和接口封装
1. 新增C端业务关键字层 (JoyHubC/)： - LoginManage.py - C端登录管理 - UserManage.py - C端用户管理 - ProductManage.py - C端产品管理 - BannerManage.py - C端Banner管理 - AppVersionManage.py - C端版本管理等 2. 新增C端测试用例 (TestCase/接口/JoyHubC/)： - JoyhubC_UserPoint.py - 用户积分测试 - JoyhubC_Product.py - 产品测试 - JoyhubC_Banner.py - Banner测试等 3. 接口层增强： - Dlizhan_interface.py 添加C端接口封装 - 添加网易163邮箱验证码获取功能 4. 配置更新： - hh-qa.robot 添加C端登录配置
This commit is contained in:
@@ -14,6 +14,10 @@ from base_framework.public_tools import utils
|
||||
from base_framework.public_tools.pgsqlhelper import PgSqlHelper
|
||||
import requests
|
||||
import json
|
||||
import re
|
||||
import imaplib
|
||||
import email as email_parser
|
||||
from email.header import decode_header
|
||||
|
||||
obj_log = log.get_logger()
|
||||
obj_runner = Runner()
|
||||
@@ -25,7 +29,11 @@ class DlzhanInterface:
|
||||
self.domain_url = eureka.get_url_from_config()
|
||||
self.pg_db = PgSqlHelper()
|
||||
self.joyhub_domain = "https://joyhub-website-manager-api-test.best-envision.com"
|
||||
self.joyhub_c_domain = "https://joyhub-website-frontend-test.best-envision.com"
|
||||
self.joyhub_c_frontend_url = "https://joyhub-website-frontend-test.best-envision.com/"
|
||||
self.joyhub_c_session = requests.session()
|
||||
self.token = None
|
||||
self.joyhub_c_token = None
|
||||
|
||||
def _get_joyhub_headers(self):
|
||||
headers = {
|
||||
@@ -39,6 +47,47 @@ class DlzhanInterface:
|
||||
def set_joyhub_token(self, token):
|
||||
self.token = token
|
||||
|
||||
def set_joyhub_c_token(self, token):
|
||||
self.joyhub_c_token = token
|
||||
|
||||
def _get_joyhub_c_headers(self):
|
||||
headers = {
|
||||
'accept': '*/*',
|
||||
'Content-Type': 'application/json',
|
||||
'jh-appchannel': '5',
|
||||
'origin': self.joyhub_c_domain,
|
||||
'referer': self.joyhub_c_frontend_url,
|
||||
'tenant-id': '126'
|
||||
}
|
||||
if self.joyhub_c_token:
|
||||
headers['Authorization'] = 'Bearer ' + self.joyhub_c_token
|
||||
return headers
|
||||
|
||||
def _joyhub_c_request(self, method, path, is_check='', note='', return_json=True, **kwargs):
|
||||
url = path if path.startswith('http') else "{}{}".format(self.joyhub_c_domain, path)
|
||||
headers = self._get_joyhub_c_headers()
|
||||
obj_log.info("=========== {} ===========".format(note or path))
|
||||
|
||||
req_params = {}
|
||||
for key, value in kwargs.items():
|
||||
if value is not None and value != '':
|
||||
req_params[key] = value
|
||||
|
||||
req_map = {
|
||||
'GET': lambda: self.joyhub_c_session.get(url, headers=headers, params=req_params, verify=False),
|
||||
'POST': lambda: self.joyhub_c_session.post(url, headers=headers, json=req_params, verify=False),
|
||||
'PUT': lambda: self.joyhub_c_session.put(url, headers=headers, json=req_params, verify=False),
|
||||
'DELETE': lambda: self.joyhub_c_session.delete(url, headers=headers, verify=False)
|
||||
}
|
||||
|
||||
resp = req_map.get(method.upper(), lambda: None)()
|
||||
self._check_resp(is_check, resp)
|
||||
|
||||
if return_json:
|
||||
return resp.json()
|
||||
else:
|
||||
return resp
|
||||
|
||||
def _joyhub_request(self, method, path, is_check='', note='', return_json=True, **kwargs):
|
||||
url = "{}{}".format(self.joyhub_domain, path)
|
||||
headers = self._get_joyhub_headers()
|
||||
@@ -684,12 +733,165 @@ class DlzhanInterface:
|
||||
def kw_in_joyhub_product_payment_recommend_page_get(self, is_check='', **kwargs):
|
||||
return self._joyhub_request('GET', '/admin-api/jh/product-payment-recommend/page', is_check, '获得支付页产品推荐分页', **kwargs)
|
||||
|
||||
# ============ C端-登录公共接口 ============
|
||||
def kw_in_joyhub_c_login_post(self, path, is_check='', **kwargs):
|
||||
return self._joyhub_c_request('POST', path, is_check, 'JoyHub C端登录', **kwargs)
|
||||
|
||||
def kw_in_joyhub_c_get_email_code(self, email, code_pattern=r'\d{4,8}'):
|
||||
auth_code = os.environ.get('JOYHUB_C_EMAIL_AUTH_CODE') or os.environ.get('NETEASE_163_AUTH_CODE')
|
||||
if not auth_code:
|
||||
raise Exception("网易163邮箱授权码不能为空，请先设置环境变量 JOYHUB_C_EMAIL_AUTH_CODE")
|
||||
|
||||
obj_log.info("开始连接网易163邮箱获取验证码 - email: {}".format(email))
|
||||
mail = imaplib.IMAP4_SSL('imap.163.com', 993)
|
||||
try:
|
||||
mail.login(email, auth_code)
|
||||
status, _ = mail.select('INBOX')
|
||||
if status != 'OK':
|
||||
raise Exception("邮箱 {} 无法选择收件箱".format(email))
|
||||
status, data = mail.search(None, 'ALL')
|
||||
if status != 'OK' or not data or not data[0]:
|
||||
raise Exception("邮箱 {} 未查询到邮件".format(email))
|
||||
|
||||
email_ids = data[0].split()
|
||||
for email_id in reversed(email_ids[-20:]):
|
||||
status, msg_data = mail.fetch(email_id, '(RFC822)')
|
||||
if status != 'OK' or not msg_data:
|
||||
continue
|
||||
|
||||
msg = email_parser.message_from_bytes(msg_data[0][1])
|
||||
subject = self._decode_email_header(msg.get('Subject', ''))
|
||||
content = self._get_email_content(msg)
|
||||
match = re.search(code_pattern, '{}\n{}'.format(subject, content or ''))
|
||||
if match:
|
||||
obj_log.info("网易163邮箱验证码获取成功")
|
||||
return match.group(0)
|
||||
|
||||
raise Exception("邮箱 {} 最近20封邮件中未匹配到验证码".format(email))
|
||||
finally:
|
||||
try:
|
||||
mail.logout()
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
def _decode_email_header(self, value):
|
||||
decoded_parts = decode_header(value)
|
||||
result = ''
|
||||
for part, charset in decoded_parts:
|
||||
if isinstance(part, bytes):
|
||||
result += part.decode(charset or 'utf-8', errors='ignore')
|
||||
else:
|
||||
result += part
|
||||
return result
|
||||
|
||||
def _get_email_content(self, msg):
|
||||
contents = []
|
||||
if msg.is_multipart():
|
||||
for part in msg.walk():
|
||||
content_type = part.get_content_type()
|
||||
content_disposition = str(part.get('Content-Disposition'))
|
||||
if content_type in ('text/plain', 'text/html') and 'attachment' not in content_disposition:
|
||||
payload = part.get_payload(decode=True)
|
||||
if payload:
|
||||
charset = part.get_content_charset() or 'utf-8'
|
||||
contents.append(payload.decode(charset, errors='ignore'))
|
||||
else:
|
||||
payload = msg.get_payload(decode=True)
|
||||
if payload:
|
||||
charset = msg.get_content_charset() or 'utf-8'
|
||||
contents.append(payload.decode(charset, errors='ignore'))
|
||||
return '\n'.join(contents)
|
||||
|
||||
# ============ C端-用户接口 ============
|
||||
def kw_in_joyhub_c_client_get_point_get(self, is_check=''):
|
||||
return self._joyhub_c_request('GET', '/web-api/jh/client/get/point', is_check, '查询当前用户积分')
|
||||
|
||||
# ============ C端-Banner信息接口 ============
|
||||
def kw_in_joyhub_c_banner_get_get(self, is_check='', **kwargs):
|
||||
return self._joyhub_c_request('GET', '/web-api/jh/banner/get', is_check, '获得Banner管理', **kwargs)
|
||||
|
||||
def kw_in_joyhub_c_banner_page_get(self, is_check='', **kwargs):
|
||||
return self._joyhub_c_request('GET', '/web-api/jh/banner/page', is_check, '获得Banner管理分页', **kwargs)
|
||||
|
||||
# ============ C端-blog信息接口 ============
|
||||
def kw_in_joyhub_c_blog_get_detail_get(self, is_check='', **kwargs):
|
||||
return self._joyhub_c_request('GET', '/web-api/jh/blog/get-detail', is_check, '获得blog详情', **kwargs)
|
||||
|
||||
def kw_in_joyhub_c_blog_get_next_get(self, is_check='', **kwargs):
|
||||
return self._joyhub_c_request('GET', '/web-api/jh/blog/get-next', is_check, '获得blog下一条', **kwargs)
|
||||
|
||||
def kw_in_joyhub_c_blog_page_get(self, is_check='', **kwargs):
|
||||
return self._joyhub_c_request('GET', '/web-api/jh/blog/page', is_check, '获得blog管理分页', **kwargs)
|
||||
|
||||
# ============ C端-blog分类接口 ============
|
||||
def kw_in_joyhub_c_blog_cate_list_get(self, is_check='', **kwargs):
|
||||
return self._joyhub_c_request('GET', '/web-api/jh/blog-cate/list', is_check, '获得blog分类列表', **kwargs)
|
||||
|
||||
# ============ C端-FAQ接口 ============
|
||||
def kw_in_joyhub_c_faq_cate_list_get(self, is_check='', **kwargs):
|
||||
return self._joyhub_c_request('GET', '/web-api/jh/faq/cate-list', is_check, '获得FAQ分类树', **kwargs)
|
||||
|
||||
def kw_in_joyhub_c_faq_list_get(self, is_check='', **kwargs):
|
||||
return self._joyhub_c_request('GET', '/web-api/jh/faq/list', is_check, '获得FAQ列表', **kwargs)
|
||||
|
||||
# ============ C端-news分类接口 ============
|
||||
def kw_in_joyhub_c_news_cate_list_get(self, is_check='', **kwargs):
|
||||
return self._joyhub_c_request('GET', '/web-api/jh/news-cate/list', is_check, '获得news分类列表', **kwargs)
|
||||
|
||||
# ============ C端-news管理接口 ============
|
||||
def kw_in_joyhub_c_news_get_detail_get(self, is_check='', **kwargs):
|
||||
return self._joyhub_c_request('GET', '/web-api/jh/news/get-detail', is_check, '获得news详情', **kwargs)
|
||||
|
||||
def kw_in_joyhub_c_news_get_next_get(self, is_check='', **kwargs):
|
||||
return self._joyhub_c_request('GET', '/web-api/jh/news/get-next', is_check, '获得news下一条', **kwargs)
|
||||
|
||||
def kw_in_joyhub_c_news_page_get(self, is_check='', **kwargs):
|
||||
return self._joyhub_c_request('GET', '/web-api/jh/news/page', is_check, '获得news管理分页', **kwargs)
|
||||
|
||||
# ============ C端-APP版本接口 ============
|
||||
def kw_in_joyhub_c_web_appversion_page_get(self, is_check='', **kwargs):
|
||||
return self._joyhub_c_request('GET', '/web-api/jh/web/appversion/page', is_check, '获取APP版本列表', **kwargs)
|
||||
|
||||
# ============ C端-产品接口 ============
|
||||
def kw_in_joyhub_c_product_get_get(self, is_check='', **kwargs):
|
||||
return self._joyhub_c_request('GET', '/web-api/jh/product/get', is_check, '产品详情', **kwargs)
|
||||
|
||||
def kw_in_joyhub_c_product_page_get(self, is_check='', **kwargs):
|
||||
return self._joyhub_c_request('GET', '/web-api/jh/product/page', is_check, '获得产品分页', **kwargs)
|
||||
|
||||
# ============ C端-支付页产品推荐接口 ============
|
||||
def kw_in_joyhub_web_product_payment_recommend_page_get(self, is_check='', **kwargs):
|
||||
return self._joyhub_request('GET', '/web-api/jh/product-payment-recommend/page', is_check, '获得支付页产品推荐分页', **kwargs)
|
||||
def kw_in_joyhub_c_product_payment_recommend_page_get(self, is_check='', **kwargs):
|
||||
return self._joyhub_c_request('GET', '/web-api/jh/product-payment-recommend/page', is_check, '获得支付页产品推荐分页', **kwargs)
|
||||
|
||||
# ============ C端-合作联系接口 ============
|
||||
def kw_in_joyhub_c_cooperation_create_post(self, is_check='', **kwargs):
|
||||
return self._joyhub_c_request('POST', '/web-api/jh/cooperation/create', is_check, '提交合作联系信息', **kwargs)
|
||||
|
||||
# ============ C端-二维码访问统计接口 ============
|
||||
def kw_in_joyhub_c_download_qrcode_get_get(self, is_check='', **kwargs):
|
||||
return self._joyhub_c_request('GET', '/web-api/jh/download-qrcode/get', is_check, '获取二维码信息', **kwargs)
|
||||
|
||||
def kw_in_joyhub_c_download_qrcode_increment_post(self, is_check='', **kwargs):
|
||||
return self._joyhub_c_request('POST', '/web-api/jh/download-qrcode/increment', is_check, '增加二维码访问/点击次数', **kwargs)
|
||||
|
||||
# ============ C端-FAQ联系我们接口 ============
|
||||
def kw_in_joyhub_c_faq_contact_us_create_post(self, is_check='', **kwargs):
|
||||
return self._joyhub_c_request('POST', '/web-api/jh/faq-contact-us/create', is_check, '提交FAQ联系信息', **kwargs)
|
||||
|
||||
# ============ C端-点赞记录接口 ============
|
||||
def kw_in_joyhub_c_like_info_create_post(self, is_check='', **kwargs):
|
||||
return self._joyhub_c_request('POST', '/web-api/jh/like-info/create', is_check, '创建点赞记录', **kwargs)
|
||||
|
||||
def kw_in_joyhub_c_like_info_delete_post(self, is_check='', **kwargs):
|
||||
return self._joyhub_c_request('POST', '/web-api/jh/like-info/delete', is_check, '取消点赞', **kwargs)
|
||||
|
||||
def kw_in_joyhub_c_like_info_get_get(self, is_check='', **kwargs):
|
||||
return self._joyhub_c_request('GET', '/web-api/jh/like-info/get', is_check, '获得点赞记录', **kwargs)
|
||||
|
||||
def kw_in_joyhub_c_like_info_page_get(self, is_check='', **kwargs):
|
||||
return self._joyhub_c_request('GET', '/web-api/jh/like-info/page', is_check, '获得点赞记录分页', **kwargs)
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
test = DlzhanInterface()
|
||||
a = test.kw_in_zhyy_purchase_todo_get(user="purchase")
|
||||
print(a)
|
||||
print()
|
||||
|
||||
Reference in New Issue
Block a user