#!/usr/bin/env python3

#

# bugzilla - a commandline frontend for the python bugzilla module

#

# Copyright (C) 2007-2017 Red Hat Inc.

# Author: Will Woods <wwoods@redhat.com>

# Author: Cole Robinson <crobinso@redhat.com>

#

# This work is licensed under the GNU GPLv2 or later.

# See the COPYING file in the top-level directory.

import argparse

import base64

import datetime

import errno

import json

import locale

from logging import getLogger, DEBUG, INFO, WARN, StreamHandler, Formatter

import os

import re

import socket

import sys

import tempfile

import urllib.parse

import xmlrpc.client

import requests.exceptions

import bugzilla

DEFAULT_BZ = 'https://bugzilla.redhat.com'

format_field_re = re.compile("%{([a-z0-9_]+)(?::([^}]*))?}")

log = getLogger(bugzilla.__name__)

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

# Util helpers #

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

def _is_unittest_debug():

return bool(os.getenv("__BUGZILLA_UNITTEST_DEBUG"))

def open_without_clobber(name, *args):

"""

Try to open the given file with the given mode; if that filename exists,

try "name.1", "name.2", etc. until we find an unused filename.

"""

fd = None

count = 1

orig_name = name

while fd is None:

try:

fd = os.open(name, os.O_CREAT | os.O_EXCL, 0o666)

except OSError as err:

if err.errno == errno.EEXIST:

name = "%s.%i" % (orig_name, count)

count += 1

else: # pragma: no cover

raise IOError(err.errno, err.strerror, err.filename) from None

fobj = open(name, *args)

if fd != fobj.fileno():

os.close(fd)

return fobj

def setup_logging(debug, verbose):

handler = StreamHandler(sys.stderr)

handler.setFormatter(Formatter(

"[%(asctime)s] %(levelname)s (%(module)s:%(lineno)d) %(message)s",

"%H:%M:%S"))

log.addHandler(handler)

if debug:

log.setLevel(DEBUG)

elif verbose:

log.setLevel(INFO)

else:

log.setLevel(WARN)

if _is_unittest_debug():

log.setLevel(DEBUG) # pragma: no cover

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

# Option parsing #

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

def _setup_root_parser():

epilog = 'Try "bugzilla COMMAND --help" for command-specific help.'

p = argparse.ArgumentParser(epilog=epilog)

default_url = bugzilla.Bugzilla.get_rcfile_default_url()

if not default_url:

default_url = DEFAULT_BZ

# General bugzilla connection options

p.add_argument('--bugzilla', default=default_url,

help="bugzilla URI. default: %s" % default_url)

p.add_argument("--nosslverify", dest="sslverify",

action="store_false", default=True,

help="Don't error on invalid bugzilla SSL certificate")

p.add_argument('--cert',

help="client side certificate file needed by the webserver")

p.add_argument('--login', action="store_true",

help='Run interactive "login" before performing the '

'specified command.')

p.add_argument('--username', help="Log in with this username")

p.add_argument('--password', help="Log in with this password")

p.add_argument('--restrict-login', action="store_true",

help="The session (login token) will be restricted to "

"the current IP address.")

p.add_argument('--ensure-logged-in', action="store_true",

help="Raise an error if we aren't logged in to bugzilla. "

"Consider using this if you are depending on "

"cached credentials, to ensure that when they expire the "

"tool errors, rather than subtly change output.")

p.add_argument('--no-cache-credentials',

action='store_false', default=True, dest='cache_credentials',

help="Don't save any bugzilla cookies or tokens to disk, and "

"don't use any pre-existing credentials.")

p.add_argument('--cookiefile', default=None, help=argparse.SUPPRESS)

p.add_argument('--tokenfile', default=None,

help="token file to use for bugzilla authentication")

p.add_argument('--verbose', action='store_true',

help="give more info about what's going on")

p.add_argument('--debug', action='store_true',

help="output bunches of debugging info")

p.add_argument('--version', action='version',

version=bugzilla.__version__)

# Allow user to specify BZClass to initialize. Kinda weird for the

# CLI, I'd rather people file bugs about this so we can fix our detection.

# So hide it from the help output but keep it for back compat

p.add_argument('--bztype', default='auto', help=argparse.SUPPRESS)

return p

