# pylint: disable=E1103

# E1103: Instance of 'bool' has no '%s' member

# pylint confuses mimemagic to be of type 'bool'

global mimemagic

if mimemagic is None:

try:

# pylint: disable=F0401

# F0401: Unable to import 'magic' (import-error)

import magic

mimemagic = magic.open(getattr(magic, "MAGIC_MIME_TYPE", 16))

mimemagic.load()

except ImportError:

e = sys.exc_info()[1]

log.debug("Could not load python-magic: %s", e)

mimemagic = False

if mimemagic is False:

return None

if not os.path.isabs(fname):

return None

try:

return mimemagic.file(fname)

except Exception:

e = sys.exc_info()[1]

log.debug("Could not detect content_type: %s", e)

cj = MozillaCookieJar(cookiefile)

if cookiefile is None:

return cj

if not os.path.exists(cookiefile):

# Make sure a new file has correct permissions

open(cookiefile, 'a').close()

os.chmod(cookiefile, 0o600)

cj.save()

return cj

# We always want to use Mozilla cookies, but we previously accepted

# LWP cookies. If we see the latter, convert it to former

try:

cj.load()

return cj

except LoadError:

pass

try:

cj = LWPCookieJar(cookiefile)

cj.load()

except LoadError:

raise BugzillaError("cookiefile=%s not in LWP or Mozilla format" %

cookiefile)

retcj = MozillaCookieJar(cookiefile)

for cookie in cj:

retcj.set_cookie(cookie)

retcj.save()

def __init__(self, uri, tokenfilename):

self.tokenfilename = tokenfilename

self.tokenfile = SafeConfigParser()

self.domain = urlparse(uri)[1]

if self.tokenfilename:

self.tokenfile.read(self.tokenfilename)

if self.domain not in self.tokenfile.sections():

self.tokenfile.add_section(self.domain)

@property

def value(self):

if self.tokenfile.has_option(self.domain, 'token'):

return self.tokenfile.get(self.domain, 'token')

else:

return None

@value.setter

def value(self, value):

if self.value == value:

return

if value is None:

self.tokenfile.remove_option(self.domain, 'token')

else:

self.tokenfile.set(self.domain, 'token', value)

if self.tokenfilename:

with open(self.tokenfilename, 'w') as tokenfile:

log.debug("Saving to tokenfile")

self.tokenfile.write(tokenfile)

def __repr__(self):

def __init__(self, uri, tokenfile, *args, **kwargs):

# pylint: disable=super-init-not-called

# No idea why pylint complains here, must be a bug

ServerProxy.__init__(self, uri, *args, **kwargs)

self.token = _BugzillaToken(uri, tokenfile)

def clear_token(self):

self.token.value = None

def _ServerProxy__request(self, methodname, params):

if self.token.value is not None:

if len(params) == 0:

params = ({}, )

if 'Bugzilla_token' not in params[0]:

params[0]['Bugzilla_token'] = self.token.value

# pylint: disable=maybe-no-member

ret = ServerProxy._ServerProxy__request(self, methodname, params)

# pylint: enable=maybe-no-member

if isinstance(ret, dict) and 'token' in ret.keys():

self.token.value = ret.get('token')

user_agent = 'Python/Bugzilla'

def __init__(self, url, cookiejar=None,

sslverify=True, sslcafile=None, debug=0):

# pylint: disable=W0231

# pylint does not handle multiple import of Transport well

if hasattr(Transport, "__init__"):

Transport.__init__(self, use_datetime=False)

self.verbose = debug

self._cookiejar = cookiejar

# transport constructor needs full url too, as xmlrpc does not pass

# scheme to request

self.scheme = urlparse(url)[0]

if self.scheme not in ["http", "https"]:

raise Exception("Invalid URL scheme: %s (%s)" % (self.scheme, url))

self.use_https = self.scheme == 'https'

self.request_defaults = {

'cert': sslcafile if self.use_https else None,

'cookies': cookiejar,

'verify': sslverify,

'headers': {

'Content-Type': 'text/xml',

'User-Agent': self.user_agent,

}

}

