# 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()