• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1#!/usr/bin/env python
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
17from dataclasses import dataclass
18from os import PathLike
19from typing import List
20
21
22class FakeAsyncProcess:
23    predefined_returncode = 0
24    predefined_communicate_return = (b"", b"")
25
26    def __init__(self):
27        pass
28
29    @property
30    def returncode(self):
31        return self.predefined_returncode
32
33    @staticmethod
34    def add_file(file_path: str, content: str = "disassembly output"):
35        with open(file_path, mode="w", encoding="utf-8") as f:
36            f.write(content)
37
38    async def communicate(self, *args, **kwargs):
39        _ = args, kwargs
40        return self.predefined_communicate_return
41
42
43@dataclass
44class FakeCommand:
45    opts: list
46    expected: str = ""
47    return_code: int = 0
48    stdout: bytes = b""
49    stderr: bytes = b""
50
51
52class MockAsyncSubprocess:
53    def __init__(self, commands: List[FakeCommand]):
54        self.commands = commands
55
56    @staticmethod
57    def create_fake_process(command: FakeCommand, program: str, *args: str | bytes | PathLike):
58        FakeAsyncProcess.predefined_returncode = command.return_code
59        FakeAsyncProcess.predefined_communicate_return = (command.stdout, command.stderr)
60        if program.endswith("ark_disasm") and command.return_code == 0:
61            FakeAsyncProcess.add_file(args[-1], )
62        return FakeAsyncProcess()
63
64    @staticmethod
65    def _match_call(command: FakeCommand, program: str, *args):
66        if command.expected != program:
67            return False
68        for option in command.opts:
69            if option not in args:
70                return False
71        return True
72
73    def create_subprocess_exec(self):
74        async def mock_subproc(program: str | bytes | PathLike, *args: str | bytes | PathLike, **kwargs):
75            _ = kwargs
76            for command in self.commands:
77                if self._match_call(command, program, *args):
78                    return self.create_fake_process(command, program, *args)
79            raise RuntimeError(f"'{program}' doesn't not match any mocks'")
80
81        return mock_subproc
82