#!/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