# -*- 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):
pass
def tearDown(self):
pass
def testJobRename(self):
"""jobs are renamed for p4dtg"""
self.server = P4Server()
trigpath = os.path.join(parent_dir, "CheckFixes.py")
config_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "~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'] = ['check-fixes fix-add fix " python ' + trigpath +
" -p %quote%" + port + "%quote% -u " + p4.user +
" -c %s " % config_path +
' %change% %jobs% "',
'check-fixes fix-delete fix " python ' + trigpath +
" -p %quote%" + port + "%quote% -u " + p4.user +
" -c %s " % config_path +
"--delete " +
' %change% %jobs% "']
self.logger.debug(triggers)
p4.save_triggers(triggers)
# Reconnect to pick up changes
p4.disconnect()
p4.connect()
fix_state_field = "JiraStatus"
with open(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."
""" % fix_state_field)
# Setup up jobspec to approximate P4DTG
job_spec = p4.fetch_jobspec()
job_spec['Fields'].append("106 %s line 32 optional" % fix_state_field)
self.logger.debug(job_spec)
p4.save_jobspec(job_spec)
# Reconnect to pick up changes
p4.disconnect()
p4.connect()
# Create a job, add and delete a fix
job = p4.fetch_job()
job['Description'] = "Description1"
job[fix_state_field] = 'In Work'
result = p4.save_job(job)
job1 = self.get_jobname(result)
self.assertEquals("job000001", job1)
job = p4.fetch_job()
job['Description'] = "Description2"
job[fix_state_field] = 'Another status'
result = p4.save_job(job)
job2 = self.get_jobname(result)
self.assertEquals("job000002", job2)
inside = localDirectory(self.server.client_root, "inside")
file1 = os.path.join(inside, "file1")
file2 = os.path.join(inside, "file2")
create_file(file1, "Some content")
create_file(file2, "Some content2")
p4.run('add', file1)
result = p4.run('submit', '-d', 'file1 added')
self.assertEquals("1", result[-1]['submittedChange'])
p4.run('add', file2)
result = p4.run('submit', '-d', 'file2 added')
self.assertEquals("2", result[-1]['submittedChange'])
p4.run('fix', '-c', '1', job1)
fixes = p4.run('fixes', '-c', '1')
self.assertEquals(1, len(fixes))
p4.run('fix', '-d', '-c', '1', job1)
fixes = p4.run('fixes', '-c', '1')
self.assertEquals(0, len(fixes))
p4.run('fix', '-c', '1', job1)
fixes = p4.run('fixes', '-c', '1')
self.assertEquals(1, len(fixes))
job = p4.fetch_job(job1)
job[fix_state_field] = 'Closed'
p4.save_job(job)
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")
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")
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()