First iteration to automate smoke tests

test/run_tests.py

@@ -0,0 +1,91 @@
+import getopt, sys, requests, time, json, subprocess, os
+
+def main():
+    opts, args = getopt.getopt(sys.argv[1:], "h", ["help", "host="])
+
+    if len(args) < 2:
+      usage()
+      sys.exit()
+
+    host = "cartodb.com"
+    username = args[0]
+    api_key = args[1]
+    table_name = "geocoder_api_test_dataset_".format(int(time.time()))
+    for o, a in opts:
+        if o in ("-h", "--help"):
+            usage()
+            sys.exit()
+        elif o in ("--host"):
+          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()
+    finally:
+      clean_environment_variables()
+      clean_test_dataset(username, api_key, table_name, host)
+
+def usage():
+  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"""
+
+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)
+
+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/"])
+  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
+
+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"]
+
+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)
+
+if __name__ == "__main__":
+    main()