P4Server.py #9

#!/usr/local/bin/python
# Copyright (c) 2017, Charles McLouth
# All rights reserved.
# 
# Redistribution and use in source and binary forms, with or without
# modification, are permitted provided that the following conditions are met:
# 
#    # Redistributions of source code must retain the above copyright
#       notice, this list of conditions and the following disclaimer.
# 
#    # Redistributions in binary form must reproduce the above copyright
#       notice, this list of conditions and the following disclaimer in the
#       documentation and/or other materials provided with the distribution.
# 
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
# ARE DISCLAIMED. IN NO EVENT SHALL STEWART LORD BE LIABLE FOR ANY
# DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
# (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
# LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND
# ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
# (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF
# THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
from P4 import P4Exception
 
'''P4Server.py
 
 This python script provides an encapsulation of a Perforce Server (P4D.)
 It is intended for use with python UnitTests that require access to a Perforce
 Server.
 
 For example usage see accompanying UnitTest P4ServerTest.py

 $Id: //guest/cmclouth/python-utils/P4Server/src/P4Server.py#9 $
 */
'''
import logging
import tempfile
import os
import shutil
import P4
import time
import unittest

logger = logging.getLogger(__name__)

p4dbin = "p4d"
if os.name == "nt":
    p4dbin = "p4d.exe"

testConfiguration = {#"P4USER": "p4tester",
                     #"P4CLIENT": "c_p4tester",
                     #"P4D":p4dbin, #windows path
                     #"P4D":"/usr/local/bin/p4d",  #linux path
                     #"TESTDIR": "test", #windows path
                     #"TESTDIR": "/Users/cmclouth/P4Server/src/p4test",  #linux path
                     }

def rmtree(path, ignore_errors=False, onerror=None):
    def roerror(func, path, exc_info):
        '''
        Error handler for ``shutil.rmtree``. lifted from StackOverflow.com
    
        If the error is due to an access error (read only file)
        it attempts to add write permission and then retries.
    
        If the error is for another reason it re-raises the error.
    
        Usage : ``shutil.rmtree(path, onerror=onerror)``
        '''
        import stat
        if not os.access(path, os.W_OK):
            # Is the error an access error ?
            os.chmod(path, stat.S_IWUSR)
            func(path)
        elif onerror is not None: onerror(func, path, exc_info)
        else:
            raise
    return shutil.rmtree(path, ignore_errors, roerror)

class P4ServerError(Exception):
    '''
    Exception thrown by the class P4Server
    '''
    def __str__(self):
        if len(self.args) == 1:
            if isinstance(self.args[0], P4.P4Exception):
                return self.args[0].__str__()
        return super(P4ServerError, self).__str__()

