企业官网建设流程全解析

前言：

代码同步分享平台：

【Pytest框架学习】分层架构

Version 01链接：

【Pytest框架学习】0-1搭建API测试框架-Version01跑通test case-CSDN博客

Github地址：后续补充

思路：通过分层架构，将“业务逻辑”、“数据驱动”和“环境配置”解耦，实现测试用例的高内聚、低耦合与可复用性。

我们先跑通核心的测试流程，这个在Version01我们已经做到了。跑通简单的测试案例之后，你就会发现，有一些数据不能写死，人工添加案例（可能是上百条）太累了、断言还可以细分，还一些重复的动作，比如读取文件、读取参数、批量生成案例、环境隔离、可视化页面等等；于是我们再看看Version01的代码里，哪里有这些功能，把这些动作剥离出来单独成为一个类/单元（util）、相互调用，哪里没有这些功能，可以拓展补充；从而实现框架复用，做到后续尽可能只要维护设计到参数，具体内容的模块... ...

此版本改动

1. 优化数据驱动

数据驱动由Json文件改为Yaml文件：方便维护；

2. 新增日志模块

将输出日志保存为文件，且在终端打印结果；

3. 新增可视化模块

（1）可视化展示：

包括：测试通过统计、测试案例对应的请求数据、响应数据等信息；

（2）为什么选择选择Allure？

Allure Report 是一个测试报告工具，核心特点：

• 可视化测试报告（图表、趋势、用例分布）

• 支持多种测试框架（Pytest / JUnit / TestNG / Cucumber / Appium）

• 能记录失败步骤、截图、日志

• 支持步骤级别的测试过程展示（不像传统报告只看结果）

• 可生成历史趋势分析（CI/CD 很常用）

（3）配置环境

安装python依赖：

pip install allure-pytest

命令行安装：

mac安装allure命令行工具：

brew install allure

安装过程中发现本机环境太老，没有办法安装X code，选择手动下载安装包安装：

Releases · allure-framework/allure2 · GitHub

下载.zip文件解压后移动到系统的以下目录下，先创建，然后移入，指令如下：

sudo mkdir -p /opt/allure

sudo mv allure-2.* /opt/allure

Mac电脑zsh终端环境变量配置文件：.zshrc文件中添加如下字段，然后保存：

#allure export ALLURE_HOME=/opt/allure/allure-2.38.1 export PATH=$ALLURE_HOME/bin:$PATH

终端输入指令， 出现allure即可。

allure --version

conda环境配置：

使用conda环境无法找到allure指令，依次执行如下指令：

conda activate 你的conda环境

mkdir -p $CONDA_PREFIX/etc/conda/activate.d

nano $CONDA_PREFIX/etc/conda/activate.d/allure.sh

->nano/PICO 编辑器写入：

export PATH=/opt/allure/allure-2.38.1/bin:$PATH

退出conda环境：

conda deactivate

再进入conda环境：

conda activate 你的conda环境

验证：

which allureallure --version

Allure可视化：（在跑完测试生成report文件后）：

a. 在项目根目录运行(本地)：

allure serve reports/allure-results

b.或者（本地）

cd /Users/xxx/Desktop/PythonFloder/PytestProject/PytestProjectLocal02/report/allure-html（html目录）

python -m http.server 8080

浏览器打开:

http://127.0.0.1:8080

c.或者（本地）

python -m http.server 8000

http://127.0.0.1:8000/report/allure-html/

d. 在线运行：容易出现loading

使用浏览器打开该网址： open report/allure-html/index.html

问题：浏览器打开——加载静态资源遇到了loarding的情况

解释：Allure 报告依赖很多静态资源文件，file://模式下浏览器常常限制这些资源加载；用本地服务变成http:// 后，浏览器才会把它当成正常网页完整加载。

4. 新增测试结果输出到飞书群聊模块

将输出测试结果发送到飞书：

在飞书创建群聊，添加机器人（获取webhook链接（不要泄漏）放入环境变量配置），给机器人设置关键词触发验证（发送给机器人的消息要包括这条验证）

5. 优化断言

——

6. 添加统一执行入口：run.py

cd tests

python run.py

代码示例

安装依赖：

pip install -e .

(1) case_loader.py

yaml文件示例：

cases: - case_id: home_article_list_page_0 name: 首页文章列表-第1页 enabled: true request: method: GET path: /article/list/0/json params: {} json: null data: null headers: {} expect: status_code: 200 error_code: 0 data_keys: - datas - curPage - pageCount path_equals: {}

from pathlib import Path #只负责读取yaml文件 import yaml BASE_DIR = Path(__file__).resolve().parents[3] / "tests" def read_yaml(file_path, file_name): """加载 YAML 测试用例""" file_path = BASE_DIR / file_path / file_name with open(file_path, encoding="utf-8") as f: return yaml.safe_load(f) class DataUtil: def __init__(self): self.base_dir = "data" def get_yaml_data(self, file_name, key=None): data = read_yaml(self.base_dir, file_name) return [ case for case in data["cases"] if case.get("enabled", True) ]

(2) logger.py

在需要输出日志的地方引入日志模块即可

# 日志工具类-统一创建logger # 日志会同时输出到控制台和 logs/test_YYYYMMDD.log import logging from datetime import datetime from pathlib import Path #获取根目录，然后创建日志文件目录 BASE_DIR = Path(__file__).resolve().parents[3] LOG_DIR = BASE_DIR / "logs" LOG_DIR.mkdir(parents=True, exist_ok=True) #日志处理器-将日志写入文件 def _build_file_handler() -> logging.Handler: log_file = LOG_DIR / f"test_{datetime.now():%Y%m%d}.log" handler = logging.FileHandler(log_file, encoding="utf-8") handler.setLevel(logging.INFO) return handler #终端打印日志 def _build_stream_handler() -> logging.Handler: handler = logging.StreamHandler() handler.setLevel(logging.INFO) return handler def get_logger(name: str = "pytest_project") -> logging.Logger: #这个函数返回一个日志处理器logging.Logger对象（类型提示） logger = logging.getLogger(name) #按名字取日志 if logger.handlers: #如果日志处理器已经存在，直接返回，避免重复创建日志处理器 return logger logger.setLevel(logging.INFO) logger.propagate = False #设置为False，避免日志处理器向父日志处理器传递日志 #日志格式化器-定义日志的格式，时间 | 级别 | logger名称 | 消息 formatter = logging.Formatter( "%(asctime)s | %(levelname)s | %(name)s | %(message)s" ) file_handler = _build_file_handler() stream_handler = _build_stream_handler() file_handler.setFormatter(formatter) stream_handler.setFormatter(formatter) logger.addHandler(file_handler) #将文件日志处理器添加到日志处理器中 logger.addHandler(stream_handler) #将终端日志处理器添加到日志处理器中 return logger

(3) allure_report.py

