path: root/src/VBox/ValidationKit/testmanager/core/testresultfailures.py

# -*- 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.