VirtualBox

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

Last change on this file since 98655 was 98655, checked in by vboxsync, 22 months ago

ValKit: Pylint 2.16.2 adjustments.

  • 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 98655 2023-02-20 15:05:40Z vboxsync $
3
4"""
5Test Manager - Test groups management.
6"""
7
8__copyright__ = \
9"""
10Copyright (C) 2012-2023 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: 98655 $"
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, idTestCase in enumerate(aoNewMembers):
370 for j in range(i + 1, len(aoNewMembers)):
371 if aoNewMembers[j].idTestCase == idTestCase:
372 asErrors.append('Duplicate testcase #%d!' % (idTestCase, ));
373 break;
374
375 return (aoNewMembers, None if not asErrors else '<br>\n'.join(asErrors));
376
377
378class TestGroupLogic(ModelLogicBase):
379 """
380 Test case management logic.
381 """
382
383 def __init__(self, oDb):
384 ModelLogicBase.__init__(self, oDb)
385 self.dCache = None;
386
387 #
388 # Standard methods.
389 #
390
391 def fetchForListing(self, iStart, cMaxRows, tsNow, aiSortColumns = None):
392 """
393 Fetches test groups.
394
395 Returns an array (list) of TestGroupDataEx items, empty list if none.
396 Raises exception on error.
397 """
398 _ = aiSortColumns;
399 if tsNow is None:
400 self._oDb.execute('SELECT *\n'
401 'FROM TestGroups\n'
402 'WHERE tsExpire = \'infinity\'::TIMESTAMP\n'
403 'ORDER BY sName ASC\n'
404 'LIMIT %s OFFSET %s\n'
405 , (cMaxRows, iStart,));
406 else:
407 self._oDb.execute('SELECT *\n'
408 'FROM TestGroups\n'
409 'WHERE tsExpire > %s\n'
410 ' AND tsEffective <= %s\n'
411 'ORDER BY sName ASC\n'
412 'LIMIT %s OFFSET %s\n'
413 , (tsNow, tsNow, cMaxRows, iStart,));
414
415 aoRet = [];
416 for aoRow in self._oDb.fetchAll():
417 aoRet.append(TestGroupDataEx().initFromDbRowEx(aoRow, self._oDb, tsNow));
418 return aoRet;
419
420 def addEntry(self, oData, uidAuthor, fCommit = False):
421 """
422 Adds a testgroup to the database.
423 """
424
425 #
426 # Validate inputs.
427 #
428 assert isinstance(oData, TestGroupDataEx);
429 dErrors = oData.validateAndConvert(self._oDb, oData.ksValidateFor_Add);
430 if dErrors:
431 raise TMInvalidData('addEntry invalid input: %s' % (dErrors,));
432 self._assertUniq(oData, None);
433
434 #
435 # Do the job.
436 #
437 self._oDb.execute('INSERT INTO TestGroups (uidAuthor, sName, sDescription, sComment)\n'
438 'VALUES (%s, %s, %s, %s)\n'
439 'RETURNING idTestGroup\n'
440 , ( uidAuthor,
441 oData.sName,
442 oData.sDescription,
443 oData.sComment ));
444 idTestGroup = self._oDb.fetchOne()[0];
445 oData.idTestGroup = idTestGroup;
446
447 for oMember in oData.aoMembers:
448 oMember.idTestGroup = idTestGroup;
449 self._insertTestGroupMember(uidAuthor, oMember)
450
451 self._oDb.maybeCommit(fCommit);
452 return True;
453
454 def editEntry(self, oData, uidAuthor, fCommit = False):
455 """
456 Modifies a test group.
457 """
458
459 #
460 # Validate inputs and read in the old(/current) data.
461 #
462 assert isinstance(oData, TestGroupDataEx);
463 dErrors = oData.validateAndConvert(self._oDb, oData.ksValidateFor_Edit);
464 if dErrors:
465 raise TMInvalidData('editEntry invalid input: %s' % (dErrors,));
466 self._assertUniq(oData, oData.idTestGroup);
467
468 oOldData = TestGroupDataEx().initFromDbWithId(self._oDb, oData.idTestGroup);
469
470 #
471 # Update the data that needs updating.
472 #
473
474 if not oData.isEqualEx(oOldData, [ 'aoMembers', 'tsEffective', 'tsExpire', 'uidAuthor', ]):
475 self._historizeTestGroup(oData.idTestGroup);
476 self._oDb.execute('INSERT INTO TestGroups\n'
477 ' (uidAuthor, idTestGroup, sName, sDescription, sComment)\n'
478 'VALUES (%s, %s, %s, %s, %s)\n'
479 , ( uidAuthor,
480 oData.idTestGroup,
481 oData.sName,
482 oData.sDescription,
483 oData.sComment ));
484
485 # Create a lookup dictionary for old entries.
486 dOld = {};
487 for oOld in oOldData.aoMembers:
488 dOld[oOld.idTestCase] = oOld;
489 assert len(dOld) == len(oOldData.aoMembers);
490
491 # Add new members, updated existing ones.
492 dNew = {};
493 for oNewMember in oData.aoMembers:
494 oNewMember.idTestGroup = oData.idTestGroup;
495 if oNewMember.idTestCase in dNew:
496 raise TMRowAlreadyExists('Duplicate test group member: idTestCase=%d (%s / %s)'
497 % (oNewMember.idTestCase, oNewMember, dNew[oNewMember.idTestCase],));
498 dNew[oNewMember.idTestCase] = oNewMember;
499
500 oOldMember = dOld.get(oNewMember.idTestCase, None);
501 if oOldMember is not None:
502 if oNewMember.isEqualEx(oOldMember, [ 'uidAuthor', 'tsEffective', 'tsExpire' ]):
503 continue; # Skip, nothing changed.
504 self._historizeTestGroupMember(oData.idTestGroup, oNewMember.idTestCase);
505 self._insertTestGroupMember(uidAuthor, oNewMember);
506
507 # Expire members that have been removed.
508 sQuery = self._oDb.formatBindArgs('UPDATE TestGroupMembers\n'
509 'SET tsExpire = CURRENT_TIMESTAMP\n'
510 'WHERE idTestGroup = %s\n'
511 ' AND tsExpire = \'infinity\'::TIMESTAMP\n'
512 , ( oData.idTestGroup, ));
513 if dNew:
514 sQuery += ' AND idTestCase NOT IN (%s)' % (', '.join([str(iKey) for iKey in dNew]),);
515 self._oDb.execute(sQuery);
516
517 self._oDb.maybeCommit(fCommit);
518 return True;
519
520 def removeEntry(self, uidAuthor, idTestGroup, fCascade = False, fCommit = False):
521 """
522 Deletes a test group.
523 """
524 _ = uidAuthor; ## @todo record uidAuthor.
525
526 #
527 # Cascade.
528 #
529 if fCascade is not True:
530 self._oDb.execute('SELECT SchedGroups.idSchedGroup, SchedGroups.sName\n'
531 'FROM SchedGroupMembers, SchedGroups\n'
532 'WHERE SchedGroupMembers.idTestGroup = %s\n'
533 ' AND SchedGroupMembers.tsExpire = \'infinity\'::TIMESTAMP\n'
534 ' AND SchedGroups.idSchedGroup = SchedGroupMembers.idSchedGroup\n'
535 ' AND SchedGroups.tsExpire = \'infinity\'::TIMESTAMP\n'
536 , ( idTestGroup, ));
537 aoGroups = self._oDb.fetchAll();
538 if aoGroups:
539 asGroups = ['%s (#%d)' % (sName, idSchedGroup) for idSchedGroup, sName in aoGroups];
540 raise TMRowInUse('Test group #%d is member of one or more scheduling groups: %s'
541 % (idTestGroup, ', '.join(asGroups),));
542 else:
543 self._oDb.execute('UPDATE SchedGroupMembers\n'
544 'SET tsExpire = CURRENT_TIMESTAMP\n'
545 'WHERE idTestGroup = %s\n'
546 ' AND tsExpire = \'infinity\'::TIMESTAMP\n'
547 , ( idTestGroup, ));
548
549 #
550 # Remove the group.
551 #
552 self._oDb.execute('UPDATE TestGroupMembers\n'
553 'SET tsExpire = CURRENT_TIMESTAMP\n'
554 'WHERE idTestGroup = %s\n'
555 ' AND tsExpire = \'infinity\'::TIMESTAMP\n'
556 , (idTestGroup,))
557 self._oDb.execute('UPDATE TestGroups\n'
558 'SET tsExpire = CURRENT_TIMESTAMP\n'
559 'WHERE idTestGroup = %s\n'
560 ' AND tsExpire = \'infinity\'::TIMESTAMP\n'
561 , (idTestGroup,))
562
563 self._oDb.maybeCommit(fCommit)
564 return True;
565
566 def cachedLookup(self, idTestGroup):
567 """
568 Looks up the most recent TestGroupDataEx object for idTestGroup
569 via an object cache.
570
571 Returns a shared TestGroupDataEx object. None if not found.
572 Raises exception on DB error.
573 """
574 if self.dCache is None:
575 self.dCache = self._oDb.getCache('TestGroupDataEx');
576 oEntry = self.dCache.get(idTestGroup, None);
577 if oEntry is None:
578 fNeedTsNow = False;
579 self._oDb.execute('SELECT *\n'
580 'FROM TestGroups\n'
581 'WHERE idTestGroup = %s\n'
582 ' AND tsExpire = \'infinity\'::TIMESTAMP\n'
583 , (idTestGroup, ));
584 if self._oDb.getRowCount() == 0:
585 # Maybe it was deleted, try get the last entry.
586 self._oDb.execute('SELECT *\n'
587 'FROM TestGroups\n'
588 'WHERE idTestGroup = %s\n'
589 'ORDER BY tsExpire DESC\n'
590 'LIMIT 1\n'
591 , (idTestGroup, ));
592 fNeedTsNow = True;
593 elif self._oDb.getRowCount() > 1:
594 raise self._oDb.integrityException('%s infinity rows for %s' % (self._oDb.getRowCount(), idTestGroup));
595
596 if self._oDb.getRowCount() == 1:
597 aaoRow = self._oDb.fetchOne();
598 oEntry = TestGroupDataEx();
599 tsNow = oEntry.initFromDbRow(aaoRow).tsEffective if fNeedTsNow else None;
600 oEntry.initFromDbRowEx(aaoRow, self._oDb, tsNow);
601 self.dCache[idTestGroup] = oEntry;
602 return oEntry;
603
604
605 #
606 # Other methods.
607 #
608
609 def fetchOrderedByName(self, tsNow = None):
610 """
611 Return list of objects of type TestGroupData ordered by name.
612 May raise exception on database error.
613 """
614 if tsNow is None:
615 self._oDb.execute('SELECT *\n'
616 'FROM TestGroups\n'
617 'WHERE tsExpire = \'infinity\'::TIMESTAMP\n'
618 'ORDER BY sName ASC\n');
619 else:
620 self._oDb.execute('SELECT *\n'
621 'FROM TestGroups\n'
622 'WHERE tsExpire > %s\n'
623 ' AND tsEffective <= %s\n'
624 'ORDER BY sName ASC\n'
625 , (tsNow, tsNow,));
626 aoRet = []
627 for _ in range(self._oDb.getRowCount()):
628 aoRet.append(TestGroupData().initFromDbRow(self._oDb.fetchOne()));
629 return aoRet;
630
631 def getMembers(self, idTestGroup):
632 """
633 Fetches all test case records from DB which are
634 belong to current Test Group.
635 Returns list of objects of type TestGroupMemberData2 (!).
636 """
637 self._oDb.execute('SELECT TestCases.*,\n'
638 ' TestGroupMembers.idTestGroup,\n'
639 ' TestGroupMembers.aidTestCaseArgs\n'
640 'FROM TestCases, TestGroupMembers\n'
641 'WHERE TestCases.tsExpire = \'infinity\'::TIMESTAMP\n'
642 ' AND TestGroupMembers.tsExpire = \'infinity\'::TIMESTAMP\n'
643 ' AND TestGroupMembers.idTestCase = TestCases.idTestCase\n'
644 ' AND TestGroupMembers.idTestGroup = %s\n'
645 'ORDER BY TestCases.idTestCase ASC;',
646 (idTestGroup,))
647
648 aaoRows = self._oDb.fetchAll()
649 aoRet = []
650 for aoRow in aaoRows:
651 aoRet.append(TestGroupMemberData2().initFromDbRowEx(aoRow))
652
653 return aoRet
654
655 def getAll(self, tsNow=None):
656 """Return list of objects of type TestGroupData"""
657
658 if tsNow is None:
659 self._oDb.execute('SELECT *\n'
660 'FROM TestGroups\n'
661 'WHERE tsExpire = \'infinity\'::TIMESTAMP\n'
662 'ORDER BY idTestGroup ASC;')
663 else:
664 self._oDb.execute('SELECT *\n'
665 'FROM TestGroups\n'
666 'WHERE tsExpire > %s\n'
667 ' AND tsEffective <= %s\n'
668 'ORDER BY idTestGroup ASC;',
669 (tsNow, tsNow))
670
671 aaoRows = self._oDb.fetchAll()
672 aoRet = []
673 for aoRow in aaoRows:
674 aoRet.append(TestGroupData().initFromDbRow(aoRow))
675
676 return aoRet
677
678 def getById(self, idTestGroup, tsNow=None):
679 """Get Test Group data by its ID"""
680
681 if tsNow is None:
682 self._oDb.execute('SELECT *\n'
683 'FROM TestGroups\n'
684 'WHERE tsExpire = \'infinity\'::timestamp\n'
685 ' AND idTestGroup = %s\n'
686 'ORDER BY idTestGroup ASC;'
687 , (idTestGroup,))
688 else:
689 self._oDb.execute('SELECT *\n'
690 'FROM TestGroups\n'
691 'WHERE tsExpire > %s\n'
692 ' AND tsEffective <= %s\n'
693 ' AND idTestGroup = %s\n'
694 'ORDER BY idTestGroup ASC;'
695 , (tsNow, tsNow, idTestGroup))
696
697 aRows = self._oDb.fetchAll()
698 if len(aRows) not in (0, 1):
699 raise TMTooManyRows('Found more than one test groups with the same credentials. Database structure is corrupted.')
700 try:
701 return TestGroupData().initFromDbRow(aRows[0])
702 except IndexError:
703 return None
704
705 #
706 # Helpers.
707 #
708
709 def _assertUniq(self, oData, idTestGroupIgnore):
710 """ Checks that the test group name is unique, raises exception if it isn't. """
711 self._oDb.execute('SELECT idTestGroup\n'
712 'FROM TestGroups\n'
713 'WHERE sName = %s\n'
714 ' AND tsExpire = \'infinity\'::TIMESTAMP\n'
715 + ('' if idTestGroupIgnore is None else ' AND idTestGroup <> %d\n' % (idTestGroupIgnore,))
716 , ( oData.sName, ))
717 if self._oDb.getRowCount() > 0:
718 raise TMRowAlreadyExists('A Test group with name "%s" already exist.' % (oData.sName,));
719 return True;
720
721 def _historizeTestGroup(self, idTestGroup):
722 """ Historize Test Group record. """
723 self._oDb.execute('UPDATE TestGroups\n'
724 'SET tsExpire = CURRENT_TIMESTAMP\n'
725 'WHERE idTestGroup = %s\n'
726 ' AND tsExpire = \'infinity\'::TIMESTAMP\n'
727 , ( idTestGroup, ));
728 return True;
729
730 def _historizeTestGroupMember(self, idTestGroup, idTestCase):
731 """ Historize Test Group Member record. """
732 self._oDb.execute('UPDATE TestGroupMembers\n'
733 'SET tsExpire = CURRENT_TIMESTAMP\n'
734 'WHERE idTestGroup = %s\n'
735 ' AND idTestCase = %s\n'
736 ' AND tsExpire = \'infinity\'::timestamp\n'
737 , (idTestGroup, idTestCase,));
738 return True;
739
740 def _insertTestGroupMember(self, uidAuthor, oMember):
741 """ Inserts a test group member. """
742 self._oDb.execute('INSERT INTO TestGroupMembers\n'
743 ' (uidAuthor, idTestGroup, idTestCase, iSchedPriority, aidTestCaseArgs)\n'
744 'VALUES (%s, %s, %s, %s, %s)\n'
745 , ( uidAuthor,
746 oMember.idTestGroup,
747 oMember.idTestCase,
748 oMember.iSchedPriority,
749 oMember.aidTestCaseArgs, ));
750 return True;
751
752
753
754#
755# Unit testing.
756#
757
758# pylint: disable=missing-docstring
759class TestGroupMemberDataTestCase(ModelDataBaseTestCase):
760 def setUp(self):
761 self.aoSamples = [TestGroupMemberData(),];
762
763class TestGroupDataTestCase(ModelDataBaseTestCase):
764 def setUp(self):
765 self.aoSamples = [TestGroupData(),];
766
767if __name__ == '__main__':
768 unittest.main();
769 # not reached.
770
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