# 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 = None

if not mimemagic:

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)

"""

homepath = os.path.expanduser("~/.%s" % filename)

xdgpath = os.path.expanduser("~/.cache/python-bugzilla/%s" % filename)

if os.path.exists(xdgpath):

return xdgpath

if os.path.exists(homepath):

return homepath

if not os.path.exists(os.path.dirname(xdgpath)):

os.makedirs(os.path.dirname(xdgpath), 0o700)

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

try:

cj.load()

return cj

except LoadError:

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

'/etc/bugzillarc',

'~/.bugzillarc',

'~/.config/python-bugzilla/bugzillarc',

if configpaths == -1:

configpaths = _default_configpaths[:]

# pylint: disable=protected-access

configpaths = [os.path.expanduser(p) for p in

Bugzilla._listify(configpaths)]

# pylint: enable=protected-access

cfg = SafeConfigParser()

read_files = cfg.read(configpaths)

if not read_files:

return

log.info("Found bugzillarc files: %s", read_files)

"""

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

self.newname = newname

self.oldname = oldname

self.is_api = is_api

"""

def __init__(self):

self.products = []

self.bugfields = []

self.components = {}

"""

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

# subclasses

bz_ver_major = 0

bz_ver_minor = 0

@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

@staticmethod

def _listify(val):

if val is None:

return val

if isinstance(val, list):

return val

return [val]

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

sslverify=True, tokenfile=-1, use_creds=True, api_key=None):

"""

if url is -1:

raise TypeError("Specify a valid bugzilla url, or pass url=None")

# Settings the user might want to tweak

self.user = user or ''

self.password = password or ''

self.api_key = api_key

self.url = ''

self._proxy = None

self._transport = None

self._cookiejar = None

self._sslverify = sslverify

self._cache = _BugzillaAPICache()

self._bug_autorefresh = False

self._field_aliases = []

self._init_field_aliases()

self.configpath = _default_configpaths[:]

if not use_creds:

cookiefile = None

tokenfile = None

self.configpath = []

if cookiefile == -1:

cookiefile = _default_auth_location("bugzillacookies")

if tokenfile == -1:

tokenfile = _default_auth_location("bugzillatoken")

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

self.cookiefile = cookiefile

self.tokenfile = tokenfile

if url:

self.connect(url)

self._init_class_from_url()

self._init_class_state()

def _init_class_from_url(self):

"""

from bugzilla import RHBugzilla

if isinstance(self, RHBugzilla):

return

c = None

if "bugzilla.redhat.com" in self.url:

log.info("Using RHBugzilla for URL containing bugzilla.redhat.com")

c = RHBugzilla

else:

try:

extensions = self._proxy.Bugzilla.extensions()

if "RedHat" in extensions.get('extensions', {}):

log.info("Found RedHat bugzilla extension, "

"using RHBugzilla")

c = RHBugzilla

except Fault:

log.debug("Failed to fetch bugzilla extensions", exc_info=True)

if not c:

return

self.__class__ = c

def _init_class_state(self):

"""

pass

def _init_field_aliases(self):

# List of field aliases. Maps old style RHBZ parameter

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

# query include_fields at least.

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')

def _get_user_agent(self):

return 'python-bugzilla/%s' % __version__

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 _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 #

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

def readconfig(self, configpath=None):

"""

cfg = _open_bugzillarc(configpath or self.configpath)

if not cfg:

return

section = ""

log.debug("bugzillarc: Searching for config section matching %s",

self.url)

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

# Substring match - prefer the longest match found

if s in self.url:

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

section = s

if not section:

log.debug("bugzillarc: No section found")

return

for key, val in cfg.items(section):

if key == "api_key":

log.debug("bugzillarc: setting api_key")

self.api_key = val

elif key == "user":

log.debug("bugzillarc: setting user=%s", val)

self.user = val

elif key == "password":

log.debug("bugzillarc: setting password")

self.password = val

else:

