• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1#!/usr/bin/env python3
2# coding: utf-8
3
4"""
5Copyright (c) 2023 Huawei Device Co., Ltd.
6Licensed under the Apache License, Version 2.0 (the "License");
7you may not use this file except in compliance with the License.
8You may obtain a copy of the License at
9
10    http://www.apache.org/licenses/LICENSE-2.0
11
12Unless required by applicable law or agreed to in writing, software
13distributed under the License is distributed on an "AS IS" BASIS,
14WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15See the License for the specific language governing permissions and
16limitations under the License.
17
18Description: utils for test suite
19"""
20
21import datetime
22import json
23import logging
24import os
25import shutil
26import time
27import subprocess
28import sys
29
30import gzip
31import httpx
32import requests
33import tqdm
34
35
36def get_log_level(arg_log_level):
37    log_level_dict = {
38        'debug': logging.DEBUG,
39        'info': logging.INFO,
40        'warn': logging.WARN,
41        'error': logging.ERROR
42    }
43    if arg_log_level not in log_level_dict.keys():
44        return logging.ERROR  # use error as default log level
45    else:
46        return log_level_dict[arg_log_level]
47
48
49def init_logger(log_level, log_file):
50    logging.basicConfig(filename=log_file,
51                        level=get_log_level(log_level),
52                        encoding=get_encoding(),
53                        format='[%(asctime)s %(filename)s:%(lineno)d]: [%(levelname)s] %(message)s')
54    logging.info("Test command:")
55    logging.info(" ".join(sys.argv))
56
57
58def get_encoding():
59    if is_windows():
60        return 'utf-8'
61    else:
62        return sys.getfilesystemencoding()
63
64
65def is_windows():
66    return sys.platform == 'win32' or sys.platform == 'cygwin'
67
68
69def is_mac():
70    return sys.platform == 'darwin'
71
72
73def is_linux():
74    return sys.platform == 'linux'
75
76
77def get_time_string():
78    return time.strftime('%Y%m%d-%H%M%S')
79
80
81def is_esmodule(hap_type):
82    # if hap_type is stage, it's esmodule.
83    # if hap_type is js, fa, compatible 8, it's js_bundle
84    return 'stage' in hap_type
85
86
87def get_sdk_url():
88    now_time = datetime.datetime.now().strftime('%Y%m%d%H%M%S')
89    last_hour = (datetime.datetime.now() +
90                 datetime.timedelta(hours=-24)).strftime('%Y%m%d%H%M%S')
91    url = 'http://ci.openharmony.cn/api/ci-backend/ci-portal/v1/dailybuilds'
92    downnload_job = {
93        'pageNum': 1,
94        'pageSize': 1000,
95        'startTime': '',
96        'endTime': '',
97        'projectName': 'openharmony',
98        'branch': 'master',
99        'component': '',
100        'deviceLevel': '',
101        'hardwareBoard': '',
102        'buildStatus': '',
103        'buildFailReason': '',
104        'testResult': '',
105    }
106    downnload_job['startTime'] = str(last_hour)
107    downnload_job['endTime'] = str(now_time)
108    post_result = requests.post(url, data=downnload_job)
109    post_data = json.loads(post_result.text)
110    sdk_url_suffix = ''
111    for ohos_sdk_list in post_data['result']['dailyBuildVos']:
112        try:
113            if get_remote_sdk_name() in ohos_sdk_list['obsPath']:
114                sdk_url_suffix = ohos_sdk_list['obsPath']
115                break
116        except BaseException as err:
117            logging.error(err)
118    sdk_url = 'http://download.ci.openharmony.cn/' + sdk_url_suffix
119    return sdk_url
120
121
122def get_api_version(json_path):
123    with open(json_path, 'r') as uni:
124        uni_cont = uni.read()
125        uni_data = json.loads(uni_cont)
126        api_version = uni_data['apiVersion']
127    return api_version
128
129
130def check_gzip_file(file_path):
131    try:
132        with gzip.open(file_path, 'rb') as gzfile:
133            gzfile.read(1)
134    except Exception as e:
135        logging.exception(e)
136        return False
137    return True
138
139
140def is_file_timestamps_same(file_a, file_b):
141    file_a_mtime = os.stat(file_a).st_mtime
142    file_b_mtime = os.stat(file_b).st_mtime
143    return file_a_mtime == file_b_mtime
144
145
146def download(url, temp_file, temp_file_name):
147    with httpx.stream('GET', url) as response:
148        with open(temp_file, "wb") as temp:
149            total_length = int(response.headers.get("content-length"))
150            with tqdm.tqdm(total=total_length, unit="B", unit_scale=True) as pbar:
151                pbar.set_description(temp_file_name)
152                chunk_sum = 0
153                count = 0
154                for chunk in response.iter_bytes():
155                    temp.write(chunk)
156                    chunk_sum += len(chunk)
157                    percentage = chunk_sum / total_length * 100
158                    while str(percentage).startswith(str(count)):
159                        if str(percentage).startswith('100'):
160                            logging.info(f'SDK Download Complete {percentage: .1f}%')
161                            break
162                        else:
163                            logging.info(f'SDK Downloading... {percentage: .1f}%')
164                        count += 1
165                    pbar.update(len(chunk))
166
167
168def add_executable_permission(file_path):
169    current_mode = os.stat(file_path).st_mode
170    new_mode = current_mode | 0o111
171    os.chmod(file_path, new_mode)
172
173
174def get_remote_sdk_name():
175    if is_windows():
176        return 'ohos-sdk-full.tar.gz'
177    elif is_mac():
178        return 'L2-MAC-SDK-FULL.tar.gz'
179    else:
180        logging.error('Unsuport platform to get sdk from daily build')
181        return ''
182