def _parser_add_output_options(p):

outg = p.add_argument_group("Output format options")

outg.add_argument('--full', action='store_const', dest='output',

const='full', default='normal',

help="output detailed bug info")

outg.add_argument('-i', '--ids', action='store_const', dest='output',

const='ids', help="output only bug IDs")

outg.add_argument('-e', '--extra', action='store_const',

dest='output', const='extra',

help="output additional bug information "

"(keywords, Whiteboards, etc.)")

outg.add_argument('--oneline', action='store_const', dest='output',

const='oneline',

help="one line summary of the bug (useful for scripts)")

outg.add_argument('--json', action='store_const', dest='output',

const='json', help="output contents in json format")

outg.add_argument("--includefield", action="append",

help="Pass the field name to bugzilla include_fields list. "

"Only the fields passed to include_fields are returned "

"by the bugzilla server. "

"This can be specified multiple times.")

outg.add_argument("--extrafield", action="append",

help="Pass the field name to bugzilla extra_fields list. "

"When used with --json this can be used to request "

"bugzilla to return values for non-default fields. "

"This can be specified multiple times.")

outg.add_argument("--excludefield", action="append",

help="Pass the field name to bugzilla exclude_fields list. "

"When used with --json this can be used to request "

"bugzilla to not return values for a field. "

"This can be specified multiple times.")

outg.add_argument('--raw', action='store_const', dest='output',

const='raw', help="raw output of the bugzilla contents. This "

"format is unstable and difficult to parse. Use --json instead.")

outg.add_argument('--outputformat',

help="Print output in the form given. "

"You can use RPM-style tags that match bug "

"fields, e.g.: '%%{id}: %%{summary}'. See the man page "

"section 'Output options' for more details.")

def _parser_add_field_passthrough_opts(p):

p.add_argument('--field',

metavar="FIELD=VALUE", action="append", dest="fields",

help="Manually specify a bugzilla API field. FIELD is "

"the raw name used by the bugzilla instance. For example, if your "

"bugzilla instance has a custom field cf_my_field, do:\n"

" --field cf_my_field=VALUE")

p.add_argument('--field-json',

metavar="JSONSTRING", action="append", dest="field_jsons",

help="Specify --field data as a JSON string. Example: --field-json "

'\'{"cf_my_field": "VALUE", "cf_array_field": [1, 2]}\'')

def _parser_add_bz_fields(rootp, command):

cmd_new = (command == "new")

cmd_query = (command == "query")

cmd_modify = (command == "modify")

if cmd_new:

comment_help = "Set initial bug comment/description"

elif cmd_query:

comment_help = "Search all bug comments"

else:

comment_help = "Add new bug comment"

p = rootp.add_argument_group("Standard bugzilla options")

p.add_argument('-p', '--product', help="Product name")

p.add_argument('-v', '--version', help="Product version")

p.add_argument('-c', '--component', help="Component name")

p.add_argument('-t', '--summary', '--short_desc', help="Bug summary")

p.add_argument('-l', '--comment', '--long_desc', help=comment_help)

if not cmd_query:

p.add_argument("--comment-tag", action="append",

help="Comment tag for the new comment")

p.add_argument("--sub-component", action="append",

help="RHBZ sub component field")

p.add_argument('-o', '--os', help="Operating system")

p.add_argument('--arch', help="Arch this bug occurs on")

p.add_argument('-x', '--severity', help="Bug severity")

p.add_argument('-z', '--priority', help="Bug priority")

p.add_argument('--alias', help='Bug alias (name)')

p.add_argument('-s', '--status', '--bug_status',

help='Bug status (NEW, ASSIGNED, etc.)')

p.add_argument('-u', '--url', help="URL field")

p.add_argument('-m', '--target_milestone', help="Target milestone")

p.add_argument('--target_release', help="RHBZ Target release")

p.add_argument('--blocked', action="append",

help="Bug IDs that this bug blocks")

p.add_argument('--dependson', action="append",

help="Bug IDs that this bug depends on")

p.add_argument('--keywords', action="append",

help="Bug keywords")

p.add_argument('--groups', action="append",

help="Which user groups can view this bug")

p.add_argument('--cc', action="append", help="CC list")

p.add_argument('-a', '--assigned_to', '--assignee', help="Bug assignee")

