Rds Operator pass custom conn_id to superclass (apache#51196)

* Adjuste RDSHook to pass aws_conn_id to Hooks

* Added tests for changes and changed name of defaul conn_id in in RdsBaseOperator

* Fixed pre-commit errors

* Removed unnecessary args from RDSBaseOperator

Author: dominikhei

providers/amazon/src/airflow/providers/amazon/aws/operators/rds.py

@@ -54,12 +54,8 @@ class RdsBaseOperator(AwsBaseOperator[RdsHook]):
     def __init__(
         self,
         *args,
-        aws_conn_id: str | None = "aws_conn_id",
-        region_name: str | None = None,
         **kwargs,
     ):
-        self.aws_conn_id = aws_conn_id
-        self.region_name = region_name
         super().__init__(*args, **kwargs)
 
         self._await_interval = 60  # seconds

providers/amazon/tests/unit/amazon/aws/operators/test_rds.py

@@ -174,6 +174,18 @@ def test_hook_attribute(self):
         assert hasattr(self.op, "hook")
         assert self.op.hook.__class__.__name__ == "RdsHook"
 
+    def test_overwritten_conn_passed_to_hook(self):
+        OVERWRITTEN_CONN = "new-conn-id"
+        op = RdsBaseOperator(
+            task_id="test_overwritten_conn_passed_to_hook_task", aws_conn_id=OVERWRITTEN_CONN, dag=self.dag
+        )
+        assert op.hook.aws_conn_id == OVERWRITTEN_CONN
+
+    def test_no_conn_passed_to_hook(self):
+        DEFAULT_CONN = "aws_default"
+        op = RdsBaseOperator(task_id="test_no_conn_passed_to_hook_task", dag=self.dag)
+        assert op.hook.aws_conn_id == DEFAULT_CONN
+
 
 class TestRdsCreateDbSnapshotOperator:
     @classmethod