class P4Server(object):
    '''
    Encapsulates the startup and shutdown of a P4D process.
    Quite useful when performing repeatable tests that need access to a 
    Perforce server
    '''

    def __init__(self, p4user, p4client, testdir, p4d):
        '''
        Constructor for P4Server required inputs:
            p4user: A valid perforce user id
            p4client: A valid perforce client name
            testdir: A valid existing directory
            p4d: fully qualified path to a p4d binary
        '''
        logger.debug('Enter')
        # validate input parameters
        if p4user is None or not isinstance(p4user, str) or len(p4user) < 1 \
                or p4user.isspace():
            raise P4ServerError("Invalid p4user '" + str(p4user) + "'.")
        if p4client is None or not isinstance(p4client, str) \
                or len(p4client) < 1 or p4client.isspace():
            raise P4ServerError("Invalid p4client '" + str(p4client) + "'.")
        if p4d is None or not isinstance(p4d, str) or len(p4d) < 1 \
                or p4d.isspace():
            raise P4ServerError("Invalid p4d '" + str(p4d) + "'.")
        if testdir is None or not isinstance(testdir, str) or len(testdir) < 1 \
                or testdir.isspace():
            raise P4ServerError("Invalid testdir '" + str(testdir) + "'.")
        # validate testdir is a dir that exists
        if not os.path.isdir(testdir):
            raise P4ServerError("Invalid testdir '" + str(testdir) + \
                                "' not found.")
        # convert relative path to absolute path
        testdir = os.path.abspath(testdir)
        P4ROOT = tempfile.mkdtemp('', p4client + '_', testdir)
        P4PORT = "rsh:%s -r \"%s\" -L log -i" % (p4d, P4ROOT)
        P4PROG = "P4Server"
        P4CLIENTROOT = os.path.join(P4ROOT, p4client)
        os.mkdir(P4CLIENTROOT)
        # setup P4 connection
        self.p4api = P4.P4()
        self.p4api.user = p4user
        self.p4api.client = p4client
        self.p4d = p4d
        self.p4root = P4ROOT
        self.p4api.port = P4PORT
        self.clientroot = P4CLIENTROOT
        self.p4api.prog = P4PROG
        # Test connection
        self.p4api.connect()
        self.p4api.disconnect()
        logger.debug('Exit')

    def dispose(self):
        '''
        clean up
        '''
        logger.debug('Enter')
        # delete the directory
        logger.debug('dispose ' + self.p4root)
        if self.p4api.connected(): self.p4api.disconnect()
        time.sleep(.5)
        if os.path.isdir(self.p4root): rmtree(self.p4root)
        else: logger.warn('not a directory?:' + self.p4root)
        del(self.p4api)
        logger.debug('Exit')

class P4Testcase(unittest.TestCase):
    '''Base class that contains init and dispose of P4Server'''
    p4server = None

    def setUpServer(self):
        '''initialize P4Server Object'''
        logger.debug(testConfiguration)
        testConfiguration["TESTDIR"] = os.path.abspath(testConfiguration.get(
                                                "TESTDIR", __name__ + "test"))
        self.removeTestDir = False
        if not os.path.isdir(testConfiguration["TESTDIR"]):
            os.mkdir(testConfiguration["TESTDIR"])
            self.removeTestDir = True
        self.p4server = P4Server(testConfiguration.get("P4USER", "p4tester"),
                            testConfiguration.get("P4CLIENT", "c_p4tester"), 
                            testConfiguration.get("TESTDIR"), 
                            testConfiguration.get("P4D", p4dbin))
        if not self.p4server.p4api.connected(): self.p4server.p4api.connect()
        result = self.p4server.p4api.run("info")
        self.assertTrue(result is not None and len(result) > 0, 
                        "info returned no results")
        logger.debug("Exit")

    def tearDownServer(self):
        '''dispose P4Server Object'''
        logger.debug("Enter")
        if self.p4server is not None and isinstance(self.p4server, P4Server):
            try:
                self.p4server.p4api.run_logout()
            except P4Exception:
#                 logger.exception("ignoring non-fatal exception", exc_info=True)
                pass
            if self.p4server.p4api.connected(): 
                self.p4server.p4api.disconnect()
            self.p4server.dispose()
            if self.removeTestDir:
                rmtree(os.path.abspath(testConfiguration["TESTDIR"]))
            self.p4server = None
        logger.debug("Exit")

    def initTestData(self, p4api):
        raise NotImplementedError

    def setUp(self):
        self.setUpServer()
        self.initTestData(self.p4server.p4api)

    def tearDown(self):
        self.tearDownServer()


class P4TestcaseShared(P4Testcase):
    '''Base class that contains setup and tear down on a per-class basis'''
    @classmethod
    def setUpClass(cls):
        cls.setUpServer(cls)

    @classmethod
    def tearDownClass(cls):
        cls.tearDownServer(cls)

    def setUp(self):
        '''override to call initTestData only once'''
        testclass = self.id().rsplit(".", 1)[0]
        if int(self.p4server.p4api.run_counter(testclass)[0]["value"]) < 1:
            self.initTestData(self.p4server.p4api)
            self.p4server.p4api.run_counter("-i", testclass)

    def tearDown(self):
        '''override to prevent from doing anything'''
        pass