• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1#!/usr/bin/env python3
2# -*- coding: utf-8 -*-
3
4# Copyright (c) 2021-2023 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# This file does only contain a selection of the most common options. For a
18# full list see the documentation:
19# http://www.sphinx-doc.org/en/master/config
20
21import logging
22import re
23import subprocess
24from glob import glob
25from os import getenv, path, makedirs
26from typing import Optional
27
28from runner.logger import Log
29from runner.options.config import Config
30from runner.utils import generate, wrap_with_function
31from runner.plugins.work_dir import WorkDir
32
33HERMES_URL = "HERMES_URL"
34HERMES_REVISION = "HERMES_REVISION"
35
36_LOGGER = logging.getLogger("runner.hermes.util_hermes")
37
38
39class UtilHermes:
40    def __init__(self, config: Config, work_dir: WorkDir) -> None:
41        self.check_expr = re.compile(r"^\s*//\s?(?:CHECK-NEXT|CHECK-LABEL|CHECK):(.+)", re.MULTILINE)
42
43        self.show_progress = config.general.show_progress
44        self.force_download = config.general.force_download
45        self.jit = config.ark.jit.enable
46        self.jit_preheat_repeats = config.ark.jit.num_repeats
47
48        self.work_dir = work_dir
49
50        self.hermes_url: Optional[str] = getenv(HERMES_URL)
51        self.hermes_revision: Optional[str] = getenv(HERMES_REVISION)
52        if self.hermes_url is None:
53            Log.exception_and_raise(_LOGGER, f"No {HERMES_URL} environment variable set", EnvironmentError)
54        if self.hermes_revision is None:
55            Log.exception_and_raise(_LOGGER, f"No {HERMES_REVISION} environment variable set", EnvironmentError)
56
57    def generate(self) -> str:
58        stamp_name = f"hermes-{self.hermes_revision}"
59        if self.jit and self.jit_preheat_repeats > 1:
60            stamp_name += f"-jit-{self.jit_preheat_repeats}"
61        assert self.hermes_url is not None
62        assert self.hermes_revision is not None
63        return generate(
64            name="hermes",
65            stamp_name=stamp_name,
66            url=self.hermes_url,
67            revision=self.hermes_revision,
68            generated_root=self.work_dir.gen,
69            test_subdir="test/hermes",
70            show_progress=self.show_progress,
71            process_copy=self.process_copy,
72            force_download=self.force_download
73        )
74
75    def process_copy(self, src_path: str, dst_path: str) -> None:
76        Log.all(_LOGGER, "Generating tests")
77
78        glob_expression = path.join(src_path, "**/*.js")
79        files = glob(glob_expression, recursive=True)
80
81        for src_file in files:
82            dest_file = src_file.replace(src_path, dst_path)
83            makedirs(path.dirname(dest_file), exist_ok=True)
84            self.create_file(src_file, dest_file)
85
86    def create_file(self, src_file: str, dest_file: str) -> None:
87        with open(src_file, 'r', encoding="utf-8") as file_pointer:
88            input_str = file_pointer.read()
89
90        if self.jit and self.jit_preheat_repeats > 1:
91            out_str = wrap_with_function(input_str, self.jit_preheat_repeats)
92        else:
93            out_str = input_str
94
95        with open(dest_file, 'w', encoding="utf-8") as output:
96            output.write(out_str)
97
98    def run_filecheck(self, test_file: str, actual_output: str) -> bool:
99        with open(test_file, 'r', encoding="utf-8") as file_pointer:
100            input_str = file_pointer.read()
101        if not re.match(self.check_expr, input_str):
102            return True
103
104        assert actual_output is not None, "Expected some output to check"
105        cmd = ['FileCheck-14', test_file]
106        with subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, stdin=subprocess.PIPE) as process:
107            try:
108                if process.stdin is None:
109                    Log.exception_and_raise(_LOGGER, f"Unexpected error on checking {test_file}")
110                process.stdin.write(actual_output.encode('utf-8'))
111                process.communicate(timeout=10)
112                return_code = process.returncode
113            except subprocess.TimeoutExpired as ex:
114                _LOGGER.error(f"{' '.join(cmd)} failed with {ex}")
115                process.kill()
116                return_code = -1
117            except Exception as ex:  # pylint: disable=broad-except
118                _LOGGER.error(f"{' '.join(cmd)} failed with {ex}")
119                return_code = -1
120
121        return return_code == 0
122