# qingcloud-playwright

**Repository Path**: tython/qingcloud-playwright

## Basic Information

- **Project Name**: qingcloud-playwright
- **Description**: Pitrix-UI E2E自动化测试框架
- **Primary Language**: Python
- **License**: Not specified
- **Default Branch**: main
- **Homepage**: https://gitee.com/tython/qingcloud-playwright
- **GVP Project**: No

## Statistics

- **Stars**: 4
- **Forks**: 2
- **Created**: 2024-05-20
- **Last Updated**: 2025-11-03

## Categories & Tags

**Categories**: Uncategorized

**Tags**: None

## README

# Pitrix-UI E2E自动化测试框架

## 1.安装部署
### 1.1 开发环境
```text
python版本: 3.9
包管理工具: poetry

dev阶段:
black 代码格式化: black .
isort import语句排序: isort .
flake8 代码质量分析: flake8 .

安装浏览器插件: playwright install
录制: playwright codegen https://account.qingcloud.com/login
trace追踪: playwright show-trace trace.zip
```

## 2.框架结构说明
- [auth](auth) 登录认证相关会话保存目录
- [business](business) 业务层的操作封装
- [components](components) 前端公共组件组封装
- [config](config) 配置文件
- [constant](constant) 一些常量
- [databases](databases) 数据库
- [datas](datas) 测试数据
- [deploy](deploy) 测试部署
- [logs](logs) 执行测试相关日志
- [models](models) 页面对象封装,采用传统pom
- [public](public) 公共
- [reports](reports) 测试报告
- [test-results](test-results) 测试记录trace
- [testcases](testcases) 测试用例目录
- [utils](utils) 自定义工具库
- [Dockerfile](Dockerfile) 运行在linux系统下的docker打包文件
- [Jenkinsfile](Jenkinsfile) 在kubesphere上运行的流水线文件
- [main.py](main.py) 测试主入口
- [pyproject.toml](pyproject.toml) poetry 配置文件和一些python依赖配置

## 3.业务层
### 3.1.业务层封装说明
- [models](models) 中为单独的某个页面的元素操作封装，子目录根据业务来分；
- [business](business) 中为多个连续的页面操作和业务组合封装，其中子目录按业务模块分，分为计算、存储、网络、安全、公共服务等；
- [testcases](testcases) 中主要调用business中封装好的步骤组成场景的测试用例；

### 3.2 POM说明
一个页面算作一个pom,以VPC产品为例，一共只有3个页面   
主页算一个pom
![vpc_home.png](datas%2Fframe%2Fvpc_home.png)
创建vpc算一个pom
![create_vpc.png](datas%2Fframe%2Fcreate_vpc.png)
VPC详情页算一个pom
![vpc_detail.png](datas%2Fframe%2Fvpc_detail.png)

```python
# vpc页面需要将创建页和详情页进行实例化

class RouterPage(BasePage):
    def __init__(self, page: Page):
        super().__init__(page)
        self.route = "routers"
        self.create_page = CreateRouterPage(page)
        self.detail_page = RouterDetailPage(page)
```

```python
# 业务层面 router_page 作为唯一入口

class RouterBusiness(BasePage):
    def __init__(self, page: Page):
        super().__init__(page)
        self.router_page = RouterPage(page)
```
注意：页面上的弹框不算一个单独的POM，以vxnet页面为例, 因为页面路由 https://console.qingcloud.com/pek3/vxnets 不变，不算一个单独页面 
![vxnet_home.png](datas%2Fframe%2Fvxnet_home.png)

### 3.3 数据库说明
[pitrix-ui.db](databases%2Fpitrix-ui.db) 数据库记录本次测试是所有数据，在开始测试时自动创建，测试结束时保留，直到下次运行测试再次自动创建   
测试用例表   
![testcase_table.png](datas%2Fframe%2Ftestcase_table.png)   
测试配置表   
![test_config_table.png](datas%2Fframe%2Ftest_config_table.png)   
请求记录表   
![request_table.png](datas%2Fframe%2Frequest_table.png)   
响应记录表   
![response_table.png](datas%2Fframe%2Fresponse_table.png)   


### 3.4.异步任务及租赁信息处理
#### 3.4.1 异步任务
1.通过QingCloud API 发送请求轮询异步任务的状态，直到任务完成（成功或失败）；   

#### 3.4.2 租赁信息
1.界面UI创建资源完成时，通过截取响应获取到资源ID，通过QingCloud API 发送请求轮询租赁信息状态，达到预期状态后退出轮询；   

