• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1#!/usr/bin/env python3
2# coding=utf-8
3
4#
5# Copyright (c) 2020-2022 Huawei Device Co., Ltd.
6# Licensed under the Apache License, Version 2.0 (the "License");
7# you may not use this file except in compliance with the License.
8# You may obtain a copy of the License at
9#
10#     http://www.apache.org/licenses/LICENSE-2.0
11#
12# Unless required by applicable law or agreed to in writing, software
13# distributed under the License is distributed on an "AS IS" BASIS,
14# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15# See the License for the specific language governing permissions and
16# limitations under the License.
17#
18import types
19
20from queue import Queue
21from _core.interface import IParser
22from _core.report.encrypt import check_pub_key_exist
23from _core.logger import platform_logger
24
25__all__ = ["ShellHandler", "driver_output_method"]
26
27LOG = platform_logger("ParserLite")
28
29
30def driver_output_method(handler, output, end_mark="\n"):
31    """
32    Some driver output should remove last line when last line is not finish
33    Such as ltp driver, cpp driver, new js driver
34    """
35    content = output
36    if handler.unfinished_line:
37        content = "".join((handler.unfinished_line, content))
38        handler.unfinished_line = ""
39    lines = content.split(end_mark)
40    if content.endswith(end_mark):
41        # get rid of the tail element of this list contains empty str
42        return lines[:-1]
43    else:
44        handler.unfinished_line = lines[-1]
45        # not return the tail element of this list contains unfinished str,
46        # so we set position -1
47        return lines[:-1]
48
49
50class ShellHandler:
51    def __init__(self, parsers):
52        self.parsers = []
53        self.unfinished_line = ""
54        self.output_queue = Queue()
55        self.process_output_methods = []
56        for parser in parsers:
57            if isinstance(parser, IParser):
58                self.parsers.append(parser)
59            else:
60                raise TypeError(
61                    "Parser {} must implement IOutputParser interface.".format(
62                        parser, ))
63
64    def _process_output(self, output, end_mark="\n"):
65        if self.process_output_methods:
66            method = self.process_output_methods[0]
67            if callable(method):
68                return method(self, output, end_mark)
69        else:
70            content = output
71            if self.unfinished_line:
72                content = "".join((self.unfinished_line, content))
73                self.unfinished_line = ""
74            lines = content.split(end_mark)
75            if content.endswith(end_mark):
76                # get rid of the tail element of this list contains empty str
77                return lines[:-1]
78            else:
79                self.unfinished_line = lines[-1]
80                # not return the tail element of this list contains unfinished str,
81                # so we set position -1
82                return lines
83
84    def add_process_method(self, func):
85        if isinstance(func, types.FunctionType):
86            self.process_output_methods.clear()
87            self.process_output_methods.append(func)
88
89    def __read__(self, output):
90        lines = self._process_output(output)
91        for line in lines:
92            for parser in self.parsers:
93                try:
94                    parser.__process__([line])
95                except (ValueError, TypeError, SyntaxError, AttributeError) \
96                        as error:
97                    LOG.debug("Parse %s line error: %s" % (line, error))
98
99    def __error__(self, message):
100        if message:
101            for parser in self.parsers:
102                parser.__process__([message])
103
104    def __done__(self, result_code="", message=""):
105        msg_fmt = ""
106        if message:
107            msg_fmt = ", message is {}".format(message)
108            for parser in self.parsers:
109                parser.__process__([message])
110        if not check_pub_key_exist():
111            LOG.debug("Result code is: {}{}".format(result_code, msg_fmt))
112        for parser in self.parsers:
113            parser.__done__()
114