p.add_argument('-q', '--qa_contact', help='QA contact')

if cmd_modify:

p.add_argument("--minor-update", action="store_true",

help="Request bugzilla to not send any "

"email about this change")

if not cmd_new:

p.add_argument('-f', '--flag', action='append',

help="Bug flags state. Ex:\n"

" --flag needinfo?\n"

" --flag dev_ack+ \n"

" clear with --flag needinfoX")

p.add_argument("--tags", action="append",

help="Tags/Personal Tags field.")

p.add_argument('-w', "--whiteboard", '--status_whiteboard',

action="append", help='Whiteboard field')

p.add_argument("--devel_whiteboard", action="append",

help='RHBZ devel whiteboard field')

p.add_argument("--internal_whiteboard", action="append",

help='RHBZ internal whiteboard field')

p.add_argument("--qa_whiteboard", action="append",

help='RHBZ QA whiteboard field')

p.add_argument('-F', '--fixed_in',

help="RHBZ 'Fixed in version' field")

_parser_add_field_passthrough_opts(p)

if not cmd_modify:

_parser_add_output_options(rootp)

def _setup_action_new_parser(subparsers):

description = ("Create a new bug report. "

"--product, --component, --version, --summary, and --comment "

"must be specified. "

"Options that take multiple values accept comma separated lists, "

"including --cc, --blocks, --dependson, --groups, and --keywords.")

p = subparsers.add_parser("new", description=description)

_parser_add_bz_fields(p, "new")

g = p.add_argument_group("'new' specific options")

g.add_argument('--private', action='store_true', default=False,

help='Mark new comment as private')

def _setup_action_query_parser(subparsers):

description = ("List bug reports that match the given criteria. "

"Certain options can accept a comma separated list to query multiple "

"values, including --status, --component, --product, --version, --id.")

epilog = ("Note: querying via explicit command line options will only "

"get you so far. See the --from-url option for a way to use powerful "

"Web UI queries from the command line.")

p = subparsers.add_parser("query",

description=description, epilog=epilog)

_parser_add_bz_fields(p, "query")

g = p.add_argument_group("'query' specific options")

g.add_argument('-b', '--id', '--bug_id',

help="specify individual bugs by IDs, separated with commas")

g.add_argument('-r', '--reporter',

help="Email: search reporter email for given address")

g.add_argument('--quicksearch',

help="Search using bugzilla's quicksearch functionality.")

g.add_argument('--savedsearch',

help="Name of a bugzilla saved search. If you don't own this "

"saved search, you must passed --savedsearch_sharer_id.")

g.add_argument('--savedsearch-sharer-id',

help="Owner ID of the --savedsearch. You can get this ID from "

"the URL bugzilla generates when running the saved search "

"from the web UI.")

# Keep this at the end so it sticks out more

g.add_argument('--from-url', metavar="WEB_QUERY_URL",

help="Make a working query via bugzilla's 'Advanced search' web UI, "

"grab the url from your browser (the string with query.cgi or "

"buglist.cgi in it), and --from-url will run it via the "

"bugzilla API. Don't forget to quote the string! "

"This only works for Bugzilla 5 and Red Hat bugzilla")

# Deprecated options

p.add_argument('-E', '--emailtype', help=argparse.SUPPRESS)

p.add_argument('--components_file', help=argparse.SUPPRESS)

p.add_argument('-U', '--url_type',

help=argparse.SUPPRESS)

p.add_argument('-K', '--keywords_type',

help=argparse.SUPPRESS)

p.add_argument('-W', '--status_whiteboard_type',

help=argparse.SUPPRESS)

p.add_argument('--fixed_in_type', help=argparse.SUPPRESS)

def _setup_action_info_parser(subparsers):

description = ("List products or component information about the "

"bugzilla server.")

p = subparsers.add_parser("info", description=description)

x = p.add_mutually_exclusive_group(required=True)

x.add_argument('-p', '--products', action='store_true',

help='Get a list of products')

x.add_argument('-c', '--components', metavar="PRODUCT",

help='List the components in the given product')

x.add_argument('-o', '--component_owners', metavar="PRODUCT",

help='List components (and their owners)')

x.add_argument('-v', '--versions', metavar="PRODUCT",

help='List the versions for the given product')

