#Python 自动化测试最佳实践

# tests/ui/test_first_test.py
from playwright.sync_api import Page, expect

def test_homepage_loads(page: Page):
    """验证首页能正常加载"""
    page.goto("http://localhost:3000")

    # 断言页面标题
    expect(page).to_have_title("我的电商网站")

    # 断言关键元素可见
    expect(page.locator("[data-testid='logo']")).to_be_visible()
    expect(page.locator("[data-testid='search-input']")).to_be_visible()

def test_search_functionality(page: Page):
    """验证搜索功能"""
    page.goto("http://localhost:3000")

    # 输入搜索关键词
    page.fill("[data-testid='search-input']", "蓝牙耳机")
    page.click("[data-testid='search-btn']")

    # 等待结果加载
    page.wait_for_selector("[data-testid='product-card']")

    # 验证搜索结果
    product_names = page.locator("[data-testid='product-name']").all_inner_texts()
    assert len(product_names) > 0

# 1. 角色定位（最稳定，反映语义）
page.get_by_role("button", name="登录")
page.get_by_role("heading", name="商品列表")
page.get_by_role("textbox", name="用户名")

# 2. 标签定位（表单元素）
page.get_by_label("邮箱地址")
page.get_by_label("记住我")

# 3. 占位符定位
page.get_by_placeholder("请输入搜索关键词")

# 4. 测试 ID（团队约定，最稳定的自定义方案）
page.locator("[data-testid='login-btn']")
page.locator("[data-testid='product-card']")

# 5. 文本内容（适合按钮/链接）
page.get_by_text("加入购物车")
page.get_by_text("立即购买", exact=True)

# 6. CSS / XPath（最后选择——容易因 UI 变更而断裂）
page.locator(".btn-primary")
page.locator("//button[@class='submit-btn']")

# 点击
page.click("[data-testid='submit-btn']")
page.get_by_role("button", name="提交").click()

# 输入
page.fill("[data-testid='username']", "admin")
page.get_by_label("密码").fill("secret")

# 下拉选择
page.select_option("[data-testid='category']", "electronics")

# 勾选/取消勾选
page.check("[data-testid='agree-terms']")
page.uncheck("[data-testid='subscribe']")

# 等待
page.wait_for_selector("[data-testid='result-list']")
page.wait_for_url("**/dashboard")
page.wait_for_load_state("networkidle")

# 断言可见性
expect(page.locator("[data-testid='error-msg']")).to_be_visible()
expect(page.locator("[data-testid='loading']")).not_to_be_visible()

# 断言文本
expect(page.locator("[data-testid='title']")).to_have_text("商品详情")
expect(page.locator("[data-testid='price']")).to_contain_text("99")

# 断言 URL
expect(page).to_have_url("**/login")

# conftest.py 中配置自动截图
@pytest.fixture(scope="function")
def page(context: "BrowserContext") -> "Page":
    page = context.new_page()
    yield page
    # 测试失败时自动截图（简化示例）
    page.close()

# 或使用 Playwright 内置的视频录制
@pytest.fixture(scope="function")
def context(browser: "Browser") -> "BrowserContext":
    context = browser.new_context(
        record_video_dir="videos/",
        record_video_size={"width": 1280, "height": 720},
    )
    yield context
    context.close()

# conftest.py（项目根）
import pytest
import os
from dotenv import load_dotenv

load_dotenv()

@pytest.fixture(scope="session")
def base_url():
    return os.getenv("BASE_URL", "http://localhost:3000")

@pytest.fixture(scope="session")
def api_base_url():
    return os.getenv("API_BASE_URL", "http://localhost:8000")

# tests/ui/conftest.py
import pytest
from playwright.sync_api import Page
from pages.login_page import LoginPage

@pytest.fixture(scope="function")
def login_page(page: Page, base_url: str) -> LoginPage:
    """返回已导航至登录页的 LoginPage 对象"""
    lp = LoginPage(page, base_url)
    lp.navigate().wait_for_loaded()
    return lp

# tests/api/conftest.py
import pytest
from httpx import AsyncClient

@pytest.fixture(scope="session")
async def admin_token(api_base_url: str):
    """管理员 token——session 级别，只登录一次"""
    async with AsyncClient(base_url=api_base_url) as client:
        resp = await client.post("/api/auth/login", json={
            "username": "admin",
            "password": "admin_secret",
        })
        return resp.json()["access_token"]

