Add tap tester (#14)

* changes to tap entrypoint to work with rest of tap

* changes to discover.py to work with rest of tap

* changes to schema.py to work with rest of the tap

* changes to sync.py to work with rest of the tap

* updated README links to documentation

* changes to client.py:
* actually raise 429 error
* pylint fixes

* initial working version of streams.py

* adding pagination to streams.py

* adding links object as part of client return for pagination changes

* adjusting indentation to match PEP8:
* __init__.py
* client.py
* sync.py

* removing extraneous keys and related functions from discover.py

* changes to streams.py:
* indentation fixes to match PEP8
* unpacking result of client.get with `records, _`

* adding tap-tester base.py

* adding tap-tester sync canary

* adding tap-tester discovery

* adding tap-tester start date

* adding tap-tester automated fields

* changes to setup.py:
* bumping singer-python
* adding dev requirements

* adding circleci config

* adding tap-tester pagination

* fix for onetimes stream not supporting cursor based pagination

* updating cirlce config

* adding additional assertions per PR feedback

* add additional streams to pagination test

* switch customers stream to page based
* Link header was not honoring original call params and duplicating records

* Make pylint happy

* Update tap-tester invocation

* pylint fixes

* remove extraneous string interpolation

* remove extraneous class attributes for shop stream

* fixes to start_date test:
* adjust start_date_2 to work with data
* add stream replication methods for tests
* modify/add assertions based on stream replication methods and data

* adding assertion for valid-replication-keys to discover test

Co-authored-by: Andy Lu <andy@stitchdata.com>

.circleci/config.yml

@@ -0,0 +1,44 @@
+version: 2
+jobs:
+  build:
+    docker:
+      - image: 218546966473.dkr.ecr.us-east-1.amazonaws.com/circle-ci:stitch-tap-tester
+    steps:
+      - checkout
+      - run:
+          name: 'Setup virtual env'
+          command: |
+            python3 -mvenv /usr/local/share/virtualenvs/tap-recharge
+            source /usr/local/share/virtualenvs/tap-recharge/bin/activate
+            pip install .[dev]
+      - run:
+          name: 'pylint'
+          command: |
+            source /usr/local/share/virtualenvs/tap-recharge/bin/activate
+            pylint tap_recharge --disable 'missing-module-docstring,missing-function-docstring,missing-class-docstring,no-else-raise,raise-missing-from,inconsistent-return-statements'
+      - run:
+          when: always
+          name: 'Integration Tests'
+          command: |
+            aws s3 cp s3://com-stitchdata-dev-deployment-assets/environments/tap-tester/tap_tester_sandbox dev_env.sh
+            source dev_env.sh
+            source /usr/local/share/virtualenvs/tap-tester/bin/activate
+            run-test --tap=tap-recharge tests
+
+workflows:
+  version: 2
+  commit:
+    jobs:
+      - build:
+          context: circleci-user
+  build_daily:
+    triggers:
+      - schedule:
+          cron: "0 14 * * *"
+          filters:
+            branches:
+              only:
+                - master
+    jobs:
+      - build:
+          context: circleci-user

README.md

@@ -14,10 +14,10 @@ This tap:
   - [Customers](https://developer.rechargepayments.com/#list-customers)
   - [Discounts](https://developer.rechargepayments.com/#list-discounts)
   - [Metafields for Store, Customers, Subscriptions](https://developer.rechargepayments.com/#list-metafields)
-  - [One-time Products](https://developer.rechargepayments.com/#list-onetimes-alpha)
+  - [One-time Products](https://developer.rechargepayments.com/#list-onetimes)
   - [Orders](https://developer.rechargepayments.com/#list-orders)
-  - [Products](https://developer.rechargepayments.com/#list-products-beta)
-  - [Shop](https://developer.rechargepayments.com/#retrieve-shop)
+  - [Products](https://developer.rechargepayments.com/#list-products)
+  - [Shop](https://developer.rechargepayments.com/#retrieve-a-shop)
   - [Subscriptions](https://developer.rechargepayments.com/#list-subscriptions)
 - Outputs the schema for each resource
 - Incrementally pulls data based on the input state
@@ -94,7 +94,7 @@ This tap:
   - Bookmark: updated_at (date-time)
 - Transformations: None
 
-[**onetimes**](https://developer.rechargepayments.com/#list-onetimes-alpha)
+[**onetimes**](https://developer.rechargepayments.com/#list-onetimes)
 - Endpoint: https://api.rechargeapps.com/onetimes
 - Primary keys: id
 - Foreign keys: address_id (addresses), customer_id (customers), recharge_product_id (products), shopify_product_id, shopify_variant_id
@@ -112,15 +112,15 @@ This tap:
   - Bookmark: updated_at (date-time)
 - Transformations: None
 
-[**products**](https://developer.rechargepayments.com/#list-products-beta)
+[**products**](https://developer.rechargepayments.com/#list-products)
 - Endpoint: https://api.rechargeapps.com/products
 - Primary keys: id
 - Foreign keys: collection_id (collections), shopify_product_id
 - Replication strategy: Incremental (query all, filter results)
   - Bookmark: updated_at (date-time)
 - Transformations: None
 
-[**shop**](https://developer.rechargepayments.com/#retrieve-shop)
+[**shop**](https://developer.rechargepayments.com/#retrieve-a-shop)
 - Endpoint: https://api.rechargeapps.com/shop
 - Primary keys: id
 - Foreign keys: None

setup.py

@@ -11,7 +11,7 @@
       install_requires=[
           'backoff==1.8.0',
           'requests==2.23.0',
-          'singer-python==5.9.0'
+          'singer-python==5.10.0'
       ],
       entry_points='''
           [console_scripts]
@@ -22,4 +22,9 @@
           'tap_recharge': [
               'schemas/*.json'
           ]
+      },
+      extras_require={
+          'dev': [
+              'pylint'
+          ]
       })

tap_recharge/__init__.py

@@ -1,15 +1,12 @@
 #!/usr/bin/env python3
 
-import sys
-import json
-import argparse
-import singer
-from singer import metadata, utils
+from singer import get_logger, utils
+
 from tap_recharge.client import RechargeClient
 from tap_recharge.discover import discover
 from tap_recharge.sync import sync
 
-LOGGER = singer.get_logger()
+LOGGER = get_logger()
 
 REQUIRED_CONFIG_KEYS = [
     'access_token',
@@ -21,17 +18,19 @@ def do_discover():
 
     LOGGER.info('Starting discover')
     catalog = discover()
-    json.dump(catalog.to_dict(), sys.stdout, indent=2)
+    catalog.dump()
     LOGGER.info('Finished discover')
 
 
-@singer.utils.handle_top_exception(LOGGER)
+@utils.handle_top_exception(LOGGER)
 def main():
+    """Entrypoint function for tap."""
 
-    parsed_args = singer.utils.parse_args(REQUIRED_CONFIG_KEYS)
+    parsed_args = utils.parse_args(REQUIRED_CONFIG_KEYS)
 
-    with RechargeClient(parsed_args.config['access_token'],
-                        parsed_args.config['user_agent']) as client:
+    with RechargeClient(
+        parsed_args.config['access_token'],
+        parsed_args.config['user_agent']) as client:
 
         state = {}
         if parsed_args.state:
@@ -40,10 +39,11 @@ def main():
         if parsed_args.discover:
             do_discover()
         elif parsed_args.catalog:
-            sync(client=client,
-                 catalog=parsed_args.catalog,
-                 state=state,
-                 start_date=parsed_args.config['start_date'])
+            sync(
+                client=client,
+                catalog=parsed_args.catalog,
+                state=state,
+                config=parsed_args.config)
 
 if __name__ == '__main__':
     main()

tap_recharge/client.py

@@ -1,8 +1,8 @@
 import backoff
 import requests
-from requests.exceptions import ConnectionError
-from singer import metrics, utils
+
 import singer
+from singer import metrics, utils
 
 LOGGER = singer.get_logger()
 
@@ -82,8 +82,8 @@ def raise_for_error(response):
                 return
             response = response.json()
             if ('error' in response) or ('errorCode' in response):
-                message = '%s: %s' % (response.get('error', str(error)),
-                                      response.get('message', 'Unknown Error'))
+                message = f"{response.get('error', str(error))}: \
+                    {response.get('message', 'Unknown Error')}"
                 error_code = response.get('status')
                 ex = get_exception_for_error_code(error_code)
                 if response.status_code == 401 and 'Expired access token' in message:
@@ -97,10 +97,11 @@ def raise_for_error(response):
             raise RechargeError(error)
 
 
-class RechargeClient(object):
-    def __init__(self,
-                 access_token,
-                 user_agent=None):
+class RechargeClient:
+    def __init__(
+            self,
+            access_token,
+            user_agent=None):
         self.__access_token = access_token
         self.__user_agent = user_agent
         self.__session = requests.Session()
@@ -114,10 +115,11 @@ def __enter__(self):
     def __exit__(self, exception_type, exception_value, traceback):
         self.__session.close()
 
-    @backoff.on_exception(backoff.expo,
-                          Server5xxError,
-                          max_tries=5,
-                          factor=2)
+    @backoff.on_exception(
+        backoff.expo,
+        Server5xxError,
+        max_tries=5,
+        factor=2)
     def check_access_token(self):
         if self.__access_token is None:
             raise Exception('Error: Missing access_token.')
@@ -131,16 +133,17 @@ def check_access_token(self):
             url='https://api.rechargeapps.com',
             headers=headers)
         if response.status_code != 200:
-            LOGGER.error('Error status_code = {}'.format(response.status_code))
+            LOGGER.error('Error status_code = %s', response.status_code)
             raise_for_error(response)
         else:
             return True
 
 
-    @backoff.on_exception(backoff.expo,
-                          (Server5xxError, ConnectionError, Server429Error),
-                          max_tries=5,
-                          factor=2)
+    @backoff.on_exception(
+        backoff.expo,
+        (Server5xxError, requests.ConnectionError, Server429Error),
+        max_tries=5,
+        factor=2)
     # Call/rate limit: https://developer.rechargepayments.com/#rate-limits
     @utils.ratelimit(120, 60)
     def request(self, method, path=None, url=None, **kwargs):
@@ -177,17 +180,20 @@ def request(self, method, path=None, url=None, **kwargs):
         if response.status_code >= 500:
             raise Server5xxError()
 
+        if response.status_code == 429:
+            raise Server429Error()
+
         if response.status_code != 200:
             raise_for_error(response)
 
         # Log invalid JSON (e.g. unterminated string errors)
         try:
             response_json = response.json()
         except Exception as err:
-            LOGGER.error('{}'.format(err))
+            LOGGER.error(err)
             raise Exception(err)
 
-        return response_json
+        return response_json, response.links
 
     def get(self, path, **kwargs):
         return self.request('GET', path=path, **kwargs)

tap_recharge/discover.py

@@ -1,20 +1,24 @@
-from singer.catalog import Catalog, CatalogEntry, Schema
-from tap_recharge.schema import get_schemas, STREAMS
+from singer.catalog import Catalog
+from tap_recharge.schema import get_schemas
+
 
 def discover():
+    """
+    Constructs a singer Catalog object based on the schemas and metadata.
+    """
     schemas, field_metadata = get_schemas()
-    catalog = Catalog([])
+    streams = []
+
+    for schema_name, schema in schemas.items():
+        schema_meta = field_metadata[schema_name]
 
-    for stream_name, schema_dict in schemas.items():
-        schema = Schema.from_dict(schema_dict)
-        mdata = field_metadata[stream_name]
+        catalog_entry = {
+            'stream': schema_name,
+            'tap_stream_id': schema_name,
+            'schema': schema,
+            'metadata': schema_meta
+        }
 
-        catalog.streams.append(CatalogEntry(
-            stream=stream_name,
-            tap_stream_id=stream_name,
-            key_properties=STREAMS[stream_name]['key_properties'],
-            schema=schema,
-            metadata=mdata
-        ))
+        streams.append(catalog_entry)
 
-    return catalog
+    return Catalog.from_dict({'streams': streams})