# -*- coding: utf-8 -*- # $Id: testresultfailures.py $ # pylint: disable=C0302 ## @todo Rename this file to testresult.py! """ Test Manager - Test result failures. """ __copyright__ = \ """ Copyright (C) 2012-2019 Oracle Corporation This file is part of VirtualBox Open Source Edition (OSE), as available from http://www.virtualbox.org. This file is free software; you can redistribute it and/or modify it under the terms of the GNU General Public License (GPL) as published by the Free Software Foundation, in version 2 as it comes in the "COPYING" file of the VirtualBox OSE distribution. VirtualBox OSE is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY of any kind. The contents of this file may alternatively be used under the terms of the Common Development and Distribution License Version 1.0 (CDDL) only, as it comes in the "COPYING.CDDL" file of the VirtualBox OSE distribution, in which case the provisions of the CDDL are applicable instead of those of the GPL. You may elect to license modified versions of this file under the terms and conditions of either the GPL or the CDDL or both. """ __version__ = "$Revision: 127855 $" # Standard python imports. import sys; import unittest; # Validation Kit imports. from testmanager.core.base import ModelDataBase, ModelLogicBase, ModelDataBaseTestCase, TMInvalidData, TMRowNotFound, \ TMRowAlreadyExists, ChangeLogEntry, AttributeChangeEntry; from testmanager.core.failurereason import FailureReasonData; from testmanager.core.useraccount import UserAccountLogic; # Python 3 hacks: if sys.version_info[0] >= 3: xrange = range; # pylint: disable=redefined-builtin,invalid-name class TestResultFailureData(ModelDataBase): """ Test result failure reason data. """ ksIdAttr = 'idTestResult'; kfIdAttrIsForForeign = True; # Modifies the 'add' validation. ksParam_idTestResult = 'TestResultFailure_idTestResult'; ksParam_tsEffective = 'TestResultFailure_tsEffective'; ksParam_tsExpire = 'TestResultFailure_tsExpire'; ksParam_uidAuthor = 'TestResultFailure_uidAuthor'; ksParam_idTestSet = 'TestResultFailure_idTestSet'; ksParam_idFailureReason = 'TestResultFailure_idFailureReason'; ksParam_sComment = 'TestResultFailure_sComment'; kasAllowNullAttributes = ['tsEffective', 'tsExpire', 'uidAuthor', 'sComment', 'idTestSet' ]; kcDbColumns = 7; def __init__(self): ModelDataBase.__init__(self) self.idTestResult = None; self.tsEffective = None; self.tsExpire = None; self.uidAuthor = None; self.idTestSet = None; self.idFailureReason = None; self.sComment = None; def initFromDbRow(self, aoRow): """ Reinitialize from a SELECT * FROM TestResultFailures. Return self. Raises exception if no row. """ if aoRow is None: raise TMRowNotFound('Test result file record not found.') self.idTestResult = aoRow[0]; self.tsEffective = aoRow[1]; self.tsExpire = aoRow[2]; self.uidAuthor = aoRow[3]; self.idTestSet = aoRow[4]; self.idFailureReason = aoRow[5]; self.sComment = aoRow[6]; return self; def initFromDbWithId(self, oDb, idTestResult, tsNow = None, sPeriodBack = None): """ Initialize the object from the database. """ oDb.execute(self.formatSimpleNowAndPeriodQuery(oDb, 'SELECT *\n' 'FROM TestResultFailures\n' 'WHERE idTestResult = %s\n' , ( idTestResult,), tsNow, sPeriodBack)); aoRow = oDb.fetchOne() if aoRow is None: raise TMRowNotFound('idTestResult=%s not found (tsNow=%s, sPeriodBack=%s)' % (idTestResult, tsNow, sPeriodBack)); assert len(aoRow) == self.kcDbColumns; return self.initFromDbRow(aoRow); def initFromValues(self, idTestResult, idFailureReason, uidAuthor, tsExpire = None, tsEffective = None, idTestSet = None, sComment = None): """ Initialize from values. """ self.idTestResult = idTestResult; self.tsEffective = tsEffective; self.tsExpire = tsExpire; self.uidAuthor = uidAuthor; self.idTestSet = idTestSet; self.idFailureReason = idFailureReason; self.sComment = sComment; return self; class TestResultFailureDataEx(TestResultFailureData): """ Extends TestResultFailureData by resolving reasons and user. """ def __init__(self): TestResultFailureData.__init__(self); self.oFailureReason = None; self.oAuthor = None; def initFromDbRowEx(self, aoRow, oFailureReasonLogic, oUserAccountLogic): """ Reinitialize from a SELECT * FROM TestResultFailures. Return self. Raises exception if no row. """ self.initFromDbRow(aoRow); self.oFailureReason = oFailureReasonLogic.cachedLookup(self.idFailureReason); self.oAuthor = oUserAccountLogic.cachedLookup(self.uidAuthor); return self; class TestResultListingData(ModelDataBase): # pylint: disable=R0902 """ Test case result data representation for table listing """ def __init__(self): """Initialize""" ModelDataBase.__init__(self) self.idTestSet = None self.idBuildCategory = None; self.sProduct = None self.sRepository = None; self.sBranch = None self.sType = None self.idBuild = None; self.sVersion = None; self.iRevision = None self.sOs = None; self.sOsVersion = None; self.sArch = None; self.sCpuVendor = None; self.sCpuName = None; self.cCpus = None; self.fCpuHwVirt = None; self.fCpuNestedPaging = None; self.fCpu64BitGuest = None; self.idTestBox = None self.sTestBoxName = None self.tsCreated = None self.tsElapsed = None self.enmStatus = None self.cErrors = None; self.idTestCase = None self.sTestCaseName = None self.sBaseCmd = None self.sArgs = None self.sSubName = None; self.idBuildTestSuite = None; self.iRevisionTestSuite = None; self.oFailureReason = None; self.oFailureReasonAssigner = None; self.tsFailureReasonAssigned = None; self.sFailureReasonComment = None; def initFromDbRowEx(self, aoRow, oFailureReasonLogic, oUserAccountLogic): """ Reinitialize from a database query. Return self. Raises exception if no row. """ if aoRow is None: raise TMRowNotFound('Test result record not found.') self.idTestSet = aoRow[0]; self.idBuildCategory = aoRow[1]; self.sProduct = aoRow[2]; self.sRepository = aoRow[3]; self.sBranch = aoRow[4]; self.sType = aoRow[5]; self.idBuild = aoRow[6]; self.sVersion = aoRow[7]; self.iRevision = aoRow[8]; self.sOs = aoRow[9]; self.sOsVersion = aoRow[10]; self.sArch = aoRow[11]; self.sCpuVendor = aoRow[12]; self.sCpuName = aoRow[13]; self.cCpus = aoRow[14]; self.fCpuHwVirt = aoRow[15]; self.fCpuNestedPaging = aoRow[16]; self.fCpu64BitGuest = aoRow[17]; self.idTestBox = aoRow[18]; self.sTestBoxName = aoRow[19]; self.tsCreated = aoRow[20]; self.tsElapsed = aoRow[21]; self.enmStatus = aoRow[22]; self.cErrors = aoRow[23]; self.idTestCase = aoRow[24]; self.sTestCaseName = aoRow[25]; self.sBaseCmd = aoRow[26]; self.sArgs = aoRow[27]; self.sSubName = aoRow[28]; self.idBuildTestSuite = aoRow[29]; self.iRevisionTestSuite = aoRow[30]; self.oFailureReason = None; if aoRow[31] is not None: self.oFailureReason = oFailureReasonLogic.cachedLookup(aoRow[31]); self.oFailureReasonAssigner = None; if aoRow[32] is not None: self.oFailureReasonAssigner = oUserAccountLogic.cachedLookup(aoRow[32]); self.tsFailureReasonAssigned = aoRow[33]; self.sFailureReasonComment = aoRow[34]; return self class TestResultFailureLogic(ModelLogicBase): # pylint: disable=R0903 """ Test result failure reason logic. """ def __init__(self, oDb): ModelLogicBase.__init__(self, oDb) def fetchForChangeLog(self, idTestResult, iStart, cMaxRows, tsNow): # pylint: disable=R0914 """ Fetches change log entries for a failure reason. Returns an array of ChangeLogEntry instance and an indicator whether there are more entries. Raises exception on error. """ if tsNow is None: tsNow = self._oDb.getCurrentTimestamp(); # 1. Get a list of the changes from both TestResultFailures and assoicated # FailureReasons. The latter is useful since the failure reason # description may evolve along side the invidiual failure analysis. self._oDb.execute('( SELECT trf.tsEffective AS tsEffectiveChangeLog,\n' ' trf.uidAuthor AS uidAuthorChangeLog,\n' ' trf.*,\n' ' fr.*\n' ' FROM TestResultFailures trf,\n' ' FailureReasons fr\n' ' WHERE trf.idTestResult = %s\n' ' AND trf.tsEffective <= %s\n' ' AND trf.idFailureReason = fr.idFailureReason\n' ' AND fr.tsEffective <= trf.tsEffective\n' ' AND fr.tsExpire > trf.tsEffective\n' ')\n' 'UNION\n' '( SELECT fr.tsEffective AS tsEffectiveChangeLog,\n' ' fr.uidAuthor AS uidAuthorChangeLog,\n' ' trf.*,\n' ' fr.*\n' ' FROM TestResultFailures trf,\n' ' FailureReasons fr\n' ' WHERE trf.idTestResult = %s\n' ' AND trf.tsEffective <= %s\n' ' AND trf.idFailureReason = fr.idFailureReason\n' ' AND fr.tsEffective > trf.tsEffective\n' ' AND fr.tsEffective < trf.tsExpire\n' ')\n' 'ORDER BY tsEffectiveChangeLog DESC\n' 'LIMIT %s OFFSET %s\n' , ( idTestResult, tsNow, idTestResult, tsNow, cMaxRows + 1, iStart, )); aaoRows = []; for aoChange in self._oDb.fetchAll(): oTrf = TestResultFailureDataEx().initFromDbRow(aoChange[2:]); oFr = FailureReasonData().initFromDbRow(aoChange[(2+TestResultFailureData.kcDbColumns):]); oTrf.oFailureReason = oFr; aaoRows.append([aoChange[0], aoChange[1], oTrf, oFr]); # 2. Calculate the changes. oFailureCategoryLogic = None; aoEntries = []; for i in xrange(0, len(aaoRows) - 1): aoNew = aaoRows[i]; aoOld = aaoRows[i + 1]; aoChanges = []; oNew = aoNew[2]; oOld = aoOld[2]; for sAttr in oNew.getDataAttributes(): if sAttr not in [ 'tsEffective', 'tsExpire', 'uidAuthor', 'oFailureReason', 'oAuthor' ]: oOldAttr = getattr(oOld, sAttr); oNewAttr = getattr(oNew, sAttr); if oOldAttr != oNewAttr: if sAttr == 'idFailureReason': oNewAttr = '%s (%s)' % (oNewAttr, oNew.oFailureReason.sShort, ); oOldAttr = '%s (%s)' % (oOldAttr, oOld.oFailureReason.sShort, ); aoChanges.append(AttributeChangeEntry(sAttr, oNewAttr, oOldAttr, str(oNewAttr), str(oOldAttr))); if oOld.idFailureReason == oNew.idFailureReason: oNew = aoNew[3]; oOld = aoOld[3]; for sAttr in oNew.getDataAttributes(): if sAttr not in [ 'tsEffective', 'tsExpire', 'uidAuthor', ]: oOldAttr = getattr(oOld, sAttr); oNewAttr = getattr(oNew, sAttr); if oOldAttr != oNewAttr: if sAttr == 'idFailureCategory': if oFailureCategoryLogic is None: from testmanager.core.failurecategory import FailureCategoryLogic; oFailureCategoryLogic = FailureCategoryLogic(self._oDb); oCat = oFailureCategoryLogic.cachedLookup(oNewAttr); if oCat is not None: oNewAttr = '%s (%s)' % (oNewAttr, oCat.sShort, ); oCat = oFailureCategoryLogic.cachedLookup(oOldAttr); if oCat is not None: oOldAttr = '%s (%s)' % (oOldAttr, oCat.sShort, ); aoChanges.append(AttributeChangeEntry(sAttr, oNewAttr, oOldAttr, str(oNewAttr), str(oOldAttr))); tsExpire = aaoRows[i - 1][0] if i > 0 else aoNew[2].tsExpire; aoEntries.append(ChangeLogEntry(aoNew[1], None, aoNew[0], tsExpire, aoNew[2], aoOld[2], aoChanges)); # If we're at the end of the log, add the initial entry. if len(aaoRows) <= cMaxRows and aaoRows: aoNew = aaoRows[-1]; tsExpire = aaoRows[-1 - 1][0] if len(aaoRows) > 1 else aoNew[2].tsExpire; aoEntries.append(ChangeLogEntry(aoNew[1], None, aoNew[0], tsExpire, aoNew[2], None, [])); return (UserAccountLogic(self._oDb).resolveChangeLogAuthors(aoEntries), len(aaoRows) > cMaxRows); def getById(self, idTestResult): """Get Test result failure reason data by idTestResult""" self._oDb.execute('SELECT *\n' 'FROM TestResultFailures\n' 'WHERE tsExpire = \'infinity\'::timestamp\n' ' AND idTestResult = %s;', (idTestResult,)) aRows = self._oDb.fetchAll() if len(aRows) not in (0, 1): raise self._oDb.integrityException( 'Found more than one failure reasons with the same credentials. Database structure is corrupted.') try: return TestResultFailureData().initFromDbRow(aRows[0]) except IndexError: return None def addEntry(self, oData, uidAuthor, fCommit = False): """ Add a test result failure reason record. """ # # Validate inputs and read in the old(/current) data. # assert isinstance(oData, TestResultFailureData); dErrors = oData.validateAndConvert(self._oDb, oData.ksValidateFor_AddForeignId); if dErrors: raise TMInvalidData('editEntry invalid input: %s' % (dErrors,)); # Check if it exist first (we're adding, not editing, collisions not allowed). oOldData = self.getById(oData.idTestResult); if oOldData is not None: raise TMRowAlreadyExists('TestResult %d already have a failure reason associated with it:' '%s\n' 'Perhaps someone else beat you to it? Or did you try resubmit?' % (oData.idTestResult, oOldData)); oData = self._resolveSetTestIdIfMissing(oData); # # Add record. # self._readdEntry(uidAuthor, oData); self._oDb.maybeCommit(fCommit); return True; def editEntry(self, oData, uidAuthor, fCommit = False): """ Modifies a test result failure reason. """ # # Validate inputs and read in the old(/current) data. # assert isinstance(oData, TestResultFailureData); dErrors = oData.validateAndConvert(self._oDb, oData.ksValidateFor_Edit); if dErrors: raise TMInvalidData('editEntry invalid input: %s' % (dErrors,)); oOldData = self.getById(oData.idTestResult) oData.idTestSet = oOldData.idTestSet; # # Update the data that needs updating. # if not oData.isEqualEx(oOldData, [ 'tsEffective', 'tsExpire', 'uidAuthor', ]): self._historizeEntry(oData.idTestResult); self._readdEntry(uidAuthor, oData); self._oDb.maybeCommit(fCommit); return True; def removeEntry(self, uidAuthor, idTestResult, fCascade = False, fCommit = False): """ Deletes a test result failure reason. """ _ = fCascade; # Not applicable. oData = self.getById(idTestResult) (tsCur, tsCurMinusOne) = self._oDb.getCurrentTimestamps(); if oData.tsEffective != tsCur and oData.tsEffective != tsCurMinusOne: self._historizeEntry(idTestResult, tsCurMinusOne); self._readdEntry(uidAuthor, oData, tsCurMinusOne); self._historizeEntry(idTestResult); self._oDb.execute('UPDATE TestResultFailures\n' 'SET tsExpire = CURRENT_TIMESTAMP\n' 'WHERE idTestResult = %s\n' ' AND tsExpire = \'infinity\'::TIMESTAMP\n' , (idTestResult,)); self._oDb.maybeCommit(fCommit); return True; # # Helpers. # def _readdEntry(self, uidAuthor, oData, tsEffective = None): """ Re-adds the TestResultFailure entry. Used by addEntry, editEntry and removeEntry. """ if tsEffective is None: tsEffective = self._oDb.getCurrentTimestamp(); self._oDb.execute('INSERT INTO TestResultFailures (\n' ' uidAuthor,\n' ' tsEffective,\n' ' idTestResult,\n' ' idTestSet,\n' ' idFailureReason,\n' ' sComment)\n' 'VALUES (%s, %s, %s, %s, %s, %s)\n' , ( uidAuthor, tsEffective, oData.idTestResult, oData.idTestSet, oData.idFailureReason, oData.sComment,) ); return True; def _historizeEntry(self, idTestResult, tsExpire = None): """ Historizes the current entry. """ if tsExpire is None: tsExpire = self._oDb.getCurrentTimestamp(); self._oDb.execute('UPDATE TestResultFailures\n' 'SET tsExpire = %s\n' 'WHERE idTestResult = %s\n' ' AND tsExpire = \'infinity\'::TIMESTAMP\n' , (tsExpire, idTestResult,)); return True; def _resolveSetTestIdIfMissing(self, oData): """ Resolve any missing idTestSet reference (it's a duplicate for speed efficiency). """ if oData.idTestSet is None and oData.idTestResult is not None: self._oDb.execute('SELECT idTestSet FROM TestResults WHERE idTestResult = %s', (oData.idTestResult,)); oData.idTestSet = self._oDb.fetchOne()[0]; return oData; # # Unit testing. # # pylint: disable=C0111 class TestResultFailureDataTestCase(ModelDataBaseTestCase): def setUp(self): self.aoSamples = [TestResultFailureData(),]; if __name__ == '__main__': unittest.main(); # not reached.