1
1
import subprocess
2
2
from argparse import ArgumentParser
3
3
import time
4
+ import sys
5
+ import os
6
+ from datetime import datetime
7
+ import traceback
8
+
9
+ import logging
10
+ import threading
11
+
12
+ pathlogs = 'testsLog'
13
+
14
+ def LOG (* args , mode = 'INFO' ):
15
+ print (args )
16
+ if (mode == 'INFO' ):
17
+ logging .info (args )
18
+ elif (mode == 'ERR' ):
19
+ logging .error (args )
20
+ elif (mode == 'ELF' ):
21
+ logging .info (f'<CPP>{ args } ' )
22
+
4
23
5
24
6
25
class DuplicatedTestError (Exception ):
@@ -16,14 +35,27 @@ def __init__(self, executable):
16
35
self ._executable = executable
17
36
18
37
def __enter__ (self ):
38
+ self ._executable = f"stdbuf -oL { self ._executable } "
19
39
self ._process = subprocess .Popen (
20
40
self ._executable ,
21
41
shell = True ,
22
42
stdout = subprocess .PIPE ,
23
43
stderr = subprocess .PIPE ,
24
- text = True
44
+ text = True ,
45
+ bufsize = 1 ,
25
46
)
26
47
48
+ self ._stdout_thread = threading .Thread (target = self .stream_output )
49
+ self ._stdout_thread .daemon = True
50
+ self ._stdout_thread .start ()
51
+
52
+ def stream_output (self ):
53
+ while True :
54
+ line = self ._process .stdout .readline ()
55
+ if not line : break
56
+
57
+ LOG (line , mode = 'ELF' )
58
+
27
59
def __exit__ (self , * args ):
28
60
try :
29
61
out , err = self ._process .communicate (timeout = 1 )
@@ -35,10 +67,12 @@ def __exit__(self, *args):
35
67
out = "Error recovering stdout"
36
68
err = "Error recovering stderr"
37
69
70
+ self ._stdout_thread .join (timeout = 1 )
71
+
38
72
if out :
39
- print (f" * UUT stdout:\n { out } " )
73
+ LOG (f" * UUT stdout:\n { out } " )
40
74
if err :
41
- print (f" * UUT stderr:\n { err } " )
75
+ LOG (f" * UUT stderr:\n { err } " )
42
76
43
77
44
78
class Test :
@@ -99,27 +133,49 @@ def decorator(test_func):
99
133
100
134
# Runs all the registered tests, cleaning up after each test
101
135
def run (self ):
136
+
137
+ failed_tests = 0
138
+
139
+ if not os .path .exists (f'./{ pathlogs } ' ):
140
+ os .makedirs (f'./{ pathlogs } ' )
141
+ date = datetime .now ()
142
+ logging .basicConfig (level = logging .INFO , filename = f'{ pathlogs } /{ date } log.log' , filemode = 'w' , format = "%(levelname)s - %(message)s" )
143
+
102
144
for name , test in self ._tests .items ():
103
145
try :
104
146
test .run_prepare ()
105
147
106
148
with self ._uut :
107
149
time .sleep (0.1 )
108
150
try :
109
- print (f"[{ name } ] Running..." )
151
+ LOG (f"[{ name } ] Running..." )
110
152
result = test ()
111
- print (f"[{ name } ] Succesfull!" )
153
+ LOG (f"[{ name } ] Succesfull!" )
112
154
if result is not None :
113
- print (f" * Result: { result } " )
155
+ LOG (f" * Result: { result } " )
114
156
except Exception as reason :
115
- print (f"[{ name } ] Failed!" )
116
- print (f" * Reason: { reason } " )
157
+ tb = traceback .extract_tb (reason .__traceback__ )
158
+ failed_tests += 1
159
+
160
+ # Obtener la última llamada (donde ocurrió el error)
161
+ file1 , line1 , _ , _ = tb [- 2 ]
162
+ file2 , line2 , _ , _ = tb [- 1 ]
163
+ LOG (f"[{ name } ] Failed!" , mode = 'ERR' )
164
+ LOG (f'file: { file1 } line: { line1 } , file: { file2 } line: { line2 } ' , mode = 'ERR' )
165
+ LOG (f" * Reason: { reason } " , mode = 'ERR' )
117
166
118
167
test .run_cleanup ()
119
168
except KeyboardInterrupt :
120
- print (f"[{ name } ] Keyboard Interrupt. Aborted." )
121
-
122
-
169
+ LOG (f"[{ name } ] Keyboard Interrupt. Aborted." )
170
+ import sys
171
+ sys .exit (130 )
172
+
173
+ if failed_tests > 0 :
174
+ import sys
175
+ LOG (f"[{ failed_tests } ] Tests Failed!" , mode = 'ERR' )
176
+ sys .exit (1 )
177
+
178
+
123
179
124
180
parser = ArgumentParser (
125
181
prog = "test" ,
0 commit comments