# DTG # Python interface to P4DTG - like the p4dtg-test program. from ctypes import * class DTGError(Structure): """ struct DTGError { char *message; int can_continue; // 0: connection failure }; """ _fields_ = [("message", c_char_p), ("can_continue", c_int)] @classmethod def from_param(cls, obj): if isinstance(obj, cls): return obj raise TypeError class DTGAttribute(Structure): """ struct DTGAttribute { char *name; // Key, will be the name field in DTGField struct char *label; // Displayed in UI char *desc; // Displayed in UI char *def; // default, May be null int required; // 0:Not required, o/w:Required struct DTGAttribute *next; }; """ @classmethod def from_param(cls, obj): if isinstance(obj, cls): return obj raise TypeError DTGAttribute._fields_ = [("name", c_char_p), ("label", c_char_p), ("desc", c_char_p), ("default", c_char_p), ("required", c_int), ("next", POINTER(DTGAttribute))] class Attribute(object): "Pythonic attr class" name = None label = None desc = None default = None required = None def __init__(self, name=None, label=None, desc=None, default=None, required=None): if name: self.name = name if label: self.label = label if desc: self.desc = desc if default: self.default = default if required: self.required = required def __repr__(self): return "Name: %s, label: %s, default: %s, required: %s\ndesc: %s" % ( self.name, self.label, self.default, self.required, self.desc) def __str__(self): return self.__repr__() class Field(object): "Makes DTGFieldDesc more pythonic" name = None type = None readonly = None select_values = [] def __init__(self, name = None, type = None, readonly = None, select_values = []): if name: self.name = name if type: self.type = type if readonly: self.readonly = readonly if select_values: self.select_values = select_values def __repr__(self): return "Name: %s, type: %s, readonly: %s, select_values: %s" % ( self.name, self.type, self.readonly, str(self.select_values)) def __str__(self): return self.__repr__() def decode_attrs(attr_p): "Convert to an array" attrs = [] while attr_p: attr = Attribute(attr_p.contents.name, attr_p.contents.label, attr_p.contents.desc, attr_p.contents.default, attr_p.contents.required) attrs.append(attr) attr_p = attr_p.contents.next return attrs class DTGField(Structure): """ struct DTGField { char *name; char *value; struct DTGField *next; """ @classmethod def from_param(cls, obj): if isinstance(obj, cls): return obj raise TypeError DTGField._fields_ = [("name", c_char_p), ("value", c_char_p), ("next", POINTER(DTGField))] class DTGStrList(Structure): """ struct DTGStrList { char *value; struct DTGStrList *next; """ @classmethod def from_param(cls, obj): if isinstance(obj, cls): return obj raise TypeError DTGStrList._fields_ = [("value", c_char_p), ("next", POINTER(DTGStrList))] def array_from_strlist(list_p): "Return array" arr = [] while list_p: arr.append(list_p.contents.value) list_p = list_p.contents.next return arr def check_error(err): "Checks for returned error messages" if err and err.message: print("\n\n****", err.message, "\n\n") if not err.can_continue: raise Exception("%s: Can't continue" % err.message) class DTGFieldDesc(Structure): """ struct DTGFieldDesc { char *name; char *type; // word, date, line, text, select int readonly; // 0:rw, >= 1:ro, 2:mod-date, 3:mod-user, 4:defectid struct DTGStrList *select_values; struct DTGFieldDesc *next; """ @classmethod def from_param(cls, obj): if isinstance(obj, cls): return obj raise TypeError DTGFieldDesc._fields_ = [("name", c_char_p), ("type", c_char_p), ("readonly", c_int), ("select_values", POINTER(DTGStrList)), ("next", POINTER(DTGFieldDesc))] class DTGDate(Structure): """ struct DTGDate { int year; int month; int day; int hour; int minute; int second; """ _fields_ = [("year", c_int), ("month", c_int), ("day", c_int), ("hour", c_int), ("minute", c_int), ("second", c_int)] @classmethod def from_param(cls, obj): if isinstance(obj, cls): return obj raise TypeError def __str__(self): return "%4d/%02d/%02d %02d:%02d:%02d" % (self.year, self.month, self.day, self.hour, self.minute, self.second) def __repr__(self): return self.__str__() class DTG(object): "Interface to DLL" def __init__(self, dll_path): self.dtg = CDLL(dll_path) self.err = DTGError() # Various function descriptions self.dt_get_name = self.dtg.dt_get_name self.dt_get_name.argtypes = [POINTER(DTGError)] self.dt_get_name.restype = c_char_p self.dt_get_module_version = self.dtg.dt_get_module_version self.dt_get_module_version.argtypes = [POINTER(DTGError)] self.dt_get_module_version.restype = c_char_p self.dt_list_attrs = self.dtg.dt_list_attrs self.dt_list_attrs.restype = POINTER(DTGAttribute) """ typedef void *(dt_connect_ftn)( const char *server, const char *user, const char *pass, const struct DTGField *attrs, struct DTGError *error ); """ self.dt_connect = self.dtg.dt_connect self.dt_connect.argtypes = [c_char_p, c_char_p, c_char_p, POINTER(DTGField), POINTER(DTGError)] self.dt_connect.restype = c_void_p self.dt_get_server_version = self.dtg.dt_get_server_version self.dt_get_server_version.argtypes = [c_void_p, POINTER(DTGError)] self.dt_get_server_version.restype = c_char_p # struct DTGDate *dt_get_server_date( void *dtID, struct DTGError *error ) self.dt_get_server_date = self.dtg.dt_get_server_date self.dt_get_server_date.argtypes = [c_void_p, POINTER(DTGError)] self.dt_get_server_date.restype = POINTER(DTGDate) # struct DTGStrList *dt_list_projects( void *dtID, struct DTGError *error ) self.dt_list_projects = self.dtg.dt_list_projects self.dt_list_projects.argtypes = [c_void_p, POINTER(DTGError)] self.dt_list_projects.restype = POINTER(DTGStrList) # void *dt_get_project( void *dtID, const char *project, struct DTGError *error ) self.dt_get_project = self.dtg.dt_get_project self.dt_get_project.argtypes = [c_void_p, c_char_p, POINTER(DTGError)] self.dt_get_project.restype = c_void_p # struct DTGFieldDesc *proj_list_fields( void *projID, struct DTGError *error ) self.proj_list_fields = self.dtg.proj_list_fields self.proj_list_fields.argtypes = [c_void_p, POINTER(DTGError)] self.proj_list_fields.restype = POINTER(DTGFieldDesc) # struct DTGStrList *proj_list_changed_defects( void *projID, # int max_rows, # struct DTGDate *since, # const char *mod_date_field, # const char *mod_by_field, # const char *exclude_user, # struct DTGError *error ) self.proj_list_changed_defects = self.dtg.proj_list_changed_defects self.proj_list_changed_defects.argtypes = [c_void_p, c_int, POINTER(DTGDate), c_char_p, c_char_p, c_char_p, POINTER(DTGError)] self.proj_list_changed_defects.restype = POINTER(DTGStrList) # void *proj_get_defect( void *projID, const char *defect, struct DTGError *error ) self.proj_get_defect = self.dtg.proj_get_defect self.proj_get_defect.argtypes = [c_void_p, c_char_p, POINTER(DTGError)] self.proj_get_defect.restype = c_void_p # char *defect_get_field( void *defectID, # const char *field, # struct DTGError *error ) self.defect_get_field = self.dtg.defect_get_field self.defect_get_field.argtypes = [c_void_p, c_char_p, POINTER(DTGError)] self.defect_get_field.restype = c_char_p # void defect_set_field( void *defectID, # const char *name, const char *value, # struct DTGError *error ) self.defect_set_field = self.dtg.defect_set_field self.defect_set_field.argtypes = [c_void_p, c_char_p, c_char_p, POINTER(DTGError)] self.defect_set_field.restype = None # char *defect_save( void *defectID, struct DTGError *error ) self.defect_save = self.dtg.defect_save self.defect_save.argtypes = [c_void_p, POINTER(DTGError)] self.defect_save.restype = c_char_p def check_err(self): "Check for errors" pass def get_name(self): "Get plugin name" name = c_char_p() name.value = self.dt_get_name(byref(self.err)) check_error(self.err) return name.value def list_attrs(self): "Return array of attributes" attr = self.dt_list_attrs() return decode_attrs(attr) class DTGServer(object): "Interface to server object" def __init__(self, dtg): self.dtg = dtg # basic interface self.server = None self.err = DTGError() def connect(self, server, user, passwd, attrs): "Connect to server" self.server = self.dtg.dt_connect(server, user, passwd, attrs, byref(self.err)) check_error(self.err) if self.err.message: raise Exception(self.err.message) def get_server_version(self): ver = c_char_p() ver.value = self.dtg.dt_get_server_version(self.server, byref(self.err)) check_error(self.err) return ver.value def get_server_date(self): date = self.dtg.dt_get_server_date(self.server, byref(self.err)) check_error(self.err) return date def list_projects(self): proj_p = self.dtg.dt_list_projects(self.server, byref(self.err)) check_error(self.err) projects = array_from_strlist(proj_p) return projects def get_project(self, project_name): "Get project object" # void *dt_get_project( void *dtID, const char *project, struct DTGError *error ) proj_p = self.dtg.dt_get_project(self.server, project_name, byref(self.err)) check_error(self.err) if proj_p: return DTGProject(self, proj_p) return None class DTGProject(object): "Interface to a project object" def __init__(self, server, proj_p): self.server = server self.dtg = server.dtg self.project = proj_p self.err = DTGError() def list_fields(self): "List our fields" field_p = self.dtg.proj_list_fields(self.project, byref(self.err)) check_error(self.err) fields = [] while field_p: select_values = array_from_strlist(field_p.contents.select_values) fields.append(Field(field_p.contents.name, field_p.contents.type, field_p.contents.readonly, select_values)) field_p = field_p.contents.next return fields # struct DTGStrList *proj_list_changed_defects( void *projID, # int max_rows, # struct DTGDate *since, # const char *mod_date_field, # const char *mod_by_field, # const char *exclude_user, # struct DTGError *error ) def list_changed_defects(self, since, max_rows=0, mod_date_field="", mod_by_field="", exclude_user=""): "Wrapper" defects = self.dtg.proj_list_changed_defects(self.project, max_rows, since, mod_date_field, mod_by_field, exclude_user, byref(self.err)) check_error(self.err) return array_from_strlist(defects) def get_defect(self, defect_id): "Creates specified defect" defect = self.dtg.proj_get_defect(self.project, defect_id, byref(self.err)) check_error(self.err) if defect: return DTGDefect(self, defect) return None # defect = proj_get_defect(project, defect_list[0], byref(err)) # print "Defect: %s " % defect class DTGDefect(object): "Interface to a defect object" def __init__(self, project, defect): self.project = project self.dtg = project.dtg self.defect = defect self.err = DTGError() def get_field(self, field_name): "Read specific field" value = self.dtg.defect_get_field(self.defect, field_name, byref(self.err)) check_error(self.err) return value def set_field(self, field_name, value): "Set field" self.dtg.defect_set_field(self.defect, field_name, value, byref(self.err)) check_error(self.err) def save(self): "Save updates" result = self.dtg.defect_save(self.defect, byref(self.err)) check_error(self.err) def main(): pass if __name__ == "__main__": main()