praekelt / go-http-api / 211

Committed 23 Jun 2016 - 6:31 coverage: 97.986% (-1.4%) from 99.392%

Build # 211

Build Type

Pull #21

travis-ci

Committed by

web-flow

Commit Message

Add test for update_routing_table.

Pull Request Pull Request #21: Add support for the JSON-RPC Go API.

Run Details

156 of 171 new or added lines in 5 files covered. (91.23%)

973 of 993 relevant lines covered (97.99%)

1.96 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

97.3

/go_http/account.py

"""
Client for Vumi Go's JSON-RPC based account, conversations, channel and
routing API.
"""

import json

import requests

from go_http.exceptions import JsonRpcException


class AccountApiClient(object):
    """
    Client for Vumi Go's JSON-RPC based account, conversations, channel and
    routing API.

    :param str auth_token:
        An OAuth 2 access token. NOTE: This will be replaced by a proper
        authentication system at some point.

    :param str api_url:
        The full URL of the HTTP API. Defaults to
        ``https://go.vumi.org/api/v1/go``.

    :type session:
        :class:`requests.Session`
    :param session:
        Requests session to use for HTTP requests. Defaults to a new session.
    """

    def __init__(self, auth_token, api_url=None, session=None):
        self.auth_token = auth_token
        if api_url is None:
            api_url = "https://go.vumi.org/api/v1/go"
        self.api_url = api_url.rstrip('/')
        if session is None:
            session = requests.Session()
        self.session = session

    def _api_request(self, method, params):
        url = "%s/api/" % (self.api_url,)
        headers = {
            "Content-Type": "application/json; charset=utf-8",
            "Authorization": "Bearer %s" % (self.auth_token,),
        }
        data = {
            "method": method,
            "params": params,
            "jsonrpc": "2.0",
            "id": 0,
        }
        r = self.session.post(url, data=json.dumps(data), headers=headers)
        r.raise_for_status()
        rpc_response = r.json()
        rpc_error = rpc_response['error']
        if rpc_error is not None:
            raise JsonRpcException(
                fault=rpc_error['fault'], fault_code=rpc_error['faultCode'],
                fault_string=rpc_error['fault_string'])
        return rpc_response['result']

    def campaigns(self):
        """
        Return a list of campaigns accessible by the account.

        Note: The server-side implementation of this API method is a stub. It
        always returns a single campaign whose name is 'Your Campaign' and
        whose key is the account key.
        """
        return self._api_request("campaigns", [])

    def conversations(self, campaign_id):
        """
        Return a list of conversations for the campaign.

        :param str campaign_id:
            The campaign or account id.
        """
        return self._api_request("conversations", [campaign_id])

    def channels(self, campaign_id):
        """
        Return a list of channels for the campaign.

        :param str campaign_id:
            The campaign or account id.
        """
        return self._api_request("channels", [campaign_id])

    def routers(self, campaign_id):
        """
        Return a list of routers for the campaign.

        :param str campaign_id:
            The campaign or account id.
        """
        return self._api_request("routers", [campaign_id])

    def routing_entries(self, campaign_id):
        """
        Return a list of routing entries for the campaign.

        :param str campaign_id:
            The campaign or account id.
        """
        return self._api_request("routing_entries", [campaign_id])

    def routing_table(self, campaign_id):
        """
        Return the complete routing table for the campaign.

        :param str campaign_id:
            The campaign or account id.
        """
        return self._api_request("routing_table", [campaign_id])

    def update_routing_table(self, campaign_id, routing_table):
        """
        Update the routing table for the campaign.

        :param str campaign_id:
            The campaign or account id.
        :param dict routing_table:
            The complete new routing table.
        """
        return self._api_request(
            "update_routing_table", [campaign_id, routing_table])

1	"""
2	Client for Vumi Go's JSON-RPC based account, conversations, channel and
3	routing API.
4	"""
5
6	import json	2×
7
8	import requests	2×
9
10	from go_http.exceptions import JsonRpcException	2×
11
12
13	class AccountApiClient(object):	2×
14	"""
15	Client for Vumi Go's JSON-RPC based account, conversations, channel and
16	routing API.
17
18	:param str auth_token:
19	An OAuth 2 access token. NOTE: This will be replaced by a proper
20	authentication system at some point.
21
22	:param str api_url:
23	The full URL of the HTTP API. Defaults to
24	``https://go.vumi.org/api/v1/go``.
25
26	:type session:
27	:class:`requests.Session`
28	:param session:
29	Requests session to use for HTTP requests. Defaults to a new session.
30	"""
31
32	def __init__(self, auth_token, api_url=None, session=None):	2×
33	self.auth_token = auth_token	2×
34	if api_url is None:	2×
35	api_url = "https://go.vumi.org/api/v1/go"	2×
36	self.api_url = api_url.rstrip('/')	2×
37	if session is None:	2×
38	session = requests.Session()	2×
39	self.session = session	2×
40
41	def _api_request(self, method, params):	2×
42	url = "%s/api/" % (self.api_url,)	2×
43	headers = {	2×
44	"Content-Type": "application/json; charset=utf-8",
45	"Authorization": "Bearer %s" % (self.auth_token,),
46	}
47	data = {	2×
48	"method": method,
49	"params": params,
50	"jsonrpc": "2.0",
51	"id": 0,
52	}
53	r = self.session.post(url, data=json.dumps(data), headers=headers)	2×
54	r.raise_for_status()	2×
55	rpc_response = r.json()	2×
56	rpc_error = rpc_response['error']	2×
57	if rpc_error is not None:	2×
NEW 58	raise JsonRpcException(	!
59	fault=rpc_error['fault'], fault_code=rpc_error['faultCode'],
60	fault_string=rpc_error['fault_string'])
61	return rpc_response['result']	2×
62
63	def campaigns(self):	2×
64	"""
65	Return a list of campaigns accessible by the account.
66
67	Note: The server-side implementation of this API method is a stub. It
68	always returns a single campaign whose name is 'Your Campaign' and
69	whose key is the account key.
70	"""
71	return self._api_request("campaigns", [])	2×
72
73	def conversations(self, campaign_id):	2×
74	"""
75	Return a list of conversations for the campaign.
76
77	:param str campaign_id:
78	The campaign or account id.
79	"""
80	return self._api_request("conversations", [campaign_id])	2×
81
82	def channels(self, campaign_id):	2×
83	"""
84	Return a list of channels for the campaign.
85
86	:param str campaign_id:
87	The campaign or account id.
88	"""
89	return self._api_request("channels", [campaign_id])	2×
90
91	def routers(self, campaign_id):	2×
92	"""
93	Return a list of routers for the campaign.
94
95	:param str campaign_id:
96	The campaign or account id.
97	"""
98	return self._api_request("routers", [campaign_id])	2×
99
100	def routing_entries(self, campaign_id):	2×
101	"""
102	Return a list of routing entries for the campaign.
103
104	:param str campaign_id:
105	The campaign or account id.
106	"""
107	return self._api_request("routing_entries", [campaign_id])	2×
108
109	def routing_table(self, campaign_id):	2×
110	"""
111	Return the complete routing table for the campaign.
112
113	:param str campaign_id:
114	The campaign or account id.
115	"""
116	return self._api_request("routing_table", [campaign_id])	2×
117
118	def update_routing_table(self, campaign_id, routing_table):	2×
119	"""
120	Update the routing table for the campaign.
121
122	:param str campaign_id:
123	The campaign or account id.
124	:param dict routing_table:
125	The complete new routing table.
126	"""
127	return self._api_request(	2×
128	"update_routing_table", [campaign_id, routing_table])

praekelt / go-http-api / 211

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous