• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1#!/usr/bin/env python3
2# -*- coding: utf-8 -*-
3
4# Copyright (c) 2024-2025 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                                  dev_host=args.device_host,
55                                  timeout=args.timeout,
56                                  tmp_dir=args.device_dir)
57        ToolBase.sh_ = self.sh
58        ToolBase.andb_ = self.andb
59        ToolBase.hdc_ = self.hdc
60        ToolBase.dev_dir = self.dev_dir
61        # dir with shared libs (init before the suite)
62        ToolBase.libs = Path(args.get_shared_path()).joinpath('libs').resolve()
63        ToolBase.libs.mkdir(parents=True, exist_ok=True)
64        # init all the tools
65        tool_plugins = get_plugins(
66            'tools',
67            self.required_tools,
68            extra=args.extra_plugins)
69        self.flags = args.get_opts_flags()
70        self.tools = {}
71        for n, t in tool_plugins.items():
72            tool: ToolBase = t.Tool(self.target,
73                                    self.flags,
74                                    args.custom_opts.get(n, []))
75            self.tools[n] = tool
76            log.info('%s %s', tool.name, tool.version)
77
78    @property
79    @abstractmethod
80    def name(self) -> str:
81        """Name of platform."""
82        return ''
83
84    @property
85    @abstractmethod
86    def required_tools(self) -> List[str]:
87        """Expose tools required by platform."""
88        return []
89
90    @property
91    @abstractmethod
92    def langs(self) -> List[str]:
93        """Each platform should declare which templates it could use."""
94        return []
95
96    @property
97    def template(self) -> Optional[GenSettings]:
98        """Override generation settings. Default is use of `lang`."""
99        return None
100
101    @property
102    def required_hooks(self) -> List[str]:
103        """List of hooks requred by platform."""
104        return []
105
106    @property
107    def dev_dir(self) -> Path:
108        """Remote working dirrectory."""
109        return self.__dev_dir
110
111    @property
112    def sh(self) -> ShellUnix:
113        """Posix shell."""
114        return self.__sh
115
116    @property
117    def andb(self) -> ShellDevice:
118        """HOS remote shell."""
119        if self.__andb is not None:
120            return self.__andb
121        # fake object for host platforms
122        return ShellAdb.__new__(ShellAdb)
123
124    @property
125    def hdc(self) -> ShellDevice:
126        """OHOS remote shell."""
127        if self.__hdc is not None:
128            return self.__hdc
129        # fake object for host platforms
130        return ShellHdc.__new__(ShellHdc)
131
132    @property
133    @abstractmethod
134    def target(self) -> Target:
135        """Default target is host."""
136        return Target.HOST
137
138    @property
139    def gc_parcer(self) -> Optional[Type]:
140        """GC parcer class."""
141        return None
142
143    @staticmethod
144    def search_units(paths: List[Path]) -> List[BenchUnit]:
145        """Find bench units."""
146        bus = []
147        for bu_path in paths:
148            if not bu_path.exists():
149                log.warning('Requested unexisting path: %s', bu_path)
150                continue
151            # if file name provided add it unconditionally
152            if bu_path.is_file():
153                bus.append(BenchUnit(Path(bu_path).parent))
154                continue
155            # in case of dir search by file extention
156            for p in bu_path.glob(f'**/{UNIT_PREFIX}*'):
157                if p.is_dir():
158                    bus.append(BenchUnit(p.resolve()))
159        return bus
160
161    @classmethod
162    def create(cls, args: Args) -> PlatformBase:
163        try:
164            platform_module = get_plugin(
165                'platforms',
166                args.platform,
167                extra=args.extra_plugins)
168            platform = platform_module.Platform(args)
169        except Exception as e:  # pylint: disable=broad-exception-caught
170            die(True, 'Plugin load error: %s', e)
171        return platform
172
173    @abstractmethod
174    def run_unit(self, bu: BenchUnit) -> None:
175        pass
176
177    def cleanup(self, bu: BenchUnit) -> None:
178        """Do default cleanup."""
179        for binary in bu.binaries:
180            log.trace('Deleting: %s', binary)
181            binary.unlink(missing_ok=True)
182        if Target.HOST != self.target:
183            self.device_cleanup(bu)
184
185    def push_unit(self, bu: BenchUnit, *ext) -> None:
186        """Push bench unit to device."""
187        if Target.HOST == self.target:
188            return
189        bu.device_path = self.dev_dir.joinpath(bu.path.name)
190        self.x_sh.run(f'mkdir -p {bu.device_path}')
191        for f in bu.path.glob('*'):
192            # skip if suffix filter provided
193            if ext and f.suffix not in ext:
194                continue
195            p = f.resolve()
196            self.x_sh.push(p, bu.device_path.joinpath(f.name))
197        resources = bu.path.joinpath('resources')
198        if resources.exists():
199            device_resources = bu.device_path.joinpath('resources')
200            self.x_sh.run(f'mkdir -p {device_resources}')
201            for f in resources.glob('*'):
202                p = f.resolve()
203                self.x_sh.push(p, device_resources.joinpath(f.name))
204
205    def push_libs(self) -> None:
206        if Target.HOST == self.target:
207            return
208        self.x_sh.push(ToolBase.libs, self.dev_dir)
209
210    def device_cleanup(self, bu: BenchUnit) -> None:
211        if bu.device_path is None:
212            return  # Bench Unit wasn't pused to device
213        log.trace('Cleaning: %s', bu.device_path)
214        self.x_sh.run(f'rm -rf {bu.device_path}')
215
216    def set_affinity(self, arg: str) -> None:
217        self.x_sh.set_affinity(arg)
218
219    def dry_run_stop(self, bu: BenchUnit) -> bool:
220        if OptFlags.DRY_RUN in self.flags:
221            for binary in bu.binaries:
222                log.info('Path to binary: %s', binary)
223            return True
224        return False
225
226    def tools_get(self, name: str) -> ToolBase:
227        tool = self.tools.get(name)
228        if not tool:
229            raise RuntimeError(f'Tool {name} is not available')
230        return tool
231