""" 真实的端到端（E2E）测试 - 使用Playwright测试前后端联通 """ import pytest import time from playwright.async_api import async_playwright, Page, Browser, BrowserContext from httpx import AsyncClient from config.settings import settings @pytest.mark.e2e @pytest.mark.playwright class TestRealE2E: """真实的端到端测试类""" @pytest.fixture async def browser(self): """浏览器fixture - headless模式""" async with async_playwright() as p: browser = await p.chromium.launch(headless=True) yield browser await browser.close() @pytest.fixture async def context(self, browser): """浏览器上下文fixture""" context = await browser.new_context() yield context await context.close() @pytest.fixture async def page(self, context): """页面fixture""" page = await context.new_page() page.set_default_timeout(30000) yield page await page.close() @pytest.fixture async def authenticated_client(self): """已认证的HTTP客户端""" async with AsyncClient(base_url=settings.API_BASE_URL) as client: response = await client.post( "/api/auth/login", json={ "username": settings.TEST_USERNAME, "password": settings.TEST_PASSWORD } ) assert response.status_code == 200 token = response.json().get("token") client.headers.update({"Authorization": f"Bearer {token}"}) yield client @pytest.mark.asyncio async def test_complete_user_lifecycle_e2e(self, page, authenticated_client): """测试完整的用户生命周期 - 前后端联通""" timestamp = int(time.time() * 1000) username = f"e2e_user_{timestamp}" email = f"e2e_{timestamp}@example.com" # 1. 通过前端登录 await page.goto("http://localhost:3002/login") await page.wait_for_load_state("networkidle") await page.fill('input[placeholder="请输入用户名"]', settings.TEST_USERNAME) await page.fill('input[placeholder="请输入密码"]', settings.TEST_PASSWORD) await page.click('button[type="submit"]') await page.wait_for_url("**/") await page.wait_for_load_state("networkidle") # 2. 通过前端创建用户 await page.click('text=用户管理') await page.wait_for_url("**/users") await page.wait_for_load_state("networkidle") await page.click('text=新增用户') await page.wait_for_load_state("networkidle") await page.fill('input[placeholder=""]', username) await page.fill('input[placeholder=""]', 'Test123!@#') await page.fill('input[placeholder=""]', email) await page.fill('input[placeholder=""]', '13800138000') await page.click('button:has-text("确定")') await page.wait_for_load_state("networkidle") # 3. 通过API验证用户已创建 response = await authenticated_client.get("/api/users") assert response.status_code == 200 users = response.json() user_exists = any(user['username'] == username for user in users) assert user_exists, f"User {username} not found in API response" @pytest.mark.asyncio async def test_role_assignment_e2e(self, page, authenticated_client): """测试角色分配 - 前后端联通""" timestamp = int(time.time() * 1000) role_name = f"E2E_Role_{timestamp}" role_key = f"e2e_role_{timestamp}" # 1. 通过API创建角色 role_response = await authenticated_client.post( "/api/roles", json={ "roleName": role_name, "roleKey": role_key, "roleSort": 1, "status": 1 } ) assert role_response.status_code == 201 role_id = role_response.json()["id"] # 2. 通过前端登录 await page.goto("http://localhost:3002/login") await page.fill('input[placeholder="请输入用户名"]', settings.TEST_USERNAME) await page.fill('input[placeholder="请输入密码"]', settings.TEST_PASSWORD) await page.click('button[type="submit"]') await page.wait_for_url("**/") await page.wait_for_load_state("networkidle") # 3. 通过前端创建用户 await page.click('text=用户管理') await page.wait_for_url("**/users") await page.wait_for_load_state("networkidle") await page.click('text=新增用户') await page.wait_for_load_state("networkidle") username = f"e2e_user_{timestamp}" await page.fill('input[placeholder=""]', username) await page.fill('input[placeholder=""]', 'Test123!@#') await page.fill('input[placeholder=""]', f"e2e_{timestamp}@example.com") await page.click('button:has-text("确定")') await page.wait_for_load_state("networkidle") # 4. 通过API获取用户ID并分配角色 users_response = await authenticated_client.get("/api/users") users = users_response.json() user = next((u for u in users if u['username'] == username), None) assert user is not None await authenticated_client.put( f"/api/users/{user['id']}", json={"roleId": role_id} ) # 5. 通过API验证角色分配 user_response = await authenticated_client.get(f"/api/users/{user['id']}") assert user_response.status_code == 200 user_data = user_response.json() assert user_data["roleId"] == role_id # 6. 清理测试数据 await authenticated_client.delete(f"/api/users/{user['id']}") await authenticated_client.delete(f"/api/roles/{role_id}") @pytest.mark.asyncio async def test_login_and_navigation_e2e(self, page): """测试登录和导航 - 前后端联通""" # 1. 访问登录页面 await page.goto("http://localhost:3002/login") await page.wait_for_load_state("networkidle") title = await page.title() assert "登录" in title or "Login" in title.lower() # 2. 填写登录表单 await page.fill('input[placeholder="请输入用户名"]', settings.TEST_USERNAME) await page.fill('input[placeholder="请输入密码"]', settings.TEST_PASSWORD) # 3. 点击登录按钮 await page.click('button[type="submit"]') # 4. 等待跳转到首页或dashboard await page.wait_for_url("**/dashboard", timeout=15000) await page.wait_for_load_state("networkidle") # 5. 验证用户信息显示 await page.wait_for_selector('.el-card', timeout=10000) # 6. 测试导航到不同页面 - 直接导航到URL（避免菜单可见性问题） await page.goto("http://localhost:3002/users") await page.wait_for_load_state("networkidle") assert "users" in page.url await page.goto("http://localhost:3002/roles") await page.wait_for_load_state("networkidle") assert "roles" in page.url await page.goto("http://localhost:3002/config") await page.wait_for_load_state("networkidle") assert "config" in page.url @pytest.mark.asyncio async def test_system_config_e2e(self, page, authenticated_client): """测试系统配置 - 前后端联通""" # 1. 通过前端登录 await page.goto("http://localhost:3002/login") await page.fill('input[placeholder="请输入用户名"]', settings.TEST_USERNAME) await page.fill('input[placeholder="请输入密码"]', settings.TEST_PASSWORD) await page.click('button[type="submit"]') await page.wait_for_url("**/") await page.wait_for_load_state("networkidle") # 2. 通过前端访问系统配置 await page.click('text=系统配置') await page.wait_for_url("**/config") await page.wait_for_load_state("networkidle") # 3. 验证配置列表显示 await page.wait_for_selector('.el-card', timeout=10000) # 4. 通过API获取配置 config_response = await authenticated_client.get("/api/config") assert config_response.status_code == 200 configs = config_response.json() # 5. 验证前后端数据一致 page_content = await page.content() for config in configs[:3]: assert config['configKey'] in page_content or config['configName'] in page_content @pytest.mark.asyncio async def test_search_and_filter_e2e(self, page, authenticated_client): """测试搜索和过滤 - 前后端联通""" timestamp = int(time.time() * 1000) # 1. 通过API创建多个测试用户 user_ids = [] for i in range(3): username = f"search_{timestamp}_{i}" response = await authenticated_client.post( "/api/users", json={ "username": username, "password": "Test123!@#", "email": f"search_{timestamp}_{i}@example.com", "phone": f"1380013800{i}", "status": 1 } ) assert response.status_code == 201 user_ids.append(response.json()["id"]) try: # 2. 通过前端登录 await page.goto("http://localhost:3002/login") await page.fill('input[placeholder="请输入用户名"]', settings.TEST_USERNAME) await page.fill('input[placeholder="请输入密码"]', settings.TEST_PASSWORD) await page.click('button[type="submit"]') await page.wait_for_url("**/") await page.wait_for_load_state("networkidle") # 3. 通过前端搜索用户 await page.click('text=用户管理') await page.wait_for_url("**/users") await page.wait_for_load_state("networkidle") await page.fill('input[placeholder="搜索用户名或邮箱"]', f"search_{timestamp}") await page.click('button:has-text("搜索")') await page.wait_for_load_state("networkidle") # 4. 验证搜索结果显示 page_content = await page.content() assert f"search_{timestamp}" in page_content # 5. 通过API验证搜索结果 search_response = await authenticated_client.get( "/api/users/page", params={"keyword": f"search_{timestamp}", "page": 0, "size": 10} ) assert search_response.status_code == 200 search_data = search_response.json() assert len(search_data["content"]) >= 3 finally: # 6. 清理测试数据 for user_id in user_ids: try: await authenticated_client.delete(f"/api/users/{user_id}") except Exception: pass @pytest.mark.asyncio async def test_role_management_e2e(self, page, authenticated_client): """测试角色管理 - 前后端联通""" timestamp = int(time.time() * 1000) role_name = f"Role_{timestamp}" role_key = f"role_{timestamp}" # 1. 通过前端登录 await page.goto("http://localhost:3002/login") await page.fill('input[placeholder="请输入用户名"]', settings.TEST_USERNAME) await page.fill('input[placeholder="请输入密码"]', settings.TEST_PASSWORD) await page.click('button[type="submit"]') await page.wait_for_url("**/") await page.wait_for_load_state("networkidle") # 2. 通过前端访问角色管理 await page.click('text=角色管理') await page.wait_for_url("**/roles") await page.wait_for_load_state("networkidle") # 3. 通过前端创建角色 await page.click('text=新增角色') await page.wait_for_load_state("networkidle") await page.fill('input[placeholder=""]', role_name) await page.fill('input[placeholder=""]', role_key) await page.click('button:has-text("确定")') await page.wait_for_load_state("networkidle") # 4. 通过API验证角色已创建 roles_response = await authenticated_client.get("/api/roles") assert roles_response.status_code == 200 roles = roles_response.json() role_exists = any(r['roleName'] == role_name for r in roles) assert role_exists, f"Role {role_name} not found in API response" # 5. 清理测试数据 role_id = next((r['id'] for r in roles if r['roleName'] == role_name), None) if role_id: await authenticated_client.delete(f"/api/roles/{role_id}") @pytest.mark.asyncio async def test_menu_management_e2e(self, page, authenticated_client): """测试菜单管理 - 前后端联通""" # 1. 通过前端登录 await page.goto("http://localhost:3002/login") await page.fill('input[placeholder="请输入用户名"]', settings.TEST_USERNAME) await page.fill('input[placeholder="请输入密码"]', settings.TEST_PASSWORD) await page.click('button[type="submit"]') await page.wait_for_url("**/") await page.wait_for_load_state("networkidle") # 2. 通过前端访问菜单管理 await page.click('text=菜单管理') await page.wait_for_url("**/menus") await page.wait_for_load_state("networkidle") # 3. 验证菜单列表显示 await page.wait_for_selector('.el-card', timeout=10000) # 4. 通过API获取菜单 menus_response = await authenticated_client.get("/api/menus") assert menus_response.status_code == 200 menus = menus_response.json() assert len(menus) > 0, "No menus found" @pytest.mark.asyncio async def test_dict_management_e2e(self, page, authenticated_client): """测试字典管理 - 前后端联通""" # 1. 通过前端登录 await page.goto("http://localhost:3002/login") await page.fill('input[placeholder="请输入用户名"]', settings.TEST_USERNAME) await page.fill('input[placeholder="请输入密码"]', settings.TEST_PASSWORD) await page.click('button[type="submit"]') await page.wait_for_url("**/") await page.wait_for_load_state("networkidle") # 2. 通过前端访问字典管理 await page.click('text=字典管理') await page.wait_for_url("**/dicts") await page.wait_for_load_state("networkidle") # 3. 验证字典列表显示 await page.wait_for_selector('.el-card', timeout=10000) # 4. 通过API获取字典 dicts_response = await authenticated_client.get("/api/dict/types") assert dicts_response.status_code == 200 dicts = dicts_response.json() assert len(dicts) > 0, "No dictionaries found" @pytest.mark.asyncio async def test_notice_management_e2e(self, page, authenticated_client): """测试通知管理 - 前后端联通""" timestamp = int(time.time() * 1000) notice_title = f"通知_{timestamp}" # 1. 通过前端登录 await page.goto("http://localhost:3002/login") await page.fill('input[placeholder="请输入用户名"]', settings.TEST_USERNAME) await page.fill('input[placeholder="请输入密码"]', settings.TEST_PASSWORD) await page.click('button[type="submit"]') await page.wait_for_url("**/") await page.wait_for_load_state("networkidle") # 2. 通过前端访问通知管理 await page.click('text=通知管理') await page.wait_for_url("**/notices") await page.wait_for_load_state("networkidle") # 3. 通过前端创建通知 await page.click('text=新增通知') await page.wait_for_load_state("networkidle") await page.fill('input[placeholder=""]', notice_title) await page.fill('textarea[placeholder=""]', '测试通知内容') await page.click('button:has-text("确定")') await page.wait_for_load_state("networkidle") # 4. 通过API验证通知已创建 notices_response = await authenticated_client.get("/api/notices") assert notices_response.status_code == 200 notices = notices_response.json() notice_exists = any(n['title'] == notice_title for n in notices) assert notice_exists, f"Notice {notice_title} not found in API response" # 5. 清理测试数据 notice_id = next((n['id'] for n in notices if n['title'] == notice_title), None) if notice_id: await authenticated_client.delete(f"/api/notices/{notice_id}") @pytest.mark.asyncio async def test_file_management_e2e(self, page, authenticated_client): """测试文件管理 - 前后端联通""" # 1. 通过前端登录 await page.goto("http://localhost:3002/login") await page.fill('input[placeholder="请输入用户名"]', settings.TEST_USERNAME) await page.fill('input[placeholder="请输入密码"]', settings.TEST_PASSWORD) await page.click('button[type="submit"]') await page.wait_for_url("**/") await page.wait_for_load_state("networkidle") # 2. 通过前端访问文件管理 await page.click('text=文件管理') await page.wait_for_url("**/files") await page.wait_for_load_state("networkidle") # 3. 验证文件列表显示 await page.wait_for_selector('.el-card', timeout=10000) # 4. 通过API获取文件列表 files_response = await authenticated_client.get("/api/files") assert files_response.status_code == 200 files = files_response.json() # 文件列表可能为空，但API应该正常返回 @pytest.mark.asyncio async def test_audit_log_e2e(self, page, authenticated_client): """测试审计日志 - 前后端联通""" # 1. 通过前端登录 await page.goto("http://localhost:3002/login") await page.fill('input[placeholder="请输入用户名"]', settings.TEST_USERNAME) await page.fill('input[placeholder="请输入密码"]', settings.TEST_PASSWORD) await page.click('button[type="submit"]') await page.wait_for_url("**/") await page.wait_for_load_state("networkidle") # 2. 通过前端访问操作日志 await page.click('text=操作日志') await page.wait_for_url("**/operation-logs") await page.wait_for_load_state("networkidle") # 3. 验证操作日志列表显示 await page.wait_for_selector('.el-card', timeout=10000) # 4. 通过API获取操作日志 logs_response = await authenticated_client.get("/api/audit/operation-logs") assert logs_response.status_code == 200 logs = logs_response.json() # 5. 通过前端访问登录日志 await page.click('text=登录日志') await page.wait_for_url("**/login-logs") await page.wait_for_load_state("networkidle") # 6. 验证登录日志列表显示 await page.wait_for_selector('.el-card', timeout=10000) # 7. 通过API获取登录日志 login_logs_response = await authenticated_client.get("/api/audit/login-logs") assert login_logs_response.status_code == 200 login_logs = login_logs_response.json()