praekelt / go-http-api / 211

Committed 23 Jun 2016 - 13: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

90.91

/go_http/tests/test_account.py

"""
Tests for go_http.account
"""

import collections
import copy
import json
from unittest import TestCase

from requests import HTTPError
from requests.adapters import HTTPAdapter
from requests_testadapter import TestSession, Resp, TestAdapter

from go_http.account import AccountApiClient
from go_http.exceptions import JsonRpcException
from go_http.tests.fixtures import account as fixtures


class FakeAccountApiAdapter(HTTPAdapter):
    """
    Adapter providing a fake account API.

    This inherits directly from HTTPAdapter instead of using TestAdapter
    because it overrides everything TestAdaptor does.
    """

    def __init__(self, account_api):
        self.account_api = account_api
        super(FakeAccountApiAdapter, self).__init__()

    def send(self, request, stream=False, timeout=None,
             verify=True, cert=None, proxies=None):
        response = self.account_api.handle_request(request)
        r = self.build_response(request, response)
        if not stream:
            # force prefetching content unless streaming in use
            r.content
        return r


class FakeAccountApi(object):
    def __init__(self, api_path, auth_token):
        self.api_path = api_path
        self.auth_token = auth_token
        self.responses = collections.defaultdict(list)

    def http_error_response(self, http_code, error):
        return Resp("403 Forbidden", 403, headers={})

    def jsonrpc_error_response(self, fault, fault_code, fault_string):
        return Resp(json.dumps({
            "error": {
                "fault": fault, "faultCode": fault_code,
                "faultString": fault_string,
            },
        }), 200, headers={})

    def jsonrpc_success_response(self, result):
        return Resp(json.dumps({
            "error": None,
            "result": result,
        }), 200, headers={})

    def add_success_response(self, method, params, result):
        self.responses[method].append((params, copy.deepcopy(result)))

    def handle_request(self, request):
        if request.headers['Authorization'] != 'Bearer %s' % (
                self.auth_token):
            return self.http_error_response(403, "403 Forbidden")
        if request.headers['Content-Type'] != (
                'application/json; charset=utf-8'):
            return self.http_error_response(400, "Invalid Content-Type.")
        if request.method != "POST":
            return self.jsonrpc_error_response(
                "Fault", 8000, "Only POST method supported")
        data = json.loads(request.body)
        params, result = self.responses[data['method']].pop()
        assert params == data['params']
        return self.jsonrpc_success_response(result)


class TestAccountApiClient(TestCase):
    API_URL = "http://example.com/go"
    AUTH_TOKEN = "auth_token"

    def setUp(self):
        self.account_backend = FakeAccountApi("go/", self.AUTH_TOKEN)
        self.session = TestSession()
        self.adapter = FakeAccountApiAdapter(self.account_backend)
        self.simulate_api_up()

    def simulate_api_down(self):
        self.session.mount(self.API_URL, TestAdapter("API is down", 500))

    def simulate_api_up(self):
        self.session.mount(self.API_URL, self.adapter)

    def make_client(self, auth_token=AUTH_TOKEN):
        return AccountApiClient(
            auth_token, api_url=self.API_URL, session=self.session)

    def assert_http_error(self, expected_status, func, *args, **kw):
        try:
            func(*args, **kw)
        except HTTPError as err:
            self.assertEqual(err.response.status_code, expected_status)
        else:
            self.fail(
                "Expected HTTPError with status %s." % (expected_status,))

    def assert_jsonrpc_exception(self, f, *args, **kw):
        try:
            f(*args, **kw)
        except Exception as err:
            self.assertTrue(isinstance(err, JsonRpcException))
            self.assertTrue(isinstance(err.cursor, unicode))
            self.assertTrue(isinstance(err.error, Exception))
        return err

    def test_assert_http_error(self):
        self.session.mount("http://bad.example.com/", TestAdapter("", 500))

        def bad_req():
            r = self.session.get("http://bad.example.com/")
            r.raise_for_status()

        # Fails when no exception is raised.
        self.assertRaises(
            self.failureException, self.assert_http_error, 404, lambda: None)

        # Fails when an HTTPError with the wrong status code is raised.
        self.assertRaises(
            self.failureException, self.assert_http_error, 404, bad_req)

        # Passes when an HTTPError with the expected status code is raised.
        self.assert_http_error(500, bad_req)

        # Non-HTTPError exceptions aren't caught.
        def raise_error():
            raise ValueError()

        self.assertRaises(ValueError, self.assert_http_error, 404, raise_error)

    def test_default_session(self):
        import requests
        client = AccountApiClient(self.AUTH_TOKEN)
        self.assertTrue(isinstance(client.session, requests.Session))

    def test_default_api_url(self):
        client = AccountApiClient(self.AUTH_TOKEN)
        self.assertEqual(
            client.api_url, "https://go.vumi.org/api/v1/go")

    def test_auth_failure(self):
        client = self.make_client(auth_token="bogus_token")
        self.assert_http_error(403, client.campaigns)

    def test_campaigns(self):
        client = self.make_client()
        self.account_backend.add_success_response(
            "campaigns", [], fixtures.campaigns)
        self.assertEqual(client.campaigns(), fixtures.campaigns)

    def test_conversations(self):
        client = self.make_client()
        self.account_backend.add_success_response(
            "conversations", ["campaign-1"], fixtures.conversations)
        self.assertEqual(
            client.conversations("campaign-1"),
            fixtures.conversations)

    def test_channels(self):
        client = self.make_client()
        self.account_backend.add_success_response(
            "channels", ["campaign-1"], fixtures.channels)
        self.assertEqual(
            client.channels("campaign-1"),
            fixtures.channels)

    def test_routers(self):
        client = self.make_client()
        self.account_backend.add_success_response(
            "routers", ["campaign-1"], fixtures.routers)
        self.assertEqual(
            client.routers("campaign-1"),
            fixtures.routers)

    def test_routing_entries(self):
        client = self.make_client()
        self.account_backend.add_success_response(
            "routing_entries", ["campaign-1"], fixtures.routing_entries)
        self.assertEqual(
            client.routing_entries("campaign-1"),
            fixtures.routing_entries)

    def test_routing_table(self):
        client = self.make_client()
        self.account_backend.add_success_response(
            "routing_table", ["campaign-1"], fixtures.routing_table)
        self.assertEqual(
            client.routing_table("campaign-1"),
            fixtures.routing_table)

    def test_update_routing_tabel(self):
        client = self.make_client()
        self.account_backend.add_success_response(
            "update_routing_table", ["campaign-1", fixtures.routing_table],
            None)
        self.assertEqual(
            client.update_routing_table("campaign-1", fixtures.routing_table),
            None)

