132 lines
6.0 KiB
Python
132 lines
6.0 KiB
Python
# -*- coding: utf-8 -*-
|
||
import json
|
||
import logging
|
||
import os
|
||
from urllib.parse import urljoin
|
||
|
||
import allure
|
||
import requests
|
||
|
||
from joyhub_backend.library.auth import JoyhubAuth, TIMEOUT
|
||
|
||
|
||
MANAGER_BASE_URL = os.getenv(
|
||
"JOYHUB_MANAGER_BASE_URL",
|
||
"http://test-manager-api.best-envision.com",
|
||
)
|
||
|
||
|
||
class JoyhubInterface(object):
|
||
def __init__(self):
|
||
self.auth = JoyhubAuth()
|
||
self.session = requests.Session()
|
||
self.base_url = MANAGER_BASE_URL.rstrip("/") + "/"
|
||
|
||
def request(self, case_name, method, path, body=None, query=None, headers=None, expected_code=0):
|
||
url = path if path.startswith("http") else urljoin(self.base_url, path.lstrip("/"))
|
||
request_headers = self.auth.auth_headers()
|
||
if headers:
|
||
request_headers.update({key: value for key, value in headers.items() if key.lower() != "authorization"})
|
||
|
||
with allure.step("操作步骤：{}".format(case_name)):
|
||
self._attach_request(url, method, request_headers, body, query)
|
||
logging.info("case: %s", case_name)
|
||
logging.info("request url: %s", url)
|
||
logging.info("request method: %s", method)
|
||
logging.info("request headers: %s", request_headers)
|
||
logging.info("request body: %s", body)
|
||
logging.info("request query: %s", query)
|
||
|
||
request_kwargs = {
|
||
"method": method,
|
||
"url": url,
|
||
"params": query,
|
||
"headers": request_headers,
|
||
"timeout": TIMEOUT,
|
||
}
|
||
if method.upper() in ("POST", "PUT", "PATCH"):
|
||
request_kwargs["json"] = body
|
||
elif body:
|
||
request_kwargs["params"] = dict(query or {}, **body) if isinstance(body, dict) else query
|
||
response = self.session.request(**request_kwargs)
|
||
self._attach_response(response)
|
||
self._attach_log(case_name, url, method, request_headers, body, query, response)
|
||
logging.info("response status: %s", response.status_code)
|
||
logging.info("response body: %s", response.text)
|
||
|
||
is_business_api = self._is_business_api(url)
|
||
assertion_text = "HTTP状态码为200，且业务code为{}".format(expected_code) if is_business_api else "HTTP状态码为2xx，兼容非JSON/SSE响应"
|
||
allure.attach(assertion_text, "断言内容", allure.attachment_type.TEXT)
|
||
with allure.step("断言内容：{}".format(assertion_text)):
|
||
try:
|
||
if is_business_api:
|
||
assert response.status_code == 200, "HTTP状态码期望200，实际{}，响应{}".format(
|
||
response.status_code, response.text
|
||
)
|
||
data = self._safe_json(response)
|
||
assert isinstance(data, dict), "业务接口响应不是JSON对象，响应{}".format(response.text)
|
||
assert "code" in data, "响应缺少 code 字段，响应{}".format(data)
|
||
assert data.get("code") == expected_code, "业务code期望{}，实际{}，msg={}，响应{}".format(
|
||
expected_code, data.get("code"), data.get("msg"), data
|
||
)
|
||
else:
|
||
assert 200 <= response.status_code < 300, "HTTP状态码期望2xx，实际{}，响应{}".format(
|
||
response.status_code, response.text
|
||
)
|
||
data = self._non_json_result(response)
|
||
allure.attach("无", "断言失败原因", allure.attachment_type.TEXT)
|
||
return data
|
||
except AssertionError as error:
|
||
allure.attach(str(error), "断言失败原因", allure.attachment_type.TEXT)
|
||
raise AssertionError("断言失败原因：{}".format(error))
|
||
|
||
@staticmethod
|
||
def _attach_request(url, method, headers, body, query):
|
||
allure.attach(str(url), "请求url", allure.attachment_type.TEXT)
|
||
allure.attach(str(method), "请求方式", allure.attachment_type.TEXT)
|
||
allure.attach(json.dumps(headers, ensure_ascii=False, indent=2), "请求头", allure.attachment_type.JSON)
|
||
allure.attach(json.dumps(query or {}, ensure_ascii=False, indent=2), "请求query", allure.attachment_type.JSON)
|
||
allure.attach(json.dumps(body or {}, ensure_ascii=False, indent=2), "请求体", allure.attachment_type.JSON)
|
||
|
||
@staticmethod
|
||
def _attach_response(response):
|
||
allure.attach(str(response.status_code), "响应状态码", allure.attachment_type.TEXT)
|
||
content_type = response.headers.get("Content-Type", "")
|
||
attachment_type = allure.attachment_type.JSON if "json" in content_type.lower() else allure.attachment_type.TEXT
|
||
allure.attach(response.text, "响应体/日志", attachment_type)
|
||
|
||
def _is_business_api(self, url):
|
||
return url.startswith(self.base_url)
|
||
|
||
@staticmethod
|
||
def _safe_json(response):
|
||
if not response.text.strip():
|
||
return None
|
||
return response.json()
|
||
|
||
@staticmethod
|
||
def _non_json_result(response):
|
||
try:
|
||
data = response.json()
|
||
except ValueError:
|
||
data = {
|
||
"status_code": response.status_code,
|
||
"content_type": response.headers.get("Content-Type", ""),
|
||
"text": response.text,
|
||
}
|
||
return data
|
||
|
||
@staticmethod
|
||
def _attach_log(case_name, url, method, headers, body, query, response):
|
||
log_text = "\n".join([
|
||
"case: {}".format(case_name),
|
||
"request url: {}".format(url),
|
||
"request method: {}".format(method),
|
||
"request headers: {}".format(headers),
|
||
"request query: {}".format(query or {}),
|
||
"request body: {}".format(body or {}),
|
||
"response status: {}".format(response.status_code),
|
||
"response body: {}".format(response.text),
|
||
])
|
||
allure.attach(log_text, "日志", allure.attachment_type.TEXT)
|