def parse_response(self, response):

""" Parse XMLRPC response """

parser, unmarshaller = self.getparser()

parser.feed(response.text.encode('utf-8'))

parser.close()

return unmarshaller.close()

def _request_helper(self, url, request_body):

"""

response = None

try:

response = requests.post(

url, data=request_body, **self.request_defaults)

# We expect utf-8 from the server

response.encoding = 'UTF-8'

# update/set any cookies

if self._cookiejar is not None:

for cookie in response.cookies:

self._cookiejar.set_cookie(cookie)

if self._cookiejar.filename is not None:

# Save is required only if we have a filename

self._cookiejar.save()

response.raise_for_status()

return self.parse_response(response)

except requests.RequestException:

e = sys.exc_info()[1]

if not response:

raise

raise ProtocolError(

url, response.status_code, str(e), response.headers)

except Fault:

raise sys.exc_info()[1]

except Exception:

# pylint: disable=W0201

e = BugzillaError(str(sys.exc_info()[1]))

e.__traceback__ = sys.exc_info()[2]

raise e

def request(self, host, handler, request_body, verbose=0):

self.verbose = verbose

url = "%s://%s%s" % (self.scheme, host, handler)

# xmlrpclib fails to escape \r

request_body = request_body.replace(b'\r', b'
')

"""

def __init__(self, newname, oldname, is_api=True, is_bug=True):

self.newname = newname

self.oldname = oldname

self.is_api = is_api

'''An object which represents the data and methods exported by a Bugzilla

# bugzilla version that the class is targetting. filled in by

# subclasses

bz_ver_major = 0

bz_ver_minor = 0

# Intended to be the API version of the class, but historically is

# unused and basically worthless since we don't plan on breaking API.

version = "0.1"

@staticmethod

def url_to_query(url):

'''

q = {}

# pylint: disable=unpacking-non-sequence

(ignore, ignore, path,

ignore, query, ignore) = urlparse(url)

base = os.path.basename(path)

if base not in ('buglist.cgi', 'query.cgi'):

return {}

for (k, v) in parse_qsl(query):

if k not in q:

q[k] = v

elif isinstance(q[k], list):

q[k].append(v)

else:

oldv = q[k]

q[k] = [oldv, v]

# Handle saved searches

if base == "buglist.cgi" and "namedcmd" in q and "sharer_id" in q:

q = {

"sharer_id": q["sharer_id"],

"savedsearch": q["namedcmd"],

}

return q

@staticmethod

def fix_url(url):

"""

if '://' not in url:

log.debug('No scheme given for url, assuming https')

url = 'https://' + url

if url.count('/') < 3:

log.debug('No path given for url, assuming /xmlrpc.cgi')

url = url + '/xmlrpc.cgi'

return url

def __init__(self, url=None, user=None, password=None, cookiefile=-1,

sslverify=True, tokenfile=-1):

# Hook to allow Bugzilla autodetection without weirdly overriding

# __init__

if self._init_class_from_url(url, sslverify):

kwargs = locals().copy()

del(kwargs["self"])

# pylint: disable=non-parent-init-called

self.__class__.__init__(self, **kwargs)

return

# Settings the user might want to tweak

self.user = user or ''

self.password = password or ''

self.url = ''

self._transport = None

self._cookiejar = None

self._sslverify = bool(sslverify)

self.bug_autorefresh = True

# Bugzilla object state info that users shouldn't mess with

self._proxy = None

self._products = None

self._bugfields = None

self._components = {}

self._components_details = {}

self._init_private_data()

if cookiefile == -1:

cookiefile = os.path.expanduser('~/.bugzillacookies')

if tokenfile == -1:

tokenfile = os.path.expanduser("~/.bugzillatoken")

log.debug("Using tokenfile=%s", tokenfile)

self.cookiefile = cookiefile

self.tokenfile = tokenfile

# List of field aliases. Maps old style RHBZ parameter

# names to actual upstream values. Used for createbug() and

# query include_fields at least.

self._field_aliases = []

self._add_field_alias('summary', 'short_desc')

self._add_field_alias('description', 'comment')

self._add_field_alias('platform', 'rep_platform')

self._add_field_alias('severity', 'bug_severity')

self._add_field_alias('status', 'bug_status')

self._add_field_alias('id', 'bug_id')

self._add_field_alias('blocks', 'blockedby')

self._add_field_alias('blocks', 'blocked')

self._add_field_alias('depends_on', 'dependson')

self._add_field_alias('creator', 'reporter')

self._add_field_alias('url', 'bug_file_loc')

self._add_field_alias('dupe_of', 'dupe_id')

self._add_field_alias('dupe_of', 'dup_id')

self._add_field_alias('comments', 'longdescs')

self._add_field_alias('creation_time', 'opendate')

self._add_field_alias('creation_time', 'creation_ts')

self._add_field_alias('whiteboard', 'status_whiteboard')

self._add_field_alias('last_change_time', 'delta_ts')

if url:

self.connect(url)

def _init_class_from_url(self, url, sslverify):

ignore = url

ignore = sslverify

def _init_private_data(self):

'''initialize private variables used by this bugzilla instance.'''

self._proxy = None

self._products = None

self._bugfields = None

self._components = {}

self._components_details = {}

def _get_user_agent(self):

ret = ('Python-urllib bugzilla.py/%s %s' %

(__version__, str(self.__class__.__name__)))

return ret

user_agent = property(_get_user_agent)

###################

# Private helpers #

###################

def _check_version(self, major, minor):

"""

