# Errors # # p4.error A Perforce command returned with error code. # ???? Failed to run a Perforce command. # teamtrack.error # teamtrack.tsapi_error import types def is_action(action): return (action in ["replicate", "wait", "keep", "discard"]) # defect_tracker is an abstract class of defect trackers. class defect_tracker: # A human-readable name for the defect tracker, for use in log and error # messages. name = 'Generic' # The information needed to connect to a server (typically server address, # port, user name and password). server_address_type = None # A connection to the defect tracking server. connection_type = None # An issue record. issue_type = None # A changelist record. changelist_type = None # A fix record. fix_type = None # A marker in the defect tracker's history of changes, or some equivalent. # This contains enough information to be able to mark a list of changed # issues as having been replicated. marker_type = None # Errors that this class may raise. privilege_error = translate_error = server_error = # open_connection : server_address -> connection def open_connection(self, server_address): assert(isinstance(server_address, self.server_address_type)) assert(0) # changed_entities : connection -> issue list * changelist list * marker # # Return a list of the issues and changelists that are new or have changed # in the defect tracker. Also return a marker in the defect tracker's # history of changes that can be used to record that these issues have been # replicated. def changed_entities(self, connection): assert(isinstance(connection, self.connection_type)) assert(0) # mark_changed_entities_done : (connection, marker) -> unit # # Update the defect tracker's history of changes so that all the changes up # to the marker are recorded as having been replicated. def mark_changed_entities_done(self, connection, marker): assert(isinstance(connection, self.connection_type)) assert(isinstance(marker, self.marker_type)) assert(0) # issue_human_name : connection * issue -> string # # Return a human-readable name for the issue. For example, in TeamTrack # this will be something like "BUG00034"; in Perforce, something like # "job000034". def issue_human_name(self, connection, issue): assert(isinstance(connection, self.connection_type)) assert(isinstance(issue, self.issue_type)) assert(0) # issue_id : connection * issue -> string # # Return an identifier that uniquely identifies the issue (at least among # the issues in the given defect tracker) and can be used to find it again. # For example, in TeamTrack this is a string form of the database id # ("33"). In Perforce, this is the jobname ("job000033"). def issue_id(self, connection, issue): assert(isinstance(connection, self.connection_type)) assert(isinstance(issue, self.issue_type)) assert(0) # issue_action : connection * issue -> string # # The action setting for the issue ("replicate", "wait", "keep" or # "discard"). def issue_action(self, connection, issue): assert(isinstance(connection, self.connection_type)) assert(isinstance(issue, self.issue_type)) assert(0) # set_issue_action : connection * issue * string -> unit # # Set the action for the issue ("replicate", "wait", "keep" or "discard"). def issue_action(self, connection, issue, action): assert(isinstance(connection, self.connection_type)) assert(isinstance(issue, self.issue_type)) assert(isinstance(action, types.StringType)) assert(is_action(action)) assert(0) # issue_rid : connection * issue -> string # # The identifier of the replicator replicating the issue, or the empty # string if the issue is not replicated. def issue_rid(self, connection, issue): assert(isinstance(connection, self.connection_type)) assert(isinstance(issue, self.issue_type)) assert(0) # set_issue_rid : connection * issue * string -> unit # # Set the replicator identifier for the issue. def issue_rid(self, connection, issue, rid): assert(isinstance(connection, self.connection_type)) assert(isinstance(issue, self.issue_type)) assert(isinstance(rid, types.StringType)) assert(0) # id_to_issue : connection * string -> issue # # Return the issue whose identifier has the given identifier, or None if # there is no such issue. def id_to_issue(self, connection, id): assert(isinstance(connection, self.connection_type)) assert(isinstance(id, types.StringType)) assert(0) # copy_issue : self, connection, issue -> issue # # Return a copy of the issue. def copy_issue(self, connection, issue): assert(isinstance(connection, self.connection_type)) assert(isinstance(issue, self.issue_type)) assert(0) # issue_field : connection * issue * string -> value # # Return the value in an issue field. # # Raise KeyError if there is no such field. def issue_field(self, connection, issue, field_name): assert(isinstance(connection, self.connection_type)) assert(isinstance(issue, self.issue_type)) assert(isinstance(field_name, types.StringType)) assert(0) # set_issue_field : connection * issue * string * value -> unit # # Set the value in an issue field. # # Raise KeyError if there is no such field. def set_issue_field(self, connection, issue, field_name, value): assert(isinstance(connection, self.connection_type)) assert(isinstance(issue, self.issue_type)) assert(isinstance(field_name, types.StringType)) assert(0) # commit_issue : connection * issue -> unit # # Update the issue in the database. def commit_issue(self, connection, issue): assert(isinstance(connection, self.connection_type)) assert(isinstance(issue, self.issue_type)) assert(0) # new_issue : connection -> issue # # Construct a new issue (not in the database). def new_issue(self, connection): assert(isinstance(connection, self.connection_type)) assert(0) # submit_issue : connection * issue -> unit # # Submit a new issue to the database. def submit_issue(self, connection, issue): assert(isinstance(connection, self.connection_type)) assert(isinstance(issue, self.issue_type)) assert(0) # issue_fixes : connection * Issue -> fix list # # Return a list of fixes for the issue. def issue_fixes(self, connection, issue): assert(isinstance(connection, self.connection_type)) assert(isinstance(issue, self.issue_type)) assert(0) class replicator: # The pair of defect trackers which are being replicated. dt = (None, None) # Connections to the defect tracking servers. connection = (None, None) def poll(self): issues = ([], []) changelists = ([], []) marker = ([], []) for i in range(2): (issues[i], changelists[i], marker[i]) = \ dt[i].changed_entities(connection[i]) self.log("%s: %d changed issue%s", (dt[i].name, len(changed_issues), ('', 's')[len(changed_issues) == 1])) # Tell the defect trackers that we've finished replicating these # entities. for i in range(2): dt[i].mark_changed_entities_done(connection[i], marker[i]) def replicate_issue(self, issue, dt):