p.add_argument('--active-components', action="store_true",

help='Only show active components. Combine with --components*')

def _setup_action_modify_parser(subparsers):

usage = ("bugzilla modify [options] BUGID [BUGID...]\n"

"Fields that take multiple values have a special input format.\n"

"Append: --cc=foo@example.com\n"

"Overwrite: --cc==foo@example.com\n"

"Remove: --cc=-foo@example.com\n"

"Options that accept this format: --cc, --blocked, --dependson,\n"

" --groups, --tags, whiteboard fields.")

p = subparsers.add_parser("modify", usage=usage)

_parser_add_bz_fields(p, "modify")

g = p.add_argument_group("'modify' specific options")

g.add_argument("ids", nargs="+", help="Bug IDs to modify")

g.add_argument('-k', '--close', metavar="RESOLUTION",

help='Close with the given resolution (WONTFIX, NOTABUG, etc.)')

g.add_argument('-d', '--dupeid', metavar="ORIGINAL",

help='ID of original bug. Implies --close DUPLICATE')

g.add_argument('--private', action='store_true', default=False,

help='Mark new comment as private')

g.add_argument('--reset-assignee', action="store_true",

help='Reset assignee to component default')

g.add_argument('--reset-qa-contact', action="store_true",

help='Reset QA contact to component default')

def _setup_action_attach_parser(subparsers):

usage = """

bugzilla attach --file=FILE --desc=DESC [--type=TYPE] BUGID [BUGID...]

bugzilla attach --get=ATTACHID --getall=BUGID [--ignore-obsolete] [...]

bugzilla attach --type=TYPE BUGID [BUGID...]"""

description = "Attach files or download attachments."

p = subparsers.add_parser("attach", description=description, usage=usage)

p.add_argument("ids", nargs="*", help="BUGID references")

p.add_argument('-f', '--file', metavar="FILENAME",

help='File to attach, or filename for data provided on stdin')

p.add_argument('-d', '--description', '--summary',

metavar="SUMMARY", dest='desc',

help="A short summary of the file being attached")

p.add_argument('-t', '--type', metavar="MIMETYPE",

help="Mime-type for the file being attached")

p.add_argument('-g', '--get', metavar="ATTACHID", action="append",

default=[], help="Download the attachment with the given ID")

p.add_argument("--getall", "--get-all", metavar="BUGID", action="append",

default=[], help="Download all attachments on the given bug")

p.add_argument('--ignore-obsolete', action="store_true",

help='Do not download attachments marked as obsolete.')

p.add_argument('-l', '--comment', '--long_desc',

help="Add comment with attachment")

p.add_argument('--private', action='store_true', default=False,

help='Mark new comment as private')

_parser_add_field_passthrough_opts(p)

def _setup_action_login_parser(subparsers):

usage = 'bugzilla login [--api-key] [username [password]]'

description = """Log into bugzilla and save a login cookie or token.

Note: These tokens are short-lived, and future Bugzilla versions will no

longer support token authentication at all. Please use a

~/.config/python-bugzilla/bugzillarc file with an API key instead, or

use 'bugzilla login --api-key' and we will save it for you."""

p = subparsers.add_parser("login", description=description, usage=usage)

p.add_argument('--api-key', action='store_true', default=False,

help='Prompt for and save an API key into bugzillarc, '

'rather than prompt for username and password.')

p.add_argument("pos_username", nargs="?", help="Optional username ",

metavar="username")

p.add_argument("pos_password", nargs="?", help="Optional password ",

metavar="password")

def setup_parser():

rootparser = _setup_root_parser()

subparsers = rootparser.add_subparsers(dest="command")

subparsers.required = True

_setup_action_new_parser(subparsers)

_setup_action_query_parser(subparsers)

_setup_action_info_parser(subparsers)

_setup_action_modify_parser(subparsers)

_setup_action_attach_parser(subparsers)

_setup_action_login_parser(subparsers)

return rootparser

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

# Command routines #

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

def _merge_field_opts(query, fields, field_jsons, parser):

values = {}

# Add any custom fields if specified

for f in (fields or []):

try:

f, v = f.split('=', 1)

values[f] = v

except Exception:

parser.error("Invalid field argument provided: %s" % (f))

for j in (field_jsons or []):

