Skip to content

Commit

Permalink
Prevent escaping of & and = in query string
Browse files Browse the repository at this point in the history
  • Loading branch information
@pmaris
pmaris committed Jun 21, 2018
1 parent 2d2d0ac commit 0eeae87
Show file tree
Hide file tree
Showing 3 changed files with 30 additions and 30 deletions.
16 changes: 8 additions & 8 deletions py_nextbus/client.py
View file
Expand Up @@ -159,13 +159,13 @@ def get_route_config(self, route_tag=None, agency=None):

return self._perform_request(params=params)

def get_predictions(self, stop_id, route_tag=None, agency=None):
def get_predictions(self, stop_tag, route_tag=None, agency=None):
"""Make a request to the NextBus API with the "predictions" command to get arrival time
predictions for a single stop. A route tag can optionally be provided to filter the
predictions down to only that particular route at the stop.
Arguments:
stop_id: (Integer) ID of a stop on NextBus.
stop_tag: (String) Tag identifying a stop on NextBus.
route_tag: (String) The NextBus route tag for a route. If this is None, predictions for
all routes that serve the given stop will be returned.
agency: (String) Name of a transit agency on NextBus. This must be provided if the
Expand All @@ -187,7 +187,7 @@ def get_predictions(self, stop_id, route_tag=None, agency=None):
params = {
'command': 'predictions',
'a': agency,
'stopId': stop_id
's': stop_tag
}

if route_tag is not None:
Expand All @@ -202,7 +202,7 @@ def get_predictions_for_multi_stops(self, stops, agency=None):
Arguments:
stops: (List or tuple) List or tuple of dictionaries identifying the combinations of
routes and stops to get predictions for. Each dictionary must contain the keys
"stop_id" and "route_tag", indicating each stop to get predictions for, and the
"stop_tag" and "route_tag", indicating each stop to get predictions for, and the
route to get predictions for at that stop, respectively.
agency: (String) Name of a transit agency on NextBus. This must be provided if the
"agency" instance attribute has not been set.
Expand All @@ -222,17 +222,17 @@ def get_predictions_for_multi_stops(self, stops, agency=None):
raise TypeError('"stops" must be a list or a tuple.')

for stop in stops:
if not isinstance(stop, dict) or 'route_tag' not in stop or 'stop_id' not in stop:
if not isinstance(stop, dict) or 'route_tag' not in stop or 'stop_tag' not in stop:
raise ValueError('"stops" must contain dictionaries with the "route_tag" and ' \
'"stop_id" keys')
'"stop_tag" keys')

agency = self._get_agency(agency)

params = {
'command': 'predictionsForMultiStops',
# Hacky way of repeating the "stops" key in the query string for each route
'stops': '&stops='.join(['%s|%d' % (stop['route_tag'],
stop['stop_id']) for stop in stops]),
stop['stop_tag']) for stop in stops]),
'a': agency
}
return self._perform_request(params=params)
Expand Down Expand Up @@ -347,7 +347,7 @@ def _perform_request(self, params):
else:
raise ValueError('Invalid output_format: %s' % self.output_format)

url = '%s?%s' % (base_url, urllib.parse.urlencode(params))
url = '%s?%s' % (base_url, urllib.parse.urlencode(params, safe='&='))

headers = {}

Expand Down
42 changes: 21 additions & 21 deletions py_nextbus/tests.py
View file
Expand Up @@ -236,17 +236,17 @@ class TestGetPredictions(unittest.TestCase):
def test_parameters_passed_to_perform_request(self, perform_request, get_agency):
"""Test that the correct parameters are passed to the _perform_request method."""

stop_id = 12345
stop_tag = 12345
agency = 'foo'
nextbus_client = client.NextBusClient(output_format='json')
nextbus_client.get_predictions(stop_id=stop_id,
nextbus_client.get_predictions(stop_tag=stop_tag,
agency=agency)

get_agency.assert_called_once_with(agency)
perform_request.assert_called_once_with(params={
'command': 'predictions',
'a': get_agency.return_value,
'stopId': stop_id
's': stop_tag
})

def test_route_tag_added_to_parameters_if_not_none(self, perform_request, get_agency):
Expand All @@ -255,7 +255,7 @@ def test_route_tag_added_to_parameters_if_not_none(self, perform_request, get_ag

route_tag = 'foo'
nextbus_client = client.NextBusClient(output_format='json')
nextbus_client.get_predictions(stop_id=12345,
nextbus_client.get_predictions(stop_tag=12345,
route_tag=route_tag)

params = perform_request.call_args[1]['params']
Expand All @@ -267,7 +267,7 @@ def test_route_tag_not_added_to_parameters_if_none(self, perform_request, get_ag
_perform_request method if the value of the route_tag argument is None."""

