diff --git a/.circleci/config.yml b/.circleci/config.yml index 6f4c16a..2e66adc 100644 --- a/.circleci/config.yml +++ b/.circleci/config.yml @@ -1,8 +1,8 @@ -version: 2 +version: 2.1 jobs: build: docker: - - image: 218546966473.dkr.ecr.us-east-1.amazonaws.com/circle-ci:tap-tester + - image: 218546966473.dkr.ecr.us-east-1.amazonaws.com/circle-ci:tap-tester-v4 steps: - checkout - run: @@ -12,12 +12,28 @@ jobs: source /usr/local/share/virtualenvs/tap-eloqua/bin/activate pip install -U 'pip<19.2' 'setuptools<51.0.0' pip install .[dev] - - add_ssh_keys + - run: + name: 'pylint' + command: | + source /usr/local/share/virtualenvs/tap-eloqua/bin/activate + pylint tap_eloqua -d C,R,W + - run: + name: 'Unit Tests' + command: | + source /usr/local/share/virtualenvs/tap-eloqua/bin/activate + pip install nose coverage + nosetests --with-coverage --cover-erase --cover-package=tap_eloqua --cover-html-dir=htmlcov tests/unittests + coverage html + - store_test_results: + path: test_output/report.xml + - store_artifacts: + path: htmlcov workflows: version: 2 commit: jobs: - - build + - build: + context: circleci-user build_daily: triggers: - schedule: @@ -27,4 +43,5 @@ workflows: only: - master jobs: - - build + - build: + context: circleci-user diff --git a/.gitignore b/.gitignore index faf849b..237acac 100644 --- a/.gitignore +++ b/.gitignore @@ -102,4 +102,10 @@ test.*.json rsa-key tags singer-check-tap-data -state.json \ No newline at end of file +state.json + +catalog.json +get_catalog.py +persist/ +configs/ +current_state/ diff --git a/setup.py b/setup.py index c7056a7..5319d17 100644 --- a/setup.py +++ b/setup.py @@ -10,14 +10,15 @@ classifiers=['Programming Language :: Python :: 3 :: Only'], py_modules=['tap_eloqua'], install_requires=[ - 'backoff==1.3.2', + 'backoff==1.8.0', 'requests==2.20.1', 'pendulum==2.0.3', - 'singer-python==5.2.0' + 'singer-python==5.13.0' ], extras_require={ - 'dev': [ - 'ipdb==0.11' + "dev": [ + "pylint", + "ipdb", ] }, entry_points=''' diff --git a/tap_eloqua/__init__.py b/tap_eloqua/__init__.py index b1c05b2..3d31a34 100644 --- a/tap_eloqua/__init__.py +++ b/tap_eloqua/__init__.py @@ -2,11 +2,7 @@ import sys import json -import argparse - import singer -from singer import metadata - from tap_eloqua.client import EloquaClient from tap_eloqua.discover import discover from tap_eloqua.sync import sync @@ -27,101 +23,17 @@ def do_discover(client): json.dump(catalog.to_dict(), sys.stdout, indent=2) LOGGER.info('Finished discover') -##### TEMP - -from singer.catalog import Catalog - -def check_config(config, required_keys): - missing_keys = [key for key in required_keys if key not in config] - if missing_keys: - raise Exception("Config is missing required keys: {}".format(missing_keys)) - -def load_json(path): - with open(path) as fil: - return json.load(fil) - -def parse_args(required_config_keys): - '''Parse standard command-line args. - - Parses the command-line arguments mentioned in the SPEC and the - BEST_PRACTICES documents: - - -c,--config Config file - -s,--state State file - -d,--discover Run in discover mode - -p,--properties Properties file: DEPRECATED, please use --catalog instead - --catalog Catalog file - - Returns the parsed args object from argparse. For each argument that - point to JSON files (config, state, properties), we will automatically - load and parse the JSON file. - ''' - parser = argparse.ArgumentParser() - - parser.add_argument( - '-c', '--config', - help='Config file', - required=True) - - parser.add_argument( - '-s', '--state', - help='State file') - - parser.add_argument( - '-p', '--properties', - help='Property selections: DEPRECATED, Please use --catalog instead') - - parser.add_argument( - '--catalog', - help='Catalog file') - - parser.add_argument( - '-d', '--discover', - action='store_true', - help='Do schema discovery') - - args = parser.parse_args() - if args.config: - setattr(args, 'config_path', args.config) - args.config = load_json(args.config) - if args.state: - setattr(args, 'state_path', args.state) - args.state = load_json(args.state) - else: - args.state = {} - if args.properties: - setattr(args, 'properties_path', args.properties) - args.properties = load_json(args.properties) - if args.catalog: - setattr(args, 'catalog_path', args.catalog) - args.catalog = Catalog.load(args.catalog) - - check_config(args.config, required_config_keys) - - return args - -##### @singer.utils.handle_top_exception(LOGGER) def main(): - #parsed_args = singer.utils.parse_args(REQUIRED_CONFIG_KEYS) - parsed_args = parse_args(REQUIRED_CONFIG_KEYS) - - with EloquaClient(parsed_args.config_path, - parsed_args.config['client_id'], - parsed_args.config['client_secret'], - parsed_args.config['refresh_token'], - parsed_args.config['redirect_uri'], - parsed_args.config.get('user_agent')) as client: - - if parsed_args.discover: + args = singer.parse_args(REQUIRED_CONFIG_KEYS) + if args.dev: + LOGGER.warning("Executing Tap in Dev mode") + with EloquaClient(args.config_path, args.config, args.dev) as client: + if args.discover: do_discover(client) - elif parsed_args.catalog: - sync(client, - parsed_args.catalog, - parsed_args.state, - parsed_args.config['start_date'], - int(parsed_args.config.get('bulk_page_size', 5000))) + elif args.catalog: + sync(client, args.catalog, args.state, args.config) if __name__ == "__main__": main() \ No newline at end of file diff --git a/tap_eloqua/client.py b/tap_eloqua/client.py index 21ecadb..2139b4b 100644 --- a/tap_eloqua/client.py +++ b/tap_eloqua/client.py @@ -1,28 +1,28 @@ -import json -from datetime import datetime, timedelta +from datetime import timedelta import backoff import requests from requests.exceptions import ConnectionError -from singer import metrics +from singer import metrics,get_logger +from singer.utils import strptime_to_utc,now,strftime +from .utils import write_config +LOGGER = get_logger() + class Server5xxError(Exception): pass + class EloquaClient(object): - def __init__(self, - config_path, - client_id, - client_secret, - refresh_token, - redirect_uri, - user_agent): + def __init__(self,config_path, config, dev_mode=False): self.__config_path = config_path - self.__client_id = client_id - self.__client_secret = client_secret - self.__refresh_token = refresh_token - self.__redirect_uri = redirect_uri - self.__user_agent = user_agent + self.config = config + self.__client_id = config["client_id"] + self.__client_secret = config["client_secret"] + self.__refresh_token = config["refresh_token"] + self.__redirect_uri = config["redirect_uri"] + self.__user_agent = config.get("user_agent","") + self.dev_mode = dev_mode self.__access_token = None self.__expires = None self.__session = requests.Session() @@ -40,7 +40,16 @@ def __exit__(self, type, value, traceback): max_tries=5, factor=2) def get_access_token(self): - if self.__access_token is not None and self.__expires > datetime.utcnow(): + if self.dev_mode: + try: + self.__access_token = self.config['access_token'] + self.__expires=strptime_to_utc(self.config['expires_in']) + except KeyError as ex: + raise Exception("Unable to locate key in config") from ex + if not self.__access_token or self.__expires < now(): + raise Exception("Access Token in config is expired, unable to authenticate in dev mode") + + if self.__access_token and self.__expires > now(): return headers = {} @@ -63,31 +72,25 @@ def get_access_token(self): if response.status_code != 200: eloqua_response = response.json() - eloqua_response.update( - {'status': response.status_code}) - raise Exception( - 'Unable to authenticate (Eloqua response: `{}`)'.format( - eloqua_response)) + eloqua_response.update({'status': response.status_code}) + raise Exception('Unable to authenticate (Eloqua response: `{}`)'.format(eloqua_response)) data = response.json() - self.__access_token = data['access_token'] self.__refresh_token = data['refresh_token'] + expires_in_seconds = data['expires_in'] - 10 # pad by 10 seconds + self.__expires = now() + timedelta(seconds=expires_in_seconds) - ## refresh_token rotates on every reauth - with open(self.__config_path) as file: - config = json.load(file) - config['refresh_token'] = data['refresh_token'] - with open(self.__config_path, 'w') as file: - json.dump(config, file, indent=2) - - expires_seconds = data['expires_in'] - 10 # pad by 10 seconds - self.__expires = datetime.utcnow() + timedelta(seconds=expires_seconds) + if not self.dev_mode: + update_config_keys = { + "refresh_token":self.__refresh_token, + "access_token":self.__access_token, + "expires_in": strftime(self.__expires) + } + self.config = write_config(self.__config_path,update_config_keys) def get_base_urls(self): - data = self.request('GET', - url='https://login.eloqua.com/id', - endpoint='base_url') + data = self.request('GET',url='https://login.eloqua.com/id',endpoint='base_url') self.__base_url = data['urls']['base'] @backoff.on_exception(backoff.expo, diff --git a/tap_eloqua/sync.py b/tap_eloqua/sync.py index f346301..5d274ad 100644 --- a/tap_eloqua/sync.py +++ b/tap_eloqua/sync.py @@ -173,8 +173,8 @@ def sync_bulk_obj(client, catalog, state, start_date, stream_name, bulk_page_siz bulk_page_size, last_date, offset=last_offset) - except HTTPError as e: - if e.response.status_code in [404, 410]: + except HTTPError as ex: + if ex.response.status_code in [404, 410]: LOGGER.info('{} - Previous export expired: {}'.format(stream_name, last_sync_id)) else: raise @@ -377,12 +377,12 @@ def sync_activity_stream(client, activity_type): finished = False sync_start = pendulum.now('UTC') - end_date = sync_start + end_date, last_sync_date = sync_start, None while not finished: try: # Get latest bookmark to adjust time window from, if needed last_date_raw = get_bulk_bookmark(state, stream_name).get('datetime', start_date) - last_date = pendulum.parse(last_date_raw) + last_sync_date = pendulum.parse(last_date_raw) update_current_stream(state, stream_name) sync_bulk_obj(client, @@ -399,12 +399,14 @@ def sync_activity_stream(client, # If not done, sync again to now() end_date = sync_start except ActivityExportTooLarge as ex: - LOGGER.warn(ex) - end_date = last_date.add(seconds=(end_date - last_date).total_seconds() / 2) + LOGGER.warning(ex) + end_date = last_sync_date.add(seconds=(end_date - last_sync_date).total_seconds() / 2) if end_date > sync_start: end_date = sync_start -def sync(client, catalog, state, start_date, bulk_page_size): +def sync(client, catalog, state, config): + start_date = config['start_date'] + bulk_page_size = int(config.get('bulk_page_size', 5000)) selected_streams = get_selected_streams(catalog) if not selected_streams: diff --git a/tap_eloqua/utils.py b/tap_eloqua/utils.py new file mode 100644 index 0000000..0e6e77a --- /dev/null +++ b/tap_eloqua/utils.py @@ -0,0 +1,18 @@ +from typing import Dict +import json +from singer import get_logger + +LOGGER = get_logger() + + +def write_config(config_path,data) : + """ + Updates the provided filepath with json format of the `data` object + does a safe write by performing a read before write, updates only specific keys, does not rewrite. + """ + with open(config_path,'r') as tap_config: + config = json.load(tap_config) + config.update(data) + with open(config_path,'w') as tap_config: + json.dump(config, tap_config, indent=2) + return config diff --git a/tests/unittests/test_devmode.py b/tests/unittests/test_devmode.py new file mode 100644 index 0000000..0a477dc --- /dev/null +++ b/tests/unittests/test_devmode.py @@ -0,0 +1,118 @@ +import datetime +import json +import os +import tempfile +import unittest +from unittest.mock import patch + +import pytz + +from tap_eloqua.client import EloquaClient, strftime + + +class MockResponse: + def __init__(self, json_data, status_code): + self.json_data = json_data + self.status_code = status_code + + def json(self): + return self.json_data + + +def mocked_auth_post(*args, **kwargs): + return MockResponse({"access_token": "", "refresh_token": "", "expires_in": 5000}, status_code=200) + + +class Test_ClientDevMode(unittest.TestCase): + """Test the dev mode functionality.""" + + tmpdir = tempfile.mkdtemp() + predictable_filename = "eloqua_config.json" + + base_config = { + "client_id": "", + "client_secret": "", + "refresh_token": "", + "redirect_uri": "", + "user_agent": "", + } + + def setUp(self) -> None: + self.tmpdir = tempfile.mkdtemp() + self.predictable_filename = "eloqua_config.json" + return super().setUp() + + @property + def tmp_config_path(self): + return os.path.join(self.tmpdir, self.predictable_filename) + + @patch("requests.Session.post", side_effect=mocked_auth_post) + def test_client_without_dev_mode(self, mocked_auth_post): + """checks if the client can write refresh token and expiry to + config.""" + + with open(self.tmp_config_path, "w") as ff: + json.dump(self.base_config, ff) + + + client = EloquaClient(config_path=self.tmp_config_path,dev_mode=False,config=self.base_config) + client.get_access_token() + with open(self.tmp_config_path) as config_file: + config = json.load(config_file) + self.assertEqual(True, "expires_in" in config) + self.assertEqual(True, "access_token" in config) + self.assertEqual(True, "refresh_token" in config) + + def test_devmode_accesstoken_absent(self, *args, **kwargs): + """checks exception if access token is not present in config.""" + + with open(self.tmp_config_path, "w") as config_file: + json.dump(self.base_config, config_file) + + try: + client = EloquaClient(config_path=self.tmp_config_path,dev_mode=True,config=self.base_config) + + client.get_access_token() + except Exception as ex: + self.assertEqual(str(ex), "Unable to locate key in config") + + @patch("requests.Session.post", side_effect=mocked_auth_post) + def test_client_valid_token(self, mocked_auth_post): + """checks if the client can fetch and validate refresh token and expiry + from config.""" + + config_sample = { + "access_token": "token", + "expires_in": strftime(datetime.datetime.utcnow().replace(tzinfo=pytz.UTC) + datetime.timedelta(days=5)), + **self.base_config, + } + with open(self.tmp_config_path, "w") as config_file: + json.dump(config_sample, config_file) + client = EloquaClient(config_path=self.tmp_config_path,dev_mode=True,config=config_sample) + client.get_access_token() + + @patch("requests.Session.post", side_effect=mocked_auth_post) + def test_client_invalid_token(self, mocked_auth_post): + """checks if the client can fetch and validate refresh token and expiry + from config.""" + expiry_time = strftime(datetime.datetime.utcnow().replace(tzinfo=pytz.UTC) - datetime.timedelta(minutes=5)) + config_sample = { + "access_token": "token", + "expires_in": expiry_time, + **self.base_config, + } + with open(self.tmp_config_path, "w") as config_file: + json.dump(config_sample, config_file) + try: + client = EloquaClient(config_path=self.tmp_config_path,dev_mode=True,config=config_sample) + client.get_access_token() + except Exception as ex: + self.assertEqual(str(ex), "Access Token in config is expired, unable to authenticate in dev mode") + + def tearDown(self) -> None: + try: + os.remove(self.tmp_config_path) + os.rmdir(self.tmpdir) + except OSError: + pass + return super().tearDown()