• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1#!/usr/bin/env python3
2# -*- coding: utf-8 -*-
3
4"""
5Copyright (c) 2021 Huawei Device Co., Ltd.
6Licensed under the Apache License, Version 2.0 (the "License");
7you may not use this file except in compliance with the License.
8You may obtain a copy of the License at
9
10    http://www.apache.org/licenses/LICENSE-2.0
11
12Unless required by applicable law or agreed to in writing, software
13distributed under the License is distributed on an "AS IS" BASIS,
14WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15See the License for the specific language governing permissions and
16limitations under the License.
17
18Description: Implement the public interface in the 262 use case
19"""
20
21import os
22import sys
23import subprocess
24import datetime
25import time
26import shutil
27import platform
28import re
29
30TERM_NORMAL = '\033[0m'
31TERM_YELLOW = '\033[1;33m'
32TERM_BLUE = '\033[1;34m'
33TERM_RED = '\033[1;31m'
34TERM_FUCHSIA = '\033[1;35m'
35
36
37def current_time():
38    return datetime.datetime.now().strftime('%m-%d %H:%M:%S.%f')
39
40
41class Logging():
42    def __init__(self):
43        self.is_logging = True
44
45    def debug(self, info):
46        if self.is_logging:
47            print(
48                f'{current_time()} D:>>>  {TERM_BLUE}{str(info)}{TERM_NORMAL}')
49
50    def info(self, info):
51        if self.is_logging:
52            if len(str(info)) > 100:
53                print(f'{current_time()} I:>>> \n{str(info)} ')
54            else:
55                print(f'{current_time()} I:>>>    {str(info)} ')
56
57
58LOGGING = Logging()
59
60
61class Command():
62    def __init__(self, cmd):
63        self.cmd = cmd
64
65    def run(self):
66        LOGGING.debug("command: " + self.cmd)
67        out = os.popen(self.cmd).read()
68        LOGGING.info(out)
69        return out
70
71
72def run_cmd(command):
73    cmd = Command(command)
74    return cmd.run()
75
76
77class CommandCwd():
78    def __init__(self, cmds, cwd):
79        self.cmds = cmds
80        self.cwd = cwd
81
82    def run(self):
83        cmd = " ".join(self.cmds)
84        LOGGING.debug("command: " + cmd + " | " + "dir: " + self.cwd)
85        if platform.system() == "Windows" :
86            proc = subprocess.Popen(self.cmds, cwd=self.cwd ,shell=True)
87        else :
88            proc = subprocess.Popen(self.cmds, cwd=self.cwd)
89        return proc.wait()
90
91
92def run_cmd_cwd(commands, cwd=os.getcwd()):
93    cmd = CommandCwd(commands, cwd)
94    return cmd.run()
95
96
97def sleep(duration):
98    LOGGING.debug("sleeping %d" % duration)
99    time.sleep(duration)
100
101
102def write_file(save_file, result):
103    LOGGING.debug(f"write file:{save_file}")
104    with open(save_file, "a+") as file:
105        file.write(result + "\n")
106        file.flush()
107
108
109def remove_dir(path):
110    if os.path.exists(path):
111        shutil.rmtree(path)
112
113
114def remove_file(file):
115    if os.path.exists(file):
116        os.remove(file)
117
118
119def mkdir(path):
120    if not os.path.exists(path):
121        os.makedirs(path)
122
123
124def report_command(cmd_type, cmd, env=None):
125    sys.stderr.write(f'{TERM_BLUE}{cmd_type}{TERM_NORMAL}\n')
126    if env is not None:
127        sys.stderr.write(''.join(f'{TERM_BLUE}{var}={val} \\{TERM_NORMAL}\n'
128                                 for var, val in sorted(env.items())))
129    cmd_str = (f'{TERM_NORMAL}\n\t{TERM_BLUE}').join(cmd)
130    sys.stderr.write(f'\t{TERM_BLUE}{cmd_str}{TERM_NORMAL}\n')
131    sys.stderr.write("\n")
132
133
134def git_clone(git_url, code_dir):
135    cmd = ['git', 'clone', git_url, code_dir]
136    ret = run_cmd_cwd(cmd)
137    assert not ret, f"\n error: Cloning '{git_url}' failed."
138
139
140def git_checkout(git_bash, cwd):
141    cmd = ['git', 'checkout', git_bash]
142    ret = run_cmd_cwd(cmd, cwd)
143    assert not ret, f"\n error: git checkout '{git_bash}' failed."
144
145
146def git_apply(patch_file, cwd):
147    cmd = ['git', 'apply', patch_file]
148    ret = run_cmd_cwd(cmd, cwd)
149    assert not ret, f"\n error: Failed to apply '{patch_file}'"
150
151
152def git_clean(cwd):
153    cmd = ['git', 'checkout', '--', '.']
154    run_cmd_cwd(cmd, cwd)
155
156
157def npm_install(cwd):
158    cmd = ['npm', 'install']
159    ret = run_cmd_cwd(cmd, cwd)
160    assert not ret, f"\n error: Failed to 'npm install'"
161
162
163def search_dependency(file, directory):
164    for root, dirs, files in os.walk(directory, topdown=True):
165        for f in files:
166            if f == file:
167                return os.path.join(root, f)
168    return "FILE_NOT_FOUND"
169
170
171def collect_module_dependencies(file, directory, traversedDependencies):
172    dependencies = []
173    traversedDependencies.append(file)
174    with open(file, 'r', encoding='utf-8') as f:
175        content = f.read()
176        module_import_list = re.findall(r'(import|from)(?:\s*)\(?(\'(\.\/.*)\'|"(\.\/.*)")\)?', content)
177
178        for result in list(set(module_import_list)):
179            specifier = (result[2] if len(result[2]) != 0 else result[3]).lstrip('./')
180            if os.path.basename(file) is not specifier:
181                dependency = search_dependency(specifier, directory)
182                if dependency == "FILE_NOT_FOUND":
183                    continue
184
185                if dependency not in traversedDependencies:
186                    dependencies.extend(collect_module_dependencies(dependency, directory,
187                                                                    list(set(traversedDependencies))))
188                dependencies.append(dependency)
189
190    return dependencies