nextbus_client = client.NextBusClient(output_format='json')
nextbus_client.get_predictions(stop_id=12345,
nextbus_client.get_predictions(stop_tag=12345,
route_tag=None)

params = perform_request.call_args[1]['params']
Expand Down Expand Up @@ -323,7 +323,7 @@ def test_value_error_raised_if_stops_does_not_contain_dictionaries_with_required

with self.assertRaises(ValueError):
nextbus_client.get_predictions_for_multi_stops(stops=[{
'stop_id': 1234
'stop_tag': 1234
}])
with self.assertRaises(ValueError):
nextbus_client.get_predictions_for_multi_stops(stops=[{
Expand All @@ -333,7 +333,7 @@ def test_value_error_raised_if_stops_does_not_contain_dictionaries_with_required
nextbus_client.get_predictions_for_multi_stops(
stops=[
{
'stop_id': 1234,
'stop_tag': 1234,
'route_tag': 'foo'
},
None
Expand All @@ -353,7 +353,7 @@ def test_value_error_raised_if_stops_does_not_contain_dictionaries_with_required

with self.assertRaises(ValueError):
nextbus_client.get_predictions_for_multi_stops(stops=({
'stop_id': 1234
'stop_tag': 1234
},))
with self.assertRaises(ValueError):
nextbus_client.get_predictions_for_multi_stops(stops=({
Expand All @@ -363,7 +363,7 @@ def test_value_error_raised_if_stops_does_not_contain_dictionaries_with_required
nextbus_client.get_predictions_for_multi_stops(
stops=(
{
'stop_id': 1234,
'stop_tag': 1234,
'route_tag': 'foo'
},
None
Expand All @@ -378,12 +378,12 @@ def test_no_exceptions_raised_if_stops_is_valid(self, perform_request, get_agenc

nextbus_client.get_predictions_for_multi_stops(stops=[{
'route_tag': 'foo',
'stop_id': 1234
'stop_tag': 1234
}])

nextbus_client.get_predictions_for_multi_stops(stops=({
'route_tag': 'foo',
'stop_id': 1234
'stop_tag': 1234
},))

def test_parameters_passed_to_perform_request(self, perform_request, get_agency):
Expand All @@ -393,49 +393,49 @@ def test_parameters_passed_to_perform_request(self, perform_request, get_agency)
nextbus_client = client.NextBusClient(output_format='json')

# Test with only one stop
stop_id = 1234
stop_tag = 1234
route_tag = 'bar'
nextbus_client.get_predictions_for_multi_stops(
agency=agency,
stops=[{
'route_tag': route_tag,
'stop_id': stop_id
'stop_tag': stop_tag
}]
)

get_agency.assert_called_once_with(agency)
perform_request.assert_called_once_with(params={
'command': 'predictionsForMultiStops',
'a': get_agency.return_value,
'stops': '%s|%s' % (route_tag, stop_id)
'stops': '%s|%s' % (route_tag, stop_tag)
})

perform_request.reset_mock()

# Test with multiple stops
first_stop_id = 1234
first_stop_tag = 1234
first_route_tag = 'baz'
second_stop_id = 5678
second_stop_tag = 5678
second_route_tag = 'buz'
nextbus_client.get_predictions_for_multi_stops(
agency=agency,
stops=[
{
'route_tag': first_route_tag,
'stop_id': first_stop_id
'stop_tag': first_stop_tag
},
{
'route_tag': second_route_tag,
'stop_id': second_stop_id
'stop_tag': second_stop_tag
}
]
)

perform_request.assert_called_once_with(params={
'command': 'predictionsForMultiStops',
'a': get_agency.return_value,
'stops': '%s|%s&stops=%s|%s' % (first_route_tag, first_stop_id, second_route_tag,
second_stop_id)
'stops': '%s|%s&stops=%s|%s' % (first_route_tag, first_stop_tag, second_route_tag,
second_stop_tag)
})

@unittest.mock.patch('client.NextBusClient._get_agency')
Expand Down Expand Up @@ -561,7 +561,7 @@ def test_parameters_added_to_url_as_query_string(self, urlencode, request, loads
nextbus_client = client.NextBusClient(output_format='json')
nextbus_client._perform_request(params=params)

urlencode.assert_called_once_with(params)
urlencode.assert_called_once_with(params, safe='&=')
# Get the URL of the request to the NextBus API
request_url = request.Request.call_args[1]['url']
assert request_url.endswith('?%s' % urlencode.return_value)
Expand Down
2 changes: 1 addition & 1 deletion setup.py
View file
Expand Up @@ -5,7 +5,7 @@

setup(
name='py_nextbus',
version='0.1.0',
version='0.1.2',
author='Pierre Maris',
description='Minimalistic Python client for the NextBus public API for real-time transit ' \
'arrival data',
Expand Down

0 comments on commit 0eeae87

Please sign in to comment.