Skip to content

T
Test Xdevice

OpenHarmony / Test Xdevice
大约 1 年 前同步成功

通知 6
Star 23
Fork 0
T
Test Xdevice

体验新版 GitCode，发现更多精彩内容 >>

提交 d7902098 编写于 作者: D deveco_test
浏览文件
操作

调整xDevice目录 

Signed-off-by: Ndeveco_test <liguangjie1@huawei.com>
上级 eb9ce2fd
展开全部 隐藏空白更改
内联 并排
Showing with 12242 addition and 9873 deletion
extension/src/xdevice_extension/_core/constants.py 已删除 100644 → 0
浏览文件 @ eb9ce2fd
#!/usr/bin/env python3
# coding=utf-8
#
# Copyright (c) 2021 Huawei Device Co., Ltd.
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#
from dataclasses import dataclass
__all__ = ["DeviceOsType", "ProductForm", "TestType", "TestExecType",
"DeviceTestType", "HostTestType", "HostDrivenTestType",
"SchedulerType", "ListenerType", "ToolCommandType",
"TEST_DRIVER_SET", "LogType", "ParserType", "CKit", "ComType",
"DeviceLabelType", "DeviceLiteKernel", "GTestConst", "ManagerType",
"CommonParserType", "DeviceConnectorType", "FilePermission"]
@dataclass
class DeviceOsType(object):
"""
DeviceOsType enumeration
"""
default = "default"
lite = "lite"
@dataclass
class DeviceConnectorType:
hdc = "usb-hdc"
@dataclass
class ProductForm(object):
"""
ProductForm enumeration
"""
phone = "phone"
car = "ivi"
television = "tv"
watch = "watch"
tablet = 'tablet'
@dataclass
class TestType(object):
"""
TestType enumeration
"""
unittest = "unittest"
mst = "moduletest"
systemtest = "systemtest"
perf = "performance"
sec = "security"
reli = "reliability"
dst = "distributedtest"
all = "ALL"
@dataclass
class ComType(object):
"""
ComType enumeration
"""
cmd_com = "cmd"
deploy_com = "deploy"
@dataclass
class DeviceLabelType(object):
"""
DeviceLabelType enumeration
"""
wifiiot = "wifiiot"
ipcamera = "ipcamera"
watch = "watch"
phone = "phone"
@dataclass
class DeviceLiteKernel(object):
"""
Lite device os enumeration
"""
linux_kernel = "linux"
lite_kernel = "lite"
TEST_TYPE_DICT = {
"UT": TestType.unittest,
"MST": TestType.mst,
"ST": TestType.systemtest,
"PERF": TestType.perf,
"SEC": TestType.sec,
"RELI": TestType.reli,
"DST": TestType.dst,
"ALL": TestType.all,
}
@dataclass
class TestExecType(object):
"""
TestExecType enumeration according to test execution method
"""
# A test running on the device
device_test = "device"
# A test running on the host (pc)
host_test = "host"
# A test running on the host that interacts with one or more devices.
host_driven_test = "hostdriven"
@dataclass
class DeviceTestType(object):
"""
DeviceTestType enumeration
"""
cpp_test = "CppTest"
hap_test = "HapTest"
junit_test = "JUnitTest"
jsunit_test = "JSUnitTest"
ctest_lite = "CTestLite"
cpp_test_lite = "CppTestLite"
lite_cpp_test = "LiteUnitTest"
open_source_test = "OpenSourceTest"
build_only_test = "BuildOnlyTestLite"
ltp_posix_test = "LtpPosixTest"
oh_kernel_test = "OHKernelTest"
oh_jsunit_test = "OHJSUnitTest"
@dataclass
class HostTestType(object):
"""
HostTestType enumeration
"""
host_gtest = "HostGTest"
host_junit_test = "HostJUnitTest"
@dataclass
class HostDrivenTestType(object):
"""
HostDrivenType enumeration
"""
device_test = "DeviceTest"
TEST_DRIVER_SET = {
DeviceTestType.cpp_test,
DeviceTestType.hap_test,
DeviceTestType.junit_test,
DeviceTestType.jsunit_test,
DeviceTestType.cpp_test_lite,
DeviceTestType.ctest_lite,
DeviceTestType.lite_cpp_test,
DeviceTestType.oh_kernel_test,
HostDrivenTestType.device_test
}
@dataclass
class SchedulerType(object):
"""
SchedulerType enumeration
"""
# default scheduler
scheduler = "Scheduler"
@dataclass
class LogType:
tool = "Tool"
device = "Device"
@dataclass
class ListenerType:
log = "Log"
report = "Report"
upload = "Upload"
collect = "Collect"
collect_lite = "CollectLite"
collect_pass = "CollectPass"
@dataclass
class ParserType:
ctest_lite = "CTestLite"
cpp_test_lite = "CppTestLite"
cpp_test_list_lite = "CppTestListLite"
open_source_test = "OpenSourceTest"
build_only_test = "BuildOnlyTestLite"
@dataclass
class CommonParserType:
jsunit = "JSUnit"
cpptest = "CppTest"
cpptest_list = "CppTestList"
junit = "JUnit"
ltp_posix = "LtpPosixTest"
oh_kernel_test = "OHKernel"
oh_jsunit = "OHJSUnit"
oh_jsunit_list = "OHJSUnitList"
@dataclass
class ManagerType:
device = "device"
lite_device = "device_lite"
@dataclass
class ToolCommandType(object):
toolcmd_key_help = "help"
toolcmd_key_show = "show"
toolcmd_key_run = "run"
toolcmd_key_quit = "quit"
toolcmd_key_list = "list"
@dataclass
class CKit:
push = "PushKit"
command = "CommandKit"
config = "ConfigKit"
wifi = "WIFIKit"
propertycheck = 'PropertyCheckKit'
sts = 'STSKit'
shell = "ShellKit"
testbundle = "TestBundleKit"
appinstall = "AppInstallKit"
component = "ComponentKit"
@dataclass
class GTestConst(object):
exec_para_filter = "--gtest_filter"
exec_para_level = "--gtest_testsize"
@dataclass
class FilePermission(object):
mode_777 = 0o777
mode_755 = 0o755
mode_644 = 0o644
extension/src/xdevice_extension/_core/testkit/__init__.py 已删除 100644 → 0
浏览文件 @ eb9ce2fd
#!/usr/bin/env python3
# coding=utf-8
#
# Copyright (c) 2021 Huawei Device Co., Ltd.
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#
extension/src/xdevice_extension/_core/utils.py 已删除 100644 → 0
浏览文件 @ eb9ce2fd
#!/usr/bin/env python3
# coding=utf-8
#
# Copyright (c) 2021 Huawei Device Co., Ltd.
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#
import os
import platform
import subprocess
import signal
import stat
from tempfile import NamedTemporaryFile
from xdevice import ShellHandler
from xdevice import platform_logger
from xdevice import get_plugin
from xdevice import Plugin
LOG = platform_logger("Utils")
def get_filename_extension(file_path):
_, fullname = os.path.split(file_path)
filename, ext = os.path.splitext(fullname)
return filename, ext
def start_standing_subprocess(cmd, pipe=subprocess.PIPE, return_result=False):
"""Starts a non-blocking subprocess that is going to continue running after
this function returns.
A subprocess group is actually started by setting sid, so we can kill all
the processes spun out from the subprocess when stopping it. This is
necessary in case users pass in pipe commands.
Args:
cmd: Command to start the subprocess with.
pipe: pipe to get execution result
return_result: return execution result or not
Returns:
The subprocess that got started.
"""
sys_type = platform.system()
process = subprocess.Popen(cmd, stdout=pipe, shell=False,
preexec_fn=None if sys_type == "Windows"
else os.setsid)
if not return_result:
return process
else:
rev = process.stdout.read()
return rev.decode("utf-8").strip()
def stop_standing_subprocess(process):
"""Stops a subprocess started by start_standing_subprocess.
Catches and ignores the PermissionError which only happens on Macs.
Args:
process: Subprocess to terminate.
"""
try:
sys_type = platform.system()
signal_value = signal.SIGINT if sys_type == "Windows" \
else signal.SIGTERM
os.kill(process.pid, signal_value)
except (PermissionError, AttributeError, FileNotFoundError,
SystemError) as error:
LOG.error("stop standing subprocess error '%s'" % error)
def get_decode(stream):
if not isinstance(stream, str) and not isinstance(stream, bytes):
ret = str(stream)
else:
try:
ret = stream.decode("utf-8", errors="ignore")
except (ValueError, AttributeError, TypeError):
ret = str(stream)
return ret
def is_proc_running(pid, name=None):
if platform.system() == "Windows":
proc_sub = subprocess.Popen(["C:\\Windows\\System32\\tasklist"],
stdout=subprocess.PIPE,
shell=False)
proc = subprocess.Popen(["C:\\Windows\\System32\\findstr", "%s" % pid],
stdin=proc_sub.stdout,
stdout=subprocess.PIPE, shell=False)
elif platform.system() == "Linux":
# /bin/ps -ef | /bin/grep -v grep | /bin/grep -w pid
proc_sub = subprocess.Popen(["/bin/ps", "-ef"],
stdout=subprocess.PIPE,
shell=False)
proc_v_sub = subprocess.Popen(["/bin/grep", "-v", "grep"],
stdin=proc_sub.stdout,
stdout=subprocess.PIPE,
shell=False)
proc = subprocess.Popen(["/bin/grep", "-w", "%s" % pid],
stdin=proc_v_sub.stdout,
stdout=subprocess.PIPE, shell=False)
elif platform.system() == "Darwin":
# /bin/ps -ef | /bin/grep -v grep | /bin/grep -w pid
proc_sub = subprocess.Popen(["/bin/ps", "-ef"],
stdout=subprocess.PIPE,
shell=False)
proc_v_sub = subprocess.Popen(["/usr/bin/grep", "-v", "grep"],
stdin=proc_sub.stdout,
stdout=subprocess.PIPE,
shell=False)
proc = subprocess.Popen(["/usr/bin/grep", "-w", "%s" % pid],
stdin=proc_v_sub.stdout,
stdout=subprocess.PIPE, shell=False)
else:
raise Exception("Unknown system environment.")
(out, _) = proc.communicate()
out = get_decode(out).strip()
LOG.debug("check %s proc running output: %s", pid, out)
if out == "":
return False
else:
return True if name is None else out.find(name) != -1
def create_dir(path):
"""Creates a directory if it does not exist already.
Args:
path: The path of the directory to create.
"""
full_path = os.path.abspath(os.path.expanduser(path))
if not os.path.exists(full_path):
os.makedirs(full_path, exist_ok=True)
def modify_props(device, local_prop_file, target_prop_file, new_props):
"""To change the props if need
Args:
device: the device to modify props
local_prop_file : the local file to save the old props
target_prop_file : the target prop file to change
new_props : the new props
Returns:
True : prop file changed
False : prop file no need to change
"""
is_changed = False
device.pull_file(target_prop_file, local_prop_file)
old_props = {}
changed_prop_key = []
lines = []
flags = os.O_RDONLY
modes = stat.S_IWUSR | stat.S_IRUSR
with os.fdopen(os.open(local_prop_file, flags, modes), "r") as old_file:
lines = old_file.readlines()
if lines:
lines[-1] = lines[-1] + '\n'
for line in lines:
line = line.strip()
if not line.startswith("#") and line.find("=") > 0:
key_value = line.split("=")
if len(key_value) == 2:
old_props[line.split("=")[0]] = line.split("=")[1]
for key, value in new_props.items():
if key not in old_props.keys():
lines.append("".join([key, "=", value, '\n']))
is_changed = True
elif old_props.get(key) != value:
changed_prop_key.append(key)
is_changed = True
if is_changed:
local_temp_prop_file = NamedTemporaryFile(mode='w', prefix='build',
suffix='.tmp', delete=False)
for index, line in enumerate(lines):
if not line.startswith("#") and line.find("=") > 0:
key = line.split("=")[0]
if key in changed_prop_key:
lines[index] = "".join([key, "=", new_props[key], '\n'])
local_temp_prop_file.writelines(lines)
local_temp_prop_file.close()
device.push_file(local_temp_prop_file.name, target_prop_file)
device.execute_shell_command(" ".join(["chmod 644", target_prop_file]))
LOG.info("Changed the system property as required successfully")
os.remove(local_temp_prop_file.name)
return is_changed
def convert_ip(origin_ip):
addr = origin_ip.strip().split(".")
if len(addr) == 4:
return addr[0] + "." + '*'*len(addr[1]) + "." + '*'*len(addr[2]) + \
"." + addr[-1]
else:
return origin_ip
def convert_port(port):
_port = str(port)
if len(_port) >= 2:
return "{}{}{}".format(_port[0], "*" * (len(_port) - 2), _port[-1])
else:
return "*{}".format(_port[-1])
def convert_serial(serial):
if serial.startswith("local_"):
return "local_" + '*'*(len(serial)-6)
elif serial.startswith("remote_"):
return "remote_" + convert_ip(serial.split("_")[1])
else:
length = len(serial)//3
return serial[0:length] + "*"*(len(serial)-length*2) + serial[-length:]
def get_shell_handler(request, parser_type):
suite_name = request.root.source.test_name
parsers = get_plugin(Plugin.PARSER, parser_type)
if parsers:
parsers = parsers[:1]
parser_instances = []
for listener in request.listeners:
listener.device_sn = request.config.environment.devices[0].device_sn
for parser in parsers:
parser_instance = parser.__class__()
parser_instance.suite_name = suite_name
parser_instance.listeners = request.listeners
parser_instances.append(parser_instance)
handler = ShellHandler(parser_instances)
return handler
def check_path_legal(path):
if path and " " in path:
return "\"%s\"" % path
return path
extension/setup.py → plugins/ohos/setup.py
浏览文件 @ d7902098
#!/usr/bin/env python3
# coding=utf-8
#
# Copyright (c) 2021 Huawei Device Co., Ltd.
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#
from setuptools import setup
INSTALL_REQUIRES = [
]
def main():
setup(name='xdevice-extension',
description='xdevice extension test framework',
url='',
package_dir={'': 'src'},
packages=['xdevice_extension',
'xdevice_extension._core',
'xdevice_extension._core.driver',
'xdevice_extension._core.environment',
'xdevice_extension._core.executor',
'xdevice_extension._core.testkit'
],
package_data={
},
entry_points={
'driver': [
'drivers=xdevice_extension._core.driver.drivers',
'openharmony=xdevice_extension._core.driver.openharmony',
'kunpeng=xdevice_extension._core.driver.kunpeng'
],
'parser': [
'parser=xdevice_extension._core.driver.parser'
],
'device': [
'default=xdevice_extension._core.environment.device'
],
'manager': [
'device=xdevice_extension._core.environment.manager_device'
],
'listener': [
'listener=xdevice_extension._core.executor.listener'
],
'testkit': [
'kit=xdevice_extension._core.testkit.kit'
]
},
zip_safe=False,
install_requires=INSTALL_REQUIRES,
)
if __name__ == "__main__":
main()
#!/usr/bin/env python3
# coding=utf-8
#
# Copyright (c) 2020-2022 Huawei Device Co., Ltd.
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#
from setuptools import setup
INSTALL_REQUIRES = []
def main():
setup(name='xdevice-ohos',
description='plugin for ohos',
url='',
package_dir={'': 'src'},
packages=['ohos',
'ohos.drivers',
'ohos.environment',
'ohos.executor',
'ohos.managers',
'ohos.parser',
'ohos.testkit'
],
entry_points={
'device': [
'device=ohos.environment.device',
'device_lite=ohos.environment.device_lite'
],
'manager': [
'manager=ohos.managers.manager_device',
'manager_lite=ohos.managers.manager_lite'
],
'driver': [
'drivers=ohos.drivers.drivers',
'drivers_lite=ohos.drivers.drivers_lite',
'homevision=ohos.drivers.homevision',
'kunpeng=ohos.drivers.kunpeng',
'openharmony=ohos.drivers.openharmony'
],
'listener': [
'listener=ohos.executor.listener',
],
'testkit': [
'kit=ohos.testkit.kit',
'kit_lite=ohos.testkit.kit_lite'
],
'parser': [
'parser_lite=ohos.parser.parser_lite',
'parser=ohos.parser.parser'
]
},
zip_safe=False,
install_requires=INSTALL_REQUIRES,
)
if __name__ == "__main__":
main()
src/xdevice/_core/build/__init__.py → plugins/ohos/src/ohos/__init__.py
浏览文件 @ d7902098
......@@ -2,7 +2,7 @@
# coding=utf-8
#
# Copyright (c) 2020 Huawei Device Co., Ltd.
# Copyright (c) 2022 Huawei Device Co., Ltd.
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
......@@ -14,4 +14,4 @@
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#
#
\ No newline at end of file
plugins/ohos/src/ohos/constants.py 0 → 100644
浏览文件 @ d7902098
#!/usr/bin/env python3
# coding=utf-8
#
# Copyright (c) 2020-2022 Huawei Device Co., Ltd.
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#
from dataclasses import dataclass
__all__ = ["ComType", "HostDrivenTestType",
"ParserType", "DeviceLiteKernel", "CKit"]
@dataclass
class ComType(object):
"""
ComType enumeration
"""
cmd_com = "cmd"
deploy_com = "deploy"
@dataclass
class HostDrivenTestType(object):
"""
HostDrivenType enumeration
"""
device_test = "DeviceTest"
windows_test = "WindowsTest"
@dataclass
class ParserType:
ctest_lite = "CTestLite"
cpp_test_lite = "CppTestLite"
cpp_test_list_lite = "CppTestListLite"
open_source_test = "OpenSourceTest"
build_only_test = "BuildOnlyTestLite"
jsuit_test_lite = "JSUnitTestLite"
@dataclass
class DeviceLiteKernel(object):
"""
Lite device os enumeration
"""
linux_kernel = "linux"
lite_kernel = "lite"
@dataclass
class CKit:
push = "PushKit"
install = "ApkInstallKit"
liteinstall = "LiteAppInstallKit"
command = "CommandKit"
config = "ConfigKit"
wifi = "WIFIKit"
propertycheck = 'PropertyCheckKit'
sts = 'STSKit'
shell = "ShellKit"
deploy = 'DeployKit'
mount = 'MountKit'
liteuikit = 'LiteUiKit'
rootfs = "RootFsKit"
liteshell = "LiteShellKit"
app_install = "AppInstallKit"
deploytool = "DeployToolKit"
query = "QueryKit"
component = "ComponentKit"
extension/src/xdevice_extension/_core/executor/__init__.py → plugins/ohos/src/ohos/drivers/__init__.py
浏览文件 @ d7902098
#!/usr/bin/env python3
# coding=utf-8
#
# Copyright (c) 2021 Huawei Device Co., Ltd.
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#
#!/usr/bin/env python3
# coding=utf-8
#
# Copyright (c) 2022 Huawei Device Co., Ltd.
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#
\ No newline at end of file
src/xdevice/_core/driver/drivers_lite.py → plugins/ohos/src/ohos/drivers/drivers_lite.py
浏览文件 @ d7902098
......@@ -2,7 +2,7 @@
# coding=utf-8
#
# Copyright (c) 2020-2021 Huawei Device Co., Ltd.
# Copyright (c) 2020-2022 Huawei Device Co., Ltd.
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
......@@ -22,40 +22,41 @@ import glob
import time
import stat
from _core.config.config_manager import UserConfigManager
from _core.constants import ConfigConst
from _core.constants import DeviceTestType
from _core.constants import GTestConst
from _core.constants import DeviceLabelType
from _core.constants import ComType
from _core.constants import ParserType
from _core.constants import CKit
from _core.constants import DeviceLiteKernel
from _core.constants import FilePermission
from _core.driver.parser_lite import ShellHandler
from _core.environment.dmlib_lite import generate_report
from _core.exception import ExecuteTerminate
from _core.exception import LiteDeviceError
from _core.exception import LiteDeviceExecuteCommandError
from _core.exception import ParamError
from _core.executor.listener import CollectingLiteGTestListener
from _core.executor.listener import TestDescription
from _core.interface import IDriver
from _core.plugin import Plugin
from _core.plugin import get_plugin
from _core.logger import platform_logger
from _core.report.reporter_helper import DataHelper
from _core.testkit.json_parser import JsonParser
from _core.testkit.kit_lite import DeployKit
from _core.testkit.kit_lite import DeployToolKit
from _core.utils import get_config_value
from _core.utils import get_kit_instances
from _core.utils import check_result_report
from _core.utils import get_device_log_file
from _core.utils import get_filename_extension
from _core.utils import get_file_absolute_path
from _core.utils import get_test_component_version
from _core.report.suite_reporter import SuiteReporter
from xdevice import UserConfigManager
from xdevice import ConfigConst
from xdevice import ExecuteTerminate
from xdevice import ParamError
from xdevice import TestDescription
from xdevice import IDriver
from xdevice import Plugin
from xdevice import get_plugin
from xdevice import platform_logger
from xdevice import DataHelper
from xdevice import JsonParser
from xdevice import get_config_value
from xdevice import get_kit_instances
from xdevice import check_result_report
from xdevice import get_device_log_file
from xdevice import get_filename_extension
from xdevice import get_file_absolute_path
from xdevice import get_test_component_version
from xdevice import SuiteReporter
from xdevice import ShellHandler
from xdevice import FilePermission
from xdevice import DeviceTestType
from xdevice import GTestConst
from xdevice import DeviceLabelType
from ohos.constants import ComType
from ohos.constants import ParserType
from ohos.constants import DeviceLiteKernel
from ohos.constants import CKit
from ohos.exception import LiteDeviceError
from ohos.exception import LiteDeviceExecuteCommandError
from ohos.executor.listener import CollectingLiteGTestListener
from ohos.testkit.kit_lite import DeployKit
from ohos.testkit.kit_lite import DeployToolKit
from ohos.environment.dmlib_lite import generate_report
__all__ = ["CppTestDriver", "CTestDriver", "init_remote_server"]
LOG = platform_logger("DriversLite")
......@@ -305,7 +306,7 @@ class CppTestDriver(IDriver):
SuiteReporter.set_suite_list([])
tests = parser_instances[0].tests
if not tests:
LOG.error("collect test failed!", error_no="00402")
LOG.error("Collect test failed!", error_no="00402")
return parser_instances[0].tests
def run_cpp_test(self, command, request):
......@@ -390,9 +391,9 @@ class CppTestDriver(IDriver):
test_rerun = True
if self.check_xml_exist(self.execute_bin + ".xml"):
test_rerun = False
test_run = self.read_nfs_xml(request, self.config.device_xml_path,
test_run = self.read_nfs_xml(request,
self.config.device_xml_path,
test_rerun)
if len(test_run) < len(expected_tests):
expected_tests = TestDescription.remove_test(expected_tests,
test_run)
......@@ -485,7 +486,7 @@ class CppTestDriver(IDriver):
os.path.join(os.path.split(
self.result)[0], report_xml))
except (FileNotFoundError, IOError) as error:
LOG.error("download xml failed %s" % error, error_no="00403")
LOG.error("Download xml failed %s" % error, error_no="00403")
def check_xml_exist(self, xml_file, timeout=60):
ls_command = "ls %s" % self.config.device_report_path
......@@ -557,9 +558,9 @@ class CppTestDriver(IDriver):
if test not in tests:
tests.append(test)
except (FileNotFoundError, IOError) as error:
LOG.error("download xml failed %s" % error, error_no="00403")
LOG.error("Download xml failed %s" % error, error_no="00403")
except SyntaxError as error:
LOG.error("parse xml failed %s" % error, error_no="00404")
LOG.error("Parse xml failed %s" % error, error_no="00404")
return tests
def delete_device_xml(self, request, report_path):
......@@ -569,7 +570,7 @@ class CppTestDriver(IDriver):
format(self.remote)
LOG.error(err_msg, error_no="00403")
raise TypeError(err_msg)
LOG.info("delete xml directory {} from remote server: {}"
LOG.info("Delete xml directory {} from remote server: {}"
"".format
(report_path, remote_nfs.get("ip")))
if remote_nfs["remote"] == "true":
......@@ -651,13 +652,13 @@ class CTestDriver(IDriver):
for (kit_instance, kit_info) in zip(kit_instances,
json_config.get_kits()):
if isinstance(kit_instance, DeployToolKit):
LOG.debug("run ctest third party")
LOG.debug("Run ctest third party")
self._run_ctest_third_party(source=source, request=request,
time_out=int(
kit_instance.time_out))
break
else:
LOG.debug("run ctest")
LOG.debug("Run ctest")
self._run_ctest(source=source, request=request)
break
......@@ -735,8 +736,8 @@ class CTestDriver(IDriver):
handler = ShellHandler(parser_instances)
while True:
input_burning = input("Please enter 'y' or 'n' after "
"the burning is complete,"
input_burning = input("Please enter 'y' or 'n' "
"after the burning is complete,"
"enter 'quit' to exit:")
if input_burning.lower().strip() in ["y", "yes"]:
LOG.info("Burning succeeded.")
......@@ -746,17 +747,17 @@ class CTestDriver(IDriver):
elif input_burning.lower().strip() == "quit":
break
else:
LOG.info("The input {} parameter is incorrect,"
"please enter 'y' or 'n' after the "
"burning is complete ,enter 'quit' "
"to exit.".format(input_burning))
LOG.info("Please press the device "
"reset button when the send commmand [] appears ")
time.sleep(3)
LOG.info({"The input parameter is incorrect,please"
" enter 'y' or 'n' after the burning is "
"complete,enter 'quit' to exit."})
LOG.info("Please press the reset button on the device ")
time.sleep(5)
self.result = "%s.xml" % os.path.join(
request.config.report_path, "result", self.file_name)
self.config.device.device.com_dict.get(
ComType.deploy_com).connect()
LOG.debug("Device com:{}".format(self.config.device.device))
result, _, error = self.config.device.device. \
execute_command_with_timeout(
command=[], case_type=DeviceTestType.ctest_lite,
......@@ -766,7 +767,7 @@ class CTestDriver(IDriver):
__get_serial__())
device_log_file_open = \
os.open(device_log_file, os.O_WRONLY | os.O_CREAT |
os.O_APPEND, 0o755)
os.O_APPEND, FilePermission.mode_755)
with os.fdopen(device_log_file_open, "a") as file_name:
file_name.write("{}{}".format(
"\n".join(result.split("\n")[0:-1]), "\n"))
......@@ -924,7 +925,7 @@ class OpenSourceTestDriver(IDriver):
error, _, _ = self.run(command, listeners, timeout=60)
if error:
LOG.error(
"execute %s failed" % command, error_no="00402")
"Execute %s failed" % command, error_no="00402")
def __result__(self):
return self.result if os.path.exists(self.result) else ""
......@@ -1117,7 +1118,7 @@ class JSUnitTestLiteDriver(IDriver):
__get_serial__())
device_log_file_open =\
os.open(device_log_file, os.O_WRONLY | os.O_CREAT | os.O_APPEND,
0o755)
FilePermission.mode_755)
with os.fdopen(device_log_file_open, "a") as file_name:
file_name.write("{}{}".format(
"\n".join(result.split("\n")[0:-1]), "\n"))
......
plugins/ohos/src/ohos/drivers/homevision.py 0 → 100644
浏览文件 @ d7902098
#!/usr/bin/env python3
# coding=utf-8
#
# Copyright (c) 2022 Huawei Device Co., Ltd.
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#
import os
import platform
import subprocess
import sys
from xdevice import IDriver
from xdevice import Plugin
from xdevice import platform_logger
from xdevice import get_device_log_file
from xdevice import JsonParser
from xdevice import get_config_value
from xdevice import FilePermission
from xdevice import get_decode
HOMEVISION_TEST = "HomeVision"
LOG = platform_logger(HOMEVISION_TEST)
@Plugin(type=Plugin.DRIVER, id=HOMEVISION_TEST)
class HomeVisionTest(IDriver):
"""
HomeVisionTest is a Test that runs a driver test on given tv devices.
"""
# test driver config
config = None
result = ""
def __check_environment__(self, device_options):
pass
def __check_config__(self, config):
pass
def __execute__(self, request):
self.config = request.config
self.config.device = request.get_devices()[0]
self.config.devices = request.get_devices()
# get config file
config_file = request.get_config_file()
if not config_file:
LOG.error("Config file not exists")
return
LOG.debug("HomeVisionTest config file Path: %s" % config_file)
device_log_pipes = []
try:
for device in self.config.devices:
device_name = device.get("name", "")
device_log = get_device_log_file(
request.config.report_path, device.__get_serial__(),
"device_log", device_name)
hilog = get_device_log_file(
request.config.report_path, device.__get_serial__(),
"device_hilog", device_name)
device_log_open = os.open(device_log, os.O_WRONLY |
os.O_CREAT | os.O_APPEND,
FilePermission.mode_755)
hilog_open = os.open(hilog, os.O_WRONLY | os.O_CREAT |
os.O_APPEND, FilePermission.mode_755)
device_log_pipe = os.fdopen(device_log_open, "a")
hilog_pipe = os.fdopen(hilog_open, "a")
device.start_catch_device_log(device_log_pipe, hilog_pipe)
device_log_pipes.extend([device_log_pipe, hilog_pipe])
self._run_homevision_test(config_file)
finally:
for device_log_pipe in device_log_pipes:
device_log_pipe.flush()
device_log_pipe.close()
for device in self.config.devices:
device.stop_catch_device_log()
@classmethod
def _run_homevision_test(cls, config_file):
from xdevice import Variables
# insert RegressionTest path for loading homevision module
homevision_test_module = os.path.join(Variables.exec_dir,
"RegressionTest")
sys.path.insert(1, homevision_test_module)
json_config = JsonParser(config_file)
device_ip = get_config_value("device-ip", json_config.get_driver(),
False)
job_id = get_config_value("job-id", json_config.get_driver(), False)
home_vision_app_name = get_config_value(
"home-vision-app-name", json_config.get_driver(), False)
cmd_parts = []
if platform.system() == "Windows":
cmd_parts.append("python")
else:
cmd_parts.append("python3")
relative_path = "startAutoTest.py"
cmd_parts.append(os.path.abspath(os.path.join(homevision_test_module,
relative_path)))
cmd_parts.append(device_ip)
cmd_parts.append(job_id)
cmd_parts.append(home_vision_app_name)
cmd = " ".join(cmd_parts)
LOG.info("Start HomeVision test with cmd: %s" % cmd)
try:
proc = subprocess.Popen(cmd_parts, shell=False,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE)
(out, _) = proc.communicate()
out = get_decode(out).strip()
for line in out.split("\n"):
LOG.info(line)
except (subprocess.CalledProcessError, FileNotFoundError,
Exception) as error:
LOG.error("HomeVision test error: %s" % error)
def __result__(self):
return self.result if os.path.exists(self.result) else ""
extension/src/xdevice_extension/_core/driver/kunpeng.py → plugins/ohos/src/ohos/drivers/kunpeng.py
浏览文件 @ d7902098
#!/usr/bin/env python3
# coding=utf-8
#
# Copyright (c) 2021 Huawei Device Co., Ltd.
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#
import os
import platform
import subprocess
import sys
import uuid
from xdevice import platform_logger
from xdevice import IDriver
from xdevice import Plugin
from xdevice_extension._core.utils import get_decode
KUNPENG_TEST = "KunpengTest"
LOG = platform_logger(KUNPENG_TEST)
@Plugin(type=Plugin.DRIVER, id=KUNPENG_TEST)
class KunpengTest(IDriver):
"""
KunpengTest is a Test that runs a host-driven test on given kunpeng
servers.
"""
# test driver config
config = None
result = ""
def __check_environment__(self, device_options):
pass
def __check_config__(self, config):
pass
def __execute__(self, request):
self.config = request.config
# 1.parse config file
mainconfig_file = request.get_config_file()
if not mainconfig_file:
LOG.error("config file not exists")
return
LOG.debug("KunpengTest mainconfig_file FilePath: %s" % mainconfig_file)
# 2.set params
tmp_id = str(uuid.uuid4())
tmp_folder = os.path.join(self.config.report_path, "temp")
self.config.tmp_sub_folder = os.path.join(tmp_folder, "task_" + tmp_id)
os.makedirs(self.config.tmp_sub_folder, exist_ok=True)
# 3.test execution
self._start_kunpengtest_with_cmd(mainconfig_file)
return
def _start_kunpengtest_with_cmd(self, mainconfig_file):
from xdevice import Variables
# insert& _kunpengtest path for loading kunpengtest module
kunpengtest_module = os.path.join(Variables.modules_dir,
"_kunpengtest")
sys.path.insert(1, kunpengtest_module)
cmd_parts = []
if platform.system() == "Windows":
cmd_parts.append("python")
else:
cmd_parts.append("python3")
relative_path = "uniautos/src/Framework/Dev/bin/UniAutosScript.py"
cmd_parts.append(os.path.abspath(os.path.join(kunpengtest_module,
relative_path)))
cmd_parts.append("-c")
cmd_parts.append(mainconfig_file)
cmd_parts.append("-rp")
cmd_parts.append(self.config.tmp_sub_folder)
cmd = " ".join(cmd_parts)
LOG.info("start kunpengtest with cmd: %s" % cmd)
try:
proc = subprocess.Popen(cmd_parts, shell=False,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE)
(out, _) = proc.communicate()
out = get_decode(out).strip()
for line in out.split("\n"):
LOG.info(line)
except (subprocess.CalledProcessError, FileNotFoundError) as error:
LOG.error("kunpeng test error: %s" % error)
def __result__(self):
return self.result if os.path.exists(self.result) else ""
#!/usr/bin/env python3
# coding=utf-8
#
# Copyright (c) 2022 Huawei Device Co., Ltd.
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#
import os
import platform
import subprocess
from subprocess import Popen, PIPE, STDOUT
import sys
import uuid
from xdevice import platform_logger
from xdevice import IDriver
from xdevice import Plugin
from xdevice import get_config_value
from xdevice import ConfigConst
KUNPENG_TEST = "KunpengTest"
LOG = platform_logger(KUNPENG_TEST)
@Plugin(type=Plugin.DRIVER, id=KUNPENG_TEST)
class KunpengTest(IDriver):
"""
KunpengTest is a Test that runs a host-driven test on given kunpeng
servers.
"""
# test driver config
config = None
result = ""
def __check_environment__(self, device_options):
pass
def __check_config__(self, config):
pass
def __execute__(self, request):
self.config = request.config
result_dir = os.path.join(request.config.report_path, "result")
# status.db dir
log_dir = os.path.join(request.config.report_path, "log")
if not os.path.exists(result_dir):
os.makedirs(result_dir)
if not os.path.exists(log_dir):
os.makedirs(log_dir)
self.result = os.path.join(result_dir, 'result.xml')
testargs = request.get(ConfigConst.testargs)
mainconfig_file = testargs.get("main_config_path", [None])[0]
project_src_path = testargs.get("main_config_path", [None])[1]
if not mainconfig_file:
LOG.info('can not find mainconfig_file in '
'testargs!!, use default mainconfig_file')
return
tmp_id = str(uuid.uuid1())
tmp_folder = os.path.join(self.config.report_path, "temp")
self.config.tmp_sub_folder = os.path.join(tmp_folder, "task_" + tmp_id)
os.makedirs(self.config.tmp_sub_folder, exist_ok=True)
# 3.test execution
# mainconfig_file
self._start_kunpengtest_with_cmd(mainconfig_file, project_src_path,
log_dir)
return
def _get_driver_config(self, json_config):
self.config.main_config = get_config_value(
'main_config', json_config.get_driver(), False)
self.config.test_bed = get_config_value(
'test_bed', json_config.get_driver(), False)
self.config.test_set = get_config_value(
'test_set', json_config.get_driver(), False)
def _start_kunpengtest_with_cmd(self, mainconfig_file, project_src_path,
log_dir): # , mainconfig_file
sys.path.append(project_src_path)
cmd_parts = []
if platform.system() == "Windows":
cmd_parts.append("python")
else:
cmd_parts.append("python3")
start_script_path = os.path.join(project_src_path, 'bin', 'kprun.py')
cmd_parts.append(start_script_path)
cmd_parts.append("-c")
cmd_parts.append(mainconfig_file)
cmd_parts.append("-rp")
cmd_parts.append(self.result)
cmd_parts.append("-le")
cmd_parts.append(log_dir)
cmd = " ".join(cmd_parts)
LOG.info("start kunpengtest with cmd: %s" % cmd)
try:
with Popen(cmd, shell=False, stdout=PIPE, stderr=STDOUT) as p:
for line in p.stdout:
if line:
LOG.info(line.strip().decode('utf-8'))
else:
break
except (subprocess.CalledProcessError, FileNotFoundError) as error:
LOG.error("kunpeng test error: %s" % error)
def __result__(self):
return self.result if os.path.exists(self.result) else ""
extension/src/xdevice_extension/_core/driver/openharmony.py → plugins/ohos/src/ohos/drivers/openharmony.py
浏览文件 @ d7902098
......@@ -32,13 +32,13 @@ from xdevice import get_kit_instances
from xdevice import get_config_value
from xdevice import do_module_kit_setup
from xdevice import do_module_kit_teardown
from xdevice import DeviceTestType
from xdevice import CommonParserType
from xdevice import FilePermission
from xdevice import ShellCommandUnresponsiveException
from xdevice_extension._core.constants import DeviceTestType
from xdevice_extension._core.constants import CommonParserType
from xdevice_extension._core.constants import FilePermission
from xdevice_extension._core.executor.listener import CollectingPassListener
from xdevice_extension._core.exception import ShellCommandUnresponsiveException
from xdevice_extension._core.testkit.kit import oh_jsunit_para_parse
from ohos.testkit.kit import oh_jsunit_para_parse
from ohos.executor.listener import CollectingPassListener
__all__ = ["OHJSUnitTestDriver", "OHKernelTestDriver"]
......@@ -78,16 +78,26 @@ class OHKernelTestDriver(IDriver):
self.result = "%s.xml" % \
os.path.join(request.config.report_path,
"result", request.get_module_name())
device_log = get_device_log_file(
request.config.report_path,
request.config.device.__get_serial__(),
"device_log")
hilog = get_device_log_file(
request.config.report_path,
request.config.device.__get_serial__(),
"device_hilog")
device_log_open = os.open(device_log, os.O_WRONLY | os.O_CREAT |
os.O_APPEND, FilePermission.mode_755)
hilog_open = os.open(hilog, os.O_WRONLY | os.O_CREAT | os.O_APPEND,
FilePermission.mode_755)
with os.fdopen(hilog_open, "a") as hilog_file_pipe:
self.config.device.start_catch_device_log(hilog_file_pipe)
with os.fdopen(device_log_open, "a") as log_file_pipe, \
os.fdopen(hilog_open, "a") as hilog_file_pipe:
self.config.device.start_catch_device_log(log_file_pipe,
hilog_file_pipe)
self._run_oh_kernel(config_file, request.listeners, request)
log_file_pipe.flush()
hilog_file_pipe.flush()
except Exception as exception:
self.error_message = exception
......@@ -214,7 +224,7 @@ class OHJSUnitTestDriver(IDriver):
def __execute__(self, request):
try:
LOG.debug("Start execute xdevice extension JSUnit Test")
LOG.debug("Start execute OpenHarmony JSUnitTest")
self.result = os.path.join(
request.config.report_path, "result",
'.'.join((request.get_module_name(), "xml")))
......@@ -230,14 +240,15 @@ class OHJSUnitTestDriver(IDriver):
request.root.source.source_string, error_no="00110")
LOG.debug("Test case file path: %s" % suite_file)
hilog = get_device_log_file(request.config.report_path,
request.get_module_name(),
request.config.device.__get_serial__() + "_" + request.
get_module_name(),
"device_hilog")
hilog_open = os.open(hilog, os.O_WRONLY | os.O_CREAT | os.O_APPEND,
0o755)
self.config.device.execute_shell_command(command="hilog -r")
with os.fdopen(hilog_open, "a") as hilog_file_pipe:
self.config.device.start_catch_device_log(hilog_file_pipe)
self.config.device.start_catch_device_log(hilog_file_pipe=hilog_file_pipe)
self._run_oh_jsunit(config_file, request)
except Exception as exception:
self.error_message = exception
......@@ -250,6 +261,47 @@ class OHJSUnitTestDriver(IDriver):
self.result = check_result_report(
request.config.report_path, self.result, self.error_message)
def __dry_run_execute__(self, request):
LOG.debug("Start dry run xdevice JSUnit Test")
self.config = request.config
self.config.device = request.config.environment.devices[0]
config_file = request.root.source.config_file
suite_file = request.root.source.source_file
if not suite_file:
raise ParamError(
"test source '%s' not exists" %
request.root.source.source_string, error_no="00110")
LOG.debug("Test case file path: %s" % suite_file)
self._dry_run_oh_jsunit(config_file, request)
def _dry_run_oh_jsunit(self, config_file, request):
try:
if not os.path.exists(config_file):
LOG.error("Error: Test cases don't exist %s." % config_file)
raise ParamError(
"Error: Test cases don't exist %s." % config_file,
error_no="00102")
json_config = JsonParser(config_file)
self.kits = get_kit_instances(json_config,
self.config.resource_path,
self.config.testcases_path)
self._get_driver_config(json_config)
self.config.device.connector_command("target mount")
do_module_kit_setup(request, self.kits)
self.runner = OHJSUnitTestRunner(self.config)
self.runner.suite_name = request.get_module_name()
# execute test case
self._get_runner_config(json_config)
oh_jsunit_para_parse(self.runner, self.config.testargs)
test_to_run = self._collect_test_to_run()
LOG.info("Collected test count is: %s" % (len(test_to_run)
if test_to_run else 0))
finally:
do_module_kit_teardown(request)
def _run_oh_jsunit(self, config_file, request):
try:
if not os.path.exists(config_file):
......@@ -263,7 +315,7 @@ class OHJSUnitTestDriver(IDriver):
self.config.testcases_path)
self._get_driver_config(json_config)
self.config.device.hdc_command("target mount")
self.config.device.connector_command("target mount")
do_module_kit_setup(request, self.kits)
self.runner = OHJSUnitTestRunner(self.config)
self.runner.suite_name = request.get_module_name()
......@@ -314,6 +366,7 @@ class OHJSUnitTestDriver(IDriver):
if test_to_run else 0))
if not test_to_run:
self.runner.run(listener)
self.runner.notify_finished()
else:
self._run_with_rerun(listener, test_to_run)
......@@ -340,35 +393,43 @@ class OHJSUnitTestDriver(IDriver):
expected_tests = TestDescription.remove_test(expected_tests,
test_run)
if not expected_tests:
LOG.debug("No tests to re-run, all tests executed at least "
"once.")
if self.rerun_all:
self._rerun_all(expected_tests, listener)
LOG.warning("No tests to re-run twice,please check")
self.runner.notify_finished()
else:
self._rerun_serially(expected_tests, listener)
self._rerun_twice(expected_tests, listener)
else:
LOG.debug("Rerun once success")
self.runner.notify_finished()
def _rerun_all(self, expected_tests, listener):
def _rerun_twice(self, expected_tests, listener):
tests = []
for test in expected_tests:
tests.append("%s#%s" % (test.class_name, test.test_name))
self.runner.add_arg("class", ",".join(tests))
LOG.debug("Ready to rerun all, expect run: %s" % len(expected_tests))
LOG.debug("Ready to rerun twice, expect run: %s" % len(expected_tests))
test_run = self._run_tests(listener)
LOG.debug("Rerun all, has run: %s" % len(test_run))
LOG.debug("Rerun twice, has run: %s" % len(test_run))
if len(test_run) < len(expected_tests):
expected_tests = TestDescription.remove_test(expected_tests,
test_run)
if not expected_tests:
LOG.debug("Rerun textFile success")
self._rerun_serially(expected_tests, listener)
LOG.warning("No tests to re-run third,please check")
self.runner.notify_finished()
else:
self._rerun_third(expected_tests, listener)
else:
LOG.debug("Rerun twice success")
self.runner.notify_finished()
def _rerun_serially(self, expected_tests, listener):
LOG.debug("Rerun serially, expected run: %s" % len(expected_tests))
def _rerun_third(self, expected_tests, listener):
tests = []
for test in expected_tests:
self.runner.add_arg(
"class", "%s#%s" % (test.class_name, test.test_name))
self.runner.rerun(listener, test)
self.runner.remove_arg("class")
tests.append("%s#%s" % (test.class_name, test.test_name))
self.runner.add_arg("class", ",".join(tests))
LOG.debug("Ready to rerun third, expect run: %s" % len(expected_tests))
self._run_tests(listener)
LOG.debug("Rerun third success")
self.runner.notify_finished()
def __result__(self):
return self.result if os.path.exists(self.result) else ""
......@@ -380,6 +441,10 @@ class OHJSUnitTestRunner:
self.suite_name = None
self.config = config
self.rerun_attemp = 3
self.suite_recorder = {}
self.finished = False
self.expect_tests_dict = None
self.finished_observer = None
def dry_run(self):
parsers = get_plugin(Plugin.PARSER, CommonParserType.oh_jsunit_list)
......@@ -393,10 +458,20 @@ class OHJSUnitTestRunner:
command = self._get_dry_run_command()
self.config.device.execute_shell_command(
command, timeout=self.config.timeout, receiver=handler, retry=0)
self.expect_tests_dict = parser_instances[0].tests_dict
return parser_instances[0].tests
def run(self, listener):
handler = self._get_shell_handler(listener)
command = self._get_run_command()
self.config.device.execute_shell_command(
command, timeout=self.config.timeout, receiver=handler, retry=0)
def notify_finished(self):
if self.finished_observer:
self.finished_observer.notify_task_finished()
def _get_shell_handler(self, listener):
parsers = get_plugin(Plugin.PARSER, CommonParserType.oh_jsunit)
if parsers:
parsers = parsers[:1]
......@@ -405,54 +480,11 @@ class OHJSUnitTestRunner:
parser_instance = parser.__class__()
parser_instance.suite_name = self.suite_name
parser_instance.listeners = listener
parser_instance.runner = self
parser_instances.append(parser_instance)
self.finished_observer = parser_instance
handler = ShellHandler(parser_instances)
command = self._get_run_command()
self.config.device.execute_shell_command(
command, timeout=self.config.timeout, receiver=handler, retry=0)
def rerun(self, listener, test):
handler = None
if self.rerun_attemp:
test_tracker = CollectingPassListener()
try:
listener_copy = listener.copy()
listener_copy.append(test_tracker)
parsers = get_plugin(Plugin.PARSER, CommonParserType.oh_jsunit)
if parsers:
parsers = parsers[:1]
parser_instances = []
for parser in parsers:
parser_instance = parser.__class__()
parser_instance.suite_name = self.suite_name
parser_instance.listeners = listener_copy
parser_instances.append(parser_instance)
handler = ShellHandler(parser_instances)
command = self._get_run_command()
self.config.device.execute_shell_command(
command, timeout=self.config.timeout, receiver=handler,
retry=0)
except ShellCommandUnresponsiveException as _:
LOG.debug("Exception: ShellCommandUnresponsiveException")
finally:
if not len(test_tracker.get_current_run_results()):
LOG.debug("No test case is obtained finally")
self.rerun_attemp -= 1
handler.parsers[0].mark_test_as_blocked(test)
else:
LOG.debug("Not execute and mark as blocked finally")
parsers = get_plugin(Plugin.PARSER, CommonParserType.cpptest)
if parsers:
parsers = parsers[:1]
parser_instances = []
for parser in parsers:
parser_instance = parser.__class__()
parser_instance.suite_name = self.suite_name
parser_instance.listeners = listener
parser_instances.append(parser_instance)
handler = ShellHandler(parser_instances)
handler.parsers[0].mark_test_as_blocked(test)
return handler
def add_arg(self, name, value):
if not name or not value:
......@@ -506,4 +538,3 @@ class OHJSUnitTestRunner:
self.get_args_command())
return command
extension/src/xdevice_extension/_core/__init__.py → plugins/ohos/src/ohos/environment/__init__.py
浏览文件 @ d7902098
#!/usr/bin/env python3
# coding=utf-8
#
# Copyright (c) 2021 Huawei Device Co., Ltd.
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#
#!/usr/bin/env python3
# coding=utf-8
#
# Copyright (c) 2022 Huawei Device Co., Ltd.
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#
\ No newline at end of file
src/xdevice/_core/environment/device_lite.py → plugins/ohos/src/ohos/environment/device_lite.py
浏览文件 @ d7902098
......@@ -2,7 +2,7 @@
# coding=utf-8
#
# Copyright (c) 2020-2021 Huawei Device Co., Ltd.
# Copyright (c) 2022 Huawei Device Co., Ltd.
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
......@@ -22,31 +22,31 @@ import time
import os
import threading
from _core.constants import DeviceOsType
from _core.constants import ConfigConst
from _core.constants import ComType
from _core.constants import DeviceLabelType
from _core.constants import ModeType
from _core.environment.dmlib_lite import LiteHelper
from _core.exception import LiteDeviceConnectError
from _core.exception import LiteDeviceTimeout
from _core.exception import LiteParamError
from _core.interface import IDevice
from _core.logger import platform_logger
from _core.environment.manager_env import DeviceAllocationState
from _core.plugin import Plugin
from _core.utils import exec_cmd
from _core.utils import convert_serial
from _core.utils import convert_ip
from _core.utils import check_mode
from xdevice import DeviceOsType
from xdevice import ConfigConst
from xdevice import DeviceLabelType
from xdevice import ModeType
from xdevice import IDevice
from xdevice import platform_logger
from xdevice import DeviceAllocationState
from xdevice import Plugin
from xdevice import exec_cmd
from xdevice import convert_serial
from xdevice import convert_ip
from xdevice import check_mode
from ohos.exception import LiteDeviceConnectError
from ohos.exception import LiteDeviceTimeout
from ohos.exception import LiteParamError
from ohos.environment.dmlib_lite import LiteHelper
from ohos.constants import ComType
LOG = platform_logger("DeviceLite")
TIMEOUT = 90
RETRY_ATTEMPTS = 0
HDC = "hdc_std.exe"
HDC = "litehdc.exe"
DEFAULT_BAUD_RATE = 115200
def get_hdc_path():
from xdevice import Variables
user_path = os.path.join(Variables.exec_dir, "resource/tools")
......@@ -79,7 +79,7 @@ def parse_available_com(com_str):
def perform_device_action(func):
def device_action(self, *args, **kwargs):
if not self.get_recover_state():
LOG.debug("device %s %s is false" % (self.device_sn,
LOG.debug("Device %s %s is false" % (self.device_sn,
ConfigConst.recover_state))
return "", "", ""
......@@ -158,20 +158,22 @@ class DeviceLite(IDevice):
def _check_watchgt(device):
for item in device:
if "label" not in item.keys():
if "com" not in item.keys() or ("com" in item.keys() and
not item.get("com")):
error_message = "watchGT local com cannot be " \
"empty, please check"
raise LiteParamError(error_message, error_no="00108")
error_message = "watchGT local label does not exist"
raise LiteParamError(error_message, error_no="00108")
if "com" not in item.keys() or ("com" in item.keys() and
not item.get("com")):
error_message = "watchGT local com cannot be " \
"empty, please check"
raise LiteParamError(error_message, error_no="00108")
else:
hdc = get_hdc_path()
result = exec_cmd([hdc])
com_list = parse_available_com(result)
if item.get("com").upper() in com_list:
return True
else:
hdc = get_hdc_path()
result = exec_cmd([hdc])
com_list = parse_available_com(result)
if item.get("com").upper() in com_list:
return True
else:
error_message = "watchGT local com does not exist"
raise LiteParamError(error_message, error_no="00108")
error_message = "watchGT local com does not exist"
raise LiteParamError(error_message, error_no="00108")
@staticmethod
def _check_wifiiot_config(device):
......@@ -220,26 +222,9 @@ class DeviceLite(IDevice):
" empty, please check"
raise LiteParamError(error_message, error_no="00108")
@staticmethod
def _check_agent(device=None):
for item in device:
if "label" not in item.keys():
if "port" in item.keys() and item.get("port") and not item.get(
"port").isnumeric():
error_message = "agent port should be " \
"a number, please check"
raise LiteParamError(error_message, error_no="00108")
elif "port" not in item.keys():
error_message = "ipcamera agent port cannot be" \
" empty, please check"
raise LiteParamError(error_message, error_no="00108")
def __check_config__(self, device=None):
if "agent" == device[0].get("type"):
self.device_connect_type = "agent"
self.label = device[0].get("label")
else:
self.set_connect_type(device)
self.set_connect_type(device)
if self.label == DeviceLabelType.wifiiot:
self._check_wifiiot_config(device)
elif self.label == DeviceLabelType.ipcamera and \
......@@ -250,9 +235,6 @@ class DeviceLite(IDevice):
self._check_ipcamera_remote(device)
elif self.label == DeviceLabelType.watch_gt:
self._check_watchgt(device)
elif self.label == DeviceLabelType.ipcamera and \
self.device_connect_type == "agent":
self._check_agent(device)
def set_connect_type(self, device):
pattern = r'^((25[0-5]|2[0-4]\d|[01]?\d\d?)\.){3}(25[0-5]|2[0-4]\d|[' \
......@@ -290,8 +272,6 @@ class DeviceLite(IDevice):
self.__set_serial__(device)
if self.device_connect_type == "remote":
self.device = CONTROLLER_DICT.get("remote")(device)
elif self.device_connect_type == "agent":
self.device = CONTROLLER_DICT.get("agent")(device)
else:
self.device = CONTROLLER_DICT.get("local")(device)
......@@ -299,14 +279,14 @@ class DeviceLite(IDevice):
def connect(self):
"""
connect the device
Connect the device
"""
try:
self.device.connect()
except LiteDeviceConnectError:
except LiteDeviceConnectError as _:
if check_mode(ModeType.decc):
LOG.debug("set device %s recover state to false" %
LOG.debug("Set device %s recover state to false" %
self.device_sn)
self.device_allocation_state = DeviceAllocationState.unusable
self.set_recover_state(False)
......@@ -357,9 +337,10 @@ class DeviceLite(IDevice):
if not receiver:
LOG.debug("%s execute result:%s" % (
convert_serial(self.__get_serial__()), filter_result))
if not status:
LOG.debug("%s error_message:%s" % (
convert_serial(self.__get_serial__()), error_message))
if not status:
LOG.debug(
"%s error_message:%s" % (convert_serial(self.__get_serial__()),
error_message))
return filter_result, status, error_message
def recover_device(self):
......@@ -375,10 +356,19 @@ class DeviceLite(IDevice):
self.device_sn)
self.device_allocation_state = DeviceAllocationState.unusable
self.set_recover_state(False)
if self.ifconfig:
self.execute_command_with_timeout(command=self.ifconfig,
timeout=60)
enter_result, _, _ = self.execute_command_with_timeout(command='\r',
timeout=15)
if " #" in enter_result or "OHOS #" in enter_result:
LOG.info("Reset device %s success" % self.device_sn)
self.execute_command_with_timeout(command=self.ifconfig,
timeout=5)
elif "hisilicon #" in enter_result:
LOG.info("Reset device %s fail" % self.device_sn)
ifconfig_result, _, _ = self.execute_command_with_timeout(
command="ifconfig",
timeout=5)
def close(self):
"""
......@@ -411,7 +401,7 @@ class RemoteController:
def connect(self):
"""
connect the device server
Connect the device server
"""
try:
......@@ -450,7 +440,7 @@ class RemoteController:
return
self.telnet.close()
self.telnet = None
except (ConnectionError, Exception):
except (ConnectionError, Exception) as _:
error_message = "Remote device is disconnected abnormally"
LOG.error(error_message, error_no="00401")
......@@ -542,7 +532,7 @@ class ComController:
if self.is_open:
self.com.close()
self.is_open = False
except (ConnectionError, Exception):
except (ConnectionError, Exception) as _:
error_message = "Local device is disconnected abnormally"
LOG.error(error_message, error_no="00401")
......@@ -559,49 +549,7 @@ class ComController:
LiteHelper.execute_local_command(self.com, command)
class AgentController:
def __init__(self, device):
"""
Init serial.
Parameters:
device: local com
"""
LOG.info("AgentController begin")
self.com_dict = {}
self.host = device[1].get("ip")
self.port = str(device[1].get("port"))
self.headers = {'Content-Type': 'application/json'}
self.local_source_dir = ""
self.target_save_path = ""
self.base_url = "http" + "://" + self.host + ":" + self.port
def connect(self):
"""
Open serial.
"""
def close(self):
"""
Close serial.
"""
pass
def execute_command_with_timeout(self, command, **kwargs):
"""
Executes command on the device.
Parameters:
command: the command to execute
timeout: timeout for read result
receiver: parser handler
"""
return LiteHelper.execute_agent_cmd_with_timeout(self,
command,
**kwargs)
CONTROLLER_DICT = {
"local": LocalController,
"remote": RemoteController,
"agent": AgentController
}
src/xdevice/_core/environment/dmlib_lite.py → plugins/ohos/src/ohos/environment/dmlib_lite.py
浏览文件 @ d7902098
......@@ -2,7 +2,7 @@
# coding=utf-8
#
# Copyright (c) 2020-2021 Huawei Device Co., Ltd.
# Copyright (c) 2020-2022 Huawei Device Co., Ltd.
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
......@@ -15,18 +15,17 @@
# See the License for the specific language governing permissions and
# limitations under the License.
#
import json
import time
import re
from _core.constants import DeviceTestType
from _core.exception import ExecuteTerminate
from _core.exception import LiteDeviceTimeout
from _core.exception import LiteDeviceConnectError
from _core.exception import LiteDeviceExecuteCommandError
from _core.logger import platform_logger
from _core.utils import convert_port
from xdevice import DeviceTestType
from xdevice import ExecuteTerminate
from xdevice import platform_logger
from ohos.exception import LiteDeviceTimeout
from ohos.exception import LiteDeviceConnectError
from ohos.exception import LiteDeviceExecuteCommandError
__all__ = ["generate_report", "LiteHelper"]
......@@ -46,6 +45,7 @@ INSTALL_END_MARKER = "resultMessage is install success !"
PATTERN = re.compile(r'\x1B(\[([0-9]{1,2}(;[0-9]{1,2})*)?m)*')
TIMEOUT = 90
STATUS_OK_CODE = 200
LOG = platform_logger("DmlibLite")
......@@ -80,10 +80,10 @@ def check_read_test_end(result=None, input_command=None):
return True
if (result_output.find(CPP_TEST_MOUNT_SIGN) != -1
and result_output.find(CPP_TEST_STOP_SIGN) != -1):
LOG.info("find test stop")
LOG.info("Find test stop")
return True
if "%s%s" % (CPP_ERR_MESSAGE, input_command[2:]) in result_output:
LOG.error("execute file not exist, result is %s" % result_output,
LOG.error("Execute file not exist, result is %s" % result_output,
error_no="00402")
raise LiteDeviceExecuteCommandError("execute file not exist",
error_no="00402")
......@@ -283,7 +283,7 @@ class LiteHelper:
error_no="00402")
LOG.info("local_%s execute command shell %s with timeout %ss" %
(convert_port(com.port), command, str(timeout)))
(com.port, command, str(timeout)))
if isinstance(command, str):
command = command.encode("utf-8")
......@@ -306,89 +306,8 @@ class LiteHelper:
error_no="00402")
LOG.info(
"local_%s execute command shell %s" % (convert_port(com.port),
command))
"local_%s execute command shell %s" % (com.port, command))
command = command.encode("utf-8")
if command[-2:] != b"\r\n":
command = command.rstrip() + b'\r\n'
com.write(command)
@staticmethod
def execute_agent_cmd_with_timeout(self, command,
**kwargs):
import requests
base_url = self.base_url
headers = self.headers
local_source_dir = self.local_source_dir
target_save_path = self.target_save_path
args = kwargs
command_type = args.get("type", None)
sync = args.get("sync", 1)
response = None
try:
if "cmd" == command_type:
url = base_url + "/_processes/"
data = {
'cmd': command,
'sync': sync,
'stdoutRedirect': 'file',
'stderrRedirect': 'file'
}
response = requests.post(url=url, headers=headers,
data=json.dumps(data))
if response.status_code == 200:
LOG.info("connect OK")
else:
LOG.info("connect failed")
response.close()
elif "put" == command_type:
url = base_url + target_save_path + command
with open(local_source_dir + command, "rb") as data:
response = requests.put(url=url, headers=headers,
data=data)
if response.status_code == 200:
LOG.info("{} upload file to {} "
"success".format(local_source_dir,
target_save_path))
else:
LOG.info(
"{} upload file to {} fail".format(local_source_dir,
target_save_path))
response.close()
elif "get" == command_type:
url = base_url + target_save_path + command
response = requests.get(url=url, headers=headers, stream=True)
if response.status_code == 200:
with open(local_source_dir + command + "bak", "wb") \
as file_name:
for file_data in response.iter_content(chunk_size=512):
file_name.write(file_data)
LOG.info("from {} download file to {} success".format(
target_save_path + command,
local_source_dir))
else:
LOG.info("{} download file to {} fail".format(
target_save_path + command,
command,
local_source_dir))
elif "delete" == command_type:
url = base_url + target_save_path + command
LOG.info(url)
response = requests.delete(url)
if response.status_code == 200:
LOG.info("delete {} file success".format(
target_save_path + command))
else:
LOG.info("delete {} file fail".format(
target_save_path + command))
else:
LOG.info("type error")
except Exception as error_message:
LOG.info("error_message:{}".format(error_message))
finally:
if response:
response.close()
return "", True, ""
extension/src/xdevice_extension/_core/environment/emulator.py → plugins/ohos/src/ohos/environment/emulator.py
浏览文件 @ d7902098
#!/usr/bin/env python3
# coding=utf-8
#
# Copyright (c) 2021 Huawei Device Co., Ltd.
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#
from xdevice import IDevice
from xdevice import platform_logger
from xdevice_extension._core.environment.device_state import \
DeviceAllocationState
from xdevice_extension._core.environment.device_state import TestDeviceState
LOG = platform_logger("Emulator")
class Emulator(IDevice):
"""
Class representing an emulator.
Each object of this class represents one emulator in xDevice.
Attributes:
device_sn: A string that's the serial number of the emulator.
"""
def __get_serial__(self):
pass
def __set_serial__(self, device_sn=""):
pass
def __init__(self, device_sn=""):
self.device_sn = device_sn
self.is_timeout = False
self.device_log_proc = None
self.test_device_state = TestDeviceState.ONLINE
self.device_allocation_state = DeviceAllocationState.available
def __serial__(self):
return self.device_sn
def get_device_sn(self):
"""
Returns the serial number of the device.
"""
return self.device_sn
#!/usr/bin/env python3
# coding=utf-8
#
# Copyright (c) 2022 Huawei Device Co., Ltd.
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#
from xdevice import IDevice
from xdevice import platform_logger
from xdevice import DeviceAllocationState
from xdevice import TestDeviceState
LOG = platform_logger("Emulator")
class Emulator(IDevice):
"""
Class representing an emulator.
Each object of this class represents one emulator in xDevice.
Attributes:
device_sn: A string that's the serial number of the emulator.
"""
def __get_serial__(self):
pass
def __set_serial__(self, device_sn=""):
pass
def __init__(self, device_sn=""):
self.device_sn = device_sn
self.is_timeout = False
self.device_log_proc = None
self.test_device_state = TestDeviceState.ONLINE
self.device_allocation_state = DeviceAllocationState.available
def __serial__(self):
return self.device_sn
def get_device_sn(self):
"""
Returns the serial number of the device.
"""
return self.device_sn
extension/src/xdevice_extension/_core/exception.py → plugins/ohos/src/ohos/exception.py
浏览文件 @ d7902098
#!/usr/bin/env python3
# coding=utf-8
#
# Copyright (c) 2021 Huawei Device Co., Ltd.
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#
from xdevice import DeviceError
class HdcError(DeviceError):
"""
Raised when there is an error in hdc operations.
"""
def __init__(self, error_msg, error_no=""):
super(HdcError, self).__init__(error_msg, error_no)
self.error_msg = error_msg
self.error_no = error_no
def __str__(self):
return str(self.error_msg)
class HdcCommandRejectedException(HdcError):
"""
Exception thrown when hdc refuses a command.
"""
def __init__(self, error_msg, error_no=""):
super(HdcCommandRejectedException, self).__init__(error_msg, error_no)
self.error_msg = error_msg
self.error_no = error_no
def __str__(self):
return str(self.error_msg)
class ShellCommandUnresponsiveException(HdcError):
"""
Exception thrown when a shell command executed on a device takes too long
to send its output.
"""
def __init__(self, error_msg="ShellCommandUnresponsiveException",
error_no=""):
super(ShellCommandUnresponsiveException, self).\
__init__(error_msg, error_no)
self.error_msg = error_msg
self.error_no = error_no
def __str__(self):
return str(self.error_msg)
class DeviceUnresponsiveException(HdcError):
"""
Exception thrown when a shell command executed on a device takes too long
to send its output.
"""
def __init__(self, error_msg="DeviceUnresponsiveException", error_no=""):
super(DeviceUnresponsiveException, self).__init__(error_msg, error_no)
self.error_msg = error_msg
self.error_no = error_no
def __str__(self):
return str(self.error_msg)
class AppInstallError(DeviceError):
def __init__(self, error_msg, error_no=""):
super(AppInstallError, self).__init__(error_msg, error_no)
self.error_msg = error_msg
self.error_no = error_no
def __str__(self):
return str(self.error_msg)
class HapNotSupportTest(DeviceError):
def __init__(self, error_msg, error_no=""):
super(HapNotSupportTest, self).__init__(error_msg, error_no)
self.error_msg = error_msg
self.error_no = error_no
def __str__(self):
return str(self.error_msg)
#!/usr/bin/env python3
# coding=utf-8
#
# Copyright (c) 2020-2022 Huawei Device Co., Ltd.
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#
from xdevice import DeviceError
__all__ = ["LiteDeviceConnectError", "LiteDeviceTimeout", "LiteParamError",
"LiteDeviceError", "LiteDeviceExecuteCommandError",
"LiteDeviceMountError", "LiteDeviceReadOutputError"]
class LiteDeviceError(Exception):
def __init__(self, error_msg, error_no=""):
super(LiteDeviceError, self).__init__(error_msg, error_no)
self.error_msg = error_msg
self.error_no = error_no
def __str__(self):
return str(self.error_msg)
class LiteDeviceConnectError(LiteDeviceError):
def __init__(self, error_msg, error_no=""):
super(LiteDeviceConnectError, self).__init__(error_msg, error_no)
self.error_msg = error_msg
self.error_no = error_no
def __str__(self):
return str(self.error_msg)
class LiteDeviceTimeout(LiteDeviceError):
def __init__(self, error_msg, error_no=""):
super(LiteDeviceTimeout, self).__init__(
error_msg, error_no)
self.error_msg = error_msg
self.error_no = error_no
def __str__(self):
return str(self.error_msg)
class LiteParamError(LiteDeviceError):
def __init__(self, error_msg, error_no=""):
super(LiteParamError, self).__init__(error_msg, error_no)
self.error_msg = error_msg
self.error_no = error_no
def __str__(self):
return str(self.error_msg)
class LiteDeviceExecuteCommandError(LiteDeviceError):
def __init__(self, error_msg, error_no=""):
super(LiteDeviceExecuteCommandError, self).__init__(
error_msg, error_no)
self.error_msg = error_msg
self.error_no = error_no
def __str__(self):
return str(self.error_msg)
class LiteDeviceMountError(LiteDeviceError):
def __init__(self, error_msg, error_no=""):
super(LiteDeviceMountError, self).__init__(error_msg, error_no)
self.error_msg = error_msg
self.error_no = error_no
def __str__(self):
return str(self.error_msg)
class LiteDeviceReadOutputError(LiteDeviceError):
def __init__(self, error_msg, error_no=""):
super(LiteDeviceReadOutputError, self).__init__(error_msg, error_no)
self.error_msg = error_msg
self.error_no = error_no
def __str__(self):
return str(self.error_msg)
extension/src/xdevice_extension/_core/environment/__init__.py → plugins/ohos/src/ohos/executor/__init__.py
浏览文件 @ d7902098
#!/usr/bin/env python3
# coding=utf-8
#
# Copyright (c) 2021 Huawei Device Co., Ltd.
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#
#!/usr/bin/env python3
# coding=utf-8
#
# Copyright (c) 2022 Huawei Device Co., Ltd.
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#
\ No newline at end of file
extension/src/xdevice_extension/_core/executor/listener.py → plugins/ohos/src/ohos/executor/listener.py
浏览文件 @ d7902098
#!/usr/bin/env python3
# coding=utf-8
#
# Copyright (c) 2021 Huawei Device Co., Ltd.
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#
from xdevice import Plugin
from xdevice import LifeCycle
from xdevice import IListener
from xdevice import platform_logger
from xdevice import TestDescription
from xdevice import ResultCode
from xdevice_extension._core.constants import ListenerType
__all__ = ["CollectingTestListener", "CollectingLiteGTestListener",
"CollectingPassListener"]
LOG = platform_logger("Listener")
@Plugin(type=Plugin.LISTENER, id=ListenerType.collect)
class CollectingTestListener(IListener):
"""
listener test status information to the console
"""
def __init__(self):
self.tests = []
def __started__(self, lifecycle, test_result):
if lifecycle == LifeCycle.TestCase:
if not test_result.test_class or not test_result.test_name:
return
test = TestDescription(test_result.test_class,
test_result.test_name)
if test not in self.tests:
self.tests.append(test)
def __ended__(self, lifecycle, test_result=None, **kwargs):
pass
def __skipped__(self, lifecycle, test_result):
pass
def __failed__(self, lifecycle, test_result):
pass
def get_current_run_results(self):
return self.tests
@Plugin(type=Plugin.LISTENER, id=ListenerType.collect_lite)
class CollectingLiteGTestListener(IListener):
"""
listener test status information to the console
"""
def __init__(self):
self.tests = []
def __started__(self, lifecycle, test_result):
pass
def __ended__(self, lifecycle, test_result=None, **kwargs):
del kwargs
if lifecycle == LifeCycle.TestCase:
if not test_result.test_class or not test_result.test_name:
return
test = TestDescription(test_result.test_class,
test_result.test_name)
if test not in self.tests:
self.tests.append(test)
def __skipped__(self, lifecycle, test_result):
pass
def __failed__(self, lifecycle, test_result):
if lifecycle == LifeCycle.TestCase:
if not test_result.test_class or not test_result.test_name:
return
test = TestDescription(test_result.test_class,
test_result.test_name)
if test not in self.tests:
self.tests.append(test)
def get_current_run_results(self):
return self.tests
@Plugin(type=Plugin.LISTENER, id=ListenerType.collect_pass)
class CollectingPassListener(IListener):
"""
listener test status information to the console
"""
def __init__(self):
self.tests = []
def __started__(self, lifecycle, test_result):
pass
def __ended__(self, lifecycle, test_result=None, **kwargs):
if lifecycle == LifeCycle.TestCase:
if not test_result.test_class or not test_result.test_name:
return
if test_result.code != ResultCode.PASSED.value:
return
test = TestDescription(test_result.test_class,
test_result.test_name)
if test not in self.tests:
self.tests.append(test)
else:
LOG.warning("Duplicate testcase: %s#%s" % (
test_result.test_class, test_result.test_name))
def __skipped__(self, lifecycle, test_result):
pass
def __failed__(self, lifecycle, test_result):
pass
def get_current_run_results(self):
return self.tests
#!/usr/bin/env python3
# coding=utf-8
#
# Copyright (c) 2020-2022 Huawei Device Co., Ltd.
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#
from xdevice import Plugin
from xdevice import IListener
from xdevice import LifeCycle
from xdevice import platform_logger
from xdevice import ListenerType
from xdevice import TestDescription
from xdevice import ResultCode
__all__ = ["CollectingLiteGTestListener", "CollectingPassListener"]
LOG = platform_logger("Listener")
@Plugin(type=Plugin.LISTENER, id=ListenerType.collect_lite)
class CollectingLiteGTestListener(IListener):
"""
Listener test status information to the console
"""
def __init__(self):
self.tests = []
def __started__(self, lifecycle, test_result):
if lifecycle == LifeCycle.TestCase:
if not test_result.test_class or not test_result.test_name:
return
test = TestDescription(test_result.test_class,
test_result.test_name)
if test not in self.tests:
self.tests.append(test)
def __ended__(self, lifecycle, test_result=None, **kwargs):
pass
def __skipped__(self, lifecycle, test_result):
pass
def __failed__(self, lifecycle, test_result):
if lifecycle == LifeCycle.TestCase:
if not test_result.test_class or not test_result.test_name:
return
test = TestDescription(test_result.test_class,
test_result.test_name)
if test not in self.tests:
self.tests.append(test)
def get_current_run_results(self):
return self.tests
@Plugin(type=Plugin.LISTENER, id=ListenerType.collect_pass)
class CollectingPassListener(IListener):
"""
listener test status information to the console
"""
def __init__(self):
self.tests = []
def __started__(self, lifecycle, test_result):
pass
def __ended__(self, lifecycle, test_result=None, **kwargs):
if lifecycle == LifeCycle.TestCase:
if not test_result.test_class or not test_result.test_name:
return
if test_result.code != ResultCode.PASSED.value:
return
test = TestDescription(test_result.test_class,
test_result.test_name)
if test not in self.tests:
self.tests.append(test)
else:
LOG.warning("Duplicate testcase: %s#%s" % (
test_result.test_class, test_result.test_name))
def __skipped__(self, lifecycle, test_result):
pass
def __failed__(self, lifecycle, test_result):
pass
def get_current_run_results(self):
return self.tests
\ No newline at end of file
plugins/ohos/src/ohos/managers/__init__.py 0 → 100644
浏览文件 @ d7902098
#!/usr/bin/env python3
# coding=utf-8
#
# Copyright (c) 2022 Huawei Device Co., Ltd.
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#
\ No newline at end of file
src/xdevice/_core/environment/manager_lite.py → plugins/ohos/src/ohos/managers/manager_lite.py
浏览文件 @ d7902098
......@@ -2,7 +2,7 @@
# coding=utf-8
#
# Copyright (c) 2020-2021 Huawei Device Co., Ltd.
# Copyright (c) 2020-2022 Huawei Device Co., Ltd.
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
......@@ -19,17 +19,19 @@
import time
import threading
from _core.config.config_manager import UserConfigManager
from _core.constants import DeviceOsType
from _core.constants import ManagerType
from _core.environment.manager_env import DeviceAllocationState
from _core.exception import LiteDeviceError
from _core.plugin import Plugin
from _core.plugin import get_plugin
from _core.interface import IDeviceManager
from _core.logger import platform_logger
from _core.utils import convert_ip
from _core.utils import convert_port
from xdevice import UserConfigManager
from xdevice import DeviceOsType
from xdevice import ManagerType
from xdevice import DeviceAllocationState
from xdevice import Plugin
from xdevice import get_plugin
from xdevice import IDeviceManager
from xdevice import platform_logger
from xdevice import convert_ip
from xdevice import convert_port
from xdevice import convert_serial
from ohos.exception import LiteDeviceError
__all__ = ["ManagerLite"]
......@@ -39,7 +41,7 @@ LOG = platform_logger("ManagerLite")
@Plugin(type=Plugin.MANAGER, id=ManagerType.lite_device)
class ManagerLite(IDeviceManager):
"""
Class representing device manager
Class representing device manager that
managing the set of available devices for testing
"""
......@@ -47,6 +49,7 @@ class ManagerLite(IDeviceManager):
self.devices_list = []
self.list_con = threading.Condition()
self.support_labels = ["ipcamera", "wifiiot", "watchGT"]
self.support_types = ["device"]
def init_environment(self, environment="", user_config_file=""):
device_lite = get_plugin(plugin_type=Plugin.DEVICE,
......@@ -76,7 +79,7 @@ class ManagerLite(IDeviceManager):
Request a device for testing that meets certain criteria.
"""
del timeout
LOG.debug("lite apply_device: apply lock")
LOG.debug("Lite apply device: apply lock")
self.list_con.acquire()
try:
allocated_device = None
......@@ -84,31 +87,32 @@ class ManagerLite(IDeviceManager):
if device_option.matches(device):
device.device_allocation_state = \
DeviceAllocationState.allocated
LOG.debug("allocate device sn: %s, type: %s" % (
device.__get_serial__(), device.__class__))
LOG.debug("Allocate device sn: %s, type: %s" % (
convert_serial(device.__get_serial__()),
device.__class__))
return device
time.sleep(10)
return allocated_device
finally:
LOG.debug("lite apply_device: release lock")
LOG.debug("Lite apply device: release lock")
self.list_con.release()
def release_device(self, device):
LOG.debug("lite release_device: apply lock")
LOG.debug("Lite release device: apply lock")
self.list_con.acquire()
try:
if device.device_allocation_state == \
DeviceAllocationState.allocated:
device.device_allocation_state = \
DeviceAllocationState.available
LOG.debug("free device sn: %s, type: %s" % (
LOG.debug("Free device sn: %s, type: %s" % (
device.__get_serial__(), device.__class__))
finally:
LOG.debug("lite release_device: release lock")
LOG.debug("Lite release device: release lock")
self.list_con.release()
def list_devices(self):
print("lite devices:")
print("Lite devices:")
print("{0:<20}{1:<16}{2:<16}{3:<16}{4:<16}{5:<16}{6:<16}".
format("SerialPort/IP", "Baudrate/Port", "OsType", "Allocation",
"Product", "ConnectType", "ComType"))
......
plugins/ohos/src/ohos/parser/__init__.py 0 → 100644
浏览文件 @ d7902098
#!/usr/bin/env python3
# coding=utf-8
#
# Copyright (c) 2022 Huawei Device Co., Ltd.
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#
\ No newline at end of file
plugins/ohos/src/ohos/parser/parser_lite.py 0 → 100644
浏览文件 @ d7902098
此差异已折叠。
plugins/ohos/src/ohos/testkit/__init__.py 0 → 100644
浏览文件 @ d7902098
#!/usr/bin/env python3
# coding=utf-8
#
# Copyright (c) 2022 Huawei Device Co., Ltd.
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#
\ No newline at end of file
src/xdevice/_core/testkit/kit_lite.py → plugins/ohos/src/ohos/testkit/kit_lite.py
浏览文件 @ d7902098
......@@ -2,7 +2,7 @@
# coding=utf-8
#
# Copyright (c) 2020-2021 Huawei Device Co., Ltd.
# Copyright (c) 2022 Huawei Device Co., Ltd.
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
......@@ -26,32 +26,31 @@ import platform
import glob
import time
from _core.logger import platform_logger
from _core.plugin import Plugin
from _core.config.config_manager import UserConfigManager
from _core.constants import CKit
from _core.constants import ConfigConst
from _core.constants import ComType
from _core.constants import DeviceLiteKernel
from _core.constants import DeviceTestType
from _core.exception import LiteDeviceMountError
from _core.exception import ParamError
from _core.exception import LiteDeviceError
from _core.interface import ITestKit
from _core.utils import get_config_value
from _core.utils import get_file_absolute_path
from _core.utils import get_local_ip
from _core.utils import get_test_component_version
from _core.exception import LiteDeviceConnectError
from _core.constants import DeviceLabelType
from _core.constants import FilePermission
from _core.environment.manager_env import DeviceAllocationState
from xdevice import Plugin
from xdevice import platform_logger
from xdevice import DeviceAllocationState
from xdevice import ParamError
from xdevice import ITestKit
from xdevice import get_config_value
from xdevice import get_file_absolute_path
from xdevice import UserConfigManager
from xdevice import ConfigConst
from xdevice import get_test_component_version
from xdevice import get_local_ip
from xdevice import FilePermission
from xdevice import DeviceTestType
from xdevice import DeviceLabelType
from ohos.exception import LiteDeviceConnectError
from ohos.exception import LiteDeviceError
from ohos.exception import LiteDeviceMountError
from ohos.constants import ComType
from ohos.constants import CKit
from ohos.constants import DeviceLiteKernel
__all__ = ["DeployKit", "MountKit", "RootFsKit", "QueryKit", "LiteShellKit",
"LiteAppInstallKit", "DeployToolKit"]
LOG = platform_logger("KitLite")
RESET_CMD = "0xEF, 0xBE, 0xAD, 0xDE, 0x0C, 0x00, 0x87, 0x78, 0x00, 0x00, " \
"0x61, 0x94"
......@@ -120,7 +119,7 @@ class DeployKit(ITestKit):
LOG.info('The burn tool is running, please wait..')
return_code, out = subprocess.getstatusoutput(cmd)
LOG.info(
'Deploy kit to execute burn tool finished with return_code: {} '
'Deploy kit to execute burn tool finished with return code: {} '
'output: {}'.format(return_code, out))
os.remove(new_temp_tool_path)
if 0 != return_code:
......@@ -256,13 +255,13 @@ class MountKit(ITestKit):
else:
result, status, _ = device.execute_command_with_timeout(
command=command, case_type=case_type, timeout=timeout)
LOG.info('prepare environment success')
LOG.info('Prepare environment success')
def __setup__(self, device, **kwargs):
"""
Mount the file to the board by the nfs server.
"""
LOG.debug("start mount kit setup")
LOG.debug("Start mount kit setup")
request = kwargs.get("request", None)
if not request:
......@@ -291,7 +290,7 @@ class MountKit(ITestKit):
for mount_file in self.mount_list:
source = mount_file.get("source")
if not source:
raise TypeError("The source of MountKit can not be empty "
raise TypeError("The source of MountKit cant be empty "
"in Test.json!")
source = source.replace("$testcases/", "").\
replace("$resources/", "")
......@@ -321,11 +320,11 @@ class MountKit(ITestKit):
if (str(get_local_ip()) == linux_host) and (
linux_directory == ("/data%s" % testcases_dir)):
return
remote_ip = remote_info.get("ip", "")
ip = remote_info.get("ip", "")
port = remote_info.get("port", "")
remote_dir = remote_info.get("dir", "")
if not remote_ip or not port or not remote_dir:
LOG.warning("nfs server's ip or port or dir is empty")
if not ip or not port or not remote_dir:
LOG.warning("Nfs server's ip or port or dir is empty")
return
for _file in file_local_paths:
# remote copy
......@@ -334,7 +333,7 @@ class MountKit(ITestKit):
if not is_remote.lower() == "false":
try:
import paramiko
client = paramiko.Transport(remote_ip, int(port))
client = paramiko.Transport(ip, int(port))
client.connect(username=remote_info.get("username"),
password=remote_info.get("password"))
sftp = paramiko.SFTPClient.from_transport(client)
......@@ -361,9 +360,9 @@ class MountKit(ITestKit):
"Trying to copy the file from {} to nfs "
"server {} times".format(_file, count))
if count == 3:
msg = "copy {} to nfs server " \
msg = "Copy {} to nfs server " \
"failed {} times".format(
os.path.basename(_file), count)
os.path.basename(_file), count)
LOG.error(msg, error_no="00403")
LOG.debug("Nfs server:{}".format(glob.glob(
os.path.join(remote_info.get("dir"), '*.*'))))
......@@ -392,11 +391,12 @@ class MountKit(ITestKit):
command="umount {}".format(mounted_dir),
timeout=2)
if result.find("Resource busy") == -1:
device.execute_command_with_timeout(
command="rm -r {}".format(mounted_dir), timeout=1)
device.execute_command_with_timeout(command="rm -r {}".
format(mounted_dir)
, timeout=1)
if status:
break
LOG.info("umount failed,try "
LOG.info("Umount failed,try "
"again {} time".format(mount_time))
time.sleep(1)
......@@ -419,7 +419,7 @@ def copy_file_as_temp(original_file, str_length):
def mkdir_on_board(device, dir_path):
"""
liteos L1 board dont support mkdir -p
Liteos L1 board dont support mkdir -p
Parameters:
device : the L1 board
dir_path: the dir path to make
......@@ -443,7 +443,7 @@ def mkdir_on_board(device, dir_path):
def get_mount_dir(mount_dir):
"""
use windows path to mount directly when the system is windows
Use windows path to mount directly when the system is windows
Parameters:
mount_dir : the dir to mount that config in user_config.xml
such as: the mount_dir is: D:\mount\root
......@@ -485,7 +485,7 @@ class RootFsKit(ITestKit):
if not self.checksum_command or not self.hash_file_name or \
not self.device_label:
msg = "The config for rootfs kit is invalid : checksum :{}" \
" hash_file_name:{} device_label:{}" \
" hash file name:{} device label:{}" \
.format(self.checksum_command, self.hash_file_name,
self.device_label)
LOG.error(msg, error_no="00108")
......@@ -496,7 +496,7 @@ class RootFsKit(ITestKit):
# check device label
if not device.label == self.device_label:
LOG.error("device label is not match '%s '" % "demo_label",
LOG.error("Device label is not match '%s '" % "demo label",
error_no="00108")
return False
else:
......@@ -570,7 +570,7 @@ class QueryKit(ITestKit):
raise TypeError(msg)
def __setup__(self, device, **kwargs):
LOG.debug("start query kit setup")
LOG.debug("Start query kit setup")
if device.label != DeviceLabelType.ipcamera:
return
request = kwargs.get("request", None)
......@@ -616,7 +616,7 @@ class LiteShellKit(ITestKit):
del kwargs
LOG.debug("LiteShellKit setup, device:{}".format(device.device_sn))
if len(self.command_list) == 0:
LOG.info("No setup_command to run, skipping!")
LOG.info("No setup command to run, skipping!")
return
for command in self.command_list:
run_command(device, command)
......@@ -624,7 +624,7 @@ class LiteShellKit(ITestKit):
def __teardown__(self, device):
LOG.debug("LiteShellKit teardown: device:{}".format(device.device_sn))
if len(self.tear_down_command) == 0:
LOG.info("No teardown_command to run, skipping!")
LOG.info("No teardown command to run, skipping!")
return
for command in self.tear_down_command:
run_command(device, command)
......@@ -700,23 +700,29 @@ class DeployToolKit(ITestKit):
def __init__(self):
self.config = None
self.auto_deploy = None
self.device_label = None
self.time_out = None
def __check_config__(self, config):
self.config = config
self.auto_deploy = get_config_value('auto_deploy', config, False)
self.time_out = get_config_value('timeout', config, False)
if self.auto_deploy or not self.time_out:
msg = "The config for deploytool kit is invalid " \
"with auto_deploy:{},timeout:{}".format(self.auto_deploy,
self.time_out)
self.auto_deploy = get_config_value('auto_deploy',
config, is_list=False)
self.device_label = get_config_value("device_label", config,
is_list=False)
self.time_out = get_config_value("time_out", config,
is_list=False)
if not self.auto_deploy or not self.device_label or not self.time_out:
msg = "The config for deploy tool kit is" \
"invalid: device label :{} time out:{}".format(
self.device_label, self.time_out)
LOG.error(msg, error_no="00108")
raise ParamError(msg, error_no="00108")
return TypeError(msg)
def __setup__(self, device, **kwargs):
args = kwargs
request = args.get("request", None)
request.confing.deploy_tool_kit = self.config
request.config.deploy_tool_kit = self.config
def __teardown__(self, device):
pass
run.bat
浏览文件 @ d7902098
......@@ -33,6 +33,7 @@ python -c "import sys; exit(1) if sys.version_info.major < 3 or sys.version_info
python -c "import pip"
@if errorlevel 1 (
@echo "Please install pip first!"
pause
goto:eof
)
......@@ -41,6 +42,10 @@ if not exist %TOOLS% (
goto:eof
)
python -m pip uninstall -y xdevice
python -m pip uninstall -y xdevice-extension
python -m pip uninstall -y xdevice-ohos
for %%a in (%TOOLS%/*.egg) do (
python -m pip install --user %TOOLS%/%%a
@if errorlevel 1 (
......
run.sh
浏览文件 @ d7902098
此差异已折叠。
setup.py
浏览文件 @ d7902098
此差异已折叠。
src/xdevice/__init__.py
浏览文件 @ d7902098
此差异已折叠。
src/xdevice/__main__.py
浏览文件 @ d7902098
此差异已折叠。
src/xdevice/_core/__init__.py
浏览文件 @ d7902098
此差异已折叠。
src/xdevice/_core/common.py
浏览文件 @ d7902098
此差异已折叠。
src/xdevice/_core/constants.py
浏览文件 @ d7902098
此差异已折叠。
src/xdevice/_core/driver/device_test.py 已删除 100644 → 0
浏览文件 @ eb9ce2fd
此差异已折叠。
src/xdevice/_core/exception.py
浏览文件 @ d7902098
此差异已折叠。
src/xdevice/_core/interface.py
浏览文件 @ d7902098
此差异已折叠。
src/xdevice/_core/logger.py
浏览文件 @ d7902098
此差异已折叠。
src/xdevice/_core/plugin.py
浏览文件 @ d7902098
此差异已折叠。
src/xdevice/_core/report/encrypt.py
浏览文件 @ d7902098
此差异已折叠。
src/xdevice/_core/resource/config/user_config.xml
浏览文件 @ d7902098
<?xml version="1.0" encoding="UTF-8"?>
<!-- Copyright (c) 2020 Huawei Device Co., Ltd.
<!-- Copyright (c) 2022 Huawei Device Co., Ltd.
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
......
src/xdevice/_core/resource/version.txt 0 → 100644
浏览文件 @ d7902098
此差异已折叠。
src/xdevice/_core/testkit/kit.py 0 → 100644
浏览文件 @ d7902098
此差异已折叠。
src/xdevice/_core/utils.py
浏览文件 @ d7902098
此差异已折叠。
src/xdevice/variables.py
浏览文件 @ d7902098
此差异已折叠。
Markdown is supported
0% 或 .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑！
想要评论请 注册 或