Use the operator's AWS settings for deferred Neptune, MWAA, SSM tasks

[SEPURI-SAI-KRISHNA]

AwsBaseOperator and AwsBaseSensor both resolve region_name, verify and
botocore_config in __init__, so every subclass carries them. When these
operators defer, they build a trigger that constructs its own hook, and the
three fields were never passed along. The deferred half of the task then talks
to AWS with different settings than the synchronous half: the default region
instead of the configured one, SSL verification silently back on, and custom
botocore timeouts/retries dropped.

The triggers already support all three. SsmRunCommandTrigger names them
explicitly, and the Neptune Analytics and MWAA triggers take **kwargs straight
through to AwsBaseWaiterTrigger, whose hook() builds the client from them.
The Neptune trigger docstrings even document :param region_name:. Only the
operator side was missing, so this is a pure call-site fix, no trigger
signatures change.

Twelve sites: eight in Neptune Analytics, one in the MWAA operator, two in the
MWAA sensors, one in the SSM sensor. The SSM operator already passed all three,
which is what the deferred sensor beside it should have been doing.

Same bug and same fix shape as #52904, #67508, #67876, #71646 and #71857.

---

Was generative AI tooling used to co-author this PR?

• Yes, Claude Code (Opus 5)

Generated-by: Claude Code (Opus 5) following the guidelines

[vincbeck]

@ramitkataria @seanghaeli

[ramitkataria]

Thanks for continuing this cleanup series. The fix itself looks correct. However, two bigger-picture thoughts, neither blocking for this PR. Since this is the 6th PR of this shape, it might be worth stepping back:

1. By my count, 43 defer sites in the provider still pass none of region_name/verify/botocore_config (bedrock, emr, dms, comprehend, glue, opensearch_serverless, several sensors) and 18 more pass only region_name (eks, rds, ecs, glue, batch). If the plan is to keep going service by service, a tracking issue with that list would help coordinate and avoid duplicate PRs.
2. Many of the remaining sites can't actually be fixed at the call site: their trigger subclasses have closed __init__ signatures without **kwargs, and hook() implementations that pass only aws_conn_id. A default hook() on AwsBaseWaiterTrigger driven by a hook_class attribute (similar to AwsBaseHookMixin._hook_parameters), plus one parametrized invariant test over the deferrable operators, would fix this class of bug in one place and catch new operators that forget the kwargs

Related small thing: the REGION_NAME/VERIFY/BOTOCORE_CONFIG test constant trio is now copy-pasted into 12 test files. A shared helper under tests/unit/amazon/aws/utils/ (following the validate_template_fields precedent) would stop that growth for the remaining PRs in the series.

[ramitkataria]

Let's add verify here as well?

[ramitkataria]

Let's assert on the serialized attributes (trigger.serialize()[1]) like test_neptune.py does, instead of the pre-serialization attributes

[SEPURI-SAI-KRISHNA]

Thanks for the review, both addressed.

verify in the docstrings: added to all seven classes in
neptune_analytics.py, using the wording already standard elsewhere in
the provider. Two of them (NeptuneDeletePrivateGraphEndpointOperator
and NeptuneDeleteGraphOperator) had a stray blank line sitting exactly
where the entry belonged, so the parameter looks like it was dropped
during authoring rather than deliberately omitted.

Serialized assertions: switched all twelve tests to
trigger.serialize()[1] with a full payload comparison, matching
test_neptune.py. You're right that this is the stronger check —
serialize() runs the values through prune_dict, so an attribute
assertion can pass while the setting never reaches the triggerer. It also
caught something the attribute form hid: the MWAA sensors declare
success_states / failure_states as lists but serialize them as sets.

I re-ran the counterfactual after the rewrite: stripping the 36 fix lines
still fails exactly the 12 new tests and nothing else, so the sensitivity
is unchanged.

On the broader points, all three sound right to me:

1. I have the full audit already, 84 sites originally, 19 now fixed
   across Use the configured region for deferred Neptune cluster tasks #71646, Use the operator's AWS settings for deferred SageMaker tasks #71857 and this PR. My grouping matches your count
   almost exactly (18 passing only region_name; I count 44 rather than
   43 passing none). Happy to open a tracking issue with the per-service
   breakdown.
2. Agreed, and this is the more interesting fix. A default hook() on
   AwsBaseWaiterTrigger driven by a hook_class attribute would cover
   the group that can't be fixed at the call site, and a parametrized
   invariant test over the deferrable operators would stop new ones
   regressing. I'd rather do that as its own PR than fold it in here.
3. Makes sense, I'll move the constants to a shared helper as part of
   that work rather than copy them a thirteenth time.

Let me know if you'd prefer the tracking issue before or after the AwsBaseWaiterTrigger change.

[SEPURI-SAI-KRISHNA]

Opened #72144 as the tracking issue.

One correction to my earlier reply: I said I counted 44 sites passing none of the three. Your 43 is right. I re-derived the audit by parsing every self.defer(trigger=...) call with ast rather than by grep, and it comes out at exactly your numbers, 110 defer sites, 49 complete, 18 passing only region_name, 43 passing none.

The sweep also sharpened your second point more than I expected. Of the 61 remaining sites, only 7 can be fixed at the call site, the other 54 have trigger subclasses whose __init__ never accepts the parameters, so a per-service PR would have to change the trigger signature at every one of them. That makes the hook_class-driven default hook() on AwsBaseWaiterTrigger clearly the right move rather than a nice-to-have, so I'd rather do that next than continue with rds/ecs as I'd planned.

Full per-site breakdown by fix shape is in the issue.