• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1#!/usr/bin/env python3
2# -*- coding: utf-8 -*-
3
4# Copyright (c) 2024 Huawei Device Co., Ltd.
5# Licensed under the Apache License, Version 2.0 (the "License");
6# you may not use this file except in compliance with the License.
7# You may obtain a copy of the License at
8#
9# http://www.apache.org/licenses/LICENSE-2.0
10#
11# Unless required by applicable law or agreed to in writing, software
12# distributed under the License is distributed on an "AS IS" BASIS,
13# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14# See the License for the specific language governing permissions and
15# limitations under the License.
16#
17
18from __future__ import annotations
19import logging
20from typing import List, Optional, Type  # noqa
21from abc import ABC, abstractmethod
22from pathlib import Path
23from vmb.tool import ToolBase, OptFlags
24from vmb.target import Target
25from vmb.unit import BenchUnit, UNIT_PREFIX
26from vmb.helpers import get_plugins, get_plugin, die
27from vmb.cli import Args
28from vmb.shell import ShellUnix, ShellAdb, ShellHdc, ShellDevice
29from vmb.result import ExtInfo
30from vmb.x_shell import CrossShell
31from vmb.gensettings import GenSettings
32
33
34log = logging.getLogger('vmb')
35
36
37# pylint: disable-next=too-many-public-methods
38class PlatformBase(CrossShell, ABC):
39    """Platform Base."""
40
41    def __init__(self, args: Args) -> None:
42        self.__sh = ShellUnix(args.timeout)
43        self.__andb = None
44        self.__hdc = None
45        self.__dev_dir = Path(args.get('device_dir', 'unknown'))
46        self.args_langs = args.get('langs', set())
47        self.ext_info: ExtInfo = {}
48        if self.target == Target.DEVICE:
49            self.__andb = ShellAdb(dev_serial=args.device,
50                                   timeout=args.timeout,
51                                   tmp_dir=args.device_dir)
52        if self.target == Target.OHOS:
53            self.__hdc = ShellHdc(dev_serial=args.device,
54                                  timeout=args.timeout,
55                                  tmp_dir=args.device_dir)
56        ToolBase.sh_ = self.sh
57        ToolBase.andb_ = self.andb
58        ToolBase.hdc_ = self.hdc
59        ToolBase.dev_dir = self.dev_dir
60        # dir with shared libs (init before the suite)
61        ToolBase.libs = Path(args.get_shared_path()).joinpath('libs').resolve()
62        ToolBase.libs.mkdir(parents=True, exist_ok=True)
63        # init all the tools
64        tool_plugins = get_plugins(
65            'tools',
66            self.required_tools,
67            extra=args.extra_plugins)
68        self.flags = args.get_opts_flags()
69        self.tools = {}
70        for n, t in tool_plugins.items():
71            tool: ToolBase = t.Tool(self.target,
72                                    self.flags,
73                                    args.get_custom_opts(n))
74            self.tools[n] = tool
75            log.info('%s %s', tool.name, tool.version)
76
77    @property
78    @abstractmethod
79    def name(self) -> str:
80        """Name of platform."""
81        return ''
82
83    @property
84    @abstractmethod
85    def required_tools(self) -> List[str]:
86        """Expose tools required by platform."""
87        return []
88
89    @property
90    @abstractmethod
91    def langs(self) -> List[str]:
92        """Each platform should declare which templates it could use."""
93        return []
94
95    @property
96    def template(self) -> Optional[GenSettings]:
97        """Override generation settings. Default is use of `lang`."""
98        return None
99
100    @property
101    def required_hooks(self) -> List[str]:
102        """List of hooks requred by platform."""
103        return []
104
105    @property
106    def dev_dir(self) -> Path:
107        """Remote working dirrectory."""
108        return self.__dev_dir
109
110    @property
111    def sh(self) -> ShellUnix:
112        """Posix shell."""
113        return self.__sh
114
115    @property
116    def andb(self) -> ShellDevice:
117        """HOS remote shell."""
118        if self.__andb is not None:
119            return self.__andb
120        # fake object for host platforms
121        return ShellAdb.__new__(ShellAdb)
122
123    @property
124    def hdc(self) -> ShellDevice:
125        """OHOS remote shell."""
126        if self.__hdc is not None:
127            return self.__hdc
128        # fake object for host platforms
129        return ShellHdc.__new__(ShellHdc)
130
131    @property
132    @abstractmethod
133    def target(self) -> Target:
134        """Default target is host."""
135        return Target.HOST
136
137    @property
138    def gc_parcer(self) -> Optional[Type]:
139        """GC parcer class."""
140        return None
141
142    @staticmethod
143    def search_units(paths: List[Path]) -> List[BenchUnit]:
144        """Find bench units."""
145        bus = []
146        for bu_path in paths:
147            if not bu_path.exists():
148                log.warning('Requested unexisting path: %s', bu_path)
149                continue
150            # if file name provided add it unconditionally
151            if bu_path.is_file():
152                bus.append(BenchUnit(Path(bu_path).parent))
153                continue
154            # in case of dir search by file extention
155            for p in bu_path.glob(f'**/{UNIT_PREFIX}*'):
156                if p.is_dir():
157                    bus.append(BenchUnit(p.resolve()))
158        return bus
159
160    @classmethod
161    def create(cls, args: Args) -> PlatformBase:
162        try:
163            platform_module = get_plugin(
164                'platforms',
165                args.platform,
166                extra=args.extra_plugins)
167            platform = platform_module.Platform(args)
168        except Exception as e:  # pylint: disable=broad-exception-caught
169            die(True, 'Plugin load error: %s', e)
170        return platform
171
172    @abstractmethod
173    def run_unit(self, bu: BenchUnit) -> None:
174        pass
175
176    def cleanup(self, bu: BenchUnit) -> None:
177        """Do default cleanup."""
178        for binary in bu.binaries:
179            log.trace('Deleting: %s', binary)
180            binary.unlink(missing_ok=True)
181        if Target.HOST != self.target:
182            self.device_cleanup(bu)
183
184    def push_unit(self, bu: BenchUnit, *ext) -> None:
185        """Push bench unit to device."""
186        if Target.HOST == self.target:
187            return
188        bu.device_path = self.dev_dir.joinpath(bu.path.name)
189        self.x_sh.run(f'mkdir -p {bu.device_path}')
190        for f in bu.path.glob('*'):
191            # skip if suffix filter provided
192            if ext and f.suffix not in ext:
193                continue
194            p = f.resolve()
195            self.x_sh.push(p, bu.device_path.joinpath(f.name))
196        resources = bu.path.joinpath('resources')
197        if resources.exists():
198            device_resources = bu.device_path.joinpath('resources')
199            self.x_sh.run(f'mkdir -p {device_resources}')
200            for f in resources.glob('*'):
201                p = f.resolve()
202                self.x_sh.push(p, device_resources.joinpath(f.name))
203
204    def push_libs(self) -> None:
205        if Target.HOST == self.target:
206            return
207        self.x_sh.push(ToolBase.libs, self.dev_dir)
208
209    def device_cleanup(self, bu: BenchUnit) -> None:
210        if bu.device_path is None:
211            return  # Bench Unit wasn't pused to device
212        log.trace('Cleaning: %s', bu.device_path)
213        self.x_sh.run(f'rm -rf {bu.device_path}')
214
215    def set_affinity(self, arg: str) -> None:
216        self.x_sh.set_affinity(arg)
217
218    def dry_run_stop(self, bu: BenchUnit) -> bool:
219        if OptFlags.DRY_RUN in self.flags:
220            for binary in bu.binaries:
221                log.info('Path to binary: %s', binary)
222            return True
223        return False
224
225    def tools_get(self, name: str) -> ToolBase:
226        tool = self.tools.get(name)
227        if not tool:
228            raise RuntimeError(f'Tool {name} is not available')
229        return tool
230