# replicator.py -- P4DTI replicator.
# Gareth Rees, Ravenbrook Limited, 2000-08-09.
# $Id: //info.ravenbrook.com/project/p4dti/version/0.3/code/replicator/replicator.py#15 $
#
# See "Perforce Defect Tracking Integration Architecture"
# for the architecture of the integration;
# "Replicator design" for the design of the
# replicator; and "Replicator classes in Python"
# for the class organization of the
# replicator.
#
# Copyright 2000 Ravenbrook Limited. This document is provided "as is",
# without any express or implied warranty. In no event will the authors
# be held liable for any damages arising from the use of this document.
# You may make and distribute copies and derivative works of this
# document provided that (1) you do not charge a fee for this document or
# for its distribution, and (2) you retain as they appear all copyright
# and licence notices and document history entries, and (3) you append
# descriptions of your modifications to the document history.
import logger
import p4
import re
import smtplib
import socket
import string
import sys
import time
import traceback
import types
# The defect_tracker_issue class is an abstract class representing an issue in
# the defect tracker. You can't use this class; you must subclass it and use
# the subclass.
class defect_tracker_issue:
# action(). Return the action field (replicate/wait/keep/discard).
# add_filespec(filespec). Add an associated filespec record to the issue.
# add_fix(p4_fix). Add a fix record to the issue corresponding to p4_fix
# (a Perforce fix).
# conflicting_p(issue). Return true iff the given issue is marked as being
# in conflict with Perforce.
# id(). Return the identifier for the given defect tracking issue, as a
# string.
# filespecs(). Return a list of filespecs for the issue. The elements of
# the list belong to the defect_tracker_filespec class.
# fixes(). Return a list of fixes for the issue. The elements of the list
# belong to the defect_tracker_fix class.
# jobname(). Return the Perforce jobname for the issue (if the issue
# specifies one), or generate and return an appropriate jobname for the
# issue (otherwise).
# replicate_p(). A policy used to set up replication for issues where
# replication is not yet specified. Return true if the issue should be
# replicated by this replicator, false if it should not. The default
# policy is to replicate all issues.
def replicate_p(self):
return 1
# rid(). Return the identifier of the replicator which is in charge of
# replicating this issue, or the empty string if the issue is not being
# replicated.
# setup_for_replication(server_id). Set up the issue for replication. The
# server_id argument is the Perforce server id that the replicator
# replicates to.
# transform_to_job(old_job). Transform the issue into a Perforce job. The
# old_job argument is the old job corresponding to the issue (or a fresh
# job description given by "p4 job -o" if there is no corresponding job).
# Return a pair consisting of (a) the new job and (b) the changes that
# needed to be made to the job, in the form of a hash mapping job field
# name to the new value for that field. It is legal to change and return
# the old_job argument.
# transform_from_job(job, fix_diffs, filespec_diffs). Transform the issue
# so that it matches the Perforce job. Don't update the record in the
# defect tracking system yet. Return the changes that needed to be made to
# the issue, and a proposed transition that is appropriate, as a pair. The
# returned changes are in the form of a hash mapping issue field name to
# the new value for that field, or otherwise (in fact the return value only
# needs to be printed). The transition should be None if the defect
# tracker doesn't support transitions or if no transition is appropriate.
# The fixes_diffs argument is a list of differences between the fixes for
# the issue and the fixes for the jobs. The filespec_diffs argument is the
# equivalent for filespecs. (These two arguments are informational.)
# update(transition). Update the issue record in the defect tracking
# system to match the issue. The transition is one that was returned from
# the transform_from_job method.
# update_action(action). Update the issue in the defect tracking system so
# that it's action field is set to the action argument.
# The defect_tracker_filespec class is an abstract class representing an
# associated filespec record in the defect tracker. You can't use this class;
# you must subclass it and use the subclass.
class defect_tracker_filespec:
None
# delete(). Delete the filespec record.
# name(). Return the filespec's name.
# The defect_tracker_fix class is an abstract class representing a fix record
# in the defect tracker. You can't use this class; you must subclass it and
# use the subclass.
class defect_tracker_fix:
None # Classes can't be empty in Python
# change(). Return the change number for the fix record.
# delete(). Delete the fix record.
# status(). Return the status for the fix record.
# update(p4_fix). Update the fix record so that it corresponds to the
# Perforce fix record.
# The defect_tracker class is an abstract class representing the interface
# between the replicator and the defect tracker. You can't use this class; you
# must subclass it and use the subclass.
class defect_tracker:
config = { 'logger' : logger.file_logger() }
rid = None
sid = None
def __init__(self, rid, sid, config = {}):
self.rid = rid
self.sid = sid
# Merge the supplied config with the default config (the former takes
# precedence).
for k in config.keys():
self.config[k] = config[k]
# all_issues(). Return a list of all defect tracking issues that are being
# replicated by this replicator, or which are not being replicated by any
# replicator. Each element of the list belongs to the defect_tracker_issue
# class (or a subclass).
# changed_issues(). Return a pair consisting of (a) a list of the issues
# that have changed in the defect tracker and which are either replicated
# by this replicator, or are new issues that are not yet replicated, and
# (b) some private data. Each element of the list belongs to the
# defect_tracker_issue class (or a subclass). The private data will be
# passed to the method changed_issues_are_replicated when that is called
# after all the issues have been replicated.
# changed_issues_are_replicated(data). Called when all the issues returned
# by changed_issues have been replicated. The argument is the second
# element of the pair returned by the call to changed_issues. (The idea
# behind this is that the defect tracker interface may have some way of
# recording that it has considered all these issues -- perhaps by recording
# the last key on a changes table. It is important not to record this
# until it is true, so that if the replicator crashes between getting the
# changed issues and replicating them then we'll consider the same set of
# changed issues the next time round, and hopefully this will give us a
# chance to either replicate them correctly or else discover that they are
# conflicting.)
# init(). Set up the defect tracking database for the integration. Set up
# issues for replication by this replicator according to the policy in the
# replicate_p method of the issue.
# issue(issue_id). Return the defect tracking issue whose identifier has
# the string form give by issue_id, or None if there is no such issue.
# log(format, arguments, priority). Write a message to the replicator's
# log.
def log(self, format, arguments = (), priority = None):
format = "%s\t" + format
if type(arguments) == types.TupleType:
arguments = (self.rid,) + arguments
else:
arguments = (self.rid, arguments)
self.config['logger'].log("P4DTI-0", format, arguments, priority)
# replicate_changelist(p4_changelist). Replicate the changelist to the
# defect tracking database. Return 1 iff the changelist was changed, 0 if
# it was already in the database and was unchanged.
# The replicator class is a generic replicator. Pass the constructor a defect
# tracking interface and (optionally) a configuration hash. The generic
# replicator assumes the Perforce server is on the same host as the replicator.
class replicator:
config = {
'administrator-address': None,
'counter' : None,
'logger' : logger.file_logger(),
'p4-client' : 'p4dti-%s' % socket.gethostname(),
'p4-client-executable': 'p4',
'p4-password' : '',
'p4-port' : '127.0.0.1:1666',
'p4-user' : None,
'poll-period' : 10,
'replicator-address' : 'p4dti',
'smtp-server' : None,
}
dt = None
rid = None
# Error object for fatal errors raised by the replicator.
error = 'P4DTI Replicator error'
# Error object for conflicts. Use raise self.conflict_error, (id, format,
# args) when raising this.
conflict_error = 'P4DTI Replicator conflict'
# The action_table maps the (dt_action, p4_action) to the action that the
# replicator should take: 'normal' means apply the normal decision rules;
# 'conflict' means that the entities are already marked as conflicting; a
# conflict should be reported if entities have changed; 'p4' means that the
# issue should be overwritten with the job; 'dt' means that the job should
# be overwritten with the issue.
action_table = {
( 'replicate', 'replicate' ): 'normal',
( 'wait', 'wait' ): 'conflict',
( 'wait', 'keep' ): 'p4',
( 'wait', 'discard' ): 'dt',
( 'keep', 'wait' ): 'dt',
( 'keep', 'discard' ): 'dt',
( 'discard', 'wait' ): 'p4',
( 'discard', 'keep' ): 'p4',
}
def __init__(self, rid, dt, config = {}):
self.dt = dt
self.rid = rid
# Merge the supplied config with the default config (the former takes
# precedence).
for k in config.keys():
self.config[k] = config[k]
# Make a counter name for this replicator.
if not self.config['counter']:
self.config['counter'] = 'P4DTI-%s' % self.rid
# Make a userid for the replicator.
if not self.config['p4-user']:
self.config['p4-user'] = 'P4DTI-%s' % self.rid
# changed_p4_entities(). Return a 3-tuple consisting of (a) changed jobs
# (b) changed changelists and (c) the last log entry that was considered.
# The changed jobs are those that are due for replication by this
# replicator (that is, the P4DTI-rid field of the job matches the
# replicator id). The last log entry will be passed to
# changes_are_replicated.
def changed_p4_entities(self):
# Get all entries from the log.
log_entries = self.p4_run('logger -t %s' % self.config['counter'])
# "p4 logger" doesn't return fully structured output. Instead, the
# 'data' element of the returned structures is a string like "15 job
# jobname" or "126 change 1543". The job_re regular expression matches
# and parses these strings.
job_re = re.compile('([0-9]+) ([a-z]+) (.*)$')
jobs = {}
changelists = []
last_log_entry = None # The last entry number in the log.
for e in log_entries:
if e.has_key('code') and e['code'] == 'info' and e.has_key('data'):
match = job_re.match(e['data'])
assert(match)
last_log_entry = int(match.group(1))
if match.group(2) == 'job':
jobname = match.group(3)
if not jobs.has_key(jobname):
job = self.job(jobname)
assert(job.has_key('P4DTI-rid'))
assert(job.has_key('P4DTI-user'))
if (job.has_key('P4DTI-rid')
and job['P4DTI-rid'] == self.rid
# We ought to make sure not to return jobs that
# were last updated by the replicator, by looking
# at the P4DTI-user field in each job. But this
# doesn't work yet: see job000014.
# and job.has_key('P4DTI-user')
# and job['P4DTI-user']!=self.config['p4-user']
):
jobs[jobname] = job
elif match.group(2) == 'change':
change_number = match.group(3)
try:
changelist = self.p4_run('change -o %s' % change_number)[0]
changelists.append(changelist)
except p4.error:
# The changelist might not exist, because it might have
# been a pending changelist that's been renumbered. So don't
# replicate it. Should it be deleted? GDR 2000-11-02.
None
return jobs, changelists, last_log_entry
# changes_are_replicated(log_entry). Update the Perforce database to
# record the fact that the replicator has replicated all changes up to
# log_entry.
def changes_are_replicated(self, log_entry):
# Update counter to last entry number in the log that we've replicated.
# If this is the last entry in the log, it has the side-effect of
# deleting the log (see "p4 help undoc").
if log_entry:
self.p4_run('logger -t %s -c %d' % (self.config['counter'], log_entry))
# check_consistency(). Run a consistency check on the two databases,
# reporting any inconsistencies.
def check_consistency(self):
print "Checking consistency for replicator '%s'" % self.rid
inconsistencies = 0 # Number of inconsistencies found.
# Get hashes of issues (by id) and jobs (by jobname).
issues = {}
for issue in self.dt.all_issues():
issues[issue.id()] = issue
jobs = {}
for j in self.p4_run('jobs -e P4DTI-rid=%s' % self.rid):
jobs[j['Job']] = j
for id, issue in issues.items():
# Progress indication.
sys.stdout.write(".")
# Report if issue has no corresponding job.
if issue.rid() != self.rid:
if issue.replicate_p():
print "Issue '%s' should be replicated but is not." % id
inconsistencies = inconsistencies + 1
continue
jobname = issue.jobname()
if not jobs.has_key(jobname):
print("Issue '%s' should be replicated to job '%s' but that "
"job either does not exists or is not replicated."
% (id, jobname))
inconsistencies = inconsistencies + 1
continue
# Get corresponding job.
job = jobs[jobname]
del jobs[jobname]
# Report if mapping is in error.
if job['P4DTI-issue-id'] != id:
print("Issue '%s' is replicated to job '%s' but that job "
"is replicated to issue '%s'."
% (id, jobname, job['P4DTI-issue-id']))
inconsistencies = inconsistencies + 1
# Report if actions aren't replicate/replicate.
issue_action = issue.action()
job_action = job['P4DTI-action']
if not self.action_table.has_key((issue_action, job_action)):
print("Issue '%s' has action '%s' and job '%s' has action "
"'%s': this combination is illegal."
% (id, issue_action, jobname, job_action))
inconsistencies = inconsistencies + 1
else:
action = self.action_table[(issue_action, job_action)]
if action != 'normal':
print("Issue '%s' has action '%s' and job '%s' has action "
"'%s'." % (id, issue_action, jobname, job_action))
inconsistencies = inconsistencies + 1
# Report if job and issue don't match.
_, changes = issue.transform_to_job(job)
if changes:
print("Job '%s' would need the following set of changes in "
"order to match issue '%s': %s."
% (jobname, id, str(changes)))
inconsistencies = inconsistencies + 1
# Report if the sets of filespecs differ.
p4_filespecs = self.job_filespecs(job)
dt_filespecs = issue.filespecs()
for p4_filespec, dt_filespec in self.filespecs_differences(dt_filespecs, p4_filespecs):
if p4_filespec and not dt_filespec:
print("Job '%s' has associated filespec '%s' but there is "
"no corresponding filespec for issue '%s'."
% (jobname, p4_filespec, id))
inconsistencies = inconsistencies + 1
elif not p4_filespec and dt_filespec:
print("Issue '%s' has associated filespec '%s' but there "
"is no corresponding filespec for job '%s'."
% (id, dt_filespec.name(), jobname))
inconsistencies = inconsistencies + 1
else:
# Corresponding filespecs can't differ (since their only
# attribute is their name).
assert(0)
# Report if the sets of fixes differ.
p4_fixes = self.job_fixes(job)
dt_fixes = issue.fixes()
for p4_fix, dt_fix in self.fixes_differences(dt_fixes, p4_fixes):
if p4_fix and not dt_fix:
print("Change %s fixes job '%s' but there is "
"no corresponding fix for issue '%s'."
% (p4_fix['Change'], jobname, id))
inconsistencies = inconsistencies + 1
elif not p4_fix and dt_fix:
print("Change %d fixes issue '%s' but there is "
"no corresponding fix for job '%s'."
% (dt_fix.change(), id, jobname))
inconsistencies = inconsistencies + 1
else:
print("Change %s fixes job '%s' with status '%s', but "
"change %d fixes issue '%s' with status '%s'."
% (p4_fix['Change'], jobname, p4_fix['Status'],
dt_fix.change(), id, dt_fix.status()))
inconsistencies = inconsistencies + 1
# There should be no remaining jobs, so any left are in error.
for job in jobs.values():
if issues[job['P4DTI-issue-id']]:
print("Job '%s' is marked as being replicated to issue '%s' "
"but that issue is being replicated to job '%s'."
% (job['Job'], job['P4DTI-issue-id'],
issues[job['P4DTI-issue-id']].jobname()))
inconsistencies = inconsistencies + 1
else:
print("Job '%s' is marked as being replicated to issue '%s' "
"but that issue either doesn't exist or is not being "
"replicated by this replicator."
% (job['Job'], job['P4DTI-issue-id']))
inconsistencies = inconsistencies + 1
# Report on success/failure.
print
print "Consistency check completed."
print len(issues), "issues checked."
if inconsistencies == 0:
print "Looks all right to me."
elif inconsistencies == 1:
print "1 inconsistency found."
else:
print inconsistencies, "inconsistencies found."
# fixes_differences(dt_fixes, p4_fixes). Each argument is a list of fixes
# for the same job/issue. Return list of pairs (p4_fix, dt_fix) of
# corresponding fixes which differ. Elements of pairs are None where there
# is no corresponding fix.
def fixes_differences(self, dt_fixes, p4_fixes):
# Make hash from change number to p4 fix.
p4_fix_by_change = {}
for p4_fix in p4_fixes:
p4_fix_by_change[int(p4_fix['Change'])] = p4_fix
# Make pairs (dt fix, corresponding p4 fix or None).
pairs = []
for dt_fix in dt_fixes:
if not p4_fix_by_change.has_key(dt_fix.change()):
pairs.append((None, dt_fix))
else:
p4_fix = p4_fix_by_change[dt_fix.change()]
del p4_fix_by_change[dt_fix.change()]
if dt_fix.status() != p4_fix['Status']:
pairs.append((p4_fix, dt_fix))
# Remaining p4 fixes are unpaired.
for p4_fix in p4_fix_by_change.values():
pairs.append((p4_fix, None))
return pairs
# filespecs_differences(dt_filespecs, p4_filespecs). Each argument is a
# list of filespecs for the same job/issue. Return list of pairs
# (p4_filespec, dt_filespec) of filespecs which differ. Elements of pairs
# are None where there is no corresponding filespec (this is always the
# case since there is no associated information with a filespec; the
# function is like this for consistency with fixes_differences, and so that
# it is easy to extend if there is ever a way to associate information with
# a filespec, for example the nature of the association -- see requirement
# 55).
def filespecs_differences(self, dt_filespecs, p4_filespecs):
# Make hash from name to p4 filespec.
p4_filespec_by_name = {}
for p4_filespec in p4_filespecs:
p4_filespec_by_name[p4_filespec] = p4_filespec
# Make pairs (dt filespec, None).
pairs = []
for dt_filespec in dt_filespecs:
if not p4_filespec_by_name.has_key(dt_filespec.name()):
pairs.append((None, dt_filespec))
else:
del p4_filespec_by_name[dt_filespec.name()]
# Make pairs (None, p4 filespec).
for p4_filespec in p4_filespec_by_name.values():
pairs.append((p4_filespec, None))
return pairs
# init(). Set up Perforce and the defect tracking system so that
# replication can proceed.
def init(self):
# Check that the Perforce server version is supported by the
# integration.
server_version_re = re.compile('Server version: '
'[^/]+/[^/]+/[^/]+/([0-9]+)')
changelevel = 0
supported_changelevel = 16895
for x in self.p4_run('info'):
if x.has_key('code') and x['code'] == 'info' and x.has_key('data'):
match = server_version_re.match(x['data'])
if match:
changelevel = int(match.group(1))
if changelevel < supported_changelevel:
raise self.error, \
("Perforce server changelevel %d is not "
"supported by P4DTI. Server must be at "
"changelevel %d or above"
% (changelevel, supported_changelevel))
else:
break
if not changelevel:
raise self.error, "p4 info didn't report a recognisable version"
# Initialize the defect tracking system.
self.dt.init()
# Make a client for the replicator.
self.p4_run('client -i', self.p4_run('client -o'))
# TODO: Initialize Perforce by adding fields to jobspec if not present.
# For the moment I can't add fields to the jobspec, so I'll just check
# that they are there. I can't even check the presence of the
# P4DTI-user field this way, since it isn't given a value for a
# non-existent job like the one we're asking for here.
job = self.job('P4DTI-no-such-job')
for field in ['P4DTI-filespecs', 'P4DTI-issue-id', 'P4DTI-rid',
'P4DTI-action']:
if not job.has_key(field):
raise self.error, ("Field '%s' not found in Perforce jobspec"
% field)
# Has the logger been started? (We must be careful not to set the
# logger counter to 0 more than once; this will confuse Perforce
# according to Chris Seiwald's e-mail .
logger_started = 0
logger_re = re.compile('logger = ([0-9]+)$')
counters = self.p4_run('counters')
for c in counters:
if (c.has_key('code') and c['code'] == 'info' and c.has_key('data')
and logger_re.match(c['data'])):
logger_started = 1
# If not, start it.
if not logger_started:
self.p4_run('counter logger 0')
# job(jobname). Return the Perforce job with the given name if it exists,
# or an empty job specification (otherwise).
def job(self, jobname):
jobs = self.p4_run('job -o %s' % jobname)
if len(jobs) != 1 or not jobs[0].has_key('Job'):
raise self.error, ("expected a job but found %s" % str(jobs))
elif jobs[0]['Job'] != jobname:
raise self.error, ("asked for job '%s' but got job '%s'"
% (jobname, jobs[0]['Job']))
else:
return jobs[0]
# job_conflicting_p(job). Return true iff the job is marked as
# conflicting.
def job_conflicting_p(self, job):
return job['P4DTI-action'] == 'wait'
# job_filespecs(job). Return a list of filespecs for the given job. Each
# element of the list is a filespec, as a string.
def job_filespecs(self, job):
filespecs = string.split(job['P4DTI-filespecs'], '\n')
# Since Perforce text fields are terminated with a newline, the last
# item of the list should be empty. Remove it.
if filespecs:
assert(filespecs[-1] == '')
filespecs = filespecs[:-1]
return filespecs
# job_fixes(job). Return a list of fixes for the given job. Each element
# of the list is a dictionary with keys Change, Client, User, Job, and
# Status.
def job_fixes(self, job):
return self.p4_run('fixes -j %s' % job['Job'])
# log(format, arguments, priority). Write the message to the replicator's
# log.
def log(self, format, arguments = (), priority = None):
format = "%s\t" + format
if type(arguments) == types.TupleType:
arguments = (self.rid,) + arguments
else:
arguments = (self.rid, arguments)
self.config['logger'].log("P4DTI-0", format, arguments, priority)
# mail(to, subject, body, cc=''). Send e-mail to the given recipient
# integration with the given subject and body.
def mail(self, to, subject, body, cc=''):
if self.config['administrator-address'] and self.config['smtp-server']:
smtp = smtplib.SMTP(self.config['smtp-server'])
from_address = self.config['replicator-address']
message = ("From: %s\nTo: %s\nCC: %s\nSubject: %s\n\n%s"
% (from_address, to, cc, subject, body))
smtp.sendmail(from_address, to, message)
smtp.quit()
# mail_administrator(subject, body). Send e-mail to the administrator of
# the integration with the given subject and body.
def mail_administrator(self, subject, body):
self.mail(self.config['administrator-address'], subject, body)
# p4_run(arguments, input). Run the p4 client with the given arguments and
# input, returning a list of results. See the p4 module for details.
def p4_run(self, arguments, input = None, verbose = 0):
return p4.run(arguments, input,
client_executable = self.config['p4-client-executable'],
port = self.config['p4-port'],
user = self.config['p4-user'],
password = self.config['p4-password'],
client = self.config['p4-client'],
verbose = verbose)
# conflict(issue, job, message). Report that the given issue conflicts
# with the given job. The message argument is a string containing
# additional detail about the conflict. Mark the issue and job as
# conflicting if they are not already.
def conflict(self, issue, job, message):
try:
# We can't just update the issue in hand because that might have changed
# in the course of replicating some fields. So fetch it again. This is quite
# unsatisfactory: we should keep better track of the old and new versions of
# the issue. GDR 2000-10-27.
old_issue = self.dt.issue(issue.id())
old_issue.update_action('wait')
# The same problem applies to the job. GDR 2000-10-27.
old_job = self.job(job['Job'])
self.update_job_action(old_job, 'wait')
finally:
self.log("Issue '%s' conflicts with corresponding job '%s'. %s",
(issue.id(), job['Job'], message))
subject = ("Issue '%s' conflicts with corresponding job '%s'."
% (issue.id(), job['Job']))
body = ("%s\n\n%s\n\nIssue:\n%s\n\nJob:\n%s"
% (subject, message, issue, job))
self.mail_administrator(subject, body)
# conflict_policy(issue, job). This method is called when both the issue
# and the corresponding job have changed since the last time they were
# consistent. Return 'p4' if the Perforce job is correct and should be
# replicated to the defect tracker. Return 'dt' if the defect tracking
# issue is correct and should be replicated to Perforce. Return 'none' if
# the replicator should take no further action. Any other result indicates
# that the replicator should treat the situation as a conflict and proceed
# accordingly. The default policy is to treat the situation as a conflict.
def conflict_policy(self, issue, job):
return 'both'
# poll(). Poll the DTS for changed issues. Poll Perforce for changed jobs
# and changelists. Replicate all of these entities.
def poll(self):
# Get the changed issues and jobs.
changed_issues, dt_data = self.dt.changed_issues()
if len(changed_issues) == 1:
self.log('1 issue has changed')
elif len(changed_issues) > 1:
self.log('%d issues have changed', len(changed_issues))
changed_jobs, changelists, last_log_entry = self.changed_p4_entities()
if len(changed_jobs) == 1:
self.log('1 job has changed')
elif len(changed_jobs) > 1:
self.log('%d jobs have changed', len(changed_jobs))
# Replicate the issues and the jobs.
self.replicate_many(changed_issues, changed_jobs)
# Replicate the affected changelists.
for c in changelists:
if self.dt.replicate_changelist(c):
self.log("Replicated changelist %s", c['Change'])
# Tell the defect tracker and Perforce that we've finished replicating
# these changes.
self.dt.changed_issues_are_replicated(dt_data)
self.changes_are_replicated(last_log_entry)
# replicate_many(issues, jobs). Replicate the issues and jobs. The issues
# argument is a list of issues (which must belong to a subclass of
# defect_tracker_issue; the jobs list is a hash from jobname to job).
#
# The reason why the arguments have different conventions (list vs hash) is
# that the algorithm for getting the changed jobs from the p4 logger outpt
# involves constructing a hash from jobname to job, and it seems silly to
# turn this hash back into a list only to immediately turn it back into a
# hash again.
def replicate_many(self, issues, jobs):
# Make a list of triples of (defect tracking issue, Perforce job,
# status). Status is 'dt' if the defect tracking issue has changed but
# not the Perforce job; 'p4' if vice versa; 'both' if both have
# changed. TODO: mitigate effects of race conditions?
triples = []
# Go through issues making triples. Set up issues for replication if
# they are not already replicated. Delete corresponding jobs from the
# jobs dictionary.
for issue in issues:
# Issue not set up for replication yet?
if not issue.rid():
# Should issue be replicated by this replicator?
if issue.replicate_p():
issue.setup_for_replication()
self.log("Set up issue '%s' to replicate to job '%s'",
(issue.id(), issue.jobname()))
else:
# Don't replicate this issue at all.
continue
jobname = issue.jobname()
if jobs.has_key(jobname):
job = jobs[jobname]
triples.append((issue, job, 'both'))
del jobs[jobname]
else:
job = self.job(jobname)
triples.append((issue, job, 'dt'))
# Now go through the remaining changed jobs.
for job in jobs.values():
issue = self.dt.issue(job['P4DTI-issue-id'])
if not issue:
raise self.error, "No issue '%s'" % job['P4DTI-issue-id']
triples.append((issue, job, 'p4'))
# Now process each triple.
for issue, job, status in triples:
try:
self.replicate(issue, job, status)
except self.conflict_error, message:
self.conflict(issue, job, message)
except: # Should catch TSAPI errors only.
type, value, tb = sys.exc_info()
tb_list = traceback.format_exception(type, value, tb)
message = ("An error occurred while trying to replicate.\n\n"
+ string.join(tb_list, ''))
# If only the Perforce job changed (so we were replicating to
# the defect tracker), then it is most likely that the error is
# due to the change being rejected by the defect tracker. So
# attempt to set the job back to the way it was and e-mail the
# person responsible. Note however that we can't be sure who
# was responsible since changes to jobs made via fixes don't
# update the user field. GDR 2000-10-31.
user = job['P4DTI-user']
if status == 'p4' and user != self.config['p4-user']:
self.log("Job '%s' could not be replicated to issue '%s'. %s",
(job['Job'], issue.id(), message))
try:
# Get the issue again, since it might have been changed
# in memory in the course of the failed replication.
# Note new variable name so as not to overwrite the old
# issue. (Can we avoid all this nonsense by keeping
# better track of old and new issues?) GDR 2000-10-31.
issue2 = self.dt.issue(job['P4DTI-issue-id'])
if not issue2:
raise self.error, "No issue '%s'" % job['P4DTI-issue-id']
self.replicate(issue2, job, 'dt')
p4_user = self.p4_run('user -o %s' % user)[0]
email = p4_user['Email']
self.log("Mailing '%s'.", (email,))
subject = "Job '%s' could not be replicated to issue '%s'." % (job['Job'], issue.id())
body = ("%s\n\n%s\n\nIssue:\n%s\n\nJob:\n%s"
% (subject, message, issue2, job))
self.mail(email, subject, body,
self.config['administrator-address'])
except:
# Replicating back failed, so it's really a conflict.
# Really I should make tracebacks for both errors here.
# GDR 2000-10-31.
type, value, tb = sys.exc_info()
tb_list = traceback.format_exception(type, value, tb)
message2 = ("An error occurred while trying to replicate.\n\n"
+ string.join(tb_list, ''))
self.conflict(issue, job, message + '\n' + message2)
else:
self.conflict(issue, job, message)
# replicate(issue, job, changed). Replicate an issue to or from the
# corresponding job. The changed argument is 'dt' if the defect tracking
# issue has changed but not the Perforce job; 'p4' if vice versa; 'both' if
# both have changed. Return true iff the replication was successful.
def replicate(self, issue, job, changed):
id = issue.id()
jobname = job['Job']
# Figure out what to do with this issue and job. Report a conflict?
# Do nothing? Overwrite issue with job? Overwrite job with issue?
# The action arguments may tell us what to do immediately.
issue_action = issue.action()
job_action = job['P4DTI-action']
if not self.action_table.has_key((issue_action, job_action)):
raise self.conflict_error, \
("Issue '%s' has action '%s' and job '%s' has action '%s': "
"this combination is illegal."
% (id, issue_action, jobname, job_action))
action = self.action_table[(issue_action, job_action)]
# Action 'conflict' means that there was already a conflict between the
# job and the issue, and now one or both have changed. Report this,
# since the administrator may need to know what's been happening to
# these entities in order to resolve the conflict.
if action == 'conflict':
if changed == 'dt':
raise self.conflict_error, "Existing conflict; issue changed."
elif changed == 'p4':
raise self.conflict_error, "Existing conflict; job changed."
else:
assert(changed == 'both')
raise self.conflict_error, "Existing conflict; both changed."
# Action 'normal' means that the ordinary decision procedure should
# apply. This involves consulting the automatic conflict resolution
# policy when both have changed. Otherwise, overwrite the unchanged
# entity with the changed entity.
elif action == 'normal':
# Both issues changed? Apply the conflict resolution policy.
if changed == 'both':
self.log("Issue '%s' and job '%s' have both changed. "
"Consulting conflict resolution policy.",
(id, jobname))
action = self.conflict_policy(issue, job)
if action == 'none':
self.log("Conflict resolution policy decided: no action.")
return None
elif action == 'dt':
self.log("Conflict resolution policy decided: "
"Overwrite job with issue.")
elif action == 'p4':
self.log("Conflict resolution policy decided: "
"Overwrite issue with job.")
else:
self.log("Conflict resolution policy decided: "
"Report conflict.")
raise self.conflict_error, "Both changed."
else:
action = changed
elif action == 'p4':
self.log("Overwrite issue '%s' with job '%s' (actions are %s/%s)",
(id, jobname, issue_action, job_action))
subject = "Issue '%s' overwritten by job '%s'" % (id, jobname)
body = ("Action field of issue '%s' is '%s' and the action field "
"of job '%s' is '%s'. This means that the issue will be "
"overwritten by the job.\n\nThe old issue was:\n%s"
% (id, issue_action, jobname, job_action, issue))
self.mail_administrator(subject, body)
elif action == 'dt':
self.log("Overwrite job '%s' with issue '%s' (actions are %s/%s)",
(jobname, id, job_action, issue_action))
subject = "Job '%s' overwritten by issue '%s'" % (jobname, id)
body = ("Action field of issue '%s' is '%s' and the action field "
"of job '%s' is '%s'. This means that the job will be "
"overwritten by the issue.\n\nThe old job was:\n%s"
% (id, issue_action, jobname, job_action, job))
self.mail_administrator(subject, body)
else:
assert(0)
# By now we should have decided which way to replicate.
assert(action == 'dt' or action == 'p4')
if action == 'dt':
self.log("Replicating issue '%s' to job '%s'",
(issue.id(), job['Job']))
return self.replicate_issue_to_job(issue, job)
else: # action == 'p4'
self.log("Replicating job '%s' to issue '%s'",
(job['Job'], issue.id()))
return self.replicate_job_to_issue(job, issue)
# replicate_issue_to_job(issue, old_job). Replicate the given issue from
# the defect tracker to Perforce. Return true iff the issue was replicated
# successfully. Otherwise throw an exception.
def replicate_issue_to_job(self, issue, old_job):
# Transform the issue into a job. This has to be done first because
# the job might be new, and we won't be able to replicate fixes or
# filespecs until the job's been created (p4 fix won't accept
# non-existent jobnames). I suppose I could create a dummy job to act
# as a placeholder here, but that's not easy at all -- you have to know
# quite a lot about the jobspec to be able to create a job.
new_job, changes = issue.transform_to_job(old_job)
if changes:
self.log("-- Changed fields: %s", `changes`)
self.p4_run('job -i', [new_job])
else:
self.log("-- No issue fields were replicated.")
# Replicate filespecs.
dt_filespecs = issue.filespecs()
p4_filespecs = self.job_filespecs(new_job)
if self.filespecs_differences(dt_filespecs, p4_filespecs):
names = map(lambda(f): f.name(), dt_filespecs)
new_job['P4DTI-filespecs'] = string.join(names, '\n')
self.p4_run('job -i', [new_job])
self.log("-- Filespecs changed to '%s'", string.join(names))
# Replicate fixes.
p4_fixes = self.job_fixes(new_job)
dt_fixes = issue.fixes()
job_status = new_job['Status']
for p4_fix, dt_fix in self.fixes_differences(dt_fixes, p4_fixes):
if p4_fix and not dt_fix:
self.p4_run('fix -d -c %s %s'
% (p4_fix['Change'], p4_fix['Job']))
self.log("-- Deleted fix for change %s", p4_fix['Change'])
elif not p4_fix and dt_fix:
self.p4_run('fix -s %s -c %d %s'
% (dt_fix.status(), dt_fix.change(),
issue.jobname()))
job_status = dt_fix.status()
self.log("-- Added fix for change %d with status %s",
(dt_fix.change(), dt_fix.status()))
elif p4_fix['Status'] != dt_fix.status():
self.p4_run('fix -s %s -c %d %s'
% (dt_fix.status(), dt_fix.change(),
issue.jobname()))
job_status = dt_fix.status()
self.log("-- Fix for change %d updated to status %s",
(dt_fix.change(), dt_fix.status()))
else:
# This should't happen, since fixes_differences returns only a
# list of pairs which differ.
assert(0)
# It might be the case that the job status has been changed by
# replicating a fix from the defect tracker. But this changed status
# won't be right. So restore the correct status if necessary.
if job_status != new_job['Status']:
self.p4_run('job -i', [new_job])
# Job and issue are up to date.
issue.update_action('replicate')
self.update_job_action(new_job, 'replicate')
return 1
# replicate_job_to_issue(job, issue). Replicate the given job from
# Perforce to the defect tracker. Return true iff the job was replicated
# successfully.
def replicate_job_to_issue(self, job, issue):
# Replicate fixes.
p4_fixes = self.job_fixes(job)
dt_fixes = issue.fixes()
fix_diffs = self.fixes_differences(dt_fixes, p4_fixes)
for p4_fix, dt_fix in fix_diffs:
self.log("p4_fix = %s", (p4_fix,))
if dt_fix and not p4_fix:
dt_fix.delete()
self.log("-- Deleted fix for change %d", dt_fix.change())
elif not dt_fix:
issue.add_fix(p4_fix)
self.log("-- Added fix for change %s with status %s",
(p4_fix['Change'], p4_fix['Status']))
elif dt_fix.status() != p4_fix['Status']:
dt_fix.update(p4_fix)
self.log("-- Fix for change %s updated to status %s",
(p4_fix['Change'], p4_fix['Status']))
else:
# This should't happen, since fixes_differences returns only a
# list of pairs which differ.
assert(0)
# Replicate filespecs.
p4_filespecs = self.job_filespecs(job)
dt_filespecs = issue.filespecs()
filespec_diffs = self.filespecs_differences(dt_filespecs, p4_filespecs)
for p4_filespec, dt_filespec in filespec_diffs:
if dt_filespec and not p4_filespec:
dt_filespec.delete()
self.log("-- Deleted filespec %s", dt_filespec.name())
elif not dt_filespec:
issue.add_filespec(p4_filespec)
self.log("-- Added filespec %s", p4_filespec)
else:
# This should't happen, since filespecs_differences returns
# only a list of pairs which differ.
assert(0)
# Transform the job into an issue and update the issue.
changes, transition = issue.transform_from_job(job, fix_diffs,
filespec_diffs)
if changes:
self.log("-- Changed fields: %s", `changes`)
if transition:
self.log("-- Transition: %s", transition)
issue.update(transition)
else:
self.log("-- No job fields were replicated.")
# Job and issue are up to date.
issue.update_action('replicate')
self.update_job_action(job, 'replicate')
return 1
def replicate_changelists(self):
# Replicate all the changelists.
self.log("Checking changelists to see if they need replicating...")
changelists = self.p4_run('changes')
self.log("-- %d changelists to check", len(changelists))
for c in changelists:
c2 = self.p4_run('change -o %s' % c['change'])[0]
if self.dt.replicate_changelist(c2):
self.log("Replicated changelist %s", c['change'])
# run() repeatedly polls the DTS.
def run(self):
while 1:
self.log("Polling...")
self.poll()
time.sleep(self.config['poll-period'])
# update_job_action(job, action).
def update_job_action(self, job, action):
# Only update the job action if it's being replicated by this
# replicator. See job000020.
if job['P4DTI-rid'] == self.rid and job['P4DTI-action'] != action:
job['P4DTI-action'] = action
self.p4_run('job -i', [job])