TestRequireJob.py #1

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

from __future__ import print_function

import sys
import unittest
import os
import re

import P4
from p4testutils import TestCase, P4Server, localDirectory, create_file, append_to_file

parent_dir = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
sys.path.insert(0, parent_dir)
from RequireJob import RequireJob

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

python3 = sys.version_info[0] >= 3

class TestRequireJob(TestCase):
    def __init__(self, methodName='runTest'):
        super(TestRequireJob, self).__init__(LOGGER_NAME, LOG_FILE, methodName=methodName)

    def setUp(self):
        pass

    def tearDown(self):
        pass


    def setupServer(self, trig_event):
        self.server = P4Server()
        trig_path = os.path.join(parent_dir, "RequireJob.py")
        config_path = os.path.join(self.server.root, "~test_config.yaml")
        p4 = self.server.p4
        p4.logger = self.logger
        # This works if no spaces in server root pathname!
        port = p4.port.replace('"', '')
        self.logger.debug("port: |%s|" % port)
        triggers = p4.fetch_triggers()
        triggers['Triggers'] = ['require-job {} //... "python {}'
                                ' -p %quote%{}%quote% -u {} -c {} %change%"'.format(
            trig_event, trig_path, port, p4.user, config_path)]
        self.logger.debug(triggers)
        p4.save_triggers(triggers)
        # Reconnect to pick up changes
        p4.disconnect()
        p4.connect()
        return config_path, p4

    def testRequireJob(self):
        """trigger fires and sends expected info to Swarm"""
        config_path, p4 = self.setupServer('change-submit')

        inside = localDirectory(self.server.client_root, "inside")
        inside_file1 = os.path.join(inside, "inside_file1")
        create_file(inside_file1, 'Test content')

        # Submit without a config file still works fine
        p4.run('add', inside_file1)
        result = p4.run('submit', '-d', 'inside_file1 added')
        self.assertEqual("1", result[-1]['submittedChange'])

        with open(config_path, "w") as f:
            f.write("""
api: "api/v6"
user: swarmtest
ticket: A123453
review_description: "Please review me!\nDon't forget to check YYYY"
submit_without_review_users:
  - jenkins
projects:
  - name:           ProjectA
    post_submit_create_review:  y
    require_job:    y
    update_review:  n
    depot_paths:
      - //depot/inside/...
      - "-//depot/inside/A..."
""")

        # Now submit fails due to no job
        p4.run('edit', inside_file1)
        append_to_file(inside_file1, "more text")
        try:
            result = p4.run('submit', '-d', 'inside_file1 added')
            self.assertEqual("2", result[-1]['submittedChange'])
            self.assertTrue(False, "Expected exception not found")
        except P4.P4Exception as e:
            self.assertRegex(str(e), r"this codeline requires a job")

        job = p4.fetch_job()
        job["Description"] = "test job"
        p4.save_job(job)

        # This time submit succeeds
        p4.run_fix("-c", "2", "job000001")
        result = p4.run('submit', '-c', '2')
        self.assertEqual("2", result[-1]['submittedChange'])

        # ---------------------
        # Submit succeeds if done by a configured user - e.g. jenkins as configured above
        p4.user = 'jenkins'
        p4.run('edit', inside_file1)
        append_to_file(inside_file1, "more text3")
        result = p4.run('submit', '-d', 'inside_file1 edited')
        self.assertEqual("3", result[-1]['submittedChange'])

    def testRequireJobShelvedChange(self):
        """make sure shelving requires a job - pre-requisite for Swarm reviews"""
        config_path, p4 = self.setupServer('shelve-submit')

        inside = localDirectory(self.server.client_root, "inside")
        inside_file1 = os.path.join(inside, "inside_file1")
        create_file(inside_file1, 'Test content')

        # Shelve without a config file still works fine
        p4.run('add', inside_file1)
        result = p4.run('submit', '-d', 'inside_file1 added')
        self.assertEqual("1", result[-1]['submittedChange'])

        p4.run('edit', inside_file1)
        append_to_file(inside_file1, "more text")
        chg = p4.fetch_change()
        chg["Description"] = "Test change for shelving"
        result = p4.save_change(chg)
        chgid = self.get_changeid(result)
        self.assertEqual("2", chgid)
        p4.run_shelve("-c", chgid, inside_file1)

        with open(config_path, "w") as f:
            f.write("""
api: "api/v6"
user: swarmtest
ticket: A123453
review_description: "Please review me!\nDon't forget to check YYYY"
projects:
  - name:           ProjectA
    post_submit_create_review:  y
    require_job:    y
    update_review:  n
    depot_paths:
      - //depot/inside/...
      - "-//depot/inside/A..."
""")

        # Now submit fails due to no job
        p4.run('edit', inside_file1)
        append_to_file(inside_file1, "more text2")
        chg = p4.fetch_change()
        chg["Description"] = "Test change for shelving2"
        result = p4.save_change(chg)
        chgid = self.get_changeid(result)
        self.assertEqual("3", chgid)
        p4.run_reopen("-c", chgid, inside_file1)
        try:
            p4.run_shelve("-c", chgid, inside_file1)
            self.assertTrue(False, "Expected exception not found")
        except P4.P4Exception as e:
            self.assertRegex(str(e), r"this codeline requires a job")

        job = p4.fetch_job()
        job["Description"] = "test job"
        p4.save_job(job)

        # This time shelve succeeds
        p4.run_fix("-c", chgid, "job000001")
        p4.run_shelve("-c", chgid, inside_file1)


    def testRequireJobSwarmShelvedChange(self):
        """make sure shelving requires a job but not when done by Swarm user"""
        config_path, p4 = self.setupServer('shelve-submit')

        inside = localDirectory(self.server.client_root, "inside")
        inside_file1 = os.path.join(inside, "inside_file1")
        create_file(inside_file1, 'Test content')

        # Shelve without a config file still works fine
        p4.run('add', inside_file1)
        result = p4.run('submit', '-d', 'inside_file1 added')
        self.assertEqual("1", result[-1]['submittedChange'])

        p4.run('edit', inside_file1)
        append_to_file(inside_file1, "more text")
        chg = p4.fetch_change()
        chg["Description"] = "Test change for shelving"
        result = p4.save_change(chg)
        chgid = self.get_changeid(result)
        self.assertEqual("2", chgid)
        p4.run_shelve("-c", chgid, inside_file1)

        with open(config_path, "w") as f:
            f.write("""
api: "api/v6"
user: swarmtest
swarm_user: swarm
ticket: A123453
review_description: "Please review me!\nDon't forget to check YYYY"
projects:
  - name:           ProjectA
    post_submit_create_review:  y
    require_job:    y
    update_review:  n
    depot_paths:
      - //depot/inside/...
      - "-//depot/inside/A..."
""")

        # Now shelve still works as it is swarm user - from config file
        p4.user = "swarm"
        p4.run('edit', inside_file1)
        append_to_file(inside_file1, "more text2")
        chg = p4.fetch_change()
        chg["Description"] = "Test change for shelving2"
        result = p4.save_change(chg)
        chgid = self.get_changeid(result)
        self.assertEqual("3", chgid)
        p4.run_reopen("-c", chgid, inside_file1)
        p4.run_shelve("-c", chgid, inside_file1)

    def get_changeid(self, result):
        m = re.search("Change ([^ ]+) created", result[0])
        self.assertTrue(m)
        change = m.group(1)
        return change

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