log.debug("bugzillarc: unknown key=%s", key)

def _set_bz_version(self, version):

try:

self.bz_ver_major, self.bz_ver_minor = [

int(i) for i in version.split(".")[0:2]]

except:

log.debug("version doesn't match expected format X.Y.Z, "

"assuming 5.0", exc_info=True)

self.bz_ver_major = 5

self.bz_ver_minor = 0

def connect(self, url=None):

'''

if self._transport:

self.disconnect()

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

if self.api_key:

log.debug("using API key")

self._proxy.use_api_key(self.api_key)

version = self._proxy.Bugzilla.version()["version"]

log.debug("Bugzilla version string: %s", version)

self._set_bz_version(version)

def disconnect(self):

'''

self._proxy = None

self._transport = None

self._cache = _BugzillaAPICache()

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 self.api_key:

raise ValueError("cannot login when using an API key")

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):

"""

ignore = force

log.debug('Calling interactive_login')

if not user:

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

sys.stdout.flush()

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

if not password:

password = getpass.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):

'''

r = self._proxy.Bug.fields({'include_fields': ['name']})

return [f['name'] for f in r['fields']]

def getbugfields(self, force_refresh=False):

'''

if force_refresh or not self._cache.bugfields:

log.debug("Refreshing bugfields")

self._cache.bugfields = self._getbugfields()

self._cache.bugfields.sort()

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

return self._cache.bugfields

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

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

def refresh_products(self, **kwargs):

"""

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

added = False

for current in self._cache.products[:]:

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

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

continue

self._cache.products.remove(current)

self._cache.products.append(product)

added = True

break

if not added:

self._cache.products.append(product)

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

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

if force_refresh or not self._cache.products:

self._cache.products = self._getproducts(**kwargs)

return self._cache.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._cache.components_details:

clist = self._getcomponentsdetails(product)

cdict = {}

for item in clist:

name = item['component']

del item['component']

cdict[name] = item

self._cache.components_details[product] = cdict

return self._cache.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._cache.components:

self._cache.components[product] = self._getcomponents(product)

return self._cache.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):

def _find_comps():

for p in self._cache.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 #

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

def _process_include_fields(self, include_fields, exclude_fields,

extra_fields):

"""

def _convert_fields(_in):

if not _in:

return _in

for newname, oldname in self._get_api_aliases():

if oldname in _in:

_in.remove(oldname)

if newname not in _in:

_in.append(newname)

return _in

ret = {}

if self._check_version(4, 0):

if include_fields:

include_fields = _convert_fields(include_fields)

if "id" not in include_fields:

include_fields.append("id")

ret["include_fields"] = include_fields

if exclude_fields:

exclude_fields = _convert_fields(exclude_fields)

ret["exclude_fields"] = exclude_fields

if self._supports_getbug_extra_fields:

if extra_fields:

ret["extra_fields"] = _convert_fields(extra_fields)

return ret

def _get_bug_autorefresh(self):

"""

return self._bug_autorefresh

def _set_bug_autorefresh(self, val):

self._bug_autorefresh = bool(val)

bug_autorefresh = property(_get_bug_autorefresh, _set_bug_autorefresh)

# getbug_extra_fields: Extra fields that need to be explicitly

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

#

# 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, permissive,

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 [])

extra_fields += self._getbug_extra_fields

getbugdata = {"ids": idlist}

if permissive:

getbugdata["permissive"] = 1

getbugdata.update(self._process_include_fields(

include_fields, exclude_fields, extra_fields))

log.debug("Calling Bug.get with: %s", getbugdata)

r = self._proxy.Bug.get(getbugdata)

if self._check_version(4, 0):

bugdict = dict([(b['id'], b) for b in r['bugs']])

else:

bugdict = dict([(b['id'], b['internals']) for b in r['bugs']])

ret = []

for i in idlist:

found = None

if i in bugdict:

found = bugdict[i]

else:

# Need to map an alias

for valdict in bugdict.values():