new joyhub_backend

joyhub_backend/library/auth.py

@@ -0,0 +1,148 @@
+# -*- 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)