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

Refactor and extract import logic to a helper

Browse Source
This commit is contained in:
Mario de Frutos 2015-12-06 15:01:46 +01:00
parent 342066f22a
commit ae65aff707
2 changed files with 50 additions and 39 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

45
test/helpers/import_helper.py Normal file
View File

@ -0,0 +1,45 @@
import os
import requests
import json
import time
class ImportHelper:
@classmethod
def import_test_dataset(cls, username, api_key, host):
url = "https://{0}.{1}/api/v1/imports/"\
"?type_guessing=false&api_key={2}".format(
username, host, api_key)
dataset = {
'file': open('fixtures/geocoder_api_test_dataset.csv', 'rb')}
response = requests.post(url, files=dataset)
response_json = json.loads(response.text)
if not response_json['success']:
print "Error importing the test dataset: {0}".format(response.text)
sys.exit(1)
while(True):
table_name = ImportHelper.get_imported_table_name(
username,
host,
api_key,
response_json['item_queue_id']
)
if table_name:
return table_name
else:
time.sleep(5)
@classmethod
def get_imported_table_name(cls, username, host, api_key, import_id):
import_url = "https://{0}.{1}/api/v1/imports/{2}?api_key={3}".format(
username, host, import_id, api_key)
import_data_response = requests.get(import_url)
if import_data_response.status_code != 200:
print "Error getting the table name from " \
"the import data: {0}".format(
import_data_response.text)
sys.exit(1)
import_data_json = json.loads(import_data_response.text)
return import_data_json['table_name']

44
test/run_tests.py
View File

@ -5,6 +5,7 @@ import time
import json
import subprocess
import os
from helpers.import_helper import ImportHelper
def main():
@ -28,9 +29,12 @@ def main():
assert False, "unhandled option"
try:
table_name = import_test_dataset(username, api_key, host)
table_name = ImportHelper.import_test_dataset(username, api_key, host)
set_environment_variables(username, api_key, table_name, host)
execute_tests()
except Exception as e:
print e.message
sys.exit(1)
finally:
clean_environment_variables()
clean_test_dataset(username, api_key, table_name, host)
@ -43,44 +47,6 @@ def usage():
--host: take that host as base (by default is cartodb.com)"""
def import_test_dataset(username, api_key, host):
url = "https://{0}.{1}/api/v1/imports/"\
"?type_guessing=false&api_key={2}".format(
username, host, api_key)
dataset = {'file': open('fixtures/geocoder_api_test_dataset.csv', 'rb')}
response = requests.post(url, files=dataset)
response_json = json.loads(response.text)
if not response_json['success']:
print "Error importing the test dataset: {0}".format(response.text)
sys.exit(1)
while(True):
table_name = get_imported_table_name(
username,
host,
api_key,
response_json['item_queue_id']
)
if table_name:
return table_name
else:
time.sleep(5)
def get_imported_table_name(username, host, api_key, import_id):
import_data_url = "https://{0}.{1}/api/v1/imports/{2}?api_key={3}".format(
username, host, import_id, api_key
)
import_data_response = requests.get(import_data_url)
if import_data_response.status_code != 200:
print "Error getting the table name from the import data: {0}".format(
import_data_response.text
)
sys.exit(1)
import_data_json = json.loads(import_data_response.text)
return import_data_json['table_name']
def execute_tests():
print "Start testing..."
process = subprocess.Popen(["nosetests", "--where=integration/"])