From f552f2db12aeb9324d3787866eb74e33b7ecade5 Mon Sep 17 00:00:00 2001 From: Mario de Frutos Date: Sun, 6 Dec 2015 13:10:54 +0100 Subject: [PATCH] Created helper to be used in all the tests --- test/__init__.py | 0 test/helpers/__init__.py | 0 test/helpers/integration_test_helper.py | 30 +++++ test/integration/__init__.py | 0 test/integration/test_admin0_functions.py | 39 +++---- test/run_tests.py | 132 +++++++++++++--------- 6 files changed, 125 insertions(+), 76 deletions(-) create mode 100644 test/__init__.py create mode 100644 test/helpers/__init__.py create mode 100644 test/helpers/integration_test_helper.py create mode 100644 test/integration/__init__.py diff --git a/test/__init__.py b/test/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/test/helpers/__init__.py b/test/helpers/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/test/helpers/integration_test_helper.py b/test/helpers/integration_test_helper.py new file mode 100644 index 0000000..63c7b9a --- /dev/null +++ b/test/helpers/integration_test_helper.py @@ -0,0 +1,30 @@ +import os +import requests +import json + +class IntegrationTestHelper: + + @classmethod + def get_environment_variables(cls): + username = os.environ["GEOCODER_API_TEST_USERNAME"] + api_key = os.environ["GEOCODER_API_TEST_API_KEY"] + host = os.environ["GEOCODER_API_TEST_HOST"] + table_name = os.environ["GEOCODER_API_TEST_TABLE_NAME"] + + return { + "username": username, + "api_key": api_key, + "host": host, + "table_name": table_name + } + + @classmethod + def execute_query(cls, sql_api_url, query): + query_url = "{0}&q={1}".format(sql_api_url, query) + print "Executing query: {0}".format(query_url) + query_response = requests.get(query_url) + if query_response.status_code != 200: + raise Exception("Error executing SQL API query") + query_response_data = json.loads(query_response.text) + + return query_response_data['rows'][0]['geometry'] diff --git a/test/integration/__init__.py b/test/integration/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/test/integration/test_admin0_functions.py b/test/integration/test_admin0_functions.py index 3b460f7..eac3c76 100644 --- a/test/integration/test_admin0_functions.py +++ b/test/integration/test_admin0_functions.py @@ -1,29 +1,22 @@ -import os, time, requests, json from unittest import TestCase from nose.tools import assert_raises +from nose.tools import assert_not_equal +from ..helpers.integration_test_helper import IntegrationTestHelper -class TestConfigHelper(TestCase): +class TestAdmin0Functions(TestCase): - def setUp(self): - username = os.environ["GEOCODER_API_TEST_USERNAME"] - api_key = os.environ["GEOCODER_API_TEST_API_KEY"] - host = os.environ["GEOCODER_API_TEST_HOST"] - self.table_name = os.environ["GEOCODER_API_TEST_TABLE_NAME"] - self.sql_api_url = "https://{0}.{1}/api/v2/sql?api_key={2}".format(username, host, api_key) + def setUp(self): + self.env_variables = IntegrationTestHelper.get_environment_variables() + self.sql_api_url = "https://{0}.{1}/api/v2/sql?api_key={2}".format( + self.env_variables['username'], + self.env_variables['host'], + self.env_variables['api_key'] + ) - def test_if_select_with_admin0_is_ok(self): - query = "SELECT cdb_geocode_admin0_polygon(name) as geometry FROM {0} LIMIT 1".format(self.table_name) - geometry = self.execute_query(query) - assert geometry != None - - def build_sql_api_query_url(self, query): - return "{0}&q={1}".format(self.sql_api_url,query) - - def execute_query(self, query): - query_url = self.build_sql_api_query_url(query) - query_response = requests.get(query_url) - if query_response.status_code != 200: - raise Exception("Error executing SQL API query") - query_response_data = json.loads(query_response.text) - return query_response_data['rows'][0]['geometry'] + def test_if_select_with_admin0_is_ok(self): + query = "SELECT cdb_geocode_admin0_polygon(name) as geometry " \ + "FROM {0} LIMIT 1".format( + self.env_variables['table_name']) + geometry = IntegrationTestHelper.execute_query(self.sql_api_url, query) + assert_not_equal(geometry, None) diff --git a/test/run_tests.py b/test/run_tests.py index 259d88d..443d3cd 100644 --- a/test/run_tests.py +++ b/test/run_tests.py @@ -1,11 +1,18 @@ -import getopt, sys, requests, time, json, subprocess, os +import getopt +import sys +import requests +import time +import json +import subprocess +import os + def main(): opts, args = getopt.getopt(sys.argv[1:], "h", ["help", "host="]) if len(args) < 2: - usage() - sys.exit() + usage() + sys.exit() host = "cartodb.com" username = args[0] @@ -16,76 +23,95 @@ def main(): usage() sys.exit() elif o in ("--host"): - host = opts[0][1] + host = opts[0][1] else: assert False, "unhandled option" try: - table_name = import_test_dataset(username, api_key, host) - set_environment_variables(username, api_key, table_name, host) - execute_tests() + table_name = import_test_dataset(username, api_key, host) + set_environment_variables(username, api_key, table_name, host) + execute_tests() finally: - clean_environment_variables() - clean_test_dataset(username, api_key, table_name, host) + clean_environment_variables() + clean_test_dataset(username, api_key, table_name, host) + def usage(): - print """Usage: run_tests.py [options] username api_key + print """Usage: run_tests.py [options] username api_key Options: -h: Show this help - --host: take that host as base (by default is cartodb.com). Eg. cartodb.com""" + --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/?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) + url = "https://{0}.{1}/api/v1/imports/?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'] + 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/"]) - process.wait() - print "Testing finished!" + print "Start testing..." + process = subprocess.Popen(["nosetests", "--where=integration/"]) + process.wait() + print "Testing finished!" + def set_environment_variables(username, api_key, table_name, host): - os.environ["GEOCODER_API_TEST_USERNAME"] = username - os.environ["GEOCODER_API_TEST_API_KEY"] = api_key - os.environ["GEOCODER_API_TEST_TABLE_NAME"] = table_name - os.environ["GEOCODER_API_TEST_HOST"] = host + os.environ["GEOCODER_API_TEST_USERNAME"] = username + os.environ["GEOCODER_API_TEST_API_KEY"] = api_key + os.environ["GEOCODER_API_TEST_TABLE_NAME"] = table_name + os.environ["GEOCODER_API_TEST_HOST"] = host + def clean_environment_variables(): - print "Cleaning test dataset environment variables..." - del os.environ["GEOCODER_API_TEST_USERNAME"] - del os.environ["GEOCODER_API_TEST_API_KEY"] - del os.environ["GEOCODER_API_TEST_TABLE_NAME"] - del os.environ["GEOCODER_API_TEST_HOST"] + print "Cleaning test dataset environment variables..." + del os.environ["GEOCODER_API_TEST_USERNAME"] + del os.environ["GEOCODER_API_TEST_API_KEY"] + del os.environ["GEOCODER_API_TEST_TABLE_NAME"] + del os.environ["GEOCODER_API_TEST_HOST"] + def clean_test_dataset(username, api_key, table_name, host): - print "Cleaning test dataset {0}...".format(table_name) - url = "https://{0}.{1}/api/v2/sql?q=drop table {2}&api_key={3}".format(username, host, table_name, api_key) - response = requests.get(url) - if response.status_code != 200: - print "Error cleaning the test dataset: {0}".format(response.text) - sys.exit(1) + print "Cleaning test dataset {0}...".format(table_name) + url = "https://{0}.{1}/api/v2/sql?q=drop table {2}&api_key={3}".format( + username, host, table_name, api_key + ) + response = requests.get(url) + if response.status_code != 200: + print "Error cleaning the test dataset: {0}".format(response.text) + sys.exit(1) if __name__ == "__main__": - main() \ No newline at end of file + main()