File: _backendxmlrpc.py

package info (click to toggle)
python-bugzilla 3.0.2-1
  • links: PTS, VCS
  • area: main
  • in suites: bookworm, bullseye, sid
  • size: 1,072 kB
  • sloc: python: 5,760; makefile: 9
file content (236 lines) | stat: -rw-r--r-- 8,654 bytes parent folder | download
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
# This work is licensed under the GNU GPLv2 or later.
# See the COPYING file in the top-level directory.

from logging import getLogger
import sys
from xmlrpc.client import (Binary, Fault, ProtocolError,
                           ServerProxy, Transport)

from requests import RequestException

from ._backendbase import _BackendBase
from .exceptions import BugzillaError
from ._util import listify


log = getLogger(__name__)


class _BugzillaXMLRPCTransport(Transport):
    def __init__(self, bugzillasession):
        if hasattr(Transport, "__init__"):
            Transport.__init__(self, use_datetime=False)

        self.__bugzillasession = bugzillasession
        self.__bugzillasession.set_xmlrpc_defaults()
        self.__seen_valid_xml = False

        # Override Transport.user_agent
        self.user_agent = self.__bugzillasession.get_user_agent()


    ############################
    # Bugzilla private helpers #
    ############################

    def __request_helper(self, url, request_body):
        """
        A helper method to assist in making a request and parsing the response.
        """
        response = None
        # pylint: disable=try-except-raise
        # pylint: disable=raise-missing-from
        try:
            response = self.__bugzillasession.request(
                "POST", url, data=request_body)

            return self.parse_response(response)
        except RequestException as e:
            if not response:
                raise
            raise ProtocolError(
                url, response.status_code, str(e), response.headers)
        except Fault:
            raise
        except Exception:
            msg = str(sys.exc_info()[1])
            if not self.__seen_valid_xml:
                msg += "\nThe URL may not be an XMLRPC URL: %s" % url
            e = BugzillaError(msg)
            # pylint: disable=attribute-defined-outside-init
            e.__traceback__ = sys.exc_info()[2]
            # pylint: enable=attribute-defined-outside-init
            raise e


    ######################
    # Tranport overrides #
    ######################

    def parse_response(self, response):
        """
        Override Transport.parse_response
        """
        parser, unmarshaller = self.getparser()
        msg = response.text.encode('utf-8')
        try:
            parser.feed(msg)
        except Exception:
            log.debug("Failed to parse this XMLRPC response:\n%s", msg)
            raise

        self.__seen_valid_xml = True
        parser.close()
        return unmarshaller.close()

    def request(self, host, handler, request_body, verbose=0):
        """
        Override Transport.request
        """
        # Setting self.verbose here matches overrided request() behavior
        # pylint: disable=attribute-defined-outside-init
        self.verbose = verbose

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

        # xmlrpclib fails to escape \r
        request_body = request_body.replace(b'\r', b'
')

        return self.__request_helper(url, request_body)


class _BugzillaXMLRPCProxy(ServerProxy, object):
    """
    Override of xmlrpc ServerProxy, to insert bugzilla API auth
    into the XMLRPC request data
    """
    def __init__(self, uri, bugzillasession, *args, **kwargs):
        self.__bugzillasession = bugzillasession
        transport = _BugzillaXMLRPCTransport(self.__bugzillasession)
        ServerProxy.__init__(self, uri, transport, *args, **kwargs)

    def _ServerProxy__request(self, methodname, params):
        """
        Overrides ServerProxy _request method
        """
        # params is a singleton tuple, enforced by xmlrpc.client.dumps
        newparams = params and params[0].copy() or {}

        log.debug("XMLRPC call: %s(%s)", methodname, newparams)
        api_key = self.__bugzillasession.get_api_key()
        token_value = self.__bugzillasession.get_token_value()

        if api_key is not None:
            if 'Bugzilla_api_key' not in newparams:
                newparams['Bugzilla_api_key'] = api_key
        elif token_value is not None:
            if 'Bugzilla_token' not in newparams:
                newparams['Bugzilla_token'] = token_value

        # pylint: disable=no-member
        ret = ServerProxy._ServerProxy__request(self, methodname, (newparams,))
        # pylint: enable=no-member

        if isinstance(ret, dict) and 'token' in ret.keys():
            self.__bugzillasession.set_token_value(ret.get('token'))
        return ret


class _BackendXMLRPC(_BackendBase):
    """
    Internal interface for direct calls to bugzilla's XMLRPC API
    """
    def __init__(self, url, bugzillasession):
        _BackendBase.__init__(self, url, bugzillasession)
        self._xmlrpc_proxy = _BugzillaXMLRPCProxy(url, self._bugzillasession)

    def get_xmlrpc_proxy(self):
        return self._xmlrpc_proxy
    def is_xmlrpc(self):
        return True

    def bugzilla_version(self):
        return self._xmlrpc_proxy.Bugzilla.version()

    def bug_attachment_get(self, attachment_ids, paramdict):
        data = paramdict.copy()
        data["attachment_ids"] = listify(attachment_ids)
        return self._xmlrpc_proxy.Bug.attachments(data)
    def bug_attachment_get_all(self, bug_ids, paramdict):
        data = paramdict.copy()
        data["ids"] = listify(bug_ids)
        return self._xmlrpc_proxy.Bug.attachments(data)
    def bug_attachment_create(self, bug_ids, data, paramdict):
        pdata = paramdict.copy()
        pdata["ids"] = listify(bug_ids)
        if data is not None and "data" not in paramdict:
            pdata["data"] = Binary(data)
        return self._xmlrpc_proxy.Bug.add_attachment(pdata)
    def bug_attachment_update(self, attachment_ids, paramdict):
        data = paramdict.copy()
        data["ids"] = listify(attachment_ids)
        return self._xmlrpc_proxy.Bug.update_attachment(data)

    def bug_comments(self, bug_ids, paramdict):
        data = paramdict.copy()
        data["ids"] = listify(bug_ids)
        return self._xmlrpc_proxy.Bug.comments(data)
    def bug_create(self, paramdict):
        return self._xmlrpc_proxy.Bug.create(paramdict)
    def bug_fields(self, paramdict):
        return self._xmlrpc_proxy.Bug.fields(paramdict)
    def bug_get(self, bug_ids, aliases, paramdict):
        data = paramdict.copy()
        data["ids"] = listify(bug_ids) or []
        data["ids"] += listify(aliases) or []
        return self._xmlrpc_proxy.Bug.get(data)
    def bug_history(self, bug_ids, paramdict):
        data = paramdict.copy()
        data["ids"] = listify(bug_ids)
        return self._xmlrpc_proxy.Bug.history(data)
    def bug_search(self, paramdict):
        return self._xmlrpc_proxy.Bug.search(paramdict)
    def bug_update(self, bug_ids, paramdict):
        data = paramdict.copy()
        data["ids"] = listify(bug_ids)
        return self._xmlrpc_proxy.Bug.update(data)
    def bug_update_tags(self, bug_ids, paramdict):
        data = paramdict.copy()
        data["ids"] = listify(bug_ids)
        return self._xmlrpc_proxy.Bug.update_tags(data)

    def component_create(self, paramdict):
        return self._xmlrpc_proxy.Component.create(paramdict)
    def component_update(self, paramdict):
        return self._xmlrpc_proxy.Component.update(paramdict)

    def externalbugs_add(self, paramdict):
        return self._xmlrpc_proxy.ExternalBugs.add_external_bug(paramdict)
    def externalbugs_update(self, paramdict):
        return self._xmlrpc_proxy.ExternalBugs.update_external_bug(paramdict)
    def externalbugs_remove(self, paramdict):
        return self._xmlrpc_proxy.ExternalBugs.remove_external_bug(paramdict)

    def group_get(self, paramdict):
        return self._xmlrpc_proxy.Group.get(paramdict)

    def product_get(self, paramdict):
        return self._xmlrpc_proxy.Product.get(paramdict)
    def product_get_accessible(self):
        return self._xmlrpc_proxy.Product.get_accessible_products()
    def product_get_enterable(self):
        return self._xmlrpc_proxy.Product.get_enterable_products()
    def product_get_selectable(self):
        return self._xmlrpc_proxy.Product.get_selectable_products()

    def user_create(self, paramdict):
        return self._xmlrpc_proxy.User.create(paramdict)
    def user_get(self, paramdict):
        return self._xmlrpc_proxy.User.get(paramdict)
    def user_login(self, paramdict):
        return self._xmlrpc_proxy.User.login(paramdict)
    def user_logout(self):
        return self._xmlrpc_proxy.User.logout()
    def user_update(self, paramdict):
        return self._xmlrpc_proxy.User.update(paramdict)