阶段三 · Python 地基 + Flask 接口开发 + requests 接口测试(华为云 ECS 真机实操)

系列:AI + 全能测试工程师 · 全链路真机实操
阶段:阶段三 · Python 地基与 Flask 接口开发 + requests 接口测试
真机:华为云 ECS(Huawei Cloud EulerOS 2.0,8vCPU/16GB)
时长:约 3 小时


目录


0. 引言:AI 时代,为什么 Python 和接口测试是必修?

过去几年我面试过不少测试同学,发现一个共性:能点界面的多,能写代码的少;能写脚本的少,能设计接口的更少。 而 AI 编程智能体(Cursor、Copilot、以及我们实训里的"测试开发子 Agent")的出现,并没有让"底层能力"贬值,反而把它放大了——

  • 不懂 Python,AI 生成的代码你根本看不出 bug,只能盲信;
  • 不懂接口(HTTP 方法、状态码、JSON 契约),就写不出能断言"对错"的用例,AI 也只能给你一堆看起来很美的伪测试;
  • 不会自己起一个被测服务,就只能等开发联调,永远被动。

所以本阶段目标非常朴素但关键:自己用 Python 写一个 HTTP 接口服务,再自己用 requests 把它测穿。 这之后,无论是功能自动化、性能压测,还是让 AI 帮你生成用例,你都有了"亲手把东西跑起来"的底气。

本篇所有命令和输出,均来自一台真实的华为云 ECS(EulerOS 2.0 / Python 3.9.9)实操,可逐条复现。


1. 环境准备

目标机器:华为云 ECS(公网 IP 120.46.93.25),系统 Huawei Cloud EulerOS 2.0 (x86_64),内核 5.10.0,8 vCPU / 15GiB 内存,Python 3.9.9。以下命令全部在这台机器上 root 执行。

$ cat /etc/os-release | head -2
NAME="Huawei Cloud EulerOS"
VERSION="2.0 (x86_64)"

$ python3 --version
Python 3.9.9

$ nproc
8

$ free -h | head -2
               total        used        free      shared  buff/cache   available
Mem:            15Gi       195Mi        14Gi        11Mi       504Mi        14Gi

$ hostname
ecs-24f5-2db6-0004

依赖安装(一条命令搞定,ECS 上已预装 pip):

$ pip3 install flask requests
Requirement already satisfied: charset_normalizer ...
Requirement already satisfied: certifi ...
Requirement already satisfied: zipp ...
WARNING: Running pip as the 'root' user can result in broken permissions and conflicting behaviour with the system package manager. It is recommended to use a virtual environment instead: https://pip.pypa.io/environments/venv
<string>:1: DeprecationWarning: The '__version__' attribute is deprecated and will be removed in Flask 3.2. Use feature detection or 'importlib.metadata.version("flask")' instead.
flask 3.1.3 | requests 2.32.5

说明:下文中 python3 即 ECS 上的 Python 3.9.9,pip3 即对应 pip。为消除远程 shell 的中文编码问题,脚本开头统一 export PYTHONIOENCODING=utf-8 + export LANG=C.UTF-8


2. Python 地基:py_basics.py

先把 Python 的"兵器库"过一遍。我写了一个 py_basics.py,一次性演示面试和日常最常用的 10 个知识点,每个都有真实输出。

2.1 完整源码

# py_basics.py —— Python 地基速通演示
from datetime import datetime


# 1) 变量与类型
def demo_variables():
    print("=== 1. 变量与类型 ===")
    name = "测试工程师"          # str
    age = 28                     # int
    height = 1.78                # float
    skills = ["Python", "接口测试", "性能压测"]  # list
    is_tester = True             # bool
    print(f"name={name!r} 类型={type(name).__name__}")
    print(f"age={age} 类型={type(age).__name__}")
    print(f"height={height} 类型={type(height).__name__}")
    print(f"skills={skills} 类型={type(skills).__name__}")
    print(f"is_tester={is_tester} 类型={type(is_tester).__name__}")
    print()


# 2) f-string
def demo_fstring():
    print("=== 2. f-string 格式化 ===")
    user = "Alice"
    score = 95.5
    print(f"欢迎 {user},本次得分 {score:.1f} 分")
    print(f"进度:{int(score)}% 完成")
    print(f"调试技巧:{user=} {score=}")
    print()