try:

jvalues = json.loads(j)

values.update(jvalues)

except Exception as e:

parser.error("Invalid field-json value=%s: %s" % (j, e))

if values:

log.debug("parsed --field* values: %s", values)

query.update(values)

def _do_query(bz, opt, parser):

q = {}

# Parse preconstructed queries.

u = opt.from_url

if u:

q = bz.url_to_query(u)

if opt.components_file:

# Components slurped in from file (one component per line)

# This can be made more robust

clist = []

f = open(opt.components_file, 'r')

for line in f.readlines():

line = line.rstrip("\n")

clist.append(line)

opt.component = clist

if opt.status:

val = opt.status

stat = val

if val == 'ALL':

# leaving this out should return bugs of any status

stat = None

elif val == 'DEV':

# Alias for all development bug statuses

stat = ['NEW', 'ASSIGNED', 'NEEDINFO', 'ON_DEV',

'MODIFIED', 'POST', 'REOPENED']

elif val == 'QE':

# Alias for all QE relevant bug statuses

stat = ['ASSIGNED', 'ON_QA', 'FAILS_QA', 'PASSES_QA']

elif val == 'EOL':

# Alias for EndOfLife bug statuses

stat = ['VERIFIED', 'RELEASE_PENDING', 'CLOSED']

elif val == 'OPEN':

# non-Closed statuses

stat = ['NEW', 'ASSIGNED', 'MODIFIED', 'ON_DEV', 'ON_QA',

'VERIFIED', 'RELEASE_PENDING', 'POST']

opt.status = stat

# Convert all comma separated list parameters to actual lists,

# which is what bugzilla wants

# According to bugzilla docs, any parameter can be a list, but

# let's only do this for options we explicitly mention can be

# comma separated.

for optname in ["severity", "id", "status", "component",

"priority", "product", "version"]:

val = getattr(opt, optname, None)

if not isinstance(val, str):

continue

setattr(opt, optname, val.split(","))

include_fields = None

if opt.output in ['raw', 'json']:

# 'raw' always does a getbug() call anyways, so just ask for ID back

include_fields = ['id']

elif opt.outputformat:

include_fields = []

for fieldname, rest in format_field_re.findall(opt.outputformat):

if fieldname == "whiteboard" and rest:

fieldname = rest + "_" + fieldname

elif fieldname == "flag":

fieldname = "flags"

elif fieldname == "cve":

fieldname = ["keywords", "blocks"]

elif fieldname == "__unicode__":

# Needs to be in sync with bug.__unicode__

fieldname = ["id", "status", "assigned_to", "summary"]

flist = isinstance(fieldname, list) and fieldname or [fieldname]

for f in flist:

if f not in include_fields:

include_fields.append(f)

if include_fields is not None:

include_fields.sort()

kwopts = {}

if opt.product:

kwopts["product"] = opt.product

if opt.component:

kwopts["component"] = opt.component

if opt.sub_component:

kwopts["sub_component"] = opt.sub_component

if opt.version:

kwopts["version"] = opt.version

if opt.reporter:

kwopts["reporter"] = opt.reporter

if opt.id:

kwopts["bug_id"] = opt.id

if opt.summary:

kwopts["short_desc"] = opt.summary

if opt.comment:

kwopts["long_desc"] = opt.comment

if opt.cc:

kwopts["cc"] = opt.cc

if opt.assigned_to:

kwopts["assigned_to"] = opt.assigned_to

if opt.qa_contact:

kwopts["qa_contact"] = opt.qa_contact

if opt.status:

kwopts["status"] = opt.status

if opt.blocked:

kwopts["blocked"] = opt.blocked

if opt.dependson:

kwopts["dependson"] = opt.dependson

if opt.keywords:

kwopts["keywords"] = opt.keywords

if opt.keywords_type:

kwopts["keywords_type"] = opt.keywords_type

if opt.url:

kwopts["url"] = opt.url

if opt.url_type:

kwopts["url_type"] = opt.url_type

if opt.whiteboard:

kwopts["status_whiteboard"] = opt.whiteboard

if opt.status_whiteboard_type:

kwopts["status_whiteboard_type"] = opt.status_whiteboard_type

if opt.fixed_in:

kwopts["fixed_in"] = opt.fixed_in