@pytest.fixture(scope="function")
async def auth_client(api_base_url: str, admin_token: str):
    """带认证头的 httpx 客户端——每个测试独立"""
    async with AsyncClient(
        base_url=api_base_url,
        timeout=30.0,
        headers={"Authorization": f"Bearer {admin_token}"},
    ) as client:
        yield client

# pages/login_page.py
from playwright.sync_api import Page, Locator

class LoginPage:
    def __init__(self, page: Page, base_url: str):
        self.page = page
        self.url = f"{base_url}/login"
        # 所有选择器在类顶部集中定义
        self.username_input: Locator = page.locator("[data-testid='username-input']")
        self.password_input: Locator = page.locator("[data-testid='password-input']")
        self.login_btn: Locator = page.locator("[data-testid='login-btn']")
        self.error_msg: Locator = page.locator("[data-testid='login-error']")

    def navigate(self) -> "LoginPage":
        self.page.goto(self.url)
        return self

    def wait_for_loaded(self) -> "LoginPage":
        self.page.wait_for_load_state("networkidle")
        return self

    def login(self, username: str, password: str) -> Page:
        """登录并返回已认证的 Page 对象"""
        self.username_input.fill(username)
        self.password_input.fill(password)
        self.login_btn.click()
        self.page.wait_for_url("**/dashboard")
        return self.page

    def login_expecting_error(self, username: str, password: str) -> "LoginPage":
        """登录失败时返回自身，便于链式调用断言错误"""
        self.username_input.fill(username)
        self.password_input.fill(password)
        self.login_btn.click()
        self.error_msg.wait_for(state="visible")
        return self

# conftest.py
import pytest
from httpx import AsyncClient

@pytest.fixture(scope="session")
def api_base_url():
    import os
    return os.getenv("API_BASE_URL", "http://localhost:8000")

@pytest.fixture(scope="function")
async def client(api_base_url: str):
    """匿名客户端——每个测试独立"""
    async with AsyncClient(base_url=api_base_url, timeout=30.0) as c:
        yield c

@pytest.fixture(scope="function")
async def auth_client(api_base_url: str):
    """带认证的客户端"""
    async with AsyncClient(base_url=api_base_url, timeout=30.0) as c:
        # 先登录获取 token
        resp = await c.post("/api/auth/login", json={
            "username": "testuser",
            "password": "test_secret",
        })
        token = resp.json()["access_token"]
        c.headers["Authorization"] = f"Bearer {token}"
        yield c

import pytest
from httpx import AsyncClient

@pytest.mark.asyncio
class TestProductsAPI:
    """商品 API 测试"""

    async def test_list_products(self, client: AsyncClient):
        """GET /api/products — 查询商品列表"""
        resp = await client.get("/api/products")

        assert resp.status_code == 200
        data = resp.json()
        assert isinstance(data["items"], list)
        assert "total" in data

    async def test_get_product(self, client: AsyncClient, existing_product_id: int):
        """GET /api/products/{id} — 查询单个商品"""
        resp = await client.get(f"/api/products/{existing_product_id}")

        assert resp.status_code == 200
        assert resp.json()["id"] == existing_product_id

    async def test_create_product(self, auth_client: AsyncClient):
        """POST /api/products — 创建商品"""
        payload = {
            "name": "测试商品",
            "price": 99.00,
            "description": "这是一个测试商品",
        }
        resp = await auth_client.post("/api/products", json=payload)

        assert resp.status_code == 201
        data = resp.json()
        assert data["name"] == "测试商品"
        assert data["price"] == 99.00
        assert "id" in data

        # 清理
        await auth_client.delete(f"/api/products/{data['id']}")

    async def test_update_product(self, auth_client: AsyncClient, existing_product_id: int):
        """PUT /api/products/{id} — 更新商品"""
        resp = await auth_client.put(
            f"/api/products/{existing_product_id}",
            json={"price": 149.00},
        )

        assert resp.status_code == 200
        assert resp.json()["price"] == 149.00

    async def test_delete_product(self, auth_client: AsyncClient):
        """DELETE /api/products/{id} — 删除商品"""
        # 先创建一个商品
        create_resp = await auth_client.post("/api/products", json={
            "name": "待删除商品",
            "price": 9.99,
        })
        product_id = create_resp.json()["id"]

        # 删除
        resp = await auth_client.delete(f"/api/products/{product_id}")
        assert resp.status_code == 204

        # 验证已删除
        get_resp = await auth_client.get(f"/api/products/{product_id}")
        assert get_resp.status_code == 404

from pydantic import BaseModel, Field
from datetime import datetime

class ProductResponse(BaseModel):
    id: int
    name: str = Field(min_length=1, max_length=200)
    price: float = Field(gt=0)
    description: str | None = None
    is_active: bool
    created_at: datetime
    updated_at: datetime

@pytest.mark.asyncio
async def test_product_schema_validation(client: AsyncClient, existing_product_id: int):
    """验证响应结构符合 ProductResponse 模型"""
    resp = await client.get(f"/api/products/{existing_product_id}")

    assert resp.status_code == 200
    # pydantic 自动校验所有字段类型和约束——类型不匹配直接抛 ValidationError
    product = ProductResponse(**resp.json())
    assert product.id == existing_product_id
    assert product.price > 0
    assert len(product.name) >= 1

@pytest.mark.asyncio
@pytest.mark.parametrize("field, value, expected_status, expected_msg", [
    ("name", "", 422, "商品名称不能为空"),
    ("price", -1, 422, "价格必须大于 0"),
    ("price", 0, 422, "价格必须大于 0"),
    ("price", "not_a_number", 422, "价格必须是数字"),
    # 合法值
    ("name", "合法商品名", 201, None),
    ("price", 99.99, 201, None),
])
async def test_create_product_validation(
    auth_client: AsyncClient, field, value, expected_status, expected_msg
):
    """商品创建——参数化覆盖边界值和异常输入"""
    # 有效默认值
    payload = {"name": "测试商品", "price": 99.00}
    payload[field] = value

    resp = await auth_client.post("/api/products", json=payload)

    assert resp.status_code == expected_status
    if expected_msg:
        assert expected_msg in str(resp.json())
    else:
        # 创建成功，清理
        await auth_client.delete(f"/api/products/{resp.json()['id']}")

@pytest.fixture(scope="function")
async def order_with_items(auth_client: AsyncClient, existing_product_id: int):
    """创建包含商品的订单——测试结束后自动取消"""
    # 创建订单
    resp = await auth_client.post("/api/orders", json={
        "items": [{"product_id": existing_product_id, "quantity": 2}],
    })
    assert resp.status_code == 201
    order = resp.json()

    yield order  # 测试用例获得 order

    # 清理（如果订单不是已完成状态则取消）
    try:
        await auth_client.post(f"/api/orders/{order['id']}/cancel")
    except Exception:
        pass  # 可能已被测试中取消

@pytest.mark.asyncio
async def test_order_status(auth_client: AsyncClient, order_with_items: dict):
    """测试订单创建后状态为 pending"""
    order_id = order_with_items["id"]
    resp = await auth_client.get(f"/api/orders/{order_id}")

    assert resp.status_code == 200
    assert resp.json()["status"] == "pending"
    assert len(resp.json()["items"]) == 1

@pytest.fixture(scope="function")   # 默认，每个测试重建
def fresh_user(auth_client):
    user = create_user(auth_client, name="func_user")
    yield user
    delete_user(auth_client, user["id"])

@pytest.fixture(scope="class")      # 类内所有测试共享
def shared_user(auth_client):
    user = create_user(auth_client, name="class_user")
    yield user
    delete_user(auth_client, user["id"])

@pytest.fixture(scope="module")     # 模块内只读数据
def product_catalog():
    """只读的测试数据，不需要清理"""
    return [
        {"name": "商品A", "price": 9.99},
        {"name": "商品B", "price": 19.99},
        {"name": "商品C", "price": 29.99},
    ]

@pytest.fixture(scope="session")    # 全局：DB 连接、admin token
def db_connection():
    conn = create_db_connection()
    yield conn
    conn.close()

# data/factories.py
import uuid
from typing import Any

