refactor: improve code formatting and readability in workflow and test scripts

Author: emmanuelmathot

scripts/test_complete_workflow.py

@@ -19,7 +19,9 @@ def s3_to_https(s3_url: str, gateway_url: str = "https://s3.explorer.eopf.copern
     return f"{gateway_base}/{bucket}{path}"
 
 
-def https_to_s3(https_url: str, gateway_url: str = "https://s3.explorer.eopf.copernicus.eu") -> str | None:
+def https_to_s3(
+    https_url: str, gateway_url: str = "https://s3.explorer.eopf.copernicus.eu"
+) -> str | None:
     """Convert https:// URL back to s3:// URL."""
     if not https_url.startswith("https://"):
         return None
@@ -121,18 +123,22 @@ def process_item_with_gateway(item_dict: dict, s3_endpoint: str) -> dict:
                 if s3_url:
                     new_href = s3_to_https(s3_url)
                     link["href"] = new_href
-                    print(f"  🔄 Converted store link:")
+                    print("  🔄 Converted store link:")
                     print(f"     From: {old_href}")
                     print(f"     To:   {new_href}")
 
     print(f"\n✅ Processed {processed_count} assets with gateway format and alternate S3 URLs")
     return item_dict
 
 
-def main():
+def main() -> int:
     """Test the complete workflow."""
     # Read the sample JSON
-    sample_file = Path(__file__).parent.parent / "stac" / "S2A_MSIL2A_20250831T103701_N0511_R008_T31TFL_20250831T145420.json"
+    sample_file = (
+        Path(__file__).parent.parent
+        / "stac"
+        / "S2A_MSIL2A_20250831T103701_N0511_R008_T31TFL_20250831T145420.json"
+    )
 
     if not sample_file.exists():
         print(f"❌ Sample file not found: {sample_file}")
@@ -156,7 +162,7 @@ def main():
             break
 
     # Process with new gateway format
-    print(f"\n🔧 Processing with new gateway format...\n")
+    print("\n🔧 Processing with new gateway format...\n")
     s3_endpoint = "https://s3.de.io.cloud.ovh.net"
     modified_item = process_item_with_gateway(item_dict, s3_endpoint)
 

scripts/test_gateway_format.py

@@ -17,7 +17,9 @@ def s3_to_https(s3_url: str, gateway_url: str = "https://s3.explorer.eopf.copern
     return f"{gateway_base}/{bucket}{path}"
 
 
-def https_to_s3(https_url: str, gateway_url: str = "https://s3.explorer.eopf.copernicus.eu") -> str | None:
+def https_to_s3(
+    https_url: str, gateway_url: str = "https://s3.explorer.eopf.copernicus.eu"
+) -> str | None:
     """Convert https:// URL back to s3:// URL."""
     if not https_url.startswith("https://"):
         return None
@@ -47,7 +49,7 @@ def https_to_s3(https_url: str, gateway_url: str = "https://s3.explorer.eopf.cop
     return None
 
 
-def main():
+def main() -> int:
     """Test the gateway format conversions."""
     print("Testing S3 Gateway Format Conversions")
     print("=" * 70)
@@ -88,28 +90,30 @@ def main():
     print("-" * 70)
     for test in test_cases:
         https_url = test["expected_https"]
-        result = https_to_s3(https_url)
-        passed = result == test["s3"]
+        s3_result = https_to_s3(https_url)
+        passed = s3_result == test["s3"]
         all_passed = all_passed and passed
         status = "✅ PASS" if passed else "❌ FAIL"
         print(f"\n{status} - {test['name']}")
         print(f"  Input:    {https_url}")
         print(f"  Expected: {test['s3']}")
-        print(f"  Got:      {result}")
+        print(f"  Got:      {s3_result}")
 
     # Test backwards compatibility with old format
     print("\n\n3. Testing backwards compatibility with old S3 format")
     print("-" * 70)
-    old_format_url = "https://esa-zarr-sentinel-explorer-fra.s3.de.io.cloud.ovh.net/tests-output/file.zarr"
+    old_format_url = (
+        "https://esa-zarr-sentinel-explorer-fra.s3.de.io.cloud.ovh.net/tests-output/file.zarr"
+    )
     expected_s3 = "s3://esa-zarr-sentinel-explorer-fra/tests-output/file.zarr"
-    result = https_to_s3(old_format_url)
-    passed = result == expected_s3
+    old_result = https_to_s3(old_format_url)
+    passed = old_result == expected_s3
     all_passed = all_passed and passed
     status = "✅ PASS" if passed else "❌ FAIL"
     print(f"\n{status} - Old S3 subdomain format")
     print(f"  Input:    {old_format_url}")
     print(f"  Expected: {expected_s3}")
-    print(f"  Got:      {result}")
+    print(f"  Got:      {old_result}")
 
     # Summary
     print("\n" + "=" * 70)

stac/S2A_MSIL2A_20250831T103701_N0511_R008_T31TFL_20250831T145420_new_gateway.json

@@ -600,4 +600,4 @@
     "https://stac-extensions.github.io/alternate-assets/v1.2.0/schema.json",
     "https://stac-extensions.github.io/storage/v2.0.0/schema.json"
   ]
-}
+}

