Fix access token refresh issue (#69)

* fixes to access token and projects data fetching

* fix integration tests

* close asana client in tests

* update testcase for asana

* remove unused test

* update tests

* add comments for projects

* handles 401 error

* handles unauthorized with build_client

* Add Asana API error handling to recursive subtask fetching by decorating fetch_children with asana_error_handling

* fix pylint

* adds unit test case for substasks error handling

* fix unit test

---------

Co-authored-by: mukeshbhatt18gl <mukesh.bhatt@qlik.com>

Author: rdeshmukh15

CHANGELOG.md

@@ -1,5 +1,9 @@
 # Changelog
 
+## 2.4.1
+  *  Add Asana API error handling to recursive subtask fetching by decorating fetch_children with asana_error_handling.[#69](https://github.com/singer-io/tap-asana/pull/69)
+
+
 ## 2.4.0
   * Upgrade asana SDK version to 5.1.0. [#61](https://github.com/singer-io/tap-asana/pull/61)
 

setup.py

@@ -3,14 +3,15 @@
 
 setup(
     name="tap-asana",
-    version="2.4.0",
+    version="2.4.1",
     description="Singer.io tap for extracting Asana data",
     author="Stitch",
     url="http://github.com/singer-io/tap-asana",
     classifiers=["Programming Language :: Python :: 3 :: Only"],
     py_modules=["tap_asana"],
     install_requires=[
         "asana==5.1.0",
+        "parameterized",
         "requests==2.32.4",
         "singer-python==6.1.1"
     ],

tap_asana/streams/projects.py

@@ -47,6 +47,7 @@ def get_objects(self):
     bookmark = self.get_bookmark()
     session_bookmark = bookmark
 
+
     workspaces = self.fetch_workspaces()
 
     # Use ProjectsApi to fetch projects

tap_asana/streams/subtasks.py

@@ -1,7 +1,7 @@
 import asana
 import singer
 from tap_asana.context import Context
-from tap_asana.streams.base import Stream
+from tap_asana.streams.base import asana_error_handling, Stream
 
 LOGGER = singer.get_logger()
 
@@ -85,6 +85,7 @@ def get_objects(self):
         # Update the bookmark after processing all subtasks
         self.update_bookmark(session_bookmark)
 
+    @asana_error_handling
     def fetch_children(self, p_task, opt_fields):
         subtasks_children = []
         resource = asana.TasksApi(Context.asana.client)

tests/test_bookmarks.py

@@ -13,11 +13,21 @@ def name(self):
 
     def test_run(self):
         # running sync with multiple date timestamps accross different streams due to differences in bookmark values
-        self.run_test("2021-11-09T00:00:00Z", "2023-11-10T00:00:00Z", {"projects",})
         self.run_test("2023-11-28T00:00:00Z", "2023-11-30T00:00:00Z", {"subtasks",})
         # Removing Portfolios as they are only available for users in an Enterprise or Business plan.
+        # Removing Projects as data is available for the same date only.
         self.run_test("2019-01-28T00:00:00Z", "2023-11-30T00:00:00Z", self.expected_streams() - {"subtasks","projects","portfolios"})
 
+    def parse_start_date_ts(self, start_date):
+        """Parse either midnight-only or full timestamp start dates into epoch seconds."""
+        for dt_format in (self.START_DATE_FORMAT, "%Y-%m-%dT%H:%M:%SZ", self.BOOKMARK_FOMAT):
+            try:
+                return self.dt_to_ts(start_date, dt_format)
+            except ValueError:
+                continue
+
+        raise ValueError(f"Unsupported start_date format: {start_date}")
+
 
     def run_test(self, start_date_1, start_date_2, streams):
         """
@@ -135,7 +145,7 @@ def run_test(self, start_date_1, start_date_2, streams):
                         # We have added 'second_start_date' as the bookmark, it is more recent than
                         # the default start date and it will work as a simulated bookmark
                         self.assertGreaterEqual(
-                            replication_key_value_parsed, self.dt_to_ts(self.second_start_date, self.START_DATE_FORMAT),
+                            replication_key_value_parsed, self.parse_start_date_ts(self.second_start_date),
                             msg="Second sync did not respect the bookmark, \
                                 a record with a smaller replication-key value was synced."
                         )

tests/test_start_date.py

@@ -12,11 +12,21 @@ def name(self):
 
     def test_run(self):
         # running sync with multiple date timestamps accross different streams due to differences in bookmark values
-        self.run_test("2021-11-09T00:00:00Z", "2023-11-10T00:00:00Z", {"projects",})
         self.run_test("2023-11-28T00:00:00Z", "2023-11-30T00:00:00Z", {"subtasks",})
         # Removing Portfolios as they are only available for users in an Enterprise or Business plan.
+        # Removing Projects as data is available for the same date only.
         self.run_test("2019-01-28T00:00:00Z", "2023-11-30T00:00:00Z", self.expected_streams() - {"subtasks","projects","portfolios"})
-        
+
+    def parse_start_date_ts(self, start_date):
+        """Parse either midnight-only or full timestamp start dates into epoch seconds."""
+        for dt_format in (self.START_DATE_FORMAT, "%Y-%m-%dT%H:%M:%SZ", self.BOOKMARK_FOMAT):
+            try:
+                return self.dt_to_ts(start_date, dt_format)
+            except ValueError:
+                continue
+
+        raise ValueError(f"Unsupported start_date format: {start_date}")
+
     def run_test(self, start_date_1, start_date_2, streams):
         """
         Testing that the tap respects the start date
@@ -39,8 +49,8 @@ def run_test(self, start_date_1, start_date_2, streams):
         expected_streams = streams
 
 
-        start_date_1_epoch = self.dt_to_ts(self.first_start_date, self.START_DATE_FORMAT)
-        start_date_2_epoch = self.dt_to_ts(self.second_start_date, self.START_DATE_FORMAT)
+        start_date_1_epoch = self.parse_start_date_ts(self.first_start_date)
+        start_date_2_epoch = self.parse_start_date_ts(self.second_start_date)
 
         ##########################################################################
         # Update Start Date for 1st sync

tests/unittests/test_asana_auth.py

@@ -0,0 +1,57 @@
+import unittest
+from unittest import mock
+
+from tap_asana.asana import Asana
+
+
+class TestAsanaAuth(unittest.TestCase):
+    def tearDown(self):
+        if hasattr(self, "asana_client") and self.asana_client.client:
+            try:
+                self.asana_client.client.pool.close()
+                self.asana_client.client.pool.join()
+            except Exception:
+                pass
+
+    @mock.patch("tap_asana.asana.requests.post")
+    def test_refresh_access_token_success_updates_client_and_token(self, mocked_post):
+        response = mock.Mock()
+        response.status_code = 200
+        response.json.return_value = {"access_token": "new_token"}
+        mocked_post.return_value = response
+
+        self.asana_client = Asana("id", "secret", "uri", "refresh", "old_token")
+
+        token = self.asana_client.refresh_access_token()
+
+        self.assertEqual(token, "new_token")
+        self.assertEqual(self.asana_client.access_token, "new_token")
+        self.assertIsNotNone(self.asana_client.client)
+
+    @mock.patch("tap_asana.asana.requests.post")
+    def test_refresh_access_token_missing_access_token_returns_none(self, mocked_post):
+        response = mock.Mock()
+        response.status_code = 200
+        response.json.return_value = {"token_type": "bearer"}
+        mocked_post.return_value = response
+
+        self.asana_client = Asana("id", "secret", "uri", "refresh", "old_token")
+
+        token = self.asana_client.refresh_access_token()
+
+        self.assertIsNone(token)
+        self.assertEqual(self.asana_client.access_token, "old_token")
+
+    @mock.patch("tap_asana.asana.requests.post")
+    def test_refresh_access_token_non_200_returns_none(self, mocked_post):
+        response = mock.Mock()
+        response.status_code = 401
+        response.text = '{"errors":[{"message":"expired"}]}'
+        mocked_post.return_value = response
+
+        self.asana_client = Asana("id", "secret", "uri", "refresh", "old_token")
+
+        token = self.asana_client.refresh_access_token()
+
+        self.assertIsNone(token)
+        self.assertEqual(self.asana_client.access_token, "old_token")

tests/unittests/test_custom_exceptions.py

@@ -16,6 +16,15 @@ def setUp(self):
         """Set up the Asana client and mock context."""
         Context.asana = Asana("test", "test", "test", "test", "test")
 
+    def tearDown(self):
+        """Close the Asana client to suppress ApiClient.__del__ warnings."""
+        if hasattr(Context.asana, 'client') and Context.asana.client:
+            try:
+                Context.asana.client.pool.close()
+                Context.asana.client.pool.join()
+            except Exception:
+                pass
+
     @mock.patch("asana.WorkspacesApi.get_workspaces")
     def test_call_api_invalid_token_error(self, mocked_get_workspaces):
         """Test call_api raises InvalidTokenError."""
@@ -80,6 +89,15 @@ def setUp(self):
         """Set up the Asana client and mock context."""
         Context.asana = Asana("test", "test", "test", "test", "test")
 
+    def tearDown(self):
+        """Close the Asana client to suppress ApiClient.__del__ warnings."""
+        if hasattr(Context.asana, 'client') and Context.asana.client:
+            try:
+                Context.asana.client.pool.close()
+                Context.asana.client.pool.join()
+            except Exception:
+                pass
+
     @mock.patch("asana.WorkspacesApi.get_workspaces")
     def test_call_api_retry_invalid_token_error(self, mocked_get_workspaces):
         """Test call_api retries on InvalidTokenError."""

tests/unittests/test_project_id_caching.py

@@ -54,6 +54,15 @@ def mock_call_api(*args, **kwargs):
 @mock.patch("tap_asana.streams.base.Stream.call_api")
 class TestProjectIdCaching(unittest.TestCase):
 
+    def tearDown(self):
+        """Close the Asana client to suppress ApiClient.__del__ warnings."""
+        if hasattr(Context, 'asana') and hasattr(Context.asana, 'client') and Context.asana.client:
+            try:
+                Context.asana.client.pool.close()
+                Context.asana.client.pool.join()
+            except Exception:
+                pass
+
     def test_sections(self, mocked_call_api, mocked_fetch_projects, mocked_fetch_workspaces, mocked_sleep, mocked_refresh_access_token):
         # Set config file
         Context.config = {'start_date': '2021-01-01T00:00:00Z'}

tests/unittests/test_refersh_access_token.py

@@ -19,6 +19,15 @@ def token_expired_error_raiser(*args, **kwargs):
 @mock.patch("time.sleep")
 class TestRefreshAccessToken(unittest.TestCase):
 
+    def tearDown(self):
+        """Close the Asana client to suppress ApiClient.__del__ warnings."""
+        if hasattr(Context, 'asana') and hasattr(Context.asana, 'client') and Context.asana.client:
+            try:
+                Context.asana.client.pool.close()
+                Context.asana.client.pool.join()
+            except Exception:
+                pass
+
     def test_invalid_token_error_for_get_initial(self, mocked_sleep, mocked_get_tasks, mocked_refresh_access_token):
         """
         Verify that refresh_access_token is called five times due to InvalidTokenError during the initial API call.