cartodb4/dataservices-api
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Mapzen geocoder and routing quota check

Browse Source
This commit is contained in:
Mario de Frutos
2016-04-14 17:50:46 +02:00
parent 923774ffe4
commit c9ddc37051
10 changed files with 218 additions and 105 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
server/lib/python/cartodb_services/test/test_helper.py
View File

@@ -46,6 +46,6 @@ def _plpy_execute_side_effect(*args, **kwargs):
if args[0] == "SELECT cartodb.CDB_Conf_GetConf('heremaps_conf') as conf":
return [{'conf': '{"geocoder": {"app_id": "app_id", "app_code": "code", "geocoder_cost_per_hit": 1}, "isolines": {"app_id": "app_id", "app_code": "code"}}'}]
elif args[0] == "SELECT cartodb.CDB_Conf_GetConf('mapzen_conf') as conf":
return [{'conf': '{"routing_app_key": "app_key", "geocoder_app_key": "app_key"}'}]
return [{'conf': '{"routing": {"api_key": "valhalla-Z61FWEs", "monthly_quota": 1500000}, "geocoder": {"api_key": "search-d744tp0", "monthly_quota": 1500000}}'}]
elif args[0] == "SELECT cartodb.CDB_Conf_GetConf('logger_conf') as conf":
return [{'conf': '{"geocoder_log_path": "/dev/null"}'}]

97
server/lib/python/cartodb_services/test/test_quota_service.py
View File

