From 813c78dd62066d650a9957e4665a61137934be12 Mon Sep 17 00:00:00 2001 From: Gigi Cheang Date: Thu, 11 Jun 2026 15:07:50 -0400 Subject: [PATCH 1/2] test: add retries + fix Signed-off-by: Gigi Cheang --- .../integration/test_cd_tekton_pipeline_v2.py | 68 ++++++++++++------- test/integration/test_cd_toolchain_v2.py | 61 +++++++++++++---- 2 files changed, 92 insertions(+), 37 deletions(-) diff --git a/test/integration/test_cd_tekton_pipeline_v2.py b/test/integration/test_cd_tekton_pipeline_v2.py index 7409671..f7d5120 100644 --- a/test/integration/test_cd_tekton_pipeline_v2.py +++ b/test/integration/test_cd_tekton_pipeline_v2.py @@ -19,6 +19,7 @@ from ibm_cloud_sdk_core import * import os +import time import pytest from environs import Env from ibm_continuous_delivery.cd_toolchain_v2 import * @@ -37,6 +38,7 @@ trigger_prop_name_link = None pipeline_run_id_link = None rerun_id_link = None +run_log_id_link = None env = Env() @@ -580,29 +582,49 @@ def test_list_tekton_pipeline_runs_with_pager(self): f"\n list_tekton_pipeline_runs() returned a total of {len(all_results)} items(s) using TektonPipelineRunsPager." ) - # @needscredentials - # def test_get_tekton_pipeline_run_logs(self): - # global run_log_id_link - # response = self.cd_pipeline_service.get_tekton_pipeline_run_logs( - # pipeline_id=pipeline_tool_id_link, - # id=pipeline_run_id_link - # ) - # assert response.get_status_code() == 200 - # logs_collection = response.get_result() - # assert logs_collection is not None - # print(f'\n logs_collection: {logs_collection}') - # print(f'\n len(logs_collection): {len(logs_collection)}') - - # @needscredentials - # def test_get_tekton_pipeline_run_log_content(self): - # response = self.cd_pipeline_service.get_tekton_pipeline_run_log_content( - # pipeline_id=pipeline_tool_id_link, - # pipeline_run_id='bf4b3abd-0c93-416b-911e-9cf42f1a1085', - # id=pipeline_tool_id_link - # ) - # assert response.get_status_code() == 200 - # step_log = response.get_result() - # assert step_log is not None + @needscredentials + def test_get_tekton_pipeline_run_logs(self): + global run_log_id_link + response = self.cd_pipeline_service.get_tekton_pipeline_run_logs( + pipeline_id=pipeline_tool_id_link, + id=pipeline_run_id_link + ) + assert response.get_status_code() == 200 + logs_collection = response.get_result() + assert logs_collection is not None + + # Store the first log ID for use in test_get_tekton_pipeline_run_log_content + if logs_collection.get('logs') and len(logs_collection['logs']) > 0: + run_log_id_link = logs_collection['logs'][0]['id'] + assert run_log_id_link is not None + + @needscredentials + def test_get_tekton_pipeline_run_log_content(self): + # Retry logic with exponential backoff to wait for log content to be available + response = None + max_retries = 10 + base_delay = 1 + max_delay = 30 + + for i in range(max_retries): + try: + response = self.cd_pipeline_service.get_tekton_pipeline_run_log_content( + pipeline_id=pipeline_tool_id_link, + pipeline_run_id=pipeline_run_id_link, + id=run_log_id_link + ) + print(f"\nget_tekton_pipeline_run_log_content() was called successfully on attempt {i + 1}.") + break + except Exception as err: + if i == max_retries - 1: + raise + retry_delay = min(base_delay * (2 ** i), max_delay) + print(f"\nAttempt {i + 1} calling get_tekton_pipeline_run_log_content() failed, retrying in {retry_delay}s... Error: {err}") + time.sleep(retry_delay) + + assert response.get_status_code() == 200 + step_log = response.get_result() + assert step_log is not None @needscredentials def test_delete_tekton_pipeline_run(self): diff --git a/test/integration/test_cd_toolchain_v2.py b/test/integration/test_cd_toolchain_v2.py index f0ec0ba..4ac593f 100644 --- a/test/integration/test_cd_toolchain_v2.py +++ b/test/integration/test_cd_toolchain_v2.py @@ -27,6 +27,7 @@ from ibm_cloud_sdk_core import * import os from datetime import datetime +import time import pytest from environs import Env from ibm_continuous_delivery.cd_toolchain_v2 import * @@ -198,13 +199,29 @@ def test_create_toolchain_event_application_json(self): 'application_json': toolchain_event_prototype_data_application_json_model, } - response = self.cd_toolchain_service.create_toolchain_event( - toolchain_id=toolchain_id_link, - title='My-custom-event', - description='This is my custom event', - content_type='application/json', - data=toolchain_event_prototype_data_model, - ) + # Retry logic with exponential backoff to wait for event creation to succeed + response = None + max_retries = 10 + base_delay = 1 + max_delay = 30 + + for i in range(max_retries): + try: + response = self.cd_toolchain_service.create_toolchain_event( + toolchain_id=toolchain_id_link, + title='My-custom-event', + description='This is my custom event', + content_type='application/json', + data=toolchain_event_prototype_data_model, + ) + print(f"\ncreate_toolchain_event() was called successfully on attempt {i + 1}.") + break + except Exception as err: + if i == max_retries - 1: + raise + retry_delay = min(base_delay * (2 ** i), max_delay) + print(f"\nAttempt {i + 1} calling create_toolchain_event() failed, retrying in {retry_delay}s... Error: {err}") + time.sleep(retry_delay) assert response.get_status_code() == 200 toolchain_event_post = response.get_result() @@ -221,13 +238,29 @@ def test_create_toolchain_event_text_plain(self): 'text_plain': toolchain_event_prototype_data_text_plain_model, } - response = self.cd_toolchain_service.create_toolchain_event( - toolchain_id=toolchain_id_link, - title='My-custom-event', - description='This is my custom event', - content_type='text/plain', - data=toolchain_event_prototype_data_model, - ) + # Retry logic with exponential backoff to wait for event creation to succeed + response = None + max_retries = 10 + base_delay = 1 + max_delay = 30 + + for i in range(max_retries): + try: + response = self.cd_toolchain_service.create_toolchain_event( + toolchain_id=toolchain_id_link, + title='My-custom-event', + description='This is my custom event', + content_type='text/plain', + data=toolchain_event_prototype_data_model, + ) + print(f"\ncreate_toolchain_event() was called successfully on attempt {i + 1}.") + break + except Exception as err: + if i == max_retries - 1: + raise + retry_delay = min(base_delay * (2 ** i), max_delay) + print(f"\nAttempt {i + 1} calling create_toolchain_event() failed, retrying in {retry_delay}s... Error: {err}") + time.sleep(retry_delay) assert response.get_status_code() == 200 toolchain_event_post = response.get_result() From 6df114abb1efd5c642d03324bed45a4a08f392e8 Mon Sep 17 00:00:00 2001 From: Gigi Cheang Date: Thu, 11 Jun 2026 15:24:48 -0400 Subject: [PATCH 2/2] test: formatting Signed-off-by: Gigi Cheang --- test/integration/test_cd_tekton_pipeline_v2.py | 15 +++++++-------- test/integration/test_cd_toolchain_v2.py | 12 ++++++++---- 2 files changed, 15 insertions(+), 12 deletions(-) diff --git a/test/integration/test_cd_tekton_pipeline_v2.py b/test/integration/test_cd_tekton_pipeline_v2.py index f7d5120..1604eb8 100644 --- a/test/integration/test_cd_tekton_pipeline_v2.py +++ b/test/integration/test_cd_tekton_pipeline_v2.py @@ -586,13 +586,12 @@ def test_list_tekton_pipeline_runs_with_pager(self): def test_get_tekton_pipeline_run_logs(self): global run_log_id_link response = self.cd_pipeline_service.get_tekton_pipeline_run_logs( - pipeline_id=pipeline_tool_id_link, - id=pipeline_run_id_link + pipeline_id=pipeline_tool_id_link, id=pipeline_run_id_link ) assert response.get_status_code() == 200 logs_collection = response.get_result() assert logs_collection is not None - + # Store the first log ID for use in test_get_tekton_pipeline_run_log_content if logs_collection.get('logs') and len(logs_collection['logs']) > 0: run_log_id_link = logs_collection['logs'][0]['id'] @@ -609,17 +608,17 @@ def test_get_tekton_pipeline_run_log_content(self): for i in range(max_retries): try: response = self.cd_pipeline_service.get_tekton_pipeline_run_log_content( - pipeline_id=pipeline_tool_id_link, - pipeline_run_id=pipeline_run_id_link, - id=run_log_id_link + pipeline_id=pipeline_tool_id_link, pipeline_run_id=pipeline_run_id_link, id=run_log_id_link ) print(f"\nget_tekton_pipeline_run_log_content() was called successfully on attempt {i + 1}.") break except Exception as err: if i == max_retries - 1: raise - retry_delay = min(base_delay * (2 ** i), max_delay) - print(f"\nAttempt {i + 1} calling get_tekton_pipeline_run_log_content() failed, retrying in {retry_delay}s... Error: {err}") + retry_delay = min(base_delay * (2**i), max_delay) + print( + f"\nAttempt {i + 1} calling get_tekton_pipeline_run_log_content() failed, retrying in {retry_delay}s... Error: {err}" + ) time.sleep(retry_delay) assert response.get_status_code() == 200 diff --git a/test/integration/test_cd_toolchain_v2.py b/test/integration/test_cd_toolchain_v2.py index 4ac593f..bcf254b 100644 --- a/test/integration/test_cd_toolchain_v2.py +++ b/test/integration/test_cd_toolchain_v2.py @@ -219,8 +219,10 @@ def test_create_toolchain_event_application_json(self): except Exception as err: if i == max_retries - 1: raise - retry_delay = min(base_delay * (2 ** i), max_delay) - print(f"\nAttempt {i + 1} calling create_toolchain_event() failed, retrying in {retry_delay}s... Error: {err}") + retry_delay = min(base_delay * (2**i), max_delay) + print( + f"\nAttempt {i + 1} calling create_toolchain_event() failed, retrying in {retry_delay}s... Error: {err}" + ) time.sleep(retry_delay) assert response.get_status_code() == 200 @@ -258,8 +260,10 @@ def test_create_toolchain_event_text_plain(self): except Exception as err: if i == max_retries - 1: raise - retry_delay = min(base_delay * (2 ** i), max_delay) - print(f"\nAttempt {i + 1} calling create_toolchain_event() failed, retrying in {retry_delay}s... Error: {err}") + retry_delay = min(base_delay * (2**i), max_delay) + print( + f"\nAttempt {i + 1} calling create_toolchain_event() failed, retrying in {retry_delay}s... Error: {err}" + ) time.sleep(retry_delay) assert response.get_status_code() == 200