def make_user(**overrides: Any) -> dict[str, Any]:
    """生成用户测试数据，overrides 可覆盖任何字段"""
    unique = uuid.uuid4().hex[:8]
    return {
        "username": f"user_{unique}",
        "email": f"user_{unique}@example.com",
        "password": "TestPass123!",
        "full_name": f"Test User {unique}",
        **overrides,  # 覆盖默认值
    }

def make_product(**overrides: Any) -> dict[str, Any]:
    unique = uuid.uuid4().hex[:8]
    return {
        "name": f"测试商品 {unique}",
        "price": 99.00,
        "description": f"自动化测试生成的商品 {unique}",
        **overrides,
    }

# 使用
def test_with_custom_user(auth_client):
    user_data = make_user(
        username="vip_user",
        email="vip@example.com",
    )
    resp = await auth_client.post("/api/users", json=user_data)
    assert resp.status_code == 201

# data/factories.py
import factory
import uuid

class UserFactory(factory.Factory):
    class Meta:
        model = dict

    username = factory.LazyFunction(lambda: f"user_{uuid.uuid4().hex[:8]}")
    email = factory.LazyAttribute(lambda o: f"{o.username}@example.com")
    password = "TestPass123!"
    full_name = factory.Faker("name")
    is_active = True

class AdminUserFactory(UserFactory):
    role = "admin"

# 使用
def test_with_factory_boy():
    user = UserFactory(username="specific_user")
    admin = AdminUserFactory()

# conftest.py
import pytest

@pytest.fixture(scope="function", autouse=True)
async def clean_database():
    """每个测试函数执行前后清理数据库——autouse 自动应用"""
    # 注意：生产项目中应指向测试数据库
    yield
    # 清理（测试后）

@pytest.fixture(scope="function")
async def seed_data(auth_client):
    """测试前种子数据，测试后清理"""
    # 创建种子数据
    products = []
    for i in range(3):
        resp = await auth_client.post("/api/products", json={
            "name": f"种子商品 {i}",
            "price": 10.0 * (i + 1),
        })
        products.append(resp.json())

    yield products

    # 清理种子数据
    for p in products:
        await auth_client.delete(f"/api/products/{p['id']}")

# conftest.py
import os
from dotenv import load_dotenv

load_dotenv()  # 自动加载 .env 文件

@pytest.fixture(scope="session")
def admin_credentials():
    return {
        "username": os.getenv("TEST_ADMIN_USERNAME", "admin"),
        "password": os.getenv("TEST_ADMIN_PASSWORD", "changeme"),
    }

# ❌ 脆弱的 CSS 选择器（UI 框架自动生成的类名会变）
page.locator(".css-1x2y3z4")

# ✅ 稳定的选择器
page.locator("[data-testid='login-btn']")
page.get_by_role("button", name="登录")
page.get_by_label("用户名")

def test_ui_with_mocked_api(page: Page, base_url: str):
    """UI 测试中 mock 后端 API 响应，确保测试独立性"""
    # Mock GET /api/products 返回固定数据
    page.route("**/api/products", lambda route: route.fulfill(
        status=200,
        content_type="application/json",
        body='''{
            "items": [
                {"id": 1, "name": "Mock商品A", "price": 99.00},
                {"id": 2, "name": "Mock商品B", "price": 199.00}
            ],
            "total": 2
        }''',
    ))

    page.goto(f"{base_url}/products")

    # UI 应该显示 mock 的 2 个商品
    products = page.locator("[data-testid='product-card']")
    expect(products).to_have_count(2)

    # 验证显示的是 mock 数据
    expect(page.locator("[data-testid='product-name']").first).to_have_text("Mock商品A")

# 按 URL 部分 mock
def test_partial_mock(page: Page):
    # 只 mock 搜索 API，其他 API 正常访问
    page.route("**/api/products?search=*", lambda route: route.fulfill(
        status=200,
        json={"items": [], "total": 0},
    ))

# 修改真实响应（不替换，仅修改部分字段）
def test_modify_response(page: Page, base_url: str):
    def handle_route(route):
        response = route.fetch()  # 获取真实响应
        body = response.json()
        body["items"][0]["price"] = 999.99  # 修改第一个商品的价格
        route.fulfill(response=response, json=body)

    page.route("**/api/products", handle_route)
    page.goto(f"{base_url}/products")