@@ -1,7 +1,7 @@
import test_helper
from mockredis import MockRedis
from cartodb_services.metrics import QuotaService
from cartodb_services.metrics import GeocoderConfig
from cartodb_services.metrics import GeocoderConfig, RoutingConfig
from unittest import TestCase
from nose.tools import assert_raises
from datetime import datetime, date
@@ -18,79 +18,124 @@ class TestQuotaService(TestCase):
self.redis_conn = MockRedis()
def test_should_return_true_if_user_quota_with_no_use(self):
qs = self.__build_quota_service('test_user')
qs = self.__build_geocoder_quota_service('test_user')
assert qs.check_user_quota() is True
def test_should_return_true_if_org_quota_with_no_use(self):
qs = self.__build_quota_service('test_user', orgname='test_org')
qs = self.__build_geocoder_quota_service('test_user',
orgname='test_org')
assert qs.check_user_quota() is True
def test_should_return_true_if_user_quota_is_not_completely_used(self):
qs = self.__build_quota_service('test_user')
qs = self.__build_geocoder_quota_service('test_user')
test_helper.increment_geocoder_uses(self.redis_conn, 'test_user')
assert qs.check_user_quota() is True
def test_should_return_true_if_org_quota_is_not_completely_used(self):
qs = self.__build_quota_service('test_user', orgname='test_org')
qs = self.__build_geocoder_quota_service('test_user',
orgname='test_org')
test_helper.increment_geocoder_uses(self.redis_conn, 'test_user',
orgname='test_org')
assert qs.check_user_quota() is True
def test_should_return_false_if_user_quota_is_surpassed(self):
qs = self.__build_quota_service('test_user')
qs = self.__build_geocoder_quota_service('test_user')
test_helper.increment_geocoder_uses(self.redis_conn, 'test_user',
amount=300)
assert qs.check_user_quota() is False
def test_should_return_false_if_org_quota_is_surpassed(self):
qs = self.__build_quota_service('test_user', orgname='test_org')
qs = self.__build_geocoder_quota_service('test_user',
orgname='test_org')
test_helper.increment_geocoder_uses(self.redis_conn, 'test_user',
orgname='test_org', amount=400)
assert qs.check_user_quota() is False
def test_should_return_true_if_user_quota_is_surpassed_but_soft_limit_is_enabled(self):
qs = self.__build_quota_service('test_user', soft_limit=True)
qs = self.__build_geocoder_quota_service('test_user', soft_limit=True)
test_helper.increment_geocoder_uses(self.redis_conn, 'test_user',
amount=300)
assert qs.check_user_quota() is True
def test_should_return_true_if_org_quota_is_surpassed_but_soft_limit_is_enabled(self):
qs = self.__build_quota_service('test_user', orgname='test_org',
soft_limit=True)
qs = self.__build_geocoder_quota_service('test_user',
orgname='test_org',
soft_limit=True)
test_helper.increment_geocoder_uses(self.redis_conn, 'test_user',
orgname='test_org', amount=400)
assert qs.check_user_quota() is True
def test_should_check_user_increment_and_quota_check_correctly(self):
qs = self.__build_quota_service('test_user', quota=2)
qs = self.__build_geocoder_quota_service('test_user', quota=2)
qs.increment_success_service_use()
assert qs.check_user_quota() is True
qs.increment_success_service_use(amount=2)
assert qs.check_user_quota() is False
month = date.today().strftime('%Y%m')
def test_should_check_org_increment_and_quota_check_correctly(self):
qs = self.__build_quota_service('test_user', orgname='test_org',
quota=2)
qs = self.__build_geocoder_quota_service('test_user', quota=2,
orgname='test_org')
qs.increment_success_service_use()
assert qs.check_user_quota() is True
qs.increment_success_service_use(amount=2)
assert qs.check_user_quota() is False
month = date.today().strftime('%Y%m')
def __build_quota_service(self, username, quota=100, service='heremaps',
orgname=None, soft_limit=False,
end_date = datetime.today()):
def test_should_check_user_mapzen_geocoder_quota_correctly(self):
qs = self.__build_geocoder_quota_service('test_user', service='mapzen')
qs.increment_success_service_use()
assert qs.check_user_quota() is True
qs.increment_success_service_use(amount=1500000)
assert qs.check_user_quota() is False
def test_should_check_org_mapzen_geocoder_quota_correctly(self):
qs = self.__build_geocoder_quota_service('test_user', orgname='testorg',
service='mapzen')
qs.increment_success_service_use()
assert qs.check_user_quota() is True
qs.increment_success_service_use(amount=1500000)
assert qs.check_user_quota() is False
def test_should_check_user_routing_quota_correctly(self):
qs = self.__build_routing_quota_service('test_user')
qs.increment_success_service_use()
assert qs.check_user_quota() is True
qs.increment_success_service_use(amount=1500000)
assert qs.check_user_quota() is False
def test_should_check_org_routing_quota_correctly(self):
qs = self.__build_routing_quota_service('test_user', orgname='testorg')
qs.increment_success_service_use()
assert qs.check_user_quota() is True
qs.increment_success_service_use(amount=1500000)
assert qs.check_user_quota() is False
def __prepare_quota_service(self, username, quota, service, orgname,
soft_limit, end_date):
test_helper.build_redis_user_config(self.redis_conn, username,
quota = quota, service = service,
soft_limit = soft_limit,
end_date = end_date)
quota=quota, service=service,
soft_limit=soft_limit,
end_date=end_date)
if orgname:
test_helper.build_redis_org_config(self.redis_conn, orgname,
quota=quota, end_date=end_date)
plpy_mock = test_helper.build_plpy_mock()
geocoder_config = GeocoderConfig(self.redis_conn, plpy_mock,
username, orgname)
return QuotaService(geocoder_config,
redis_connection = self.redis_conn)
self._plpy_mock = test_helper.build_plpy_mock()
def __build_geocoder_quota_service(self, username, quota=100,
service='heremaps', orgname=None,
soft_limit=False,
end_date=datetime.today()):
self.__prepare_quota_service(username, quota, service, orgname,
soft_limit, end_date)
geocoder_config = GeocoderConfig(self.redis_conn, self._plpy_mock,
username, orgname)
return QuotaService(geocoder_config, redis_connection=self.redis_conn)
def __build_routing_quota_service(self, username, quota=100,
service='routing_mapzen', orgname=None,
soft_limit=False,
end_date=datetime.today()):
self.__prepare_quota_service(username, quota, service, orgname,
soft_limit, end_date)
routing_config = RoutingConfig(self.redis_conn, self._plpy_mock,
username, orgname)
return QuotaService(routing_config, redis_connection=self.redis_conn)