TestCheckFixes.py #1

# -*- encoding: UTF8 -*-
# Test harness for CheckFixes.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 CheckFixes import CheckFixes

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


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

    def setUp(self):
        self.server = P4Server()
        trigpath = os.path.join(parent_dir, "CheckFixes.py")
        self.config_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "~test_config.yaml")
        p4 = self.server.p4
        self.p4 = 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'] = ['check-fixes fix-add fix " python {} -p %quote%{}%quote% '
                                '-u {} -c {} %change% %client% %jobs% "'.format(trigpath,
                                                                                port, p4.user, self.config_path),
                                'check-fixes fix-delete fix " python {} -p %quote%{}%quote% '
                                '-u {} -c {} --delete %change% %client% %jobs% "'.format(trigpath,
                                                                                         port, p4.user, self.config_path)]
        self.logger.debug(triggers)
        p4.save_triggers(triggers)
        # Reconnect to pick up changes
        p4.disconnect()
        p4.connect()

        self.fix_state_field = "JiraStatus"

        with open(self.config_path, "w") as f:
            f.write("""
fix_state_field: "%s"
fix_allowed_states:
  - "Open"
  - "In Work"
msg_cant_link_jobs:
  - ""
  - "You are not allowed to link changes to or unlink changes from these jobs"
  - "because of the state of their associated JIRA issues."
  - "Please change the state first in JIRA and try again."
""" % self.fix_state_field)

        # Setup up jobspec to approximate P4DTG
        job_spec = p4.fetch_jobspec()
        job_spec['Fields'].append("106 %s line 32 optional" % self.fix_state_field)
        self.logger.debug(job_spec)
        p4.save_jobspec(job_spec)
        # Reconnect to pick up changes
        p4.disconnect()
        p4.connect()


    def tearDown(self):
        pass


    def testCheckFixes(self):
        """check that fixes are appropriate controlled"""

        p4 = self.p4

        # Create a job, add and delete a fix

        job = p4.fetch_job()
        job['Description'] = "Description1"
        job[self.fix_state_field] = 'In Work'
        result = p4.save_job(job)
        job1 = self.get_jobname(result)
        self.assertEqual("job000001", job1)

        inside = localDirectory(self.server.client_root, "inside")
        outside = localDirectory(self.server.client_root, "outside")
        file1 = os.path.join(inside, "file1")
        file2 = os.path.join(inside, "file2")
        create_file(file1, "Some content")
        create_file(file2, "Some content2")

        # Now repeat the test but this time allow fixes for the path.
        with open(self.config_path, "w") as f:
            f.write("""
fix_state_field: "%s"
fix_allowed_states:
  - "Open"
  - "In Work"
msg_cant_link_jobs:
  - ""
  - "You are not allowed to link changes to or unlink changes from these jobs"
  - "because of the state of their associated JIRA issues."
  - "Please change the state first in JIRA and try again."

projects:
  - name: ProjectA
    depot_paths:
      - //depot/inside/...
      - "-//depot/inside/....c"
""" % self.fix_state_field)

        p4.run('add', file1)
        result = p4.run('submit', '-d', 'file1 added')
        self.assertEqual("1", result[-1]['submittedChange'])

        p4.run('add', file2)
        result = p4.run('submit', '-d', 'file2 added')
        self.assertEqual("2", result[-1]['submittedChange'])

        p4.run('fix', '-c', '1', job1)
        fixes = p4.run('fixes', '-c', '1')
        self.assertEqual(1, len(fixes))

        p4.run('fix', '-d', '-c', '1', job1)
        fixes = p4.run('fixes', '-c', '1')
        self.assertEqual(0, len(fixes))

        p4.run('fix', '-c', '1', job1)
        fixes = p4.run('fixes', '-c', '1')
        self.assertEqual(1, len(fixes))

        # Now change job status so fix actions fail
        job = p4.fetch_job(job1)
        job[self.fix_state_field] = 'Closed'
        p4.save_job(job)

        try:
            p4.run('fix', '-d', '-c', '1', job1)
            self.assertTrue(False, "Expected exception not found")
        except P4.P4Exception as e:
            self.assertRegex(str(e), r"You are not allowed to link changes")

        job = p4.fetch_job()
        job['Description'] = "Description2"
        job[self.fix_state_field] = 'Another status'
        result = p4.save_job(job)
        job2 = self.get_jobname(result)
        self.assertEqual("job000002", job2)

        try:
            p4.run('fix', '-c', '2', job2)
            self.assertTrue(False, "Expected exception not found")
        except P4.P4Exception as e:
            self.assertRegex(str(e), r"You are not allowed to link changes")

        # Now repeat the test but this time allow fixes for the path.
        with open(self.config_path, "w") as f:
            f.write("""
fix_state_field: "%s"
fix_allowed_states:
  - "Open"
  - "In Work"
msg_cant_link_jobs:
  - ""
  - "You are not allowed to link changes to or unlink changes from these jobs"
  - "because of the state of their associated JIRA issues."
  - "Please change the state first in JIRA and try again."

projects:
  - name: ProjectA
    depot_paths:
      - //depot/inside/...
      - "-//depot/inside/....c"
    fix_allowed_paths:
      - //depot/inside/file...
""" % self.fix_state_field)

        p4.run('fix', '-c', '2', job2)
        result = p4.run('fixes', '-c', '1')
        self.assertEqual(1, len(result))

        p4.run('fix', '-d', '-c', '1', job1)
        result = p4.run('fixes', '-c', '1')
        self.assertEqual(0, len(result))

        # Now test for pending changes
        # Are allowed when file matches 'fix_allowed_paths'

        p4.run('edit', file1)
        chg = p4.fetch_change()
        chg['Description'] = "test change"
        chg['Jobs'] = [job1]
        result = p4.save_change(chg)
        self.logger.debug(result)

        # But should not be allowed when it doesn't

        file3 = os.path.join(inside, "newfile")
        create_file(file3, "Some content3")

        p4.run('add', file3)
        chg = p4.fetch_change()
        chg['Description'] = "test change"
        chg['Jobs'] = [job1]

        try:
            result = p4.save_change(chg)
            self.assertTrue(False, "Expected exception not found")
        except P4.P4Exception as e:
            self.assertRegex(str(e), r"You are not allowed to link changes")

        # ---------------------------
        # Outside files shouldbe fine with fixes even for closed jobs

        outside_file1 = os.path.join(outside, "file1")
        create_file(outside_file1, "Some content")

        p4.run('revert', "//...")
        p4.run('add', outside_file1)
        opened = p4.run_opened()
        self.assertEqual(1, len(opened))
        chg = p4.fetch_change()
        chg['Description'] = "test change"
        chg['Jobs'] = [job1]
        p4.save_change(chg)


    def testCheckFixesPendingDeletes(self):
        """check that fixes are appropriate for pending changelists"""

        p4 = self.p4
        inside = localDirectory(self.server.client_root, "inside")

        # Now repeat the test but this time allow fixes for the path.
        with open(self.config_path, "w") as f:
            f.write("""
fix_state_field: "%s"
fix_allowed_states:
  - "Open"
  - "In Work"
msg_cant_link_jobs:
  - ""
  - "You are not allowed to link changes to or unlink changes from these jobs"
  - "because of the state of their associated JIRA issues."
  - "Please change the state first in JIRA and try again."

projects:
  - name: ProjectA
    depot_paths:
      - //depot/inside/...
""" % self.fix_state_field)

        # Now test for pending changes - you can't add a fix with the wrong status but you
        # can delete one
        # Create a job, add and delete a fix

        job = p4.fetch_job()
        job['Description'] = "Description1"
        job[self.fix_state_field] = 'Open'    # Valid state
        result = p4.save_job(job)
        job1 = self.get_jobname(result)
        self.assertEqual("job000001", job1)

        file3 = os.path.join(inside, "newfile")
        create_file(file3, "Some content3")

        p4.run('add', file3)
        chg = p4.fetch_change()
        chg['Description'] = "test change"
        chg['Jobs'] = [job1]

        p4.save_change(chg)

        job = p4.fetch_job(job1)
        job[self.fix_state_field] = 'Closed'    # Invalid state
        p4.save_job(job)

        p4.run('fix', '-d', '-c', '1', job1)
        result = p4.run('fixes', '-c', '1')
        self.assertEqual(0, len(result))

        p4.run_shelve('-c', '1')
        p4.run_revert('//...')

        # Can't add until status is changed
        try:
            result = p4.run_fix('-c', '1', job1)
            self.assertTrue(False, "Expected exception not found")
        except P4.P4Exception as e:
            self.assertRegex(str(e), r"You are not allowed to link changes")

        job = p4.fetch_job(job1)
        job[self.fix_state_field] = 'Open'
        p4.save_job(job)
        p4.run_fix('-c', '1', job1)

        job = p4.fetch_job(job1)
        job[self.fix_state_field] = 'Closed'    # Invalid state
        p4.save_job(job)

        p4.run('fix', '-d', '-c', '1', job1)
        result = p4.run('fixes', '-c', '1')
        self.assertEqual(0, len(result))

    def get_jobname(self, result):
        m = re.search("Job ([^ ]+) saved", result[0])
        self.assertTrue(m)
        jobname = m.group(1)
        return jobname


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