(self, **kwargs)
| 94 | class CaseSetup(object): |
| 95 | @contextmanager |
| 96 | def check_console(self, **kwargs): |
| 97 | for key, value in kwargs.items(): |
| 98 | assert hasattr(WriterThread, key) |
| 99 | setattr(WriterThread, key, value) |
| 100 | |
| 101 | self.writer = writer |
| 102 | |
| 103 | args = runner.get_command_line() |
| 104 | |
| 105 | args = runner.add_command_line_args(args) |
| 106 | |
| 107 | if SHOW_OTHER_DEBUG_INFO: |
| 108 | print("executing: %s" % (" ".join(args),)) |
| 109 | try: |
| 110 | with runner.run_process(args, writer) as dct_with_stdout_stder: |
| 111 | writer.get_stdout = lambda: "".join(dct_with_stdout_stder["stdout"]) |
| 112 | writer.get_stderr = lambda: "".join(dct_with_stdout_stder["stderr"]) |
| 113 | |
| 114 | # Make sure communication is setup. |
| 115 | writer.hello() |
| 116 | yield writer |
| 117 | finally: |
| 118 | writer.log = [] |
| 119 | |
| 120 | stdout = dct_with_stdout_stder["stdout"] |
| 121 | stderr = dct_with_stdout_stder["stderr"] |
| 122 | writer.additional_output_checks("".join(stdout), "".join(stderr)) |
| 123 | |
| 124 | return CaseSetup() |
| 125 |
no test coverage detected