if opt.fixed_in_type:

kwopts["fixed_in_type"] = opt.fixed_in_type

if opt.flag:

kwopts["flag"] = opt.flag

if opt.alias:

kwopts["alias"] = opt.alias

if opt.qa_whiteboard:

kwopts["qa_whiteboard"] = opt.qa_whiteboard

if opt.devel_whiteboard:

kwopts["devel_whiteboard"] = opt.devel_whiteboard

if opt.severity:

kwopts["bug_severity"] = opt.severity

if opt.priority:

kwopts["priority"] = opt.priority

if opt.target_release:

kwopts["target_release"] = opt.target_release

if opt.target_milestone:

kwopts["target_milestone"] = opt.target_milestone

if opt.emailtype:

kwopts["emailtype"] = opt.emailtype

if include_fields:

kwopts["include_fields"] = include_fields

if opt.quicksearch:

kwopts["quicksearch"] = opt.quicksearch

if opt.savedsearch:

kwopts["savedsearch"] = opt.savedsearch

if opt.savedsearch_sharer_id:

kwopts["savedsearch_sharer_id"] = opt.savedsearch_sharer_id

if opt.tags:

kwopts["tags"] = opt.tags

built_query = bz.build_query(**kwopts)

_merge_field_opts(built_query, opt.fields, opt.field_jsons, parser)

built_query.update(q)

q = built_query

if not q: # pragma: no cover

parser.error("'query' command requires additional arguments")

return bz.query(q)

def _do_info(bz, opt):

"""

Handle the 'info' subcommand

"""

# All these commands call getproducts internally, so do it up front

# with minimal include_fields for speed

def _filter_components(compdetails):

ret = {}

for k, v in compdetails.items():

if v.get("is_active", True):

ret[k] = v

return ret

productname = (opt.components or opt.component_owners or opt.versions)

fastcomponents = (opt.components and not opt.active_components)

include_fields = ["name", "id"]

if opt.components or opt.component_owners:

include_fields += ["components.name"]

if opt.component_owners:

include_fields += ["components.default_assigned_to"]

if opt.active_components:

include_fields += ["components.is_active"]

if opt.versions:

include_fields += ["versions"]

bz.refresh_products(names=productname and [productname] or None,

include_fields=include_fields)

if opt.products:

for name in sorted([p["name"] for p in bz.getproducts()]):

print(name)

elif fastcomponents:

for name in sorted(bz.getcomponents(productname)):

print(name)

elif opt.components:

details = bz.getcomponentsdetails(productname)

for name in sorted(_filter_components(details)):

print(name)

elif opt.versions:

proddict = bz.getproducts()[0]

for v in proddict['versions']:

print(str(v["name"] or ''))

elif opt.component_owners:

details = bz.getcomponentsdetails(productname)

for c in sorted(_filter_components(details)):

print("%s: %s" % (c, details[c]['default_assigned_to']))

def _convert_to_outputformat(output):

fmt = ""

if output == "normal":

fmt = "%{__unicode__}"

elif output == "ids":

fmt = "%{id}"

elif output == 'full':

fmt += "%{__unicode__}\n"

fmt += "Component: %{component}\n"

fmt += "CC: %{cc}\n"

fmt += "Blocked: %{blocks}\n"

fmt += "Depends: %{depends_on}\n"

fmt += "%{comments}\n"

elif output == 'extra':

fmt += "%{__unicode__}\n"

fmt += " +Keywords: %{keywords}\n"

fmt += " +QA Whiteboard: %{qa_whiteboard}\n"

fmt += " +Status Whiteboard: %{status_whiteboard}\n"

fmt += " +Devel Whiteboard: %{devel_whiteboard}\n"

elif output == 'oneline':

fmt += "#%{bug_id} %{status} %{assigned_to} %{component}\t"

fmt += "[%{target_milestone}] %{flags} %{cve}"

else: # pragma: no cover

raise RuntimeError("Unknown output type '%s'" % output)

return fmt

def _xmlrpc_converter(obj):

if "DateTime" in str(obj.__class__):

# xmlrpc DateTime object. Convert to date format that

# bugzilla REST API outputs

dobj = datetime.datetime.strptime(str(obj), '%Y%m%dT%H:%M:%S')

