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

Add mapzen isolines to the quota checker

Browse Source
This commit is contained in:
Mario de Frutos
2016-12-22 17:01:18 +01:00
parent 9ee1d045c8
commit 23a2de0321
2 changed files with 44 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
server/lib/python/cartodb_services/cartodb_services/metrics/user.py
View File

@@ -2,21 +2,25 @@ from datetime import date, timedelta
from dateutil.relativedelta import relativedelta
from calendar import monthrange
def last_day_of_month(year, month):
"""last valid day of a month"""
return monthrange(year, month)[1]
def latest_valid_date(year, month, day):
"""latest date not later than the day specified"""
valid_day = min(day, last_day_of_month(year, month))
return date(year, month, valid_day)
class UserMetricsService:
""" Class to manage all the user info """
SERVICE_GEOCODER_NOKIA = 'geocoder_here'
SERVICE_GEOCODER_CACHE = 'geocoder_cache'
SERVICE_HERE_ISOLINES = 'here_isolines'
SERVICE_MAPZEN_ISOLINES = 'mapzen_isolines'
SERVICE_MAPZEN_ROUTING = 'routing_mapzen'
SERVICE_OBSERVATORY = 'obs_general'
DAY_OF_MONTH_ZERO_PADDED = '%d'
@@ -30,6 +34,8 @@ class UserMetricsService:
def used_quota(self, service_type, date):
if service_type == self.SERVICE_HERE_ISOLINES:
return self.__used_isolines_quota(service_type, date)
if service_type == self.SERVICE_MAPZEN_ISOLINES:
return self.__used_isolines_quota(service_type, date)
elif service_type == self.SERVICE_MAPZEN_ROUTING:
return self.__used_routing_quota(service_type, date)
elif service_type == self.SERVICE_OBSERVATORY:

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

@@ -1,7 +1,7 @@
from test_helper import *
from mockredis import MockRedis
from cartodb_services.metrics import QuotaService
from cartodb_services.metrics import GeocoderConfig, RoutingConfig, ObservatorySnapshotConfig, IsolinesRoutingConfig
from cartodb_services.metrics import *
from unittest import TestCase
from nose.tools import assert_raises
from datetime import datetime, date
@@ -81,15 +81,17 @@ class TestQuotaService(TestCase):
assert qs.check_user_quota() is False
def test_should_check_user_mapzen_geocoder_quota_correctly(self):
qs = self.__build_geocoder_quota_service('test_user', provider='mapzen')
qs = self.__build_geocoder_quota_service('test_user',
provider='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',
provider='mapzen')
qs = self.__build_geocoder_quota_service('test_user',
orgname='testorg',
provider='mapzen')
qs.increment_success_service_use()
assert qs.check_user_quota() is True
qs.increment_success_service_use(amount=1500000)
@@ -111,16 +113,17 @@ class TestQuotaService(TestCase):
def test_should_check_user_isolines_quota_correctly(self):
qs = self.__build_isolines_quota_service('test_user')
qs.increment_success_service_use()
qs.increment_isolines_service_use()
assert qs.check_user_quota() is True
qs.increment_success_service_use(amount=1500000)
qs.increment_isolines_service_use(amount=1500000)
assert qs.check_user_quota() is False
def test_should_check_org_isolines_quota_correctly(self):
qs = self.__build_isolines_quota_service('test_user', orgname='testorg')
qs.increment_success_service_use()
qs = self.__build_isolines_quota_service('test_user',
orgname='testorg')
qs.increment_isolines_service_use()
assert qs.check_user_quota() is True
qs.increment_success_service_use(amount=1500000)
qs.increment_isolines_service_use(amount=1500000)
assert qs.check_user_quota() is False
def test_should_check_user_obs_snapshot_quota_correctly(self):
@@ -138,6 +141,21 @@ class TestQuotaService(TestCase):
qs.increment_success_service_use(amount=100000)
assert qs.check_user_quota() is False
def test_should_check_user_obs_quota_correctly(self):
qs = self.__build_obs_snapshot_quota_service('test_user')
qs.increment_success_service_use()
assert qs.check_user_quota() is True
qs.increment_success_service_use(amount=100000)
assert qs.check_user_quota() is False
def test_should_check_org_obs_quota_correctly(self):
qs = self.__build_obs_quota_service('test_user',
orgname='testorg')
qs.increment_success_service_use()
assert qs.check_user_quota() is True
qs.increment_success_service_use(amount=100000)
assert qs.check_user_quota() is False
def __prepare_quota_service(self, username, service, quota, provider,
orgname, soft_limit, end_date):
build_redis_user_config(self.redis_conn, username, service,
@@ -186,3 +204,14 @@ class TestQuotaService(TestCase):
do_config = ObservatorySnapshotConfig(self.redis_conn, plpy_mock,
username, orgname)
return QuotaService(do_config, redis_connection=self.redis_conn)
def __build_obs_quota_service(self, username, quota=100,
provider='obs_general',
orgname=None,
soft_limit=False,
end_date=datetime.today()):
self.__prepare_quota_service(username, 'data_observatory', quota,
None, orgname, soft_limit, end_date)
do_config = ObservatoryConfig(self.redis_conn, plpy_mock,
username, orgname)
return QuotaService(do_config, redis_connection=self.redis_conn)