1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
|
import os
from molotov.ui.app import MolotovApp
from molotov.util import printable_error
class Console:
def __init__(
self,
interval=0.3,
max_lines_displayed=25,
simple_console=False,
single_process=True,
):
self._interval = interval
self._stop = True
self._creator = os.getpid()
self._stop = False
self._max_lines_displayed = max_lines_displayed
self._simple_console = simple_console
self.ui = MolotovApp(
refresh_interval=interval,
max_lines=max_lines_displayed,
simple_console=simple_console,
single_process=single_process,
)
self.terminal = self.ui.terminal
self.errors = self.ui.errors
self.status = self.ui.status
self.started = False
async def start(self):
await self.ui.start()
self.started = True
async def stop(self):
await self.ui.stop()
self.started = False
def print_results(self, results):
self.status.update(results)
def print(self, data):
if self.terminal is None:
return
for line in data.split("\n"):
line = line.strip()
self.terminal.write_line(line)
def print_error(self, error, tb=None):
if self.errors is None:
return
if isinstance(error, str):
for line in error.split("\n"):
line = line.strip()
self.errors.write_line(line, fg="gray")
return
for line in printable_error(error, tb):
self.errors.write_line(line, fg="gray")
self.errors.write_line("", fg="gray")
def print_block(self, start, callable, end="OK"):
if self.terminal is None:
return callable()
self.terminal.write_line(f"{start}...")
try:
return callable()
finally:
self.terminal.write_line("OK")
|