import json import os import shutil from pathlib import Path from typing import Any from datetime import datetime, timezone from contextlib import nullcontext # 报告目录 BASE_DIR = Path(__file__).resolve().parents[3] REPORT_DIR = BASE_DIR / "reports" ALLURE_RESULT_DIR = REPORT_DIR / "allure-results" ALLURE_HTML_DIR = REPORT_DIR / "allure-html" # Allure 兼容层 try: import allure as _allure except ImportError: _allure = None class _NoOpAttachmentType: JSON = "json" class _NoOpDynamic: @staticmethod def title(*args, **kwargs) -> None: return None @staticmethod def label(*args, **kwargs) -> None: return None @staticmethod def parameter(*args, **kwargs) -> None: return None class _NoOpAllure: attachment_type = _NoOpAttachmentType() dynamic = _NoOpDynamic() @staticmethod def attach(*args, **kwargs) -> None: return None @staticmethod def step(*args, **kwargs): return nullcontext() @staticmethod def _decorator(*args, **kwargs): def wrapper(func): return func return wrapper epic = _decorator feature = _decorator story = _decorator severity = _decorator tag = _decorator class severity_level: BLOCKER = "blocker" CRITICAL = "critical" NORMAL = "normal" MINOR = "minor" TRIVIAL = "trivial" allure = _allure or _NoOpAllure() # 报告目录 def ensure_report_dirs() -> None: """确保 Allure 报告目录存在。""" REPORT_DIR.mkdir(parents=True, exist_ok=True) ALLURE_RESULT_DIR.mkdir(parents=True, exist_ok=True) ALLURE_HTML_DIR.mkdir(parents=True, exist_ok=True) def clean_allure_results() -> None: """清理上一次测试产生的 Allure 结果。""" if ALLURE_RESULT_DIR.exists(): shutil.rmtree(ALLURE_RESULT_DIR) ALLURE_RESULT_DIR.mkdir(parents=True, exist_ok=True) def is_allure_available() -> bool: """判断 Python Allure 库是否可用。""" return _allure is not None # 数据处理 def _safe_value(getter, default: str = "unknown") -> str: try: value = getter() except Exception: return default return str(value) if value not in (None, "") else default def to_pretty_text(data: Any) -> str: """将数据转换成适合 Allure 展示的 JSON 文本。""" if data is None: return "" if isinstance(data, str): return data try: return json.dumps( data, ensure_ascii=False, indent=2, ) except TypeError: return str(data) def attach_json(name: str, data: Any) -> None: """将数据作为 JSON 附件写入当前 Allure 测试用例。""" if not is_allure_available(): return allure.attach( to_pretty_text(data), name=name, attachment_type=allure.attachment_type.JSON, ) # Allure Environment def write_environment_properties() -> Path: """写入 Allure 环境信息。""" ensure_report_dirs() environment_file = ( ALLURE_RESULT_DIR / "environment.properties" ) rows = { "ENV": os.getenv("env", "test"), "BASE_URL": os.getenv("API_BASE_URL", "unknown"), "PYTHON": os.getenv( "PYTHON_VERSION", os.getenv("VIRTUAL_ENV", "unknown"), ), "REPORT_GENERATED_AT_UTC": datetime.now( timezone.utc ).strftime("%Y-%m-%d %H:%M:%S"), "HOSTNAME": os.getenv( "HOSTNAME", os.getenv("COMPUTERNAME", "local"), ), } environment_file.write_text( "\n".join( f"{key}={value}" for key, value in rows.items() ) + "\n", encoding="utf-8", ) return environment_file def write_executor_json() -> Path: """写入 Allure 执行器信息。""" ensure_report_dirs() executor_file = ALLURE_RESULT_DIR / "executor.json" executor = { "name": os.getenv( "CI_PLATFORM", "Local Run", ), "type": os.getenv( "CI_PLATFORM_TYPE", "local", ), "buildName": os.getenv( "BUILD_NAME", "Pytest Allure Report", ), "buildOrder": os.getenv( "BUILD_NUMBER", "1", ), "buildUrl": os.getenv( "BUILD_URL", "", ), "reportName": os.getenv( "ALLURE_REPORT_NAME", "", ), "reportUrl": os.getenv( "ALLURE_REPORT_URL", "", ), } executor_file.write_text( json.dumps( executor, ensure_ascii=False, indent=2, ), encoding="utf-8", ) return executor_file def write_categories_json() -> Path: """写入 Allure 失败分类规则。""" ensure_report_dirs() categories_file = ALLURE_RESULT_DIR / "categories.json" categories = [ { "name": "Assertion Failure", "matchedStatuses": ["failed"], "messageRegex": ".*AssertionError.*|.*assert .*", }, { "name": "Environment Or Config Error", "matchedStatuses": ["broken"], "messageRegex": ( ".*未配置.*" "|.*Missing env var.*" "|.*Placeholder variable not found.*" "|.*不支持的环境.*" ), }, { "name": "Authentication Failure", "matchedStatuses": [ "failed", "broken", ], "messageRegex": ( ".*401.*" "|.*403.*" "|.*token.*" "|.*access_token.*" ), }, { "name": "HTTP Or Dependency Error", "matchedStatuses": [ "failed", "broken", ], "messageRegex": ( ".*Timeout.*" "|.*ConnectError.*" "|.*ReadTimeout.*" "|.*HTTPStatusError.*" ), }, { "name": "Data Or Schema Error", "matchedStatuses": [ "failed", "broken", ], "messageRegex": ( ".*KeyError.*" "|.*TypeError.*" "|.*ValueError.*" "|.*JSON.*" ), }, ] categories_file.write_text( json.dumps( categories, ensure_ascii=False, indent=2, ), encoding="utf-8", ) return categories_file

(4) feishu.py

# 将测试结果发送到飞书群聊 import base64 import hashlib import hmac import os import time from typing import Any import requests from auto_tests.utils.logger import get_logger logger = get_logger("feishu_notify") # 配置 def _is_enabled() -> bool: """ 判断是否开启飞书通知。 环境变量： FEISHU_NOTIFY_ENABLED=true -> 开启 FEISHU_NOTIFY_ENABLED=false -> 关闭 """ return os.environ.get("FEISHU_NOTIFY_ENABLED", "false").lower() == "true" def _get_webhook_url() -> str: """获取飞书机器人 Webhook URL。""" return os.environ.get("FEISHU_WEBHOOK_URL", "").strip() def _get_bot_secret() -> str: """ 获取飞书机器人签名 Secret。 如果没有配置 Secret，则不进行签名。 """ return os.environ.get("FEISHU_BOT_SECRET", "").strip() # 飞书签名 def _sign(secret: str) -> tuple[str, str]: """ 根据飞书机器人 Secret 生成签名。 Returns: timestamp: 时间戳 sign: 签名 """ timestamp = str(int(time.time())) string_to_sign = f"{timestamp}\n{secret}" digest = hmac.new( string_to_sign.encode("utf-8"), digestmod=hashlib.sha256, ).digest() sign = base64.b64encode(digest).decode("utf-8") return timestamp, sign # 飞书响应处理 def _is_success_response(body: Any) -> bool: """判断飞书接口是否返回成功。""" if not isinstance(body, dict): return False if "StatusCode" in body: return body.get("StatusCode") == 0 if "code" in body: return body.get("code") == 0 return False def _stringify_body(body: Any) -> str: """将响应内容转换成字符串，方便日志输出。""" if isinstance(body, dict): return str(body) return repr(body) # 发送飞书通知 def send_feishu_notification(message: str) -> tuple[bool, str]: """ 发送文本消息到飞书群聊。 Args: message: 要发送的消息内容。 Returns: tuple[bool, str]: True -> 发送成功 False -> 未发送或发送失败 """ # 1. 判断是否开启通知 if not _is_enabled(): result = "skipped: disabled" logger.info("Feishu notify %s", result) return False, result # 2. 获取 Webhook webhook_url = _get_webhook_url() if not webhook_url: result = "skipped: FEISHU_WEBHOOK_URL is empty" logger.warning("Feishu notify %s", result) return False, result # 3. 构造消息 payload: dict[str, Any] = { "msg_type": "text", "content": { "text": message, }, } # 4. 如果配置了 Secret，则添加签名 secret = _get_bot_secret() if secret: timestamp, sign = _sign(secret) payload["timestamp"] = timestamp payload["sign"] = sign # 5. 发送请求 try: response = requests.post( webhook_url, json=payload, timeout=10, ) response.raise_for_status() body = response.json() # 6. 判断飞书返回结果 if not _is_success_response(body): result = f"failed: response body={_stringify_body(body)}" logger.warning( "Feishu notify %s", result, ) return False, result logger.info("Feishu notification sent successfully") return True, "sent successfully" except Exception as exc: result = f"failed: {exc}" logger.warning( "Feishu notify %s", result, ) return False, result

import httpx from auto_tests.utils.logger import get_logger from auto_tests.report_util.allure import attach_json class ApiClient: """ API 客户端封装类。 用于统一管理： - HTTP GET/POST 请求 - 请求 Header - Cookie - HTTP Client 生命周期 - Allure 请求/响应附件 - 根据测试数据自动分发 HTTP 请求 """ def __init__(self, base_url: str, timeout: float = 30.0): self._client = httpx.Client( base_url=base_url, timeout=timeout, ) self.logger = get_logger("api_client") self.last_request: dict | None = None self.last_response: dict | None = None self._http_seq = 0 def _next_http_seq(self) -> int: self._http_seq += 1 return self._http_seq @staticmethod def _empty(value) -> bool: return value is None or value == {} or value == [] def request(self, req: dict) -> httpx.Response: """ 数据驱动时，根据测试数据中的 method 自动发送 HTTP 请求。 Args: req: { "method": "GET", "path": "/xxx", "params": {}, "json": {}, "data": {}, "headers": {} } """ method = (req.get("method") or "GET").upper() path = req["path"] params = ( req.get("params") if not self._empty(req.get("params")) else None ) json_body = ( req.get("json") if not self._empty(req.get("json")) else None ) data = ( req.get("data") if not self._empty(req.get("data")) else None ) headers = ( req.get("headers") if not self._empty(req.get("headers")) else None ) if method == "GET": return self.get( path, params=params, headers=headers, ) if method == "POST": return self.post( path, json=json_body, data=data, headers=headers, ) raise ValueError( f"Unsupported method: {method}" ) def set_header(self, name: str, value: str) -> None: """设置公共 Header。""" self._client.headers[name] = value def set_cookie( self, name: str, value: str, domain: str | None = None, path: str = "/", ) -> None: """设置 Cookie。""" self._client.cookies.set( name, value, domain=domain, path=path, ) self.logger.info( "Cookie 已经被设置！" ) def get( self, path: str, params: dict | None = None, headers: dict | None = None, ) -> httpx.Response: http_seq = self._next_http_seq() request_payload = { "method": "GET", "path": path, "params": params, "headers": headers or dict(self._client.headers), } self.last_request = request_payload attach_json( f"[HTTP {http_seq}] GET {path} request", request_payload, ) self.logger.info( "HTTP GET %s params=%s", path, params, ) response = self._client.get( path, params=params, headers=headers, ) self._record_response( response, http_seq, ) return response def post( self, path: str, json: dict | None = None, data: dict | None = None, headers: dict | None = None, ) -> httpx.Response: http_seq = self._next_http_seq() request_payload = { "method": "POST", "path": path, "json": json, "data": data, "headers": headers or dict(self._client.headers), } self.last_request = request_payload attach_json( f"[HTTP {http_seq}] POST {path} request", request_payload, ) self.logger.info( "HTTP POST %s json=%s data=%s", path, json, data, ) response = self._client.post( path, json=json, data=data, headers=headers, ) self._record_response( response, http_seq, ) return response def _record_response( self, response: httpx.Response, http_seq: int, ) -> None: try: body = response.json() except Exception: body = response.text self.last_response = { "status_code": response.status_code, "headers": dict(response.headers), "body": body, } self.logger.info( "HTTP RESPONSE %s body=%s", response.status_code, body, ) attach_json( f"[HTTP {http_seq}] " f"{response.request.method} " f"{response.request.url.path} response", self.last_response, ) def close(self): """关闭 HTTP Client。""" self._client.close()

(6) assertion.py

from typing import Any def key_exists_anywhere(obj: Any, target_key: str) -> bool: """ 递归查找对象（dict/list 混合）中是否存在指定 key。 """ if isinstance(obj, dict): if target_key in obj: return True return any( key_exists_anywhere(value, target_key) for value in obj.values() ) if isinstance(obj, list): return any( key_exists_anywhere(item, target_key) for item in obj ) return False def get_by_path(obj: Any, path: str) -> Any: """ 按点分路径取值，支持数组索引。 例如： data.datas.0.title 缺失时安全返回 None。 """ cur = obj for part in path.split("."): if cur is None: return None if isinstance(cur, dict): cur = cur.get(part) elif isinstance(cur, list) and part.isdigit(): index = int(part) cur = cur[index] if 0 <= index < len(cur) else None else: return None return cur def assert_wan_response( resp, expected_status: int = 200, expected_error_code: int = 0, data_keys: list[str] | None = None, path_equals: dict[str, Any] | None = None, ) -> dict: """ 针对 WanAndroid 统一响应结构进行断言。 默认成功场景： errorCode == 0 如果测试预期接口返回业务错误，可以指定： expected_error_code=-1001 例如： assert_wan_response(response) # 预期登录过期 assert_wan_response( response, expected_error_code=-1001, ) # 预期其他业务错误 assert_wan_response( response, expected_error_code=-1, ) 参数： resp: httpx.Response expected_status: 预期 HTTP 状态码，默认 200 expected_error_code: 预期业务 errorCode，默认 0 data_keys: 要求响应中存在的 key，支持递归查找 path_equals: 按点分路径进行等值断言，例如： { "data.datas.0.title": "首页文章" } """ # 1. HTTP 状态码 assert resp.status_code == expected_status, ( f"HTTP status mismatch: " f"expected {expected_status}, " f"got {resp.status_code}. " f"body: {resp.text[:500]}" ) # 2. 获取响应 JSON body = resp.json() # 3. 基础字段检查 assert "errorCode" in body, ( f"Missing errorCode in response: {body}" ) assert "errorMsg" in body, ( f"Missing errorMsg in response: {body}" ) # 4. errorCode 断言 actual_error_code = body["errorCode"] assert actual_error_code == expected_error_code, ( f"errorCode mismatch: " f"expected {expected_error_code}, " f"got {actual_error_code}. " f"errorMsg: {body.get('errorMsg')}" ) # 5. 检查指定 key 是否存在 if data_keys: for key in data_keys: assert key_exists_anywhere(body, key), ( f"Expected key '{key}' not found " f"anywhere in response body" ) # 6. 检查指定路径的值 if path_equals: for path, expected in path_equals.items(): actual = get_by_path(body, path) assert actual == expected, ( f"path '{path}' mismatch: " f"expected {expected!r}, " f"got {actual!r}" ) return body

(7) run.py

import subprocess import sys import pytest from auto_tests.utils.logger import get_logger from auto_tests.report_util import allure from auto_tests.report_util.feishu import send_feishu_notification logger = get_logger("run") # 1. 初始化报告目录 allure.ensure_report_dirs() # 2. 清理上一次 Allure 测试结果 allure.clean_allure_results() # 3. 写入 Allure 元数据 allure.write_environment_properties() allure.write_executor_json() allure.write_categories_json() # 4. 执行 pytest pytest_args = [ "--alluredir", str(allure.ALLURE_RESULT_DIR), ] logger.info( "Running pytest with args: %s", pytest_args, ) pytest_exit_code = pytest.main(pytest_args) # 5. 生成 Allure HTML 报告 command = [ "allure", "generate", str(allure.ALLURE_RESULT_DIR), "-o", str(allure.ALLURE_HTML_DIR), "--clean", ] logger.info( "Generating Allure report: %s", " ".join(command), ) subprocess.run( command, check=True, ) # 6. 发送飞书通知 message = ( "自动化测试执行完成\n" f"Pytest Exit Code: {pytest_exit_code}\n" f"Allure Report: {allure.ALLURE_HTML_DIR}" ) feishu_success, feishu_message = send_feishu_notification(message) if feishu_success: logger.info( "Feishu notification result: %s", feishu_message, ) else: logger.warning( "Feishu notification result: %s", feishu_message, ) # 7. 保留 pytest 原始退出码 sys.exit(pytest_exit_code)