return dobj.isoformat() + "Z"

if "Binary" in str(obj.__class__):

# xmlrpc Binary object. Convert to base64

return base64.b64encode(obj.data).decode("utf-8")

raise RuntimeError(

"Unexpected JSON conversion class=%s" % obj.__class__)

def _format_output_json(buglist):

out = {"bugs": [b.get_raw_data() for b in buglist]}

s = json.dumps(out, default=_xmlrpc_converter, indent=2, sort_keys=True)

print(s)

def _format_output_raw(buglist):

for b in buglist:

print("Bugzilla %s: " % b.bug_id)

SKIP_NAMES = ["bugzilla"]

for attrname in sorted(b.__dict__):

if attrname in SKIP_NAMES:

continue

if attrname.startswith("_"):

continue

print("ATTRIBUTE[%s]: %s" % (attrname, b.__dict__[attrname]))

print("\n\n")

def _bug_field_repl_cb(bz, b, matchobj):

# whiteboard and flag allow doing

# %{whiteboard:devel} and %{flag:needinfo}

# That's what 'rest' matches

(fieldname, rest) = matchobj.groups()

if fieldname == "whiteboard" and rest:

fieldname = rest + "_" + fieldname

if fieldname == "flag" and rest:

val = b.get_flag_status(rest)

elif fieldname in ["flags", "flags_requestee"]:

tmpstr = []

for f in getattr(b, "flags", []):

requestee = f.get('requestee', "")

if fieldname == "flags":

requestee = ""

if fieldname == "flags_requestee":

if requestee == "":

continue

tmpstr.append("%s" % requestee)

else:

tmpstr.append("%s%s%s" %

(f['name'], f['status'], requestee))

val = ",".join(tmpstr)

elif fieldname == "cve":

cves = []

for key in getattr(b, "keywords", []):

# grab CVE from keywords and blockers

if key.find("Security") == -1:

continue

for bl in b.blocks:

cvebug = bz.getbug(bl)

for cb in cvebug.alias:

if (cb.find("CVE") != -1 and

cb.strip() not in cves):

cves.append(cb)

val = ",".join(cves)

elif fieldname == "comments":

val = ""

for c in getattr(b, "comments", []):

val += ("\n* %s - %s:\n%s\n" % (c['time'],

c.get("creator", c.get("author", "")), c['text']))

elif fieldname == "external_bugs":

val = ""

for e in getattr(b, "external_bugs", []):

url = e["type"]["full_url"].replace("%id%", e["ext_bz_bug_id"])

if not val:

val += "\n"

val += "External bug: %s\n" % url

elif fieldname == "__unicode__":

val = b.__unicode__()

else:

val = getattr(b, fieldname, "")

vallist = isinstance(val, list) and val or [val]

val = ','.join([str(v or '') for v in vallist])

return val

def _format_output(bz, opt, buglist):

if opt.output in ['raw', 'json']:

include_fields = None

exclude_fields = None

extra_fields = None

if opt.includefield:

include_fields = opt.includefield

if opt.excludefield:

exclude_fields = opt.excludefield

if opt.extrafield:

extra_fields = opt.extrafield

buglist = bz.getbugs([b.bug_id for b in buglist],

include_fields=include_fields,

exclude_fields=exclude_fields,

extra_fields=extra_fields)

if opt.output == 'json':

_format_output_json(buglist)

if opt.output == 'raw':

_format_output_raw(buglist)

return

for b in buglist:

# pylint: disable=cell-var-from-loop

def cb(matchobj):

return _bug_field_repl_cb(bz, b, matchobj)

print(format_field_re.sub(cb, opt.outputformat))

def _parse_triset(vallist, checkplus=True, checkminus=True, checkequal=True,

splitcomma=False):

add_val = []

rm_val = []

set_val = None

def make_list(v):

if not v:

return []

if splitcomma:

return v.split(",")

return [v]

for val in isinstance(vallist, list) and vallist or [vallist]:

val = val or ""

if val.startswith("+") and checkplus:

add_val += make_list(val[1:])

elif val.startswith("-") and checkminus:

rm_val += make_list(val[1:])

elif val.startswith("=") and checkequal:

# Intentionally overwrite this

set_val = make_list(val[1:])

else:

add_val += make_list(val)

