VirtualBox

source: vbox/trunk/src/VBox/ValidationKit/testmanager/core/testgroup.py@ 96407

Last change on this file since 96407 was 96407, checked in by vboxsync, 2 years ago

scm copyright and license note update

  • Property svn:eol-style set to native
  • Property svn:keywords set to Author Date Id Revision
File size: 30.8 KB
Line 
1# -*- coding: utf-8 -*-
2# $Id: testgroup.py 96407 2022-08-22 17:43:14Z vboxsync $
3
4"""
5Test Manager - Test groups management.
6"""
7
8__copyright__ = \
9"""
10Copyright (C) 2012-2022 Oracle and/or its affiliates.
11
12This file is part of VirtualBox base platform packages, as
13available from https://www.virtualbox.org.
14
15This program is free software; you can redistribute it and/or
16modify it under the terms of the GNU General Public License
17as published by the Free Software Foundation, in version 3 of the
18License.
19
20This program is distributed in the hope that it will be useful, but
21WITHOUT ANY WARRANTY; without even the implied warranty of
22MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
23General Public License for more details.
24
25You should have received a copy of the GNU General Public License
26along with this program; if not, see <https://www.gnu.org/licenses>.
27
28The contents of this file may alternatively be used under the terms
29of the Common Development and Distribution License Version 1.0
30(CDDL), a copy of it is provided in the "COPYING.CDDL" file included
31in the VirtualBox distribution, in which case the provisions of the
32CDDL are applicable instead of those of the GPL.
33
34You may elect to license modified versions of this file under the
35terms and conditions of either the GPL or the CDDL or both.
36
37SPDX-License-Identifier: GPL-3.0-only OR CDDL-1.0
38"""
39__version__ = "$Revision: 96407 $"
40
41
42# Standard python imports.
43import unittest;
44
45# Validation Kit imports.
46from testmanager.core.base import ModelDataBase, ModelDataBaseTestCase, ModelLogicBase, TMRowInUse, \
47 TMTooManyRows, TMInvalidData, TMRowNotFound, TMRowAlreadyExists;
48from testmanager.core.testcase import TestCaseData, TestCaseDataEx;
49
50
51class TestGroupMemberData(ModelDataBase):
52 """Representation of a test group member database row."""
53
54 ksParam_idTestGroup = 'TestGroupMember_idTestGroup';
55 ksParam_idTestCase = 'TestGroupMember_idTestCase';
56 ksParam_tsEffective = 'TestGroupMember_tsEffective';
57 ksParam_tsExpire = 'TestGroupMember_tsExpire';
58 ksParam_uidAuthor = 'TestGroupMember_uidAuthor';
59 ksParam_iSchedPriority = 'TestGroupMember_iSchedPriority';
60 ksParam_aidTestCaseArgs = 'TestGroupMember_aidTestCaseArgs';
61
62 kasAllowNullAttributes = ['idTestGroup', 'idTestCase', 'tsEffective', 'tsExpire', 'uidAuthor', 'aidTestCaseArgs' ];
63 kiMin_iSchedPriority = 0;
64 kiMax_iSchedPriority = 31;
65
66 kcDbColumns = 7;
67
68 def __init__(self):
69 ModelDataBase.__init__(self)
70
71 #
72 # Initialize with defaults.
73 # See the database for explanations of each of these fields.
74 #
75 self.idTestGroup = None;
76 self.idTestCase = None;
77 self.tsEffective = None;
78 self.tsExpire = None;
79 self.uidAuthor = None;
80 self.iSchedPriority = 16;
81 self.aidTestCaseArgs = None;
82
83 def initFromDbRow(self, aoRow):
84 """
85 Reinitialize from a SELECT * FROM TestCaseGroupMembers.
86 Return self. Raises exception if no row.
87 """
88 if aoRow is None:
89 raise TMRowNotFound('Test group member not found.')
90
91 self.idTestGroup = aoRow[0];
92 self.idTestCase = aoRow[1];
93 self.tsEffective = aoRow[2];
94 self.tsExpire = aoRow[3];
95 self.uidAuthor = aoRow[4];
96 self.iSchedPriority = aoRow[5];
97 self.aidTestCaseArgs = aoRow[6];
98 return self
99
100
101 def getAttributeParamNullValues(self, sAttr):
102 # Arrays default to [] as NULL currently. That doesn't work for us.
103 if sAttr == 'aidTestCaseArgs':
104 aoNilValues = [None, '-1'];
105 else:
106 aoNilValues = ModelDataBase.getAttributeParamNullValues(self, sAttr);
107 return aoNilValues;
108
109 def _validateAndConvertAttribute(self, sAttr, sParam, oValue, aoNilValues, fAllowNull, oDb):
110 if sAttr != 'aidTestCaseArgs':
111 return ModelDataBase._validateAndConvertAttribute(self, sAttr, sParam, oValue, aoNilValues, fAllowNull, oDb);
112
113 # -1 is a special value, which when present make the whole thing NULL (None).
114 (aidVariations, sError) = self.validateListOfInts(oValue, aoNilValues = aoNilValues, fAllowNull = fAllowNull,
115 iMin = -1, iMax = 0x7ffffffe);
116 if sError is None:
117 if aidVariations is None:
118 pass;
119 elif -1 in aidVariations:
120 aidVariations = None;
121 elif 0 in aidVariations:
122 sError = 'Invalid test case varation ID #0.';
123 else:
124 aidVariations = sorted(aidVariations);
125 return (aidVariations, sError);
126
127
128
129class TestGroupMemberDataEx(TestGroupMemberData):
130 """Extended representation of a test group member."""
131
132 def __init__(self):
133 """Extend parent class"""
134 TestGroupMemberData.__init__(self)
135 self.oTestCase = None; # TestCaseDataEx.
136
137 def initFromDbRowEx(self, aoRow, oDb, tsNow = None):
138 """
139 Reinitialize from a SELECT * FROM TestGroupMembers, TestCases row.
140 Will query the necessary additional data from oDb using tsNow.
141
142 Returns self. Raises exception if no row or database error.
143 """
144 TestGroupMemberData.initFromDbRow(self, aoRow);
145 self.oTestCase = TestCaseDataEx();
146 self.oTestCase.initFromDbRowEx(aoRow[TestGroupMemberData.kcDbColumns:], oDb, tsNow);
147 return self;
148
149 def initFromParams(self, oDisp, fStrict = True):
150 self.oTestCase = None;
151 return TestGroupMemberData.initFromParams(self, oDisp, fStrict);
152
153 def getDataAttributes(self):
154 asAttributes = TestGroupMemberData.getDataAttributes(self);
155 asAttributes.remove('oTestCase');
156 return asAttributes;
157
158 def _validateAndConvertWorker(self, asAllowNullAttributes, oDb, enmValidateFor = ModelDataBase.ksValidateFor_Other):
159 dErrors = TestGroupMemberData._validateAndConvertWorker(self, asAllowNullAttributes, oDb, enmValidateFor);
160 if self.ksParam_idTestCase not in dErrors:
161 self.oTestCase = TestCaseDataEx()
162 try:
163 self.oTestCase.initFromDbWithId(oDb, self.idTestCase);
164 except Exception as oXcpt:
165 self.oTestCase = TestCaseDataEx()
166 dErrors[self.ksParam_idTestCase] = str(oXcpt);
167 return dErrors;
168
169
170class TestGroupMemberData2(TestCaseData):
171 """Special representation of a Test Group Member item"""
172
173 def __init__(self):
174 """Extend parent class"""
175 TestCaseData.__init__(self)
176 self.idTestGroup = None
177 self.aidTestCaseArgs = []
178
179 def initFromDbRowEx(self, aoRow):
180 """
181 Reinitialize from this query:
182
183 SELECT TestCases.*,
184 TestGroupMembers.idTestGroup,
185 TestGroupMembers.aidTestCaseArgs
186 FROM TestCases, TestGroupMembers
187 WHERE TestCases.idTestCase = TestGroupMembers.idTestCase
188
189 Represents complete test group member (test case) info.
190 Returns object of type TestGroupMemberData2. Raises exception if no row.
191 """
192 TestCaseData.initFromDbRow(self, aoRow);
193 self.idTestGroup = aoRow[-2]
194 self.aidTestCaseArgs = aoRow[-1]
195 return self;
196
197
198class TestGroupData(ModelDataBase):
199 """
200 Test group data.
201 """
202
203 ksIdAttr = 'idTestGroup';
204
205 ksParam_idTestGroup = 'TestGroup_idTestGroup'
206 ksParam_tsEffective = 'TestGroup_tsEffective'
207 ksParam_tsExpire = 'TestGroup_tsExpire'
208 ksParam_uidAuthor = 'TestGroup_uidAuthor'
209 ksParam_sName = 'TestGroup_sName'
210 ksParam_sDescription = 'TestGroup_sDescription'
211 ksParam_sComment = 'TestGroup_sComment'
212
213 kasAllowNullAttributes = ['idTestGroup', 'tsEffective', 'tsExpire', 'uidAuthor', 'sDescription', 'sComment' ];
214
215 kcDbColumns = 7;
216
217 def __init__(self):
218 ModelDataBase.__init__(self);
219
220 #
221 # Initialize with defaults.
222 # See the database for explanations of each of these fields.
223 #
224 self.idTestGroup = None
225 self.tsEffective = None
226 self.tsExpire = None
227 self.uidAuthor = None
228 self.sName = None
229 self.sDescription = None
230 self.sComment = None
231
232 def initFromDbRow(self, aoRow):
233 """
234 Reinitialize from a SELECT * FROM TestGroups row.
235 Returns object of type TestGroupData. Raises exception if no row.
236 """
237 if aoRow is None:
238 raise TMRowNotFound('Test group not found.')
239
240 self.idTestGroup = aoRow[0]
241 self.tsEffective = aoRow[1]
242 self.tsExpire = aoRow[2]
243 self.uidAuthor = aoRow[3]
244 self.sName = aoRow[4]
245 self.sDescription = aoRow[5]
246 self.sComment = aoRow[6]
247 return self
248
249 def initFromDbWithId(self, oDb, idTestGroup, tsNow = None, sPeriodBack = None):
250 """
251 Initialize the object from the database.
252 """
253 oDb.execute(self.formatSimpleNowAndPeriodQuery(oDb,
254 'SELECT *\n'
255 'FROM TestGroups\n'
256 'WHERE idTestGroup = %s\n'
257 , ( idTestGroup,), tsNow, sPeriodBack));
258 aoRow = oDb.fetchOne()
259 if aoRow is None:
260 raise TMRowNotFound('idTestGroup=%s not found (tsNow=%s sPeriodBack=%s)' % (idTestGroup, tsNow, sPeriodBack,));
261 return self.initFromDbRow(aoRow);
262
263
264class TestGroupDataEx(TestGroupData):
265 """
266 Extended test group data.
267 """
268
269 ksParam_aoMembers = 'TestGroupDataEx_aoMembers';
270 kasAltArrayNull = [ 'aoMembers', ];
271
272 ## Helper parameter containing the comma separated list with the IDs of
273 # potential members found in the parameters.
274 ksParam_aidTestCases = 'TestGroupDataEx_aidTestCases';
275
276
277 def __init__(self):
278 TestGroupData.__init__(self);
279 self.aoMembers = []; # TestGroupMemberDataEx.
280
281 def _initExtraMembersFromDb(self, oDb, tsNow = None, sPeriodBack = None):
282 """
283 Worker shared by the initFromDb* methods.
284 Returns self. Raises exception if no row or database error.
285 """
286 self.aoMembers = [];
287 _ = sPeriodBack; ## @todo sPeriodBack
288
289 if tsNow is None:
290 oDb.execute('SELECT TestGroupMembers.*, TestCases.*\n'
291 'FROM TestGroupMembers\n'
292 'LEFT OUTER JOIN TestCases ON (\n'
293 ' TestGroupMembers.idTestCase = TestCases.idTestCase\n'
294 ' AND TestCases.tsExpire = \'infinity\'::TIMESTAMP)\n'
295 'WHERE TestGroupMembers.idTestGroup = %s\n'
296 ' AND TestGroupMembers.tsExpire = \'infinity\'::TIMESTAMP\n'
297 'ORDER BY TestCases.sName, TestCases.idTestCase\n'
298 , (self.idTestGroup,));
299 else:
300 oDb.execute('SELECT TestGroupMembers.*, TestCases.*\n'
301 'FROM TestGroupMembers\n'
302 'LEFT OUTER JOIN TestCases ON (\n'
303 ' TestGroupMembers.idTestCase = TestCases.idTestCase\n'
304 ' AND TestCases.tsExpire > %s\n'
305 ' AND TestCases.tsEffective <= %s)\n'
306 'WHERE TestGroupMembers.idTestGroup = %s\n'
307 ' AND TestGroupMembers.tsExpire > %s\n'
308 ' AND TestGroupMembers.tsEffective <= %s\n'
309 'ORDER BY TestCases.sName, TestCases.idTestCase\n'
310 , (tsNow, tsNow, self.idTestGroup, tsNow, tsNow));
311
312 for aoRow in oDb.fetchAll():
313 self.aoMembers.append(TestGroupMemberDataEx().initFromDbRowEx(aoRow, oDb, tsNow));
314 return self;
315
316 def initFromDbRowEx(self, aoRow, oDb, tsNow = None, sPeriodBack = None):
317 """
318 Reinitialize from a SELECT * FROM TestGroups row. Will query the
319 necessary additional data from oDb using tsNow.
320 Returns self. Raises exception if no row or database error.
321 """
322 TestGroupData.initFromDbRow(self, aoRow);
323 return self._initExtraMembersFromDb(oDb, tsNow, sPeriodBack);
324
325 def initFromDbWithId(self, oDb, idTestGroup, tsNow = None, sPeriodBack = None):
326 """
327 Initialize the object from the database.
328 """
329 TestGroupData.initFromDbWithId(self, oDb, idTestGroup, tsNow, sPeriodBack);
330 return self._initExtraMembersFromDb(oDb, tsNow, sPeriodBack);
331
332
333 def getAttributeParamNullValues(self, sAttr):
334 if sAttr != 'aoMembers':
335 return TestGroupData.getAttributeParamNullValues(self, sAttr);
336 return ['', [], None];
337
338 def convertParamToAttribute(self, sAttr, sParam, oValue, oDisp, fStrict):
339 if sAttr != 'aoMembers':
340 return TestGroupData.convertParamToAttribute(self, sAttr, sParam, oValue, oDisp, fStrict);
341
342 aoNewValue = [];
343 aidSelected = oDisp.getListOfIntParams(sParam, iMin = 1, iMax = 0x7ffffffe, aiDefaults = [])
344 sIds = oDisp.getStringParam(self.ksParam_aidTestCases, sDefault = '');
345 for idTestCase in sIds.split(','):
346 try: idTestCase = int(idTestCase);
347 except: pass;
348 oDispWrapper = self.DispWrapper(oDisp, '%s[%s][%%s]' % (TestGroupDataEx.ksParam_aoMembers, idTestCase,))
349 oMember = TestGroupMemberDataEx().initFromParams(oDispWrapper, fStrict = False);
350 if idTestCase in aidSelected:
351 aoNewValue.append(oMember);
352 return aoNewValue;
353
354 def _validateAndConvertAttribute(self, sAttr, sParam, oValue, aoNilValues, fAllowNull, oDb):
355 if sAttr != 'aoMembers':
356 return TestGroupData._validateAndConvertAttribute(self, sAttr, sParam, oValue, aoNilValues, fAllowNull, oDb);
357
358 asErrors = [];
359 aoNewMembers = [];
360 for oOldMember in oValue:
361 oNewMember = TestGroupMemberDataEx().initFromOther(oOldMember);
362 aoNewMembers.append(oNewMember);
363
364 dErrors = oNewMember.validateAndConvert(oDb, ModelDataBase.ksValidateFor_Other);
365 if dErrors:
366 asErrors.append(str(dErrors));
367
368 if not asErrors:
369 for i, _ in enumerate(aoNewMembers):
370 idTestCase = aoNewMembers[i];
371 for j in range(i + 1, len(aoNewMembers)):
372 if aoNewMembers[j].idTestCase == idTestCase:
373 asErrors.append('Duplicate testcase #%d!' % (idTestCase, ));
374 break;
375
376 return (aoNewMembers, None if not asErrors else '<br>\n'.join(asErrors));
377
378
379class TestGroupLogic(ModelLogicBase):
380 """
381 Test case management logic.
382 """
383
384 def __init__(self, oDb):
385 ModelLogicBase.__init__(self, oDb)
386 self.dCache = None;
387
388 #
389 # Standard methods.
390 #
391
392 def fetchForListing(self, iStart, cMaxRows, tsNow, aiSortColumns = None):
393 """
394 Fetches test groups.
395
396 Returns an array (list) of TestGroupDataEx items, empty list if none.
397 Raises exception on error.
398 """
399 _ = aiSortColumns;
400 if tsNow is None:
401 self._oDb.execute('SELECT *\n'
402 'FROM TestGroups\n'
403 'WHERE tsExpire = \'infinity\'::TIMESTAMP\n'
404 'ORDER BY sName ASC\n'
405 'LIMIT %s OFFSET %s\n'
406 , (cMaxRows, iStart,));
407 else:
408 self._oDb.execute('SELECT *\n'
409 'FROM TestGroups\n'
410 'WHERE tsExpire > %s\n'
411 ' AND tsEffective <= %s\n'
412 'ORDER BY sName ASC\n'
413 'LIMIT %s OFFSET %s\n'
414 , (tsNow, tsNow, cMaxRows, iStart,));
415
416 aoRet = [];
417 for aoRow in self._oDb.fetchAll():
418 aoRet.append(TestGroupDataEx().initFromDbRowEx(aoRow, self._oDb, tsNow));
419 return aoRet;
420
421 def addEntry(self, oData, uidAuthor, fCommit = False):
422 """
423 Adds a testgroup to the database.
424 """
425
426 #
427 # Validate inputs.
428 #
429 assert isinstance(oData, TestGroupDataEx);
430 dErrors = oData.validateAndConvert(self._oDb, oData.ksValidateFor_Add);
431 if dErrors:
432 raise TMInvalidData('addEntry invalid input: %s' % (dErrors,));
433 self._assertUniq(oData, None);
434
435 #
436 # Do the job.
437 #
438 self._oDb.execute('INSERT INTO TestGroups (uidAuthor, sName, sDescription, sComment)\n'
439 'VALUES (%s, %s, %s, %s)\n'
440 'RETURNING idTestGroup\n'
441 , ( uidAuthor,
442 oData.sName,
443 oData.sDescription,
444 oData.sComment ));
445 idTestGroup = self._oDb.fetchOne()[0];
446 oData.idTestGroup = idTestGroup;
447
448 for oMember in oData.aoMembers:
449 oMember.idTestGroup = idTestGroup;
450 self._insertTestGroupMember(uidAuthor, oMember)
451
452 self._oDb.maybeCommit(fCommit);
453 return True;
454
455 def editEntry(self, oData, uidAuthor, fCommit = False):
456 """
457 Modifies a test group.
458 """
459
460 #
461 # Validate inputs and read in the old(/current) data.
462 #
463 assert isinstance(oData, TestGroupDataEx);
464 dErrors = oData.validateAndConvert(self._oDb, oData.ksValidateFor_Edit);
465 if dErrors:
466 raise TMInvalidData('editEntry invalid input: %s' % (dErrors,));
467 self._assertUniq(oData, oData.idTestGroup);
468
469 oOldData = TestGroupDataEx().initFromDbWithId(self._oDb, oData.idTestGroup);
470
471 #
472 # Update the data that needs updating.
473 #
474
475 if not oData.isEqualEx(oOldData, [ 'aoMembers', 'tsEffective', 'tsExpire', 'uidAuthor', ]):
476 self._historizeTestGroup(oData.idTestGroup);
477 self._oDb.execute('INSERT INTO TestGroups\n'
478 ' (uidAuthor, idTestGroup, sName, sDescription, sComment)\n'
479 'VALUES (%s, %s, %s, %s, %s)\n'
480 , ( uidAuthor,
481 oData.idTestGroup,
482 oData.sName,
483 oData.sDescription,
484 oData.sComment ));
485
486 # Create a lookup dictionary for old entries.
487 dOld = {};
488 for oOld in oOldData.aoMembers:
489 dOld[oOld.idTestCase] = oOld;
490 assert len(dOld) == len(oOldData.aoMembers);
491
492 # Add new members, updated existing ones.
493 dNew = {};
494 for oNewMember in oData.aoMembers:
495 oNewMember.idTestGroup = oData.idTestGroup;
496 if oNewMember.idTestCase in dNew:
497 raise TMRowAlreadyExists('Duplicate test group member: idTestCase=%d (%s / %s)'
498 % (oNewMember.idTestCase, oNewMember, dNew[oNewMember.idTestCase],));
499 dNew[oNewMember.idTestCase] = oNewMember;
500
501 oOldMember = dOld.get(oNewMember.idTestCase, None);
502 if oOldMember is not None:
503 if oNewMember.isEqualEx(oOldMember, [ 'uidAuthor', 'tsEffective', 'tsExpire' ]):
504 continue; # Skip, nothing changed.
505 self._historizeTestGroupMember(oData.idTestGroup, oNewMember.idTestCase);
506 self._insertTestGroupMember(uidAuthor, oNewMember);
507
508 # Expire members that have been removed.
509 sQuery = self._oDb.formatBindArgs('UPDATE TestGroupMembers\n'
510 'SET tsExpire = CURRENT_TIMESTAMP\n'
511 'WHERE idTestGroup = %s\n'
512 ' AND tsExpire = \'infinity\'::TIMESTAMP\n'
513 , ( oData.idTestGroup, ));
514 if dNew:
515 sQuery += ' AND idTestCase NOT IN (%s)' % (', '.join([str(iKey) for iKey in dNew]),);
516 self._oDb.execute(sQuery);
517
518 self._oDb.maybeCommit(fCommit);
519 return True;
520
521 def removeEntry(self, uidAuthor, idTestGroup, fCascade = False, fCommit = False):
522 """
523 Deletes a test group.
524 """
525 _ = uidAuthor; ## @todo record uidAuthor.
526
527 #
528 # Cascade.
529 #
530 if fCascade is not True:
531 self._oDb.execute('SELECT SchedGroups.idSchedGroup, SchedGroups.sName\n'
532 'FROM SchedGroupMembers, SchedGroups\n'
533 'WHERE SchedGroupMembers.idTestGroup = %s\n'
534 ' AND SchedGroupMembers.tsExpire = \'infinity\'::TIMESTAMP\n'
535 ' AND SchedGroups.idSchedGroup = SchedGroupMembers.idSchedGroup\n'
536 ' AND SchedGroups.tsExpire = \'infinity\'::TIMESTAMP\n'
537 , ( idTestGroup, ));
538 aoGroups = self._oDb.fetchAll();
539 if aoGroups:
540 asGroups = ['%s (#%d)' % (sName, idSchedGroup) for idSchedGroup, sName in aoGroups];
541 raise TMRowInUse('Test group #%d is member of one or more scheduling groups: %s'
542 % (idTestGroup, ', '.join(asGroups),));
543 else:
544 self._oDb.execute('UPDATE SchedGroupMembers\n'
545 'SET tsExpire = CURRENT_TIMESTAMP\n'
546 'WHERE idTestGroup = %s\n'
547 ' AND tsExpire = \'infinity\'::TIMESTAMP\n'
548 , ( idTestGroup, ));
549
550 #
551 # Remove the group.
552 #
553 self._oDb.execute('UPDATE TestGroupMembers\n'
554 'SET tsExpire = CURRENT_TIMESTAMP\n'
555 'WHERE idTestGroup = %s\n'
556 ' AND tsExpire = \'infinity\'::TIMESTAMP\n'
557 , (idTestGroup,))
558 self._oDb.execute('UPDATE TestGroups\n'
559 'SET tsExpire = CURRENT_TIMESTAMP\n'
560 'WHERE idTestGroup = %s\n'
561 ' AND tsExpire = \'infinity\'::TIMESTAMP\n'
562 , (idTestGroup,))
563
564 self._oDb.maybeCommit(fCommit)
565 return True;
566
567 def cachedLookup(self, idTestGroup):
568 """
569 Looks up the most recent TestGroupDataEx object for idTestGroup
570 via an object cache.
571
572 Returns a shared TestGroupDataEx object. None if not found.
573 Raises exception on DB error.
574 """
575 if self.dCache is None:
576 self.dCache = self._oDb.getCache('TestGroupDataEx');
577 oEntry = self.dCache.get(idTestGroup, None);
578 if oEntry is None:
579 fNeedTsNow = False;
580 self._oDb.execute('SELECT *\n'
581 'FROM TestGroups\n'
582 'WHERE idTestGroup = %s\n'
583 ' AND tsExpire = \'infinity\'::TIMESTAMP\n'
584 , (idTestGroup, ));
585 if self._oDb.getRowCount() == 0:
586 # Maybe it was deleted, try get the last entry.
587 self._oDb.execute('SELECT *\n'
588 'FROM TestGroups\n'
589 'WHERE idTestGroup = %s\n'
590 'ORDER BY tsExpire DESC\n'
591 'LIMIT 1\n'
592 , (idTestGroup, ));
593 fNeedTsNow = True;
594 elif self._oDb.getRowCount() > 1:
595 raise self._oDb.integrityException('%s infinity rows for %s' % (self._oDb.getRowCount(), idTestGroup));
596
597 if self._oDb.getRowCount() == 1:
598 aaoRow = self._oDb.fetchOne();
599 oEntry = TestGroupDataEx();
600 tsNow = oEntry.initFromDbRow(aaoRow).tsEffective if fNeedTsNow else None;
601 oEntry.initFromDbRowEx(aaoRow, self._oDb, tsNow);
602 self.dCache[idTestGroup] = oEntry;
603 return oEntry;
604
605
606 #
607 # Other methods.
608 #
609
610 def fetchOrderedByName(self, tsNow = None):
611 """
612 Return list of objects of type TestGroupData ordered by name.
613 May raise exception on database error.
614 """
615 if tsNow is None:
616 self._oDb.execute('SELECT *\n'
617 'FROM TestGroups\n'
618 'WHERE tsExpire = \'infinity\'::TIMESTAMP\n'
619 'ORDER BY sName ASC\n');
620 else:
621 self._oDb.execute('SELECT *\n'
622 'FROM TestGroups\n'
623 'WHERE tsExpire > %s\n'
624 ' AND tsEffective <= %s\n'
625 'ORDER BY sName ASC\n'
626 , (tsNow, tsNow,));
627 aoRet = []
628 for _ in range(self._oDb.getRowCount()):
629 aoRet.append(TestGroupData().initFromDbRow(self._oDb.fetchOne()));
630 return aoRet;
631
632 def getMembers(self, idTestGroup):
633 """
634 Fetches all test case records from DB which are
635 belong to current Test Group.
636 Returns list of objects of type TestGroupMemberData2 (!).
637 """
638 self._oDb.execute('SELECT TestCases.*,\n'
639 ' TestGroupMembers.idTestGroup,\n'
640 ' TestGroupMembers.aidTestCaseArgs\n'
641 'FROM TestCases, TestGroupMembers\n'
642 'WHERE TestCases.tsExpire = \'infinity\'::TIMESTAMP\n'
643 ' AND TestGroupMembers.tsExpire = \'infinity\'::TIMESTAMP\n'
644 ' AND TestGroupMembers.idTestCase = TestCases.idTestCase\n'
645 ' AND TestGroupMembers.idTestGroup = %s\n'
646 'ORDER BY TestCases.idTestCase ASC;',
647 (idTestGroup,))
648
649 aaoRows = self._oDb.fetchAll()
650 aoRet = []
651 for aoRow in aaoRows:
652 aoRet.append(TestGroupMemberData2().initFromDbRowEx(aoRow))
653
654 return aoRet
655
656 def getAll(self, tsNow=None):
657 """Return list of objects of type TestGroupData"""
658
659 if tsNow is None:
660 self._oDb.execute('SELECT *\n'
661 'FROM TestGroups\n'
662 'WHERE tsExpire = \'infinity\'::TIMESTAMP\n'
663 'ORDER BY idTestGroup ASC;')
664 else:
665 self._oDb.execute('SELECT *\n'
666 'FROM TestGroups\n'
667 'WHERE tsExpire > %s\n'
668 ' AND tsEffective <= %s\n'
669 'ORDER BY idTestGroup ASC;',
670 (tsNow, tsNow))
671
672 aaoRows = self._oDb.fetchAll()
673 aoRet = []
674 for aoRow in aaoRows:
675 aoRet.append(TestGroupData().initFromDbRow(aoRow))
676
677 return aoRet
678
679 def getById(self, idTestGroup, tsNow=None):
680 """Get Test Group data by its ID"""
681
682 if tsNow is None:
683 self._oDb.execute('SELECT *\n'
684 'FROM TestGroups\n'
685 'WHERE tsExpire = \'infinity\'::timestamp\n'
686 ' AND idTestGroup = %s\n'
687 'ORDER BY idTestGroup ASC;'
688 , (idTestGroup,))
689 else:
690 self._oDb.execute('SELECT *\n'
691 'FROM TestGroups\n'
692 'WHERE tsExpire > %s\n'
693 ' AND tsEffective <= %s\n'
694 ' AND idTestGroup = %s\n'
695 'ORDER BY idTestGroup ASC;'
696 , (tsNow, tsNow, idTestGroup))
697
698 aRows = self._oDb.fetchAll()
699 if len(aRows) not in (0, 1):
700 raise TMTooManyRows('Found more than one test groups with the same credentials. Database structure is corrupted.')
701 try:
702 return TestGroupData().initFromDbRow(aRows[0])
703 except IndexError:
704 return None
705
706 #
707 # Helpers.
708 #
709
710 def _assertUniq(self, oData, idTestGroupIgnore):
711 """ Checks that the test group name is unique, raises exception if it isn't. """
712 self._oDb.execute('SELECT idTestGroup\n'
713 'FROM TestGroups\n'
714 'WHERE sName = %s\n'
715 ' AND tsExpire = \'infinity\'::TIMESTAMP\n'
716 + ('' if idTestGroupIgnore is None else ' AND idTestGroup <> %d\n' % (idTestGroupIgnore,))
717 , ( oData.sName, ))
718 if self._oDb.getRowCount() > 0:
719 raise TMRowAlreadyExists('A Test group with name "%s" already exist.' % (oData.sName,));
720 return True;
721
722 def _historizeTestGroup(self, idTestGroup):
723 """ Historize Test Group record. """
724 self._oDb.execute('UPDATE TestGroups\n'
725 'SET tsExpire = CURRENT_TIMESTAMP\n'
726 'WHERE idTestGroup = %s\n'
727 ' AND tsExpire = \'infinity\'::TIMESTAMP\n'
728 , ( idTestGroup, ));
729 return True;
730
731 def _historizeTestGroupMember(self, idTestGroup, idTestCase):
732 """ Historize Test Group Member record. """
733 self._oDb.execute('UPDATE TestGroupMembers\n'
734 'SET tsExpire = CURRENT_TIMESTAMP\n'
735 'WHERE idTestGroup = %s\n'
736 ' AND idTestCase = %s\n'
737 ' AND tsExpire = \'infinity\'::timestamp\n'
738 , (idTestGroup, idTestCase,));
739 return True;
740
741 def _insertTestGroupMember(self, uidAuthor, oMember):
742 """ Inserts a test group member. """
743 self._oDb.execute('INSERT INTO TestGroupMembers\n'
744 ' (uidAuthor, idTestGroup, idTestCase, iSchedPriority, aidTestCaseArgs)\n'
745 'VALUES (%s, %s, %s, %s, %s)\n'
746 , ( uidAuthor,
747 oMember.idTestGroup,
748 oMember.idTestCase,
749 oMember.iSchedPriority,
750 oMember.aidTestCaseArgs, ));
751 return True;
752
753
754
755#
756# Unit testing.
757#
758
759# pylint: disable=missing-docstring
760class TestGroupMemberDataTestCase(ModelDataBaseTestCase):
761 def setUp(self):
762 self.aoSamples = [TestGroupMemberData(),];
763
764class TestGroupDataTestCase(ModelDataBaseTestCase):
765 def setUp(self):
766 self.aoSamples = [TestGroupData(),];
767
768if __name__ == '__main__':
769 unittest.main();
770 # not reached.
771
Note: See TracBrowser for help on using the repository browser.

© 2024 Oracle Support Privacy / Do Not Sell My Info Terms of Use Trademark Policy Automated Access Etiquette