if major < self.bz_ver_major:

return True

if (major == self.bz_ver_major and minor <= self.bz_ver_minor):

return True

return False

def _listify(self, val):

if val is None:

return val

if isinstance(val, list):

return val

return [val]

def _product_id_to_name(self, productid):

'''Convert a product ID (int) to a product name (str).'''

for p in self.products:

if p['id'] == productid:

return p['name']

raise ValueError('No product with id #%i' % productid)

def _product_name_to_id(self, product):

'''Convert a product name (str) to a product ID (int).'''

for p in self.products:

if p['name'] == product:

return p['id']

raise ValueError('No product named "%s"' % product)

def _add_field_alias(self, *args, **kwargs):

self._field_aliases.append(_FieldAlias(*args, **kwargs))

def _get_bug_aliases(self):

return [(f.newname, f.oldname)

for f in self._field_aliases if f.is_bug]

def _get_api_aliases(self):

return [(f.newname, f.oldname)

for f in self._field_aliases if f.is_api]

###################

# Cookie handling #

###################

def _getcookiefile(self):

'''cookiefile is the file that bugzilla session cookies are loaded

return self._cookiejar.filename

def _delcookiefile(self):

self._cookiejar = None

def _setcookiefile(self, cookiefile):

if (self._cookiejar and cookiefile == self._cookiejar.filename):

return

if self._proxy is not None:

raise RuntimeError("Can't set cookies with an open connection, "

"disconnect() first.")

log.debug("Using cookiefile=%s", cookiefile)

self._cookiejar = _build_cookiejar(cookiefile)

cookiefile = property(_getcookiefile, _setcookiefile, _delcookiefile)

#############################

# Login/connection handling #

#############################

configpath = ['/etc/bugzillarc', '~/.bugzillarc']

def readconfig(self, configpath=None):

'''

if not configpath:

configpath = self.configpath

configpath = [os.path.expanduser(p) for p in configpath]

c = SafeConfigParser()

r = c.read(configpath)

if not r:

return

# See if we have a config section that matches this url.

section = ""

# Substring match - prefer the longest match found

log.debug("Searching for config section matching %s", self.url)

for s in sorted(c.sections()):

if s in self.url:

log.debug("Found matching section: %s", s)

section = s

if not section:

return

for k, v in c.items(section):

if k in ('user', 'password'):

log.debug("Setting '%s' from configfile", k)

setattr(self, k, v)

def connect(self, url=None):