如下所示，MixinUtil类通过继承方式注入 BasePage 类，并注入到业务层中   
```python
class MixinUtil:
    wait_leased = ctx.get_server_yaml_config_from_common('billing').get('wait_leased')

    @staticmethod
    def send_req(action, body):
        client = APIConnection(
            qy_access_key_id=ctx.ak,
            qy_secret_access_key=ctx.sk,
            zone=ctx.region,
            host=ctx.api_host,
            port=ctx.port,
            protocol=ctx.protocol,
            debug=False,
        )
        return client.send_request(action, body, url="/iaas/", verb="GET")

    @classmethod
    def __wait_jobs(
            cls, job_ids: List[str], timeout: int = 600, interval: int = 3
    ) -> [str, bool]:
        """
        等待任务完成(成功或失败)直到超时
        :param job_ids: job_id
        :param timeout: 超时时间,默认600s
        :param interval:  轮询间隔
        :return:True or False
        """
        log.info(f"=== 查询job状态: {job_ids} ===")

        def describe_job(job_ids):
            if isinstance(job_ids, str):
                job_ids = [job_ids]
            res = cls.send_req(
                action="DescribeJobs", body={"action": "DescribeJobs", "jobs": job_ids}
            )
            log.info(f":{res}")
            case_assert.assert_is_not_none(res)
            case_assert.assert_code(res)
            if not res or not res.get("job_set"):
                return None
            return res["job_set"][0]

        deadline = time.time() + timeout
        start_time = perf_counter()
        while time.time() <= deadline:
            res = describe_job(job_ids)
            if not res:
                continue
            job_status = res["status"]
            if job_status in ["successful", "failed"]:
                elapsed_time = perf_counter() - start_time
                log.info(f"job: {job_ids} 执行: {job_status} 耗时: {elapsed_time:.2f} s")
                return job_status
            else:
                time.sleep(interval)
        log.error(f"### job: {job_ids} 执行超时:{timeout} s,请注意人工确认原因!!! ###")
        return False

    @staticmethod
    def extract_action(response: Dict[str, Any]) -> [str, None]:
        """
        提取job action name
        :param response:
        :return:
        """
        if not response:
            return None
        ret_code = response.get("ret_code")
        if ret_code == 0:
            action_name: str = response.get("action").replace("Response", "")
            return action_name
        else:
            return None

    @classmethod
    def extract_job(cls, response: Dict[str, Any]) -> [str, None]:
        """
        提取job id
        :param response:
        :return:
        """
        if not response:
            return None
        log.info(response)
        action_name: str = cls.extract_action(response)
        job_id: str = response.get("job_id")
        job_ids: list = response.get("job_ids")
        if job_id:
            log.info(f"=== 执行:{action_name},提取的job_id为: {job_id} ===")
            return job_id
        if job_ids:
            log.info(f"=== 执行:{action_name},提取的job_ids为: {job_ids} ===")
            return job_ids
        return None

    @allure.step("等待job状态")
    def wait_jobs(
            self,
            response: Dict[str, Any],
            timeout: int = 600,
            interval: int = 3,
            check_status: bool = False,
    ) -> [str, None]:
        """
        自动从response中提取job_id 并执行等待
        :param response: 响应
        :param timeout: 超时时间
        :param interval: 轮询间隔
        :param check_status: 是否检查job的执行结果
        :return:
        """
        job_id = self.extract_job(response)
        if not job_id:
            return None
        job_ids = [job_id] if isinstance(job_id, list) else job_id
        job_status = self.__wait_jobs(job_ids, timeout, interval)
        if check_status:
            case_assert.assert_eq(
                job_status, "successful", f"job:{job_id} 执行:{job_status}"
            )
        return job_status

    def get_lease_info(self, resource, user=None):
        http_data = {
            "action": "GetLeaseInfo",
            "resource": resource,
            "user": user,
        }
        resp = self.send_req(action=http_data["action"], body=http_data)
        return resp

    @allure.step("等待资源租赁信息")
    def wait_lease_status(self,
                          resources: List[str],
                          status: Union[List[str], Tuple[str], Set[str]],
                          time_out: int = 600,
                          ) -> None:
        """
        等待资源租赁状态就绪
        @param resources: 等待的资源
        @param status: 状态
        @param time_out: 设置超时时间
        """
        if not isinstance(resources, list):
            resources = [resources]
        if self.wait_leased:
            log.info(f"正在等待资源:{resources} 的租赁信息中...")
            no_ready_res = deepcopy(resources)
            cnt = 0
            sleep_interval = 10
            while cnt < time_out / sleep_interval:
                for resource_id in no_ready_res:
                    rep = self.get_lease_info(resource_id)
                    case_assert.assert_code(rep)
                    log.info(rep)
                    lease_info = rep['lease_info']
                    if lease_info['status'] in to_list(status):
                        no_ready_res.remove(resource_id)
                if len(no_ready_res) == 0:
                    break
                cnt += 1
                time.sleep(sleep_interval)
            case_assert.assert_eq(0,
                                  len(no_ready_res),
                                  "等待资源 [%s] 的租赁信息为 [%s] 状态时在billing server超时" % (no_ready_res, status))
        else:
            log.info(f"当前环境:{ctx.env} 跳过租赁信息检查")
```

### 3.5 测试配置说明   
每个测试环境对应一个配置文件，保存在 [config](config) 目录下，命名规则以【环境名_config.yaml】来命名   
![test_config.png](datas%2Fframe%2Ftest_config.png)   
取对应环境配置文件的方式如下:   
```python
@staticmethod
def load_yaml_config_to_cache() -> None:
    """
    将配置文件写入缓存
    :return:
    """
    env = os.environ['TEST_ENV']
    filename = f"{env}_config.yaml"
    config_file = CONFIG_DIR / filename
    if not config_file.exists():
        log.warning(f"警告: 文件 {config_file} 不存在,请检查配置文件")
        sys.exit(1)
    log.info(f"正在加载{env} 环境的配置文件: {config_file}")
    config = YamlHandler(config_file).read_yaml()
    cache_manager.set('env', env)
    for key, value in config.items():
        cache_manager.set(key, value)
```
配置文件统一存到 [auth](auth)/[cache](auth%2Fcache) 下的数据库中，保存的内容如下：   
![test_config_from_db.png](datas%2Fframe%2Ftest_config_from_db.png)   
测试时需要取配置通过 `cache_manager.get(key)` 获取
```python
from utils.cache_util import cache_manager
regions = cache_manager.get("regions")
```


## 4.测试说明
### 4.1 测试环境
测试时指定 --env 来指定测试环境   
```python
pytest testcases/test_qingcloud -m iaas_debug --env=qingcloud
```
![run_testcase.png](datas%2Fframe%2Frun_testcase.png)   

### 4.2 测试日志
![test_step.png](datas%2Fframe%2Ftest_step.png)
![test_log.png](datas%2Fframe%2Ftest_log.png)

### 4.3 断言说明
每做一步操作都需要对操作的内容做断言   
1.例如勾选后检查元素是否已选中    
2.导航到目标页面后检查url、title等是否符合预期，页面显示出来各模块元素是否可见   
3.创建、修改、删除资源后页面检查资源是否存在、属性是否变更等   
4.打开某个弹框后检查弹框内容是否符合预期，标志性文案是否可见等   
5.元素的存在与可见性   
6.交互功能响应是否正常  
7.数据输入输出限制和规则验证  
8.按钮的启用、禁用状态，成功、错误标志验证  
9.视音频播放是否正常  
10.上传、下载是否正常  
11.鼠标、键盘事件是否正常  
12.国际化与本地化是否正常

### 4.4 测试报告配置
[notification.yaml](config%2Fnotification.yaml) 下配置 webhook 和 allure server,用于接收测试结果      
配置有效的webhook 可接收到下面的测试结果的消息通知   
![test_message.png](datas%2Fframe%2Ftest_message.png)    
配置有效的allure server端口和地址 可自动上传测试报告到allure server       
![test_report.png](datas%2Fframe%2Ftest_report.png)    


## 6.其他
### 6.1 playwright常用方法
```
playwright 一些关键的方法
1.page.wait_for_selector() 方法，如果没有传 state 参数，默认情况下是等待元素可见 visible
等待元素出现在DOM中：page.wait_for_selector("定位方法", state='attached')
等待从DOM中移除：page.wait_for_selector("定位方法", state='detached')
等待元素可见：page.wait_for_selector("定位方法", state="visible")
等待元素不可见：page.wait_for_selector("定位方法", state='hidden')

2.wait_for() 方法
另外一个先定位元素，再使用wait_for() 方法也可以等待元素到达指定的状态。
page.locator('.toast-message').wait_for(state="detached")

3.自定义异常及超时时间
expect(page.get_by_text("Name"), "should be logged in").to_be_visible(timeout=3000)
```

### 6.2 部署allure-server （可选）
```text
docker pull registry.cn-chengdu.aliyuncs.com/yiweitang/allure-server:latest
docker run -d -p 8080:8080 --restart=always registry.cn-chengdu.aliyuncs.com/yiweitang/allure-server:latest --name allure-server
```

### 6.3 部署qingcloud-playwright （可选）
```text
测试用例在Linux上运行时, 打包前需要将headless需要改为True
docker build -t qingcloud/qingcloud-playwright:v0.1 .
docker run -it --rm --ipc=host --security-opt seccomp=seccomp_profile.json qingcloud/qingcloud-playwright:v0.1 /bin/bash
```

### 6.4 浏览器接管（开发阶段）
```text
mac:
添加chrome路径到系统环境变量中的启动方式
export PATH="/Applications/Google\ Chrome.app/Contents/MacOS:$PATH"
alias chrome="Google\ Chrome"
chrome --remote-debugging-port=12345 --incognito -–start-maximized --new-window https://console.qingcloud.com/ --user-data-dir="/Users/tyw/Desktop/tmp"

未设置环境变量，通过完成CLI启动方式
/Applications/Google\ Chrome.app/Contents/MacOS/Google\ Chrome --remote-debugging-port=12345 --incognito -–start-maximized --new-window https://console.qingcloud.com/ --user-data-dir="/Users/tyw/Desktop/tmp"

win:
添加chrome路径到path环境变量中的启动方式
chrome --remote-debugging-port=12345 -–start-maximized --new-window https://account.qingcloud.com/login
未设置环境变量，通过完成CLI启动方式
chrome.exe --remote-debugging-port=12345 --incognito -–start-maximized --new-window https://account.qingcloud.com/login --user-data-dir="C:\tmp"
```