tests/test_s3_gateway.py

@@ -1,7 +1,7 @@
 """Unit tests for S3 gateway format and alternate asset extension."""
 
-import pytest
 from pystac import Asset, Item
+
 from scripts.register_v1 import (
     add_alternate_s3_assets,
     https_to_s3,
@@ -75,7 +75,9 @@ def test_new_gateway_format_simple(self):
     def test_new_gateway_format_deep_path(self):
         """Test conversion from new gateway format - deep nested path."""
         https_url = "https://s3.explorer.eopf.copernicus.eu/esa-zarr-sentinel-explorer-fra/tests-output/sentinel-2-l2a-staging/file.zarr"
-        expected = "s3://esa-zarr-sentinel-explorer-fra/tests-output/sentinel-2-l2a-staging/file.zarr"
+        expected = (
+            "s3://esa-zarr-sentinel-explorer-fra/tests-output/sentinel-2-l2a-staging/file.zarr"
+        )
         assert https_to_s3(https_url) == expected
 
     def test_new_gateway_format_bucket_only(self):
@@ -92,7 +94,9 @@ def test_new_gateway_format_bucket_with_slash(self):
 
     def test_old_s3_format_de_region(self):
         """Test backwards compatibility with old S3 format (de region)."""
-        https_url = "https://esa-zarr-sentinel-explorer-fra.s3.de.io.cloud.ovh.net/tests-output/file.zarr"
+        https_url = (
+            "https://esa-zarr-sentinel-explorer-fra.s3.de.io.cloud.ovh.net/tests-output/file.zarr"
+        )
         expected = "s3://esa-zarr-sentinel-explorer-fra/tests-output/file.zarr"
         assert https_to_s3(https_url) == expected
 
@@ -289,8 +293,13 @@ def test_add_alternate_to_gateway_url(self):
         add_alternate_s3_assets(item, s3_endpoint)
 
         # Check extensions were added
-        assert "https://stac-extensions.github.io/alternate-assets/v1.2.0/schema.json" in item.stac_extensions
-        assert "https://stac-extensions.github.io/storage/v2.0.0/schema.json" in item.stac_extensions
+        assert (
+            "https://stac-extensions.github.io/alternate-assets/v1.2.0/schema.json"
+            in item.stac_extensions
+        )
+        assert (
+            "https://stac-extensions.github.io/storage/v2.0.0/schema.json" in item.stac_extensions
+        )
 
         # Check alternate was added
         asset = item.assets["data"]
@@ -510,11 +519,11 @@ def test_extensions_not_duplicated(self):
         add_alternate_s3_assets(item, "https://s3.de.io.cloud.ovh.net")
 
         # Count occurrences of alternate-assets extension
-        alternate_count = sum(
-            1 for ext in item.stac_extensions if "alternate-assets" in ext
-        )
+        alternate_count = sum(1 for ext in item.stac_extensions if "alternate-assets" in ext)
         assert alternate_count == 1
-        assert "https://stac-extensions.github.io/storage/v2.0.0/schema.json" in item.stac_extensions
+        assert (
+            "https://stac-extensions.github.io/storage/v2.0.0/schema.json" in item.stac_extensions
+        )
 
 
 class TestEdgeCases: