[Python] Fix GCS client creation in non-default universes

[shunping]

Part of #40569

Problem

The Python GCS client library (storage.Client) only reads the universe domain from client_options.universe_domain. It ignores GOOGLE_CLOUD_UNIVERSE_DOMAIN and the universe domain on the credentials.

Beam creates the client without client_options, so it always assumes googleapis.com. With Trusted Partner Cloud (TPC) credentials, client creation fails with:

ValueError: The configured universe domain (googleapis.com) does not match
the universe domain found in the credentials (...)

This breaks:

• gcsio.create_storage_client(), which every GCS read and write uses.
• The soft-delete check in GoogleCloudOptions. This fails during pipeline
  validation, so the job can't even be submitted.

Changes

• gcsio.create_storage_client: passes client_options={"universe_domain": ...}, taking the value from GOOGLE_CLOUD_UNIVERSE_DOMAIN or, if that's unset, the credentials. Nothing changes for the default universe.
• GoogleCloudOptions._warn_if_soft_delete_policy_enabled:
  • Builds GcsIO from the user's pipeline options instead of empty defaults, so --project, --impersonate_service_account and the universe domain are respected.
  • Catches any exception and logs a warning. This check is only meant to warn, so it should never block submission.

[codecov]

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 80.02%. Comparing base (76f6e6d) to head (df10011).
⚠️ Report is 242 commits behind head on master.

Additional details and impacted files

@@              Coverage Diff              @@
##             master   #40674       +/-   ##
=============================================
+ Coverage     56.16%   80.02%   +23.85%     
=============================================
  Files          1124      490      -634     
  Lines        178336    82818    -95518     
  Branches       1489        0     -1489     
=============================================
- Hits         100170    66275    -33895     
+ Misses        75645    16543    -59102     
+ Partials       2521        0     -2521     

Flag	Coverage Δ
go	?
java	?
python	80.02% <100.00%> (+0.46%)	⬆️

Flags with carried forward coverage won't be shown. Click here to find out more.

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

[github-actions]

📋 Review Summary

This PR adds support for non-default Google Cloud universe domains (such as Trusted Partner Cloud) when creating Python GCS storage clients and ensures GoogleCloudOptions._warn_if_soft_delete_policy_enabled honors user pipeline options without blocking job submission on error. The implementation is clean and well-tested, with one minor test environment leakage issue in gcsio_test.py.

🔍 General Feedback

• The fallback resolution of universe_domain from GOOGLE_CLOUD_UNIVERSE_DOMAIN and credentials in gcsio.create_storage_client is straightforward and preserves existing behavior for the default googleapis.com universe.
• Passing pipeline_options=self into GcsIO and catching broad exceptions in _warn_if_soft_delete_policy_enabled properly ensures pipeline validation does not fail when checking soft-delete policies.

---

💡 Tip: Comment /gemini review on this pull request to trigger a new Gemini code review.

[github-actions]

🟡 Medium: test_universe_domain_from_env_var mutates os.environ["GOOGLE_CLOUD_UNIVERSE_DOMAIN"] directly, and the class-level @mock.patch.dict(os.environ, ...) only snapshots/restores os.environ at the method level if applied to the test method or if clear=False is used per-method (when decorating a unittest.TestCase class, mock.patch.dict wraps each test_* method individually, which does restore os.environ after each test, unless other code or sub-hooks run before/after). However, mutating os.environ directly in test_default_universe_domain, test_universe_domain_from_env_var, and test_universe_domain_from_credentials is brittle and inconsistent with pipeline_options_test.py. Consider using @mock.patch.dict(os.environ, {"GOOGLE_CLOUD_UNIVERSE_DOMAIN": _TPC_UNIVERSE_DOMAIN}) (or with mock.patch.dict(...)) for clarity and guaranteed isolation.

[github-actions]

Assigning reviewers:

R: @damccorm for label python.
Backup reviewer: tvalentyn (if the main reviewer is unresponsive, you can escalate to them by tagging their username)

This pull request likely touches a core component ("core" label). Please review with scrutiny.

Note: If you would like to opt out of this review, comment assign to next reviewer.

Available commands:

• stop reviewer notifications - opt out of the automated review tooling
• remind me after tests pass - tag the comment author after tests pass
• waiting on author - shift the attention set back to the author (any comment or push by the author will return the attention set to the reviewers)

The PR bot will only process comments in the main thread (not review comments).