• 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
18import multiprocessing.pool
19import sys
20from abc import abstractmethod
21from typing import Tuple, List
22
23import stress_common
24
25
26class Test:
27
28    def __init__(self, source: str, abc: str):
29        self.source = source
30        self.abc = abc
31
32
33class Result:
34
35    def __init__(self, source, result, stdout=None, stderr=None):
36        self.source = source
37        self.result = result
38        self.stdout = stdout
39        self.stderr = stderr
40
41
42class StressTest:
43
44    def run(self, tests: List[Test]):
45        logger = stress_common.create_logger()
46        logger.debug('Running ABCKit...')
47        result_all = {}
48        counter = 0
49
50        with multiprocessing.pool.ThreadPool(stress_common.NPROC) as pool:
51            for result in pool.imap(self.run_single, tests):
52                result_all[result.source] = {}
53                result_all[result.source]['error'] = result.result
54                counter += 1
55
56        return result_all
57
58    @abstractmethod
59    def run_single(self, test: Test) -> Result:
60        pass
61
62    @abstractmethod
63    def compile_single(self, src: str) -> Tuple[str, str, int]:
64        pass
65
66    @abstractmethod
67    def prepare(self) -> None:
68        pass
69
70    def build(self) -> List[Test]:
71        logger = stress_common.create_logger()
72        tests: List[str] = self.collect()
73
74        logger.debug('Running compiler...')
75        compiled_tests: List[Test] = []
76        counter = 0
77        with multiprocessing.pool.ThreadPool(stress_common.NPROC) as pool:
78            for js_path, abc_path, retcode in pool.imap(self.compile_single,
79                                                        tests,
80                                                        chunksize=20):
81                if retcode == 0:
82                    compiled_tests.append(Test(js_path, abc_path))
83                counter += 1
84
85        logger.debug('Tests successfully compiled: %s', len(compiled_tests))
86        return compiled_tests
87
88    @abstractmethod
89    def collect(self) -> List[str]:
90        pass
91
92    @abstractmethod
93    def get_compiler_path(self, build_dir) -> str:
94        pass
95
96    @abstractmethod
97    def get_fail_list_path(self, build_dir) -> str:
98        pass
99