'''

if url is None and self.url:

url = self.url

url = self.fix_url(url)

self._transport = RequestsTransport(

url, self._cookiejar, sslverify=self._sslverify)

self._transport.user_agent = self.user_agent

self._proxy = _BugzillaServerProxy(url, self.tokenfile,

self._transport)

self.url = url

# we've changed URLs - reload config

self.readconfig()

if (self.user and self.password):

log.info("user and password present - doing login()")

self.login()

def disconnect(self):

'''

# clears all the connection state

self._init_private_data()

def _login(self, user, password):

'''Backend login method for Bugzilla3'''

return self._proxy.User.login({'login': user, 'password': password})

def _logout(self):

'''Backend login method for Bugzilla3'''

return self._proxy.User.logout()

def login(self, user=None, password=None):

'''Attempt to log in using the given username and password. Subsequent

if user:

self.user = user

if password:

self.password = password

if not self.user:

raise ValueError("missing username")

if not self.password:

raise ValueError("missing password")

try:

ret = self._login(self.user, self.password)

self.password = ''

log.info("login successful for user=%s", self.user)

return ret

except Fault:

e = sys.exc_info()[1]

raise BugzillaError("Login failed: %s" % str(e.faultString))

def interactive_login(self, user=None, password=None, force=False):

"""

if not force and user is None:

auths = {

'cookies': self.cookiefile,

'token': self.tokenfile,

}

for (method, source) in auths.items():

if source and os.path.exists(source):

log.info(

'Using %s in %s for authentication', method, source)

return

elif not force:

log.error('No authentication information provided for login')

log.info('Using username/password for authentication')

if not user:

sys.stdout.write('Bugzilla Username: ')

user = sys.stdin.readline().strip()

if not password:

password = getpass('Bugzilla Password: ')

log.info('Logging in... ')

self.login(user, password)

log.info('Authorization cookie received.')

def logout(self):

'''Log out of bugzilla. Drops server connection and user info, and

self._logout()

self.disconnect()

self.user = ''

self.password = ''

@property

def logged_in(self):

"""

try:

self._proxy.User.get({'ids': []})

return True

except Fault:

e = sys.exc_info()[1]

if e.faultCode == 505:

return False

raise e

#############################################

# Fetching info about the bugzilla instance #

#############################################

def _getbugfields(self):

raise RuntimeError("This bugzilla version does not support listing "

"bug fields.")

def getbugfields(self, force_refresh=False):

'''

if force_refresh or self._bugfields is None:

log.debug("Refreshing bugfields")

self._bugfields = self._getbugfields()

self._bugfields.sort()

log.debug("bugfields = %s", self._bugfields)

return self._bugfields

bugfields = property(fget=lambda self: self.getbugfields(),

fdel=lambda self: setattr(self, '_bugfields', None))

def refresh_products(self, **kwargs):

"""

if self._products is None:

self._products = []

for product in self._getproductinfo(**kwargs):

added = False

for current in self._products[:]:

if (current.get("id", -1) != product.get("id", -2) and

current.get("name", -1) != product.get("name", -2)):

continue

self._products.remove(current)

self._products.append(product)

added = True

break

if not added:

self._products.append(product)

def getproducts(self, force_refresh=False, **kwargs):

'''Get product data: names, descriptions, etc.

if force_refresh or not self._products:

self._products = self._getproducts(**kwargs)

return self._products

products = property(fget=lambda self: self.getproducts(),

fdel=lambda self: setattr(self, '_products', None))

def getcomponentsdetails(self, product, force_refresh=False):

'''Returns a dict of dicts, containing detailed component information

if force_refresh or product not in self._components_details:

clist = self._getcomponentsdetails(product)

cdict = {}

for item in clist:

name = item['component']

del item['component']

cdict[name] = item

self._components_details[product] = cdict

return self._components_details[product]

def getcomponentdetails(self, product, component, force_refresh=False):

