149 lines
5.3 KiB
Python
149 lines
5.3 KiB
Python
# -*- coding: utf-8 -*-
|
||
import json
|
||
import logging
|
||
import os
|
||
|
||
import allure
|
||
import requests
|
||
|
||
|
||
CAPTCHA_URL = os.getenv(
|
||
"JOYHUB_CAPTCHA_URL",
|
||
"http://test-manager-api.best-envision.com/admin/login/captcha",
|
||
)
|
||
LOGIN_BASE_URL = os.getenv(
|
||
"JOYHUB_LOGIN_BASE_URL",
|
||
"http://test-manager-api.best-envision.com",
|
||
)
|
||
LOGIN_PATH = os.getenv("JOYHUB_LOGIN_PATH", "/admin/login/login")
|
||
USERNAME = os.getenv("JOYHUB_USERNAME", "guojiabao")
|
||
PASSWORD = os.getenv("JOYHUB_PASSWORD", "gjb123456")
|
||
CAPTCHA = os.getenv("JOYHUB_CAPTCHA", "1111")
|
||
TIMEOUT = int(os.getenv("JOYHUB_TIMEOUT", "20"))
|
||
TENANT_ID = os.getenv("JOYHUB_TENANT_ID", "126")
|
||
|
||
|
||
class JoyhubAuth(object):
|
||
def __init__(self):
|
||
self.session = requests.Session()
|
||
self._token = None
|
||
|
||
@property
|
||
def login_url(self):
|
||
return LOGIN_BASE_URL.rstrip("/") + LOGIN_PATH
|
||
|
||
def get_captcha_key(self):
|
||
with allure.step("前置：获取登录验证码 key"):
|
||
logging.info("GET %s", CAPTCHA_URL)
|
||
response = self.session.get(CAPTCHA_URL, timeout=TIMEOUT)
|
||
self._attach_response("captcha", response)
|
||
response.raise_for_status()
|
||
data = response.json()
|
||
key = self._extract_key(data)
|
||
assert key, "验证码接口未返回 key，响应：{}".format(data)
|
||
return key
|
||
|
||
def login(self):
|
||
if self._token:
|
||
return self._token
|
||
key = self.get_captcha_key()
|
||
body = {
|
||
"key": key,
|
||
"username": USERNAME,
|
||
"password": PASSWORD,
|
||
"captcha": CAPTCHA,
|
||
}
|
||
headers = {"Content-Type": "application/json", "tenant-id": TENANT_ID}
|
||
with allure.step("前置：登录并获取 token"):
|
||
logging.info("POST %s", self.login_url)
|
||
logging.info("request headers: %s", headers)
|
||
logging.info("request body: %s", self._safe_body(body))
|
||
response = self.session.post(self.login_url, json=body, headers=headers, timeout=TIMEOUT)
|
||
self._attach_request(self.login_url, "POST", headers, self._safe_body(body))
|
||
self._attach_response("login", response)
|
||
response.raise_for_status()
|
||
data = response.json()
|
||
token = self._extract_token(data)
|
||
assert token, "登录接口未返回 token，响应：{}".format(data)
|
||
self._token = token if token.startswith("Bearer ") else "Bearer " + token
|
||
return self._token
|
||
|
||
def auth_headers(self):
|
||
return {
|
||
"Authorization": self.login(),
|
||
"Content-Type": "application/json",
|
||
"tenant-id": TENANT_ID,
|
||
}
|
||
|
||
@staticmethod
|
||
def _extract_key(data):
|
||
if isinstance(data, dict):
|
||
for field in ("key", "captchaKey", "captcha_key", "uuid"):
|
||
if data.get(field):
|
||
return data.get(field)
|
||
nested = data.get("data")
|
||
if isinstance(nested, dict):
|
||
for field in ("key", "captchaKey", "captcha_key", "uuid"):
|
||
if nested.get(field):
|
||
return nested.get(field)
|
||
if isinstance(nested, str):
|
||
return nested
|
||
return None
|
||
|
||
@staticmethod
|
||
def _extract_token(data):
|
||
token_fields = (
|
||
"token",
|
||
"access_token",
|
||
"accessToken",
|
||
"Authorization",
|
||
"authorization",
|
||
"userToken",
|
||
"user_token",
|
||
"jwt",
|
||
)
|
||
if isinstance(data, dict):
|
||
for field in token_fields:
|
||
token = data.get(field)
|
||
if JoyhubAuth._looks_like_token(token):
|
||
return token
|
||
nested = data.get("data")
|
||
if isinstance(nested, dict):
|
||
for field in token_fields:
|
||
token = nested.get(field)
|
||
if JoyhubAuth._looks_like_token(token):
|
||
return token
|
||
return None
|
||
|
||
@staticmethod
|
||
def _looks_like_token(value):
|
||
if not isinstance(value, str):
|
||
return False
|
||
value = value.strip()
|
||
if not value:
|
||
return False
|
||
if any(ord(char) > 127 for char in value):
|
||
return False
|
||
return len(value) >= 16 or value.startswith("Bearer ")
|
||
|
||
@staticmethod
|
||
def _safe_body(body):
|
||
safe = dict(body)
|
||
if "password" in safe:
|
||
safe["password"] = "******"
|
||
return safe
|
||
|
||
@staticmethod
|
||
def _attach_request(url, method, headers, body):
|
||
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(body, ensure_ascii=False, indent=2), "请求体", allure.attachment_type.JSON)
|
||
|
||
@staticmethod
|
||
def _attach_response(name, response):
|
||
logging.info("%s response status: %s", name, response.status_code)
|
||
logging.info("%s response body: %s", name, response.text)
|
||
allure.attach(str(response.status_code), "{} 响应状态码".format(name), allure.attachment_type.TEXT)
|
||
allure.attach(response.text, "{} 响应体".format(name), allure.attachment_type.JSON)
|