TestCheckJobEditTrigger.py #2

# -*- encoding: UTF8 -*-
# Test harness for CheckJobEditTrigger.py

from __future__ import print_function

import sys
import unittest
import os
import re

sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
import testp4
import P4
import logging
import subprocess

from CheckJobEditTrigger import CheckJobEditTrigger, REPLICATOR_USER

os.environ["LOGS"] = "."
LOGGER_NAME = "TestCheckJobEditTrigger"
LOG_FILE = "log-TestCheckJobEditTrigger.log"

python3 = sys.version_info[0] >= 3


class TestCheckJobEditTrigger(unittest.TestCase):
    def __init__(self, methodName='runTest'):
        super(TestCheckJobEditTrigger, self).__init__(methodName=methodName)
        self.logger = logging.getLogger(LOGGER_NAME)
        self.logger.setLevel(logging.DEBUG)
        logformat = '%(levelname)s [%(asctime)s] [%(funcName)s : %(lineno)d] - %(message)s'
        logging.basicConfig(format=logformat, filename=LOG_FILE, level=logging.DEBUG)

    def assertRegex(self, *args, **kwargs):
        if python3:
            return super(TestCheckJobEditTrigger, self).assertRegex(*args, **kwargs)
        else:
            return super(TestCheckJobEditTrigger, self).assertRegexpMatches(*args, **kwargs)

    def setUp(self):
        pass

    def tearDown(self):
        pass

    def testCheckJobEditTrigger(self):
        """ordinary jobs command is OK"""
        self.server = testp4.P4Server()
        trigpath = os.path.join(os.path.dirname(os.path.abspath(__file__)), "CheckJobEditTrigger.py")
        p4 = self.server.p4
        # This works if no spaces in pathname!
        port = p4.port.replace('"', '')
        self.logger.debug("port: |%s|" % port)
        triggers = p4.fetch_triggers()
        # triggers['Triggers'] = ['check_job_edit form-save job "python ' + trigpath +
        #                         ' -p \'' + p4.port + '\' -u ' + p4.user +
        #                        ' %user% %formfile%"']
        triggers['Triggers'] = ['check_job_edit form-save job " python ' + trigpath +
                                " -p %quote%" + port + "%quote% -u " + p4.user +
                               ' %user% %formfile% "']
        # triggers['Triggers'] = ['check_job_edit form-save job "python ' + trigpath +
        #                        ' %user% %formfile%"']
        self.logger.debug(triggers)
        p4.save_triggers(triggers)
        p4.disconnect()
        p4.connect()

        # Make sure we are in correct directory to pick up P4CONFIG created in P4Server()
        os.chdir(self.server.root)
        # os.environ["P4PORT"] = p4.port
        # os.environ["P4USER"] = p4.user
        self.logger.debug("dir: %s" % os.getcwd())
        p = subprocess.Popen("p4 set", stdout=subprocess.PIPE, stderr=subprocess.STDOUT, universal_newlines=True, shell=True)
        output, err = p.communicate()
        self.logger.debug("Output:\n%s" % output)

        # Replicator user allowed to create jobs
        testuser = p4.user
        p4.user = REPLICATOR_USER
        job = p4.fetch_job()
        job['Description'] = "Description1"
        result = p4.save_job(job)
        self.logger.debug(result)
        m = re.search("Job (\w+) saved", result[0])
        self.assertTrue(m)
        jobname = m.group(1)

        # Ordinary user not allowed to create jobs
        p4.user = testuser
        try:
            job = p4.fetch_job()
            job['Description'] = "Description2"
            p4.save_job(job)
            self.assertTrue(False, "Expected exception not found")
        except P4.P4Exception as e:
            self.assertRegex(str(e), "You are not allowed to create new jobs")


        # Ordinary user can't change readonly fields
        try:
            job = p4.fetch_job(jobname)
            job['Description'] = "Description2"
            p4.save_job(job)
            self.assertTrue(False, "Expected exception not found")
        except P4.P4Exception as e:
            self.assertRegex(str(e), r"You have changed one or more read.*only job fields")

        # Ordinary user can change writeable fields
        job = p4.fetch_job(jobname)
        job['Status'] = "suspended"
        result = p4.save_job(job)
        self.logger.debug(result)
        self.assertRegex(result[0], "Job .* saved")


if __name__ == '__main__':
    unittest.main()