return add_val, rm_val, set_val

def _do_new(bz, opt, parser):

# Parse options that accept comma separated list

def parse_multi(val):

return _parse_triset(val, checkplus=False, checkminus=False,

checkequal=False, splitcomma=True)[0]

kwopts = {}

if opt.blocked:

kwopts["blocks"] = parse_multi(opt.blocked)

if opt.cc:

kwopts["cc"] = parse_multi(opt.cc)

if opt.component:

kwopts["component"] = opt.component

if opt.dependson:

kwopts["depends_on"] = parse_multi(opt.dependson)

if opt.comment:

kwopts["description"] = opt.comment

if opt.groups:

kwopts["groups"] = parse_multi(opt.groups)

if opt.keywords:

kwopts["keywords"] = parse_multi(opt.keywords)

if opt.os:

kwopts["op_sys"] = opt.os

if opt.arch:

kwopts["platform"] = opt.arch

if opt.priority:

kwopts["priority"] = opt.priority

if opt.product:

kwopts["product"] = opt.product

if opt.severity:

kwopts["severity"] = opt.severity

if opt.summary:

kwopts["summary"] = opt.summary

if opt.url:

kwopts["url"] = opt.url

if opt.version:

kwopts["version"] = opt.version

if opt.assigned_to:

kwopts["assigned_to"] = opt.assigned_to

if opt.qa_contact:

kwopts["qa_contact"] = opt.qa_contact

if opt.sub_component:

kwopts["sub_component"] = opt.sub_component

if opt.alias:

kwopts["alias"] = opt.alias

if opt.comment_tag:

kwopts["comment_tags"] = opt.comment_tag

if opt.private:

kwopts["comment_private"] = opt.private

ret = bz.build_createbug(**kwopts)

_merge_field_opts(ret, opt.fields, opt.field_jsons, parser)

b = bz.createbug(ret)

b.refresh()

return [b]

def _do_modify(bz, parser, opt):

bugid_list = [bugid for a in opt.ids for bugid in a.split(',')]

add_wb, rm_wb, set_wb = _parse_triset(opt.whiteboard)

add_devwb, rm_devwb, set_devwb = _parse_triset(opt.devel_whiteboard)

add_intwb, rm_intwb, set_intwb = _parse_triset(opt.internal_whiteboard)

add_qawb, rm_qawb, set_qawb = _parse_triset(opt.qa_whiteboard)

add_blk, rm_blk, set_blk = _parse_triset(opt.blocked, splitcomma=True)

add_deps, rm_deps, set_deps = _parse_triset(opt.dependson, splitcomma=True)

add_key, rm_key, set_key = _parse_triset(opt.keywords)

add_cc, rm_cc, ignore = _parse_triset(opt.cc,

checkplus=False,

checkequal=False)

add_groups, rm_groups, ignore = _parse_triset(opt.groups,

checkequal=False,

splitcomma=True)

add_tags, rm_tags, ignore = _parse_triset(opt.tags, checkequal=False)

status = opt.status or None

if opt.dupeid is not None:

opt.close = "DUPLICATE"

if opt.close:

status = "CLOSED"

flags = []

if opt.flag:

# Convert "foo+" to tuple ("foo", "+")

for f in opt.flag:

flags.append({"name": f[:-1], "status": f[-1]})

update_opts = {}

if opt.assigned_to:

update_opts["assigned_to"] = opt.assigned_to

if opt.comment:

update_opts["comment"] = opt.comment

if opt.private:

update_opts["comment_private"] = opt.private

if opt.component:

update_opts["component"] = opt.component

if opt.product:

update_opts["product"] = opt.product

if add_blk:

update_opts["blocks_add"] = add_blk

if rm_blk:

update_opts["blocks_remove"] = rm_blk

if set_blk is not None:

update_opts["blocks_set"] = set_blk

if opt.url:

update_opts["url"] = opt.url

if add_cc:

update_opts["cc_add"] = add_cc

if rm_cc:

update_opts["cc_remove"] = rm_cc

if add_deps:

update_opts["depends_on_add"] = add_deps

if rm_deps:

update_opts["depends_on_remove"] = rm_deps

if set_deps is not None:

update_opts["depends_on_set"] = set_deps

if add_groups:

update_opts["groups_add"] = add_groups