DynamoDBHook - not able to registering a custom waiter

[utkarsharma2]

Apache Airflow Provider(s)

amazon

Versions of Apache Airflow Providers

apache-airflow-providers-amazon=7.4.1

Apache Airflow version

airflow=2.5.3

Operating System

Mac

Deployment

Docker-Compose

Deployment details

No response

What happened

We can register a custom waiter by adding a JSON file to the path - airflow/airflow/providers/amazon/aws/waiters/. The should be named <client_type>.json in this case - dynamodb.json. Once registered we can use the custom waiter.

content of the file - airflow/airflow/providers/amazon/aws/waiters/dynamodb.json:

{
    "version": 2,
    "waiters": {
        "export_table": {
            "operation": "ExportTableToPointInTime",
            "delay": 30,
            "maxAttempts": 60,
            "acceptors": [
                {
                    "matcher": "path",
                    "expected": "COMPLETED",
                    "argument": "ExportDescription.ExportStatus",
                    "state": "success"
                },
                {
                    "matcher": "path",
                    "expected": "FAILED",
                    "argument": "ExportDescription.ExportStatus",
                    "state": "failure"
                },
                {
                    "matcher": "path",
                    "expected": "IN_PROGRESS",
                    "argument": "ExportDescription.ExportStatus",
                    "state": "retry"
                }
            ]
        }
    }
}

Getting below error post running test case:

class TestCustomDynamoDBServiceWaiters:
    """Test waiters from ``amazon/aws/waiters/dynamodb.json``."""

    STATUS_COMPLETED = "COMPLETED"
    STATUS_FAILED = "FAILED"
    STATUS_IN_PROGRESS = "IN_PROGRESS"

    @pytest.fixture(autouse=True)
    def setup_test_cases(self, monkeypatch):
        self.client = boto3.client("dynamodb", region_name="eu-west-3")
        monkeypatch.setattr(DynamoDBHook, "conn", self.client)

    @pytest.fixture
    def mock_export_table_to_point_in_time(self):
        """Mock ``DynamoDBHook.Client.export_table_to_point_in_time`` method."""
        with mock.patch.object(self.client, "export_table_to_point_in_time") as m:
            yield m

    def test_service_waiters(self):
        assert os.path.exists('/Users/utkarsharma/sandbox/airflow-sandbox/airflow/airflow/providers/amazon/aws/waiters/dynamodb.json')
        hook_waiters = DynamoDBHook(aws_conn_id=None).list_waiters()
        assert "export_table" in hook_waiters

Error

tests/providers/amazon/aws/waiters/test_custom_waiters.py:273 (TestCustomDynamoDBServiceWaiters.test_service_waiters)
'export_table' != ['table_exists', 'table_not_exists']

Expected :['table_exists', 'table_not_exists']
Actual :'export_table'

self = <tests.providers.amazon.aws.waiters.test_custom_waiters.TestCustomDynamoDBServiceWaiters object at 0x117f085e0>

def test_service_waiters(self):
    assert os.path.exists('/Users/utkarsharma/sandbox/airflow-sandbox/airflow/airflow/providers/amazon/aws/waiters/dynamodb.json')
    hook_waiters = DynamoDBHook(aws_conn_id=None).list_waiters()

  assert "export_table" in hook_waiters

E AssertionError: assert 'export_table' in ['table_exists', 'table_not_exists']

test_custom_waiters.py:277: AssertionError

What you think should happen instead

It should register the custom waiter and test case should pass.the

How to reproduce

Add the file mentioned above to Airflow's code base and try running the test case provided.

Anything else

No response

Are you willing to submit PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct