Spaces:
Runtime error
Runtime error
Added loading of test groups from both the DB and the local file and merging these two
Browse files- src/testing.py +29 -0
src/testing.py
CHANGED
@@ -104,6 +104,7 @@ class ArchitectureRequestRecord:
|
|
104 |
test_traces = Architecture.get_trace_records()
|
105 |
for trace in test_traces:
|
106 |
records.append(ArchitectureRequestRecord.from_dict(trace))
|
|
|
107 |
cls.all = records
|
108 |
|
109 |
|
@@ -171,6 +172,31 @@ class TestGroup:
|
|
171 |
self.end = arr.end
|
172 |
self.elapsed = self.end - self.start
|
173 |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
174 |
@classmethod
|
175 |
def load_all(cls, reload=False):
|
176 |
if cls.all is None or reload:
|
@@ -181,6 +207,9 @@ class TestGroup:
|
|
181 |
if arr.test_group not in records:
|
182 |
records[arr.test_group] = TestGroup(arr.test_group)
|
183 |
records[arr.test_group].add_record(arr)
|
|
|
|
|
|
|
184 |
cls.all = records
|
185 |
|
186 |
@classmethod
|
|
|
104 |
test_traces = Architecture.get_trace_records()
|
105 |
for trace in test_traces:
|
106 |
records.append(ArchitectureRequestRecord.from_dict(trace))
|
107 |
+
|
108 |
cls.all = records
|
109 |
|
110 |
|
|
|
172 |
self.end = arr.end
|
173 |
self.elapsed = self.end - self.start
|
174 |
|
175 |
+
@classmethod
|
176 |
+
def load_db_records(cls) -> List[TestGroup]:
|
177 |
+
db_file = os.path.join(data_dir, 'sqlite', 'test_records.db')
|
178 |
+
con = sqlite3.connect(db_file)
|
179 |
+
cur = con.cursor()
|
180 |
+
|
181 |
+
sql = "SELECT id, test_group from test_groups"
|
182 |
+
cur.execute(sql)
|
183 |
+
tg_id_names = [(r[0], r[1]) for r in cur.fetchall()]
|
184 |
+
test_groups: List[TestGroup] = []
|
185 |
+
|
186 |
+
for tg_id, tg_name in tg_id_names:
|
187 |
+
tg = TestGroup(tg_name)
|
188 |
+
sql = f"SELECT id, arch_name, response_len, start, end, comment FROM arch_requests WHERE test_group_id={tg_id}"
|
189 |
+
cur.execute(sql)
|
190 |
+
arch_requests = [(r[0], r[1], r[2], r[3], r[4], r[5]) for r in cur.fetchall()]
|
191 |
+
for ar_id, ar_arch_name, ar_resp_len, ar_start, ar_end, ar_comment in arch_requests:
|
192 |
+
sql = f"SELECT name, start, end FROM arch_req_steps WHERE arch_req_id={ar_id}"
|
193 |
+
cur.execute(sql)
|
194 |
+
steps = [ArchitectureRequestRecord.ArchStep(r[0], r[1], r[2]) for r in cur.fetchall()]
|
195 |
+
arch_req_record = ArchitectureRequestRecord(ar_arch_name, ar_resp_len, ar_start, ar_end, (ar_end - ar_start), [tg_name], tg_name, f"(DB persisted) {ar_comment}", steps)
|
196 |
+
tg.add_record(arch_req_record)
|
197 |
+
test_groups.append(tg)
|
198 |
+
return test_groups
|
199 |
+
|
200 |
@classmethod
|
201 |
def load_all(cls, reload=False):
|
202 |
if cls.all is None or reload:
|
|
|
207 |
if arr.test_group not in records:
|
208 |
records[arr.test_group] = TestGroup(arr.test_group)
|
209 |
records[arr.test_group].add_record(arr)
|
210 |
+
db_records = cls.load_db_records()
|
211 |
+
for test_group in db_records:
|
212 |
+
records[test_group.test_group] = test_group
|
213 |
cls.all = records
|
214 |
|
215 |
@classmethod
|