1	"""
2	Tests for go_http.account
3	"""
4
5	import collections	2×
6	import copy	2×
7	import json	2×
8	from unittest import TestCase	2×
9
10	from requests import HTTPError	2×
11	from requests.adapters import HTTPAdapter	2×
12	from requests_testadapter import TestSession, Resp, TestAdapter	2×
13
14	from go_http.account import AccountApiClient	2×
15	from go_http.exceptions import JsonRpcException	2×
16	from go_http.tests.fixtures import account as fixtures	2×
17
18
19	class FakeAccountApiAdapter(HTTPAdapter):	2×
20	"""
21	Adapter providing a fake account API.
22
23	This inherits directly from HTTPAdapter instead of using TestAdapter
24	because it overrides everything TestAdaptor does.
25	"""
26
27	def __init__(self, account_api):	2×
28	self.account_api = account_api	2×
29	super(FakeAccountApiAdapter, self).__init__()	2×
30
31	def send(self, request, stream=False, timeout=None,	2×
32	verify=True, cert=None, proxies=None):
33	response = self.account_api.handle_request(request)	2×
34	r = self.build_response(request, response)	2×
35	if not stream:	2×
36	# force prefetching content unless streaming in use
37	r.content	2×
38	return r	2×
39
40
41	class FakeAccountApi(object):	2×
42	def __init__(self, api_path, auth_token):	2×
43	self.api_path = api_path	2×
44	self.auth_token = auth_token	2×
45	self.responses = collections.defaultdict(list)	2×
46
47	def http_error_response(self, http_code, error):	2×
48	return Resp("403 Forbidden", 403, headers={})	2×
49
50	def jsonrpc_error_response(self, fault, fault_code, fault_string):	2×
NEW 51	return Resp(json.dumps({	!
52	"error": {
53	"fault": fault, "faultCode": fault_code,
54	"faultString": fault_string,
55	},
56	}), 200, headers={})
57
58	def jsonrpc_success_response(self, result):	2×
59	return Resp(json.dumps({	2×
60	"error": None,
61	"result": result,
62	}), 200, headers={})
63
64	def add_success_response(self, method, params, result):	2×
65	self.responses[method].append((params, copy.deepcopy(result)))	2×
66
67	def handle_request(self, request):	2×
68	if request.headers['Authorization'] != 'Bearer %s' % (	2×
69	self.auth_token):
70	return self.http_error_response(403, "403 Forbidden")	2×
71	if request.headers['Content-Type'] != (	2×
72	'application/json; charset=utf-8'):
NEW 73	return self.http_error_response(400, "Invalid Content-Type.")	!
74	if request.method != "POST":	2×
NEW 75	return self.jsonrpc_error_response(	!
76	"Fault", 8000, "Only POST method supported")
77	data = json.loads(request.body)	2×
78	params, result = self.responses[data['method']].pop()	2×
79	assert params == data['params']	2×
80	return self.jsonrpc_success_response(result)	2×
81
82
83	class TestAccountApiClient(TestCase):	2×
84	API_URL = "http://example.com/go"	2×
85	AUTH_TOKEN = "auth_token"	2×
86
87	def setUp(self):	2×
88	self.account_backend = FakeAccountApi("go/", self.AUTH_TOKEN)	2×
89	self.session = TestSession()	2×
90	self.adapter = FakeAccountApiAdapter(self.account_backend)	2×
91	self.simulate_api_up()	2×
92
93	def simulate_api_down(self):	2×
NEW 94	self.session.mount(self.API_URL, TestAdapter("API is down", 500))	!
95
96	def simulate_api_up(self):	2×
97	self.session.mount(self.API_URL, self.adapter)	2×
98
99	def make_client(self, auth_token=AUTH_TOKEN):	2×
100	return AccountApiClient(	2×
101	auth_token, api_url=self.API_URL, session=self.session)
102
103	def assert_http_error(self, expected_status, func, args, *kw):	2×
104	try:	2×
105	func(args, *kw)	2×
106	except HTTPError as err:	2×
107	self.assertEqual(err.response.status_code, expected_status)	2×
108	else:
109	self.fail(	2×
110	"Expected HTTPError with status %s." % (expected_status,))
111
112	def assert_jsonrpc_exception(self, f, args, *kw):	2×
NEW 113	try:	!
NEW 114	f(args, *kw)	!
NEW 115	except Exception as err:	!
NEW 116	self.assertTrue(isinstance(err, JsonRpcException))	!
NEW 117	self.assertTrue(isinstance(err.cursor, unicode))	!
NEW 118	self.assertTrue(isinstance(err.error, Exception))	!
NEW 119	return err	!
120
121	def test_assert_http_error(self):	2×
122	self.session.mount("http://bad.example.com/", TestAdapter("", 500))	2×
123
124	def bad_req():	2×
125	r = self.session.get("http://bad.example.com/")	2×
126	r.raise_for_status()	2×
127
128	# Fails when no exception is raised.
129	self.assertRaises(	2×
130	self.failureException, self.assert_http_error, 404, lambda: None)
131
132	# Fails when an HTTPError with the wrong status code is raised.
133	self.assertRaises(	2×
134	self.failureException, self.assert_http_error, 404, bad_req)
135
136	# Passes when an HTTPError with the expected status code is raised.
137	self.assert_http_error(500, bad_req)	2×
138
139	# Non-HTTPError exceptions aren't caught.
140	def raise_error():	2×
141	raise ValueError()	2×
142
143	self.assertRaises(ValueError, self.assert_http_error, 404, raise_error)	2×
144
145	def test_default_session(self):	2×
146	import requests	2×
147	client = AccountApiClient(self.AUTH_TOKEN)	2×
148	self.assertTrue(isinstance(client.session, requests.Session))	2×
149
150	def test_default_api_url(self):	2×
151	client = AccountApiClient(self.AUTH_TOKEN)	2×
152	self.assertEqual(	2×
153	client.api_url, "https://go.vumi.org/api/v1/go")
154
155	def test_auth_failure(self):	2×
156	client = self.make_client(auth_token="bogus_token")	2×
157	self.assert_http_error(403, client.campaigns)	2×
158
159	def test_campaigns(self):	2×
160	client = self.make_client()	2×
161	self.account_backend.add_success_response(	2×
162	"campaigns", [], fixtures.campaigns)
163	self.assertEqual(client.campaigns(), fixtures.campaigns)	2×
164
165	def test_conversations(self):	2×
166	client = self.make_client()	2×
167	self.account_backend.add_success_response(	2×
168	"conversations", ["campaign-1"], fixtures.conversations)
169	self.assertEqual(	2×
170	client.conversations("campaign-1"),
171	fixtures.conversations)
172
173	def test_channels(self):	2×
174	client = self.make_client()	2×
175	self.account_backend.add_success_response(	2×
176	"channels", ["campaign-1"], fixtures.channels)
177	self.assertEqual(	2×
178	client.channels("campaign-1"),
179	fixtures.channels)
180
181	def test_routers(self):	2×
182	client = self.make_client()	2×
183	self.account_backend.add_success_response(	2×
184	"routers", ["campaign-1"], fixtures.routers)
185	self.assertEqual(	2×
186	client.routers("campaign-1"),
187	fixtures.routers)
188
189	def test_routing_entries(self):	2×
190	client = self.make_client()	2×
191	self.account_backend.add_success_response(	2×
192	"routing_entries", ["campaign-1"], fixtures.routing_entries)
193	self.assertEqual(	2×
194	client.routing_entries("campaign-1"),
195	fixtures.routing_entries)
196
197	def test_routing_table(self):	2×
198	client = self.make_client()	2×
199	self.account_backend.add_success_response(	2×
200	"routing_table", ["campaign-1"], fixtures.routing_table)
201	self.assertEqual(	2×
202	client.routing_table("campaign-1"),
203	fixtures.routing_table)
204
205	def test_update_routing_tabel(self):	2×
206	client = self.make_client()	2×
207	self.account_backend.add_success_response(	2×
208	"update_routing_table", ["campaign-1", fixtures.routing_table],
209	None)
210	self.assertEqual(	2×
211	client.update_routing_table("campaign-1", fixtures.routing_table),
212	None)

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