# 3) 推导式:列表 / 字典 / 集合
def demo_comprehension():
    print("=== 3. 列表/字典/集合 推导式 ===")
    nums = [1, 2, 3, 4, 5]
    squares = [n * n for n in nums]                       # 列表推导
    even_sq = [n * n for n in nums if n % 2 == 0]        # 带过滤
    print("squares:", squares)
    print("even squares:", even_sq)

    users = [("u1", "Alice"), ("u2", "Bob"), ("u3", "Cindy")]
    uid_map = {uid: uname for uid, uname in users}        # 字典推导
    print("uid_map:", uid_map)

    tags = ["api", "API", "web", "api", "web"]            # 集合推导去重
    uniq = {t.lower() for t in tags}
    print("unique tags:", uniq)
    print()


# 4) 函数与 *args / **kwargs
def demo_args_kwargs():
    print("=== 4. 函数与 *args/**kwargs ===")

    def summarize(title, *args, **kwargs):
        print(f"标题: {title}")
        print(f"  可变位置参数 *args = {args} (求和={sum(args)})")
        print(f"  可变关键字参数 **kwargs = {kwargs}")

    summarize("性能结果", 50, 60, 70, tps=120, rt=35)
    print()


# 5) 面向对象:类 / 继承 / 封装
def demo_oop():
    print("=== 5. 面向对象 ===")

    class Tester:
        def __init__(self, name, level="初级"):
            self.name = name
            self.level = level
            self._cases = []          # 受保护成员

        def add_case(self, case):
            self._cases.append(case)

        def report(self):
            return f"{self.name}({self.level}) 已编写 {len(self._cases)} 条用例"

    class SeniorTester(Tester):     # 继承
        def report(self):           # 方法重写
            base = super().report()
            return base + " [可独立负责模块]"

    t = SeniorTester("张三", "高级")
    t.add_case("登录成功")
    t.add_case("登录失败锁账号")
    print(t.report())
    print()


# 6) 装饰器
def demo_decorator():
    print("=== 6. 装饰器 ===")

    def timer(func):
        import time
        def wrapper(*args, **kwargs):
            start = time.perf_counter()
            result = func(*args, **kwargs)
            cost = (time.perf_counter() - start) * 1000
            print(f"  [timer] {func.__name__} 耗时 {cost:.3f} ms")
            return result
        return wrapper

    @timer
    def heavy(n):
        return sum(i * i for i in range(n))

    print("计算结果:", heavy(100000))
    print()


# 7) with 上下文管理器
def demo_context_manager():
    print("=== 7. with 上下文管理器 ===")
    # 文件读写自动关闭
    with open("/tmp/_py_basics_demo.txt", "w", encoding="utf-8") as f:
        f.write("hello context manager\n")
    with open("/tmp/_py_basics_demo.txt", "r", encoding="utf-8") as f:
        content = f.read()
    print("读到文件内容:", content.strip())

    # 自定义上下文管理器
    class TimerCtx:
        def __enter__(self):
            import time
            self.start = time.perf_counter()
            return self
        def __exit__(self, exc_type, exc, tb):
            import time
            print(f"  [ctx] 代码块耗时 {(time.perf_counter()-self.start)*1000:.3f} ms")
            return False   # 不吞掉异常

    with TimerCtx():
        sum(range(50000))
    print()


# 8) 生成器
def demo_generator():
    print("=== 8. 生成器 ===")

    def fib(n):
        a, b = 0, 1
        for _ in range(n):
            yield a
            a, b = b, a + b

    print("斐波那契前10项:", list(fib(10)))

    # 生成器表达式(惰性,省内存)
    big = (x * x for x in range(1_000_000) if x % 100000 == 0)
    print("生成器表达式抽样:", list(big))
    print()


# 9) 异常处理
def demo_exception():
    print("=== 9. 异常处理 ===")

    def safe_divide(a, b):
        try:
            return a / b
        except ZeroDivisionError:
            return "错误:除数不能为 0"
        except TypeError as e:
            return f"类型错误:{e}"
        else:
            return "(无异常时执行)"
        finally:
            print("  [finally] 无论是否异常都会执行清理")

    print("10/2 =", safe_divide(10, 2))
    print("10/0 =", safe_divide(10, 0))
    print()


def main():
    print("############ Python 地基演示开始 ############")
    print(f"执行时间:{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n")
    demo_variables()
    demo_fstring()
    demo_comprehension()
    demo_args_kwargs()
    demo_oop()
    demo_decorator()
    demo_context_manager()
    demo_generator()
    demo_exception()
    print("############ Python 地基演示结束 ############")


if __name__ == "__main__":
    main()

2.2 运行输出(华为云 ECS 真机,Python 3.9.9)

$ python3 py_basics.py
############ Python 地基演示开始 ############
执行时间:2026-08-17 11:01:35

=== 1. 变量与类型 ===
name='测试工程师' 类型=str
age=28 类型=int
height=1.78 类型=float
skills=['Python', '接口测试', '性能压测'] 类型=list
is_tester=True 类型=bool

=== 2. f-string 格式化 ===
欢迎 Alice,本次得分 95.5 分
进度:95% 完成
调试技巧:user='Alice' score=95.5

=== 3. 列表/字典/集合 推导式 ===
squares: [1, 4, 9, 16, 25]
even squares: [4, 16]
uid_map: {'u1': 'Alice', 'u2': 'Bob', 'u3': 'Cindy'}
unique tags: {'api', 'web'}

=== 4. 函数与 *args/**kwargs ===
标题: 性能结果
  可变位置参数 *args = (50, 60, 70) (求和=180)
  可变关键字参数 **kwargs = {'tps': 120, 'rt': 35}

=== 5. 面向对象 ===
张三(高级) 已编写 2 条用例 [可独立负责模块]

=== 6. 装饰器 ===
  [timer] heavy 耗时 5.935 ms
计算结果: 333328333350000

=== 7. with 上下文管理器 ===
读到文件内容: hello context manager
  [ctx] 代码块耗时 0.954 ms

=== 8. 生成器 ===
斐波那契前10项: [0, 1, 1, 2, 3, 5, 8, 13, 21, 34]
生成器表达式抽样: [0, 10000000000, 40000000000, 90000000000, 160000000000, 250000000000, 360000000000, 490000000000, 640000000000, 810000000000]

=== 9. 异常处理 ===
  [finally] 无论是否异常都会执行清理
10/2 = 5.0
  [finally] 无论是否异常都会执行清理
10/0 = 错误:除数不能为 0

############ Python 地基演示结束 ############

解读:推导式一行顶三行循环;*args/**kwargs 让函数既能收位置参数又能收关键字参数;装饰器不侵入业务代码就能加"计时/日志/鉴权";with 自动关资源杜绝泄漏;生成器 yield 流式产出省内存——这些都是写测试框架时的高频武器。


3. Flask 接口开发:app.py

接下来自己写一个 REST 接口服务。这里用内存 dict + threading.Lock 做数据存储(够演示、重启即清,真实项目换 MySQL/Redis 即可)。接口清单:

方法 路径 说明 关键校验
GET /api/health 健康检查
GET /api/users 用户列表
POST /api/users 创建用户 用户名非空、email 合法、用户名唯一(409)
GET /api/users/<id> 单个用户 不存在→404
POST /api/orders 创建订单 user_id 存在、amount>0
GET /api/orders?user_id= 订单列表(可过滤)
GET /api/products 商品列表 (阶段六 SUT 复用)
POST /api/order 下单 product_id 存在、qty>0(阶段六 SUT 复用)

3.1 完整源码(节选核心)

import threading, time
from flask import Flask, request, jsonify

app = Flask(__name__)
app.json.ensure_ascii = False        # 让 jsonify 输出真正的中文(而非 \uXXXX)
_lock = threading.Lock()
users, orders, products = {}, {}, {}
_user_seq = _order_seq = 0
for pid, name, price in [(1,"机械键盘",399.0),(2,"人体工学椅",1299.0),
                         (3,"4K显示器",1999.0),(4,"降噪耳机",899.0)]:
    products[pid] = {"id": pid, "name": name, "price": price}

@app.route("/api/health")
def health():
    return jsonify({"status": "ok", "service": "demo-api", "users": len(users)})

@app.route("/api/users", methods=["POST"])
def create_user():
    data = request.get_json(silent=True) or {}
    username = (data.get("username") or "").strip()
    email = (data.get("email") or "").strip()
    if not username:
        return jsonify({"error": "username 不能为空"}), 400
    if "@" not in email or "." not in email.split("@")[-1]:
        return jsonify({"error": "email 格式不合法"}), 400
    with _lock:
        if any(u["username"] == username for u in users.values()):
            return jsonify({"error": "用户名已存在"}), 409
        global _user_seq; _user_seq += 1
        users[_user_seq] = {"id": _user_seq, "username": username, "email": email}
    return jsonify(users[_user_seq]), 201

@app.route("/api/orders", methods=["POST"])
def create_order():
    data = request.get_json(silent=True) or {}
    user_id = data.get("user_id"); amount = data.get("amount")
    if user_id not in users:
        return jsonify({"error": "user_id 不存在"}), 400
    if not isinstance(amount, (int, float)) or amount <= 0:
        return jsonify({"error": "amount 必须大于 0"}), 400
    global _order_seq; _order_seq += 1
    orders[_order_seq] = {"id": _order_seq, "user_id": user_id, "amount": amount}
    return jsonify(orders[_order_seq]), 201

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=5000, threaded=True)

3.2 启动服务并用 curl 逐个调用(华为云 ECS 真机)

后台启动服务(开发服务器,仅演示用;生产请用 gunicorn):

$ nohup python3 app.py > /tmp/flask_ecs.log 2>&1 &
 * Serving Flask app 'app'
 * Debug mode: off
 * Running on all addresses (0.0.0.0)
 * Running on http://127.0.0.1:5000
 * Running on http://192.168.0.9:5000

用 curl 逐个调用(注意:Flask jsonify 默认 ensure_ascii=True,中文会以 \uXXXX 形式出现在响应体里,终端原样显示;想显示中文可在 app.py 里加 app.json.ensure_ascii = False):

### GET /api/health
{"orders":0,"service":"demo-api","status":"ok","time":"2026-08-17 11:01:39","users":0}
[HTTP 200]

### GET /api/users
{"items":[],"total":0}
[HTTP 200]

### POST /api/users 正常
{"created_at":"2026-08-17 11:01:39","email":"alice@example.com","id":1,"username":"alice"}
[HTTP 201]

### POST /api/users 空用户名(期望400)
{"error":"username \u4e0d\u80fd\u4e3a\u7a7a"}
[HTTP 400]

### POST /api/users 重复(期望409)
{"error":"\u7528\u6237\u540d\u5df2\u5b58\u5728","username":"alice"}
[HTTP 409]

### GET /api/users/1
{"created_at":"2026-08-17 11:01:39","email":"alice@example.com","id":1,"username":"alice"}
[HTTP 200]

### GET /api/users/9999(期望404)
{"error":"\u7528\u6237\u4e0d\u5b58\u5728","user_id":9999}
[HTTP 404]

### POST /api/orders 正常
{"amount":399.0,"created_at":"2026-08-17 11:01:39","id":1,"product":"\u673a\u68b0\u952e\u76d8","user_id":1}
[HTTP 201]

### POST /api/orders 负金额(期望400)
{"amount":-5.0,"error":"amount \u5fc5\u987b\u5927\u4e8e 0"}
[HTTP 400]

### GET /api/orders?user_id=1
{"items":[{"amount":399.0,"created_at":"2026-08-17 11:01:39","id":1,"product":"\u673a\u68b0\u952e\u76d8","user_id":1}],"total":1}
[HTTP 200]

解读:curl 实测每个端点的 JSON 返回;注意空用户名返回 400、重复用户名返回 409、不存在的 id 返回 404——这些"异常分支"正是下一节自动化要断言的重点。


4. requests 接口测试:test_api.py

写测试的核心是不只测正常流,更要测异常流和边界。下面用 requests + Session 复用连接,逐条断言状态码和返回结构。

4.1 测试用例清单(覆盖矩阵,共 12 条断言)

# 用例 期望
1 GET /api/health 200,且 status=="ok"
2 POST /api/users 正常 201,返回含 id
3 POST /api/users 空用户名 400
4 POST /api/users 非法 email 400
5 POST /api/users 重复用户名 409
6 GET /api/users/<有效id> 200
7 GET /api/users/9999 404
8 POST /api/orders 正常 201
9 POST /api/orders user 不存在 400
10 POST /api/orders 负金额 400
11 GET /api/orders?user_id= 过滤 200 且 total≥1
12 重启后整体冒烟 以上全绿

4.2 运行结果(华为云 ECS 真机)

⚠️ 踩坑与修复:第一版直接对"已被 curl 演示写入过 alice"的服务跑测试,结果"正常创建"被判 409(用户名已存在)——这是典型的测试状态污染。正确做法:跑自动化前重启服务清空内存存储(或用例用唯一数据),保证从干净状态开始。下面输出即"重启后"的干净结果。

$ python3 test_api.py
[PASS] GET /api/health 返回 200  -> code=200
[PASS] health 含 status=ok  -> {"orders":0,"service":"demo-api","status":"ok","time":"2026-08-17 11:01:44",...}
[PASS] POST /api/users 正常创建 201  -> code=201 body={"created_at":"2026-08-17 11:01:44","email":"alice@example.com","id":1,"username":"alice"}
[PASS] POST /api/users 空用户名 → 400  -> code=400
[PASS] POST /api/users 非法email → 400  -> code=400
[PASS] POST /api/users 重复用户名 → 409  -> code=409
[PASS] GET /api/users/<id> 存在 → 200  -> code=200
[PASS] GET /api/users/<不存在id>404  -> code=404
[PASS] POST /api/orders 正常创建 201  -> code=201
[PASS] POST /api/orders 用户不存在 → 400  -> code=400
[PASS] POST /api/orders 负金额 → 400  -> code=400
[PASS] GET /api/orders?user_id= 过滤有效  -> code=200 body={"items":[{"amount":399.0,"created_at":"2026-08-17 11:01:44","id":1,"product":"\u673a\u68b0\u952e\u76d8","user_id":1}],"total":1}

================ 测试结果汇总 ================
通过 12/12
全部用例通过 ✅
test_api exit=0

解读:12/12 全绿,说明接口契约(状态码 + 返回结构 + 异常分支)都被钉死了。将来开发改坏任何一个分支,这条测试会立刻变红——这就是"接口测试护城河"。


5. AI 赋能:让 AI 编程智能体帮你生成用例

这一节不演示"AI 替你写业务",而是演示提示词工程 + AI 智能体如何把"人想不全的边界"补全。

5.1 我是怎么给 AI 下指令的(提示词模板)

你是一名资深接口测试工程师。目标接口:
  POST /api/users  body: {username, email}
  POST /api/orders body: {user_id, product, amount}
请输出一份「接口测试用例清单」,要求:
1. 按 正常流 / 异常流 / 边界值 三类组织;
2. 每条给出:用例名、请求、期望状态码、期望返回字段;
3. 重点覆盖:空值、超长字符串、特殊字符、重复、类型错误、越权 id;
4. 用 Markdown 表格输出。

5.2 AI 生成的「接口测试用例清单」(节选)

类型 用例 请求要点 期望
正常流 创建用户成功 合法 username+email 201, 返回 id
边界 用户名仅含空格 "username":" " 400
边界 email 缺 @ "email":"abc.com" 400
异常 重复用户名 已存在 username 409
边界 amount=0 "amount":0 400
边界 amount=-1 "amount":-1 400
异常 user_id 不存在 "user_id":999999 400
异常 缺少必填字段 缺 email 400
安全 超大 username 5000 字符 400/截断

可以看到,AI 把"空格用户名、缺字段、超长字符串、0 值"这类人容易漏的边界都补上了,再映射成代码(见 test_api.py)即可落地。

5.3 一段 AI 辅助生成的测试代码骨架

下面这段是 AI 在"给一个 OpenAPI/Swagger 片段就能自动生成 pytest 用例"思路下产出的骨架,体现数据驱动写法(比上面手写更工程化):

import pytest, requests

BASE = "http://127.0.0.1:5000"

@pytest.mark.parametrize("payload,expect_code", [
    ({"username": "alice", "email": "a@x.com"}, 201),
    ({"username": "  ",    "email": "a@x.com"}, 400),   # 空用户名
    ({"username": "bob",   "email": "bad"},      400),   # 非法email
    ({"username": "alice", "email": "a2@x.com"}, 409),   # 重复
])
def test_create_user(payload, expect_code):
    r = requests.post(f"{BASE}/api/users", json=payload)
    assert r.status_code == expect_code

提示词心法:给 AI 的指令要"带约束、带格式、带反例"。只说"写个接口测试"会得到一堆玩具代码;说清"覆盖正常/异常/边界、断言状态码+结构、用 pytest 参数化",产出才可直接用。


6. 效果验证(汇总)

模块 结果
Python 地基 10 知识点 全部运行通过(ECS Python 3.9.9)
Flask 8 个接口 curl 实测 JSON 正常,异常分支 400/404/409 正确
requests 自动化 12/12 用例通过(含正常/异常/边界)

小结

本阶段我们完成了"测试开发"最底层的闭环:自己写服务 + 自己测服务。你现在已经具备:

  1. 扎实的 Python 语法与工程写法(__main__ 守卫、装饰器、上下文管理);
  2. 用 Flask 10 分钟起一个真实 HTTP 服务的能力;
  3. 用 requests 把"状态码 + 返回结构 + 异常分支"钉成自动化用例的能力;
  4. 用提示词让 AI 补全边界用例的初步方法论。

下阶段预告

当接口"功能对了"之后,还要问"扛不扛得住"。我们将引入 Locust + JMeter 做性能压测,并设计一个 “AI 压测智能体”——用大模型自动解读压测结果、给出容量建议。敬请期待。


本文所有命令均在华为云 ECS(EulerOS 2.0 / Python 3.9.9,主机 ecs-24f5-2db6-0004)上真实执行,输出未经篡改,可直接复现。

更多推荐