'''Get details for a single component. See bugzilla documentation

d = self.getcomponentsdetails(product, force_refresh)

return d[component]

def getcomponents(self, product, force_refresh=False):

'''Return a dict of components:descriptions for the given product.'''

if force_refresh or product not in self._components:

self._components[product] = self._getcomponents(product)

return self._components[product]

def _component_data_convert(self, data, update=False):

if isinstance(data['product'], int):

data['product'] = self._product_id_to_name(data['product'])

# Back compat for the old RH interface

convert_fields = [

("initialowner", "default_assignee"),

("initialqacontact", "default_qa_contact"),

("initialcclist", "default_cc"),

]

for old, new in convert_fields:

if old in data:

data[new] = data.pop(old)

if update:

names = {"product": data.pop("product"),

"component": data.pop("component")}

updates = {}

for k in data.keys():

updates[k] = data.pop(k)

data["names"] = [names]

data["updates"] = updates

def addcomponent(self, data):

'''

data = data.copy()

self._component_data_convert(data)

log.debug("Calling Component.create with: %s", data)

return self._proxy.Component.create(data)

def editcomponent(self, data):

'''

data = data.copy()

self._component_data_convert(data, update=True)

log.debug("Calling Component.update with: %s", data)

return self._proxy.Component.update(data)

def _getproductinfo(self, ids=None, names=None,

include_fields=None, exclude_fields=None):

'''

if ids is None and names is None:

raise RuntimeError("Products must be specified")

kwargs = {}

if not self._check_version(4, 2):

if names:

ids = [self._product_name_to_id(name) for name in names]

names = None

include_fields = None

exclude_fields = None

if ids:

kwargs["ids"] = self._listify(ids)

if names:

kwargs["names"] = self._listify(names)

if include_fields:

kwargs["include_fields"] = include_fields

if exclude_fields:

kwargs["exclude_fields"] = exclude_fields

log.debug("Calling Product.get with: %s", kwargs)

ret = self._proxy.Product.get(kwargs)

return ret['products']

def _getproducts(self, **kwargs):

product_ids = self._proxy.Product.get_accessible_products()

r = self._getproductinfo(product_ids['ids'], **kwargs)

return r

def _getcomponents(self, product):

if isinstance(product, str):

product = self._product_name_to_id(product)

r = self._proxy.Bug.legal_values({'product_id': product,

'field': 'component'})

return r['values']

def _getcomponentsdetails(self, product):

# Originally this was a RH extension getProdCompDetails

# Upstream support has been available since 4.2

if not self._check_version(4, 2):

raise RuntimeError("This bugzilla version does not support "

"fetching component details.")

comps = None

if self._products is None:

self._products = []

def _find_comps():

for p in self._products:

if p["name"] != product:

continue

return p.get("components", None)

comps = _find_comps()

if comps is None:

self.refresh_products(names=[product],

include_fields=["name", "id", "components"])

comps = _find_comps()

if comps is None:

raise ValueError("Unknown product '%s'" % product)

# Convert to old style dictionary to maintain back compat

# with original RH bugzilla call

ret = []

for comp in comps:

row = {}

row["component"] = comp["name"]

row["initialqacontact"] = comp["default_qa_contact"]

row["initialowner"] = comp["default_assigned_to"]

row["description"] = comp["description"]

ret.append(row)

return ret

###################

# getbug* methods #

###################

# getbug_extra_fields: Extra fields that need to be explicitly

# requested from Bug.get in order for the data to be returned. This

# decides the difference between getbug() and getbugsimple().

#

# As of Dec 2012 it seems like only RH bugzilla actually has behavior

# like this, for upstream bz it returns all info for every Bug.get()

_getbug_extra_fields = []

_supports_getbug_extra_fields = False

def _getbugs(self, idlist, simple=False, permissive=True,

include_fields=None, exclude_fields=None, extra_fields=None):

'''

oldidlist = idlist

idlist = []

for i in oldidlist:

try:

idlist.append(int(i))

except ValueError:

# String aliases can be passed as well

idlist.append(i)

extra_fields = self._listify(extra_fields or [])

if not simple:

extra_fields += self._getbug_extra_fields

getbugdata = {"ids": idlist}

if permissive:

getbugdata["permissive"] = 1

if self.bz_ver_major >= 4:

if include_fields:

getbugdata["include_fields"] = self._listify(include_fields)

if exclude_fields:

getbugdata["exclude_fields"] = self._listify(exclude_fields)

if self._supports_getbug_extra_fields: