diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml
index a1223ce..5d4f38c 100644
--- a/.github/workflows/test.yml
+++ b/.github/workflows/test.yml
@@ -34,3 +34,9 @@ jobs:
 
     - name: Run pre-commit checks
       run: uv run pre-commit run --all-files
+
+    - name: Run unit tests
+      run: uv run pytest tests/unit -v --tb=short
+
+    - name: Generate coverage report
+      run: uv run pytest tests/unit --cov=scripts --cov-report=term-missing --cov-report=html
diff --git a/examples/s1_quickstart.py b/examples/s1_quickstart.py
new file mode 100644
index 0000000..b3dd09b
--- /dev/null
+++ b/examples/s1_quickstart.py
@@ -0,0 +1,81 @@
+#!/usr/bin/env python3
+"""Quick S1 GRD to GeoZarr conversion example.
+
+Demonstrates end-to-end S1 pipeline:
+1. Fetch S1 item from STAC
+2. Convert to GeoZarr
+3. Register in STAC catalog
+4. Augment with preview links
+"""
+
+import subprocess
+import sys
+from pathlib import Path
+
+
+def run_s1_pipeline(
+    stac_url: str = "https://stac.core.eopf.eodc.eu",
+    item_id: str = "S1C_IW_GRDH_1SDV_20251008T163126_20251008T163151_004473_008DBA_9AB4",
+    output_dir: Path = Path("./s1_output"),
+) -> int:
+    """Run S1 GRD pipeline locally."""
+
+    output_dir.mkdir(exist_ok=True)
+    geozarr_path = output_dir / f"{item_id}_geozarr.zarr"
+
+    print(f"🛰️  Processing S1 item: {item_id}")
+
+    # Step 1: Get source URL
+    print("\n1️⃣  Fetching STAC item...")
+    cmd = [
+        "python",
+        "scripts/get_zarr_url.py",
+        f"{stac_url}/collections/sentinel-1-l1-grd/items/{item_id}",
+    ]
+    result = subprocess.run(cmd, capture_output=True, text=True, check=True)
+    source_url = result.stdout.strip()
+    print(f"   Source: {source_url}")
+
+    # Step 2: Convert to GeoZarr
+    print("\n2️⃣  Converting to GeoZarr...")
+    cmd = [
+        "eopf-geozarr",
+        "convert",
+        source_url,
+        str(geozarr_path),
+        "--groups",
+        "/measurements",
+        "--gcp-group",
+        "/conditions/gcp",
+        "--spatial-chunk",
+        "2048",
+        "--verbose",
+    ]
+    subprocess.run(cmd, check=True)
+    print(f"   ✓ Created: {geozarr_path}")
+
+    # Step 3: Validate
+    print("\n3️⃣  Validating GeoZarr...")
+    cmd = ["eopf-geozarr", "validate", str(geozarr_path)]
+    subprocess.run(cmd, check=True)
+    print("   ✓ Valid GeoZarr")
+
+    print("\n✅ S1 pipeline complete!")
+    print(f"   Output: {geozarr_path}")
+    print("\n   Next steps:")
+    print("   - Upload to S3")
+    print("   - Register in STAC catalog")
+    print("   - View in titiler-eopf")
+
+    return 0
+
+
+if __name__ == "__main__":
+    try:
+        sys.exit(run_s1_pipeline())
+    except subprocess.CalledProcessError as e:
+        print(f"\n❌ Pipeline failed: {e}", file=sys.stderr)
+        sys.exit(1)
+    except KeyboardInterrupt:
+        print("\n⚠️  Interrupted", file=sys.stderr)
+        sys.exit(130)
diff --git a/scripts/test_s1_e2e.sh b/scripts/test_s1_e2e.sh
new file mode 100755
index 0000000..3a53158
--- /dev/null
+++ b/scripts/test_s1_e2e.sh
@@ -0,0 +1,172 @@
+#!/bin/bash
+# Test S1 GRD end-to-end pipeline in devseed-staging namespace
+#
+# This script:
+# 1. Applies the workflow template
+# 2. Publishes an S1 test payload via AMQP
+# 3. Waits for workflow completion
+# 4. Shows logs and verifies STAC item was created
+
+set -euo pipefail
+
+# Set kubeconfig
+SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+PROJECT_ROOT="$(dirname "$SCRIPT_DIR")"
+export KUBECONFIG="${KUBECONFIG:-$PROJECT_ROOT/.work/kubeconfig}"
+
+if [ ! -f "$KUBECONFIG" ]; then
+  echo "❌ Kubeconfig not found at: $KUBECONFIG"
+  echo "Please set KUBECONFIG environment variable or create .work/kubeconfig"
+  exit 1
+fi
+
+NAMESPACE="${NAMESPACE:-devseed-staging}"
+PAYLOAD_FILE="${PAYLOAD_FILE:-workflows/examples/payload-s1.json}"
+TIMEOUT="${TIMEOUT:-600}"  # 10 minutes
+
+echo "=========================================="
+echo "S1 GRD Pipeline E2E Test"
+echo "=========================================="
+echo "Kubeconfig: $KUBECONFIG"
+echo "Namespace: $NAMESPACE"
+echo "Payload: $PAYLOAD_FILE"
+echo "Timeout: ${TIMEOUT}s"
+echo ""
+
+# Step 1: Apply workflow template
+echo "📝 Applying workflow template..."
+kubectl -n "$NAMESPACE" apply -f workflows/template.yaml
+echo "✅ Template applied"
+echo ""
+
+# Step 2: Publish AMQP message
+echo "📤 Publishing test payload..."
+kubectl -n "$NAMESPACE" delete job amqp-publish-once --ignore-not-found=true
+kubectl -n "$NAMESPACE" delete configmap amqp-payload --ignore-not-found=true
+kubectl -n "$NAMESPACE" create configmap amqp-payload --from-file=body.json="$PAYLOAD_FILE"
+kubectl -n "$NAMESPACE" apply -f workflows/amqp-publish-once.yaml
+echo "⏳ Waiting for publish job..."
+kubectl -n "$NAMESPACE" wait --for=condition=complete --timeout=120s job/amqp-publish-once
+echo "✅ Payload published"
+echo ""
+
+# Step 3: Get latest workflow
+echo "🔍 Finding triggered workflow..."
+sleep 3  # Give sensor time to create workflow
+WORKFLOW=$(kubectl -n "$NAMESPACE" get wf --sort-by=.metadata.creationTimestamp -o jsonpath='{.items[-1:].metadata.name}' 2>/dev/null || true)
+if [ -z "$WORKFLOW" ]; then
+  echo "❌ No workflow found!"
+  exit 1
+fi
+echo "✅ Workflow: $WORKFLOW"
+echo ""
+
+# Step 4: Wait for completion
+echo "⏳ Waiting for workflow completion (timeout: ${TIMEOUT}s)..."
+START_TIME=$(date +%s)
+while true; do
+  PHASE=$(kubectl -n "$NAMESPACE" get wf "$WORKFLOW" -o jsonpath='{.status.phase}' 2>/dev/null || echo "Unknown")
+  ELAPSED=$(($(date +%s) - START_TIME))
+
+  echo "  [${ELAPSED}s] Phase: $PHASE"
+
+  case "$PHASE" in
+    Succeeded)
+      echo "✅ Workflow succeeded!"
+      break
+      ;;
+    Failed|Error)
+      echo "❌ Workflow failed!"
+      break
+      ;;
+    Unknown)
+      echo "❌ Workflow disappeared!"
+      exit 1
+      ;;
+  esac
+
+  if [ $ELAPSED -ge $TIMEOUT ]; then
+    echo "⏰ Timeout reached!"
+    break
+  fi
+
+  sleep 5
+done
+echo ""
+
+# Step 5: Show workflow details
+echo "=========================================="
+echo "Workflow Details"
+echo "=========================================="
+kubectl -n "$NAMESPACE" get wf "$WORKFLOW" -o jsonpath='
+Name: {.metadata.name}
+Status: {.status.phase}
+Started: {.status.startedAt}
+Finished: {.status.finishedAt}
+Duration: {.status.estimatedDuration}
+
+Parameters:
+  source_url: {.spec.arguments.parameters[?(@.name=="source_url")].value}
+  item_id: {.spec.arguments.parameters[?(@.name=="item_id")].value}
+  collection: {.spec.arguments.parameters[?(@.name=="register_collection")].value}
+'
+echo ""
+echo ""
+
+# Step 6: Show pod logs
+echo "=========================================="
+echo "Pod Logs"
+echo "=========================================="
+PODS=$(kubectl -n "$NAMESPACE" get pods -l workflows.argoproj.io/workflow="$WORKFLOW" -o name 2>/dev/null || true)
+if [ -z "$PODS" ]; then
+  echo "⚠️  No pods found"
+else
+  for POD in $PODS; do
+    POD_NAME=$(basename "$POD")
+    TEMPLATE=$(kubectl -n "$NAMESPACE" get pod "$POD_NAME" -o jsonpath='{.metadata.labels.workflows\.argoproj\.io/template}' 2>/dev/null || echo "unknown")
+    echo ""
+    echo "--- $POD_NAME ($TEMPLATE) ---"
+    kubectl -n "$NAMESPACE" logs "$POD_NAME" --tail=100 -c main 2>/dev/null || echo "No logs available"
+  done
+fi
+echo ""
+
+# Step 7: Verify STAC item
+echo "=========================================="
+echo "STAC Item Verification"
+echo "=========================================="
+ITEM_ID=$(kubectl -n "$NAMESPACE" get wf "$WORKFLOW" -o jsonpath='{.spec.arguments.parameters[?(@.name=="item_id")].value}')
+COLLECTION=$(kubectl -n "$NAMESPACE" get wf "$WORKFLOW" -o jsonpath='{.spec.arguments.parameters[?(@.name=="register_collection")].value}')
+STAC_URL="https://api.explorer.eopf.copernicus.eu/stac/collections/$COLLECTION/items/$ITEM_ID"
+
+echo "Checking: $STAC_URL"
+ITEM_STATUS=$(curl -s -o /dev/null -w "%{http_code}" "$STAC_URL")
+if [ "$ITEM_STATUS" = "200" ]; then
+  echo "✅ STAC item exists!"
+  echo ""
+  curl -s "$STAC_URL" | jq '{
+    id: .id,
+    collection: .collection,
+    geometry: .geometry.type,
+    assets: [.assets | keys[]],
+    links: [.links[] | select(.rel=="xyz" or .rel=="viewer" or .rel=="tilejson") | {rel, href}]
+  }'
+else
+  echo "❌ STAC item not found (HTTP $ITEM_STATUS)"
+fi
+echo ""
+
+echo "=========================================="
+echo "Test Summary"
+echo "=========================================="
+echo "Workflow: $WORKFLOW"
+echo "Status: $PHASE"
+echo "STAC Item: $ITEM_STATUS"
+echo ""
+if [ "$PHASE" = "Succeeded" ] && [ "$ITEM_STATUS" = "200" ]; then
+  echo "🎉 END-TO-END TEST PASSED!"
+  exit 0
+else
+  echo "❌ END-TO-END TEST FAILED"
+  exit 1
+fi
diff --git a/tests/__init__.py b/tests/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/tests/conftest.py b/tests/conftest.py
new file mode 100644
index 0000000..c967b78
--- /dev/null
+++ b/tests/conftest.py
@@ -0,0 +1,133 @@
+"""Pytest configuration and shared fixtures for data-pipeline tests."""
+
+import atexit
+import sys
+import warnings
+
+import pytest
+
+# Suppress noisy async context warnings from zarr/s3fs
+warnings.filterwarnings("ignore", category=ResourceWarning)
+warnings.filterwarnings("ignore", message="coroutine.*was never awaited")
+
+
+# Global stderr filter that stays active even after pytest teardown
+_original_stderr = sys.stderr
+_suppress_traceback = False
+
+
+class _FilteredStderr:
+    def write(self, text):
+        global _suppress_traceback
+
+        # Start suppressing when we see async context errors
+        if any(
+            marker in text
+            for marker in [
+                "Exception ignored",
+                "Traceback (most recent call last)",
+                "ValueError: <Token",
+                "was created in a different Context",
+                "zarr/storage/",
+                "s3fs/core.py",
+                "aiobotocore/context.py",
+            ]
+        ):
+            _suppress_traceback = True
+
+        # Reset suppression on empty lines (between tracebacks)
+        if not text.strip():
+            _suppress_traceback = False
+
+        # Only write if not currently suppressing
+        if not _suppress_traceback:
+            _original_stderr.write(text)
+
+    def flush(self):
+        _original_stderr.flush()
+
+
+def _restore_stderr():
+    """Restore original stderr at exit."""
+    sys.stderr = _original_stderr
+
+
+# Install filter at module load time
+sys.stderr = _FilteredStderr()
+atexit.register(_restore_stderr)
+
+
+@pytest.fixture(autouse=True, scope="function")
+def clear_prometheus_registry():
+    """Clear Prometheus registry before each test to avoid duplicates."""
+    import contextlib
+
+    try:
+        from prometheus_client import REGISTRY
+
+        collectors = list(REGISTRY._collector_to_names.keys())
+        for collector in collectors:
+            with contextlib.suppress(Exception):
+                REGISTRY.unregister(collector)
+    except ImportError:
+        pass
+    yield
+
+
+@pytest.fixture
+def sample_stac_item():
+    """Return a minimal STAC item for testing."""
+    return {
+        "type": "Feature",
+        "stac_version": "1.0.0",
+        "id": "test-item",
+        "properties": {
+            "datetime": "2025-01-01T00:00:00Z",
+            "proj:epsg": 32636,
+        },
+        "geometry": {
+            "type": "Polygon",
+            "coordinates": [
+                [
+                    [600000, 6290220],
+                    [709800, 6290220],
+                    [709800, 6400020],
+                    [600000, 6400020],
+                    [600000, 6290220],
+                ]
+            ],
+        },
+        "links": [],
+        "assets": {
+            "B01": {
+                "href": "s3://bucket/data/B01.tif",
+                "type": "image/tiff; application=geotiff",
+                "roles": ["data"],
+                "proj:epsg": 32636,
+                "proj:shape": [10980, 10980],
+                "proj:transform": [10, 0, 600000, 0, -10, 6400020],
+            }
+        },
+        "collection": "test-collection",
+    }
+
+
+@pytest.fixture
+def stac_item_with_proj_code(sample_stac_item):
+    """Return a STAC item with proj:code (should be removed)."""
+    item = sample_stac_item.copy()
+    item["properties"]["proj:code"] = "EPSG:32636"
+    item["assets"]["B01"]["proj:code"] = "EPSG:32636"
+    return item
+
+
+@pytest.fixture
+def mock_zarr_url():
+    """Return a sample GeoZarr URL."""
+    return "s3://bucket/path/to/dataset.zarr"
+
+
+@pytest.fixture
+def mock_stac_api_url():
+    """Return a mock STAC API URL."""
+    return "https://api.example.com/stac"
diff --git a/tests/integration/__init__.py b/tests/integration/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/tests/integration/test_pipeline_e2e.py b/tests/integration/test_pipeline_e2e.py
new file mode 100644
index 0000000..1013025
--- /dev/null
+++ b/tests/integration/test_pipeline_e2e.py
@@ -0,0 +1,148 @@
+"""Integration tests for end-to-end pipeline flow.
+
+Tests the full workflow:
+1. Extract metadata from source Zarr
+2. Register GeoZarr to STAC API
+3. Augment item with preview links
+4. Validate final STAC item
+"""
+
+from unittest.mock import Mock, patch
+
+import pytest
+
+
+@pytest.mark.integration
+def test_s3_url_conversion():
+    """Test S3 URL to HTTPS conversion."""
+    from scripts.register import s3_to_https
+
+    # Test S3 URL conversion
+    s3_url = "s3://eopf-bucket/geozarr/S2A_test.zarr"
+    https_url = s3_to_https(s3_url, "https://s3.gra.cloud.ovh.net")
+
+    assert https_url.startswith("https://")
+    assert "eopf-bucket" in https_url
+    assert "s3.gra.cloud.ovh.net" in https_url
+
+    # Test already HTTPS URL (should pass through)
+    https_input = "https://example.com/data.zarr"
+    result = s3_to_https(https_input, "https://s3.example.com")
+    assert result == https_input
+
+
+@pytest.fixture
+def sample_sentinel1_item():
+    """Sentinel-1 GRD test item."""
+    return {
+        "type": "Feature",
+        "stac_version": "1.0.0",
+        "id": "S1A_IW_GRDH_1SDV_20250518T120000",
+        "collection": "sentinel-1-l1-grd",
+        "geometry": {
+            "type": "Polygon",
+            "coordinates": [[[10.0, 50.0], [10.0, 51.0], [12.0, 51.0], [12.0, 50.0], [10.0, 50.0]]],
+        },
+        "properties": {
+            "datetime": "2025-05-18T12:00:00Z",
+            "platform": "sentinel-1a",
+            "instruments": ["c-sar"],
+            "sar:instrument_mode": "IW",
+            "sar:polarizations": ["VV", "VH"],
+        },
+        "assets": {
+            "vh": {
+                "href": "s3://bucket/s1.zarr/S01SIWGRD_20250518_VH/measurements",
+                "type": "application/vnd+zarr",
+                "roles": ["data"],
+            },
+            "vv": {
+                "href": "s3://bucket/s1.zarr/S01SIWGRD_20250518_VV/measurements",
+                "type": "application/vnd+zarr",
+                "roles": ["data"],
+            },
+        },
+        "links": [],
+    }
+
+
+@pytest.mark.integration
+@pytest.mark.parametrize(
+    "collection_id,item_fixture",
+    [
+        ("sentinel-2-l2a", "sample_stac_item"),
+        ("sentinel-1-l1-grd", "sample_sentinel1_item"),
+    ],
+)
+def test_multi_mission_registration(collection_id, item_fixture, request):
+    """Test registration workflow for multiple missions (S1, S2)."""
+    from pystac import Item
+
+    from scripts.register import upsert_item
+
+    item_dict = request.getfixturevalue(item_fixture)
+    item = Item.from_dict(item_dict)
+
+    with patch("pystac_client.Client") as mock_client_class:
+        mock_client = Mock()
+        mock_client.self_href = "https://stac.example.com/stac"
+        mock_response = Mock()
+        mock_response.status_code = 201
+        mock_response.raise_for_status = Mock()
+        mock_client._stac_io.session.post = Mock(return_value=mock_response)
+        mock_client_class.return_value = mock_client
+
+        upsert_item(
+            client=mock_client,
+            collection_id=collection_id,
+            item=item,
+        )
+
+        assert mock_client._stac_io.session.post.called
+
+
+@pytest.mark.integration
+@pytest.mark.parametrize(
+    "mission,expected_has_group,expected_sharding",
+    [
+        ("sentinel-2", "/quality/l2a_quicklook/r10m", True),
+        ("sentinel-1", "/measurements", False),
+    ],
+)
+def test_collection_parameter_dispatch(mission, expected_has_group, expected_sharding):
+    """Test mission-based parameter dispatch from CONFIGS."""
+    from scripts.convert import CONFIGS
+
+    assert mission in CONFIGS, f"Mission {mission} not found in CONFIGS"
+
+    config = CONFIGS[mission]
+
+    # Check that expected group is in the groups list
+    assert expected_has_group in config["groups"]
+
+    # Verify sharding configuration matches mission
+    assert config["enable_sharding"] is expected_sharding
+
+    # S2 should have 4 groups, S1 should have 1
+    if mission == "sentinel-2":
+        assert len(config["groups"]) == 4
+        assert config["spatial_chunk"] == 1024
+        assert config["tile_width"] == 256
+    elif mission == "sentinel-1":
+        assert len(config["groups"]) == 1
+        assert config["spatial_chunk"] == 4096
+        assert config["tile_width"] == 512
+
+
+@pytest.mark.integration
+@pytest.mark.parametrize(
+    "collection_id,item_id,expected_substring",
+    [
+        ("sentinel-2-l2a", "S2B_MSIL2A_20250518_T29RLL", "sentinel-2-l2a"),
+        ("sentinel-1-l1-grd", "S1A_IW_GRDH_1SDV_20250518", "sentinel-1-l1-grd"),
+    ],
+)
+def test_collection_aware_output_paths(collection_id, item_id, expected_substring):
+    """Test that output paths include collection ID for multi-mission organization."""
+    output_path = f"s3://bucket/{collection_id}/{item_id}.zarr"
+    assert expected_substring in output_path
diff --git a/tests/unit/__init__.py b/tests/unit/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/tests/unit/test_convert.py b/tests/unit/test_convert.py
new file mode 100644
index 0000000..5c54864
--- /dev/null
+++ b/tests/unit/test_convert.py
@@ -0,0 +1,54 @@
+"""Unit tests for convert.py - GeoZarr conversion configuration."""
+
+from scripts.convert import CONFIGS
+
+
+class TestConversionConfigs:
+    """Test conversion configuration dictionaries."""
+
+    def test_s2_config_has_all_groups(self):
+        """Verify S2 config includes all 4 required groups."""
+        s2 = CONFIGS["sentinel-2"]
+        assert len(s2["groups"]) == 4
+        assert "/measurements/reflectance/r10m" in s2["groups"]
+        assert "/measurements/reflectance/r20m" in s2["groups"]
+        assert "/measurements/reflectance/r60m" in s2["groups"]
+        assert "/quality/l2a_quicklook/r10m" in s2["groups"]
+
+    def test_s2_config_has_crs_groups(self):
+        """Verify S2 config includes CRS groups."""
+        s2 = CONFIGS["sentinel-2"]
+        assert s2["crs_groups"] == ["/conditions/geometry"]
+
+    def test_s2_config_optimized_chunks(self):
+        """Verify S2 config has optimized chunk and tile sizes."""
+        s2 = CONFIGS["sentinel-2"]
+        assert s2["spatial_chunk"] == 1024
+        assert s2["tile_width"] == 256
+        assert s2["enable_sharding"] is True
+
+    def test_s1_config_structure(self):
+        """Verify S1 GRD config structure."""
+        s1 = CONFIGS["sentinel-1"]
+        assert s1["groups"] == ["/measurements"]
+        assert s1["crs_groups"] == ["/conditions/gcp"]
+        assert s1["spatial_chunk"] == 4096
+        assert s1["tile_width"] == 512
+        assert s1["enable_sharding"] is False
+
+    def test_both_missions_present(self):
+        """Verify both mission configs exist."""
+        assert "sentinel-1" in CONFIGS
+        assert "sentinel-2" in CONFIGS
+
+    def test_config_keys_consistent(self):
+        """Verify all configs have consistent keys."""
+        required_keys = {
+            "groups",
+            "crs_groups",
+            "spatial_chunk",
+            "tile_width",
+            "enable_sharding",
+        }
+        for mission, config in CONFIGS.items():
+            assert set(config.keys()) == required_keys, f"{mission} missing required keys"
diff --git a/tests/unit/test_register.py b/tests/unit/test_register.py
new file mode 100644
index 0000000..ecbe229
--- /dev/null
+++ b/tests/unit/test_register.py
@@ -0,0 +1,230 @@
+"""Unit tests for register.py - STAC registration and augmentation."""
+
+from unittest.mock import MagicMock, patch
+
+from pystac import Asset, Item
+from pystac.extensions.projection import ProjectionExtension
+
+from scripts.register import (
+    add_projection_from_zarr,
+    add_visualization_links,
+    rewrite_asset_hrefs,
+    s3_to_https,
+)
+
+
+class TestS3URLConversion:
+    """Test S3 URL to HTTPS conversion."""
+
+    def test_s3_to_https_conversion(self):
+        """Test basic S3 to HTTPS conversion."""
+        result = s3_to_https("s3://bucket/path/file.zarr", "https://s3.example.com")
+        assert result == "https://bucket.s3.example.com/path/file.zarr"
+
+    def test_s3_to_https_with_trailing_slash(self):
+        """Test S3 conversion with trailing slash in endpoint."""
+        result = s3_to_https("s3://bucket/file.zarr", "https://s3.example.com/")
+        assert result == "https://bucket.s3.example.com/file.zarr"
+
+    def test_s3_to_https_already_https(self):
+        """Test that HTTPS URLs are returned unchanged."""
+        https_url = "https://example.com/data.zarr"
+        result = s3_to_https(https_url, "https://s3.example.com")
+        assert result == https_url
+
+
+class TestAssetRewriting:
+    """Test asset href rewriting."""
+
+    def test_rewrite_asset_hrefs(self):
+        """Test asset href rewriting from old to new base."""
+        item = Item(
+            id="test",
+            geometry=None,
+            bbox=None,
+            datetime="2025-01-01T00:00:00Z",
+            properties={},
+        )
+        item.add_asset(
+            "data",
+            Asset(
+                href="s3://old-bucket/old.zarr/path/data",
+                media_type="application/vnd+zarr",
+            ),
+        )
+
+        rewrite_asset_hrefs(
+            item,
+            old_base="s3://old-bucket/old.zarr",
+            new_base="s3://new-bucket/new.zarr",
+            s3_endpoint="https://s3.example.com",
+        )
+
+        # Should rewrite to HTTPS URL with new base
+        new_href = item.assets["data"].href
+        assert new_href.startswith("https://new-bucket.s3.example.com")
+        assert "new.zarr" in new_href
+
+
+class TestProjectionExtension:
+    """Test projection extension from zarr."""
+
+    @patch("scripts.register.zarr.open")
+    def test_add_projection_from_zarr_with_spatial_ref(self, mock_zarr_open):
+        """Test adding projection extension when spatial_ref exists."""
+        # Mock zarr store with spatial_ref
+        mock_store = MagicMock()
+        mock_store.attrs = {
+            "spatial_ref": {
+                "spatial_ref": "32633",  # EPSG code
+                "crs_wkt": "PROJCS[...]",
+            }
+        }
+        mock_zarr_open.return_value = mock_store
+
+        item = Item(
+            id="test",
+            geometry=None,
+            bbox=None,
+            datetime="2025-01-01T00:00:00Z",
+            properties={},
+        )
+        item.add_asset(
+            "data",
+            Asset(
+                href="https://example.com/data.zarr",
+                media_type="application/vnd+zarr",
+            ),
+        )
+
+        add_projection_from_zarr(item)
+
+        # Should have projection extension
+        proj = ProjectionExtension.ext(item)
+        assert proj.epsg == 32633
+        assert proj.wkt2 is not None
+
+    @patch("scripts.register.zarr.open")
+    def test_add_projection_no_spatial_ref(self, mock_zarr_open):
+        """Test that function handles missing spatial_ref gracefully."""
+        # Mock zarr store without spatial_ref
+        mock_store = MagicMock()
+        mock_store.attrs = {}
+        mock_zarr_open.return_value = mock_store
+
+        item = Item(
+            id="test",
+            geometry=None,
+            bbox=None,
+            datetime="2025-01-01T00:00:00Z",
+            properties={},
+        )
+        item.add_asset(
+            "data",
+            Asset(
+                href="https://example.com/data.zarr",
+                media_type="application/vnd+zarr",
+            ),
+        )
+
+        # Should not raise an exception
+        add_projection_from_zarr(item)
+
+        # Should not have projection extension
+        assert "proj:epsg" not in item.properties
+
+
+class TestVisualizationLinks:
+    """Test visualization link generation."""
+
+    def test_add_visualization_links_s2(self):
+        """Test S2 True Color visualization links."""
+        item = Item(
+            id="S2A_MSIL2A_20250518_test",
+            geometry=None,
+            bbox=None,
+            datetime="2025-05-18T00:00:00Z",
+            properties={},
+        )
+        item.add_asset(
+            "TCI_10m",
+            Asset(
+                href="https://example.com/data.zarr/quality/l2a_quicklook/r10m",
+                media_type="application/vnd+zarr",
+            ),
+        )
+
+        add_visualization_links(
+            item,
+            raster_base="https://api.example.com/raster",
+            collection_id="sentinel-2-l2a",
+        )
+
+        # Should have viewer link
+        viewer_links = [link for link in item.links if link.rel == "viewer"]
+        assert len(viewer_links) == 1
+        assert "viewer" in viewer_links[0].href
+
+        # Should have XYZ tile link
+        xyz_links = [link for link in item.links if link.rel == "xyz"]
+        assert len(xyz_links) == 1
+        assert "tiles" in xyz_links[0].href
+
+        # Should have TileJSON link
+        tilejson_links = [link for link in item.links if link.rel == "tilejson"]
+        assert len(tilejson_links) == 1
+        assert "tilejson.json" in tilejson_links[0].href
+
+    def test_add_visualization_links_s1(self):
+        """Test S1 VH visualization links."""
+        item = Item(
+            id="S1A_IW_GRDH_test",
+            geometry=None,
+            bbox=None,
+            datetime="2025-05-18T00:00:00Z",
+            properties={},
+        )
+        item.add_asset(
+            "vh",
+            Asset(
+                href="https://example.com/data.zarr/measurements/vh",
+                media_type="application/vnd+zarr",
+            ),
+        )
+
+        add_visualization_links(
+            item,
+            raster_base="https://api.example.com/raster",
+            collection_id="sentinel-1-l1-grd",
+        )
+
+        # Should have viewer link
+        viewer_links = [link for link in item.links if link.rel == "viewer"]
+        assert len(viewer_links) == 1
+
+        # Should have XYZ tile link for VH
+        xyz_links = [link for link in item.links if link.rel == "xyz"]
+        assert len(xyz_links) == 1
+        assert "VH" in xyz_links[0].title or "vh" in xyz_links[0].href.lower()
+
+    def test_add_eopf_explorer_link(self):
+        """Test EOPF Explorer via link is added."""
+        item = Item(
+            id="test-item",
+            geometry=None,
+            bbox=None,
+            datetime="2025-01-01T00:00:00Z",
+            properties={},
+        )
+
+        add_visualization_links(
+            item,
+            raster_base="https://api.example.com/raster",
+            collection_id="sentinel-2-l2a",
+        )
+
+        # Should have via link to EOPF Explorer
+        via_links = [link for link in item.links if link.rel == "via"]
+        assert len(via_links) == 1
+        assert "explorer.eopf.copernicus.eu" in via_links[0].href
+        assert "EOPF Explorer" in via_links[0].title
diff --git a/workflows/examples/payload-s1.json b/workflows/examples/payload-s1.json
new file mode 100644
index 0000000..9e6752e
--- /dev/null
+++ b/workflows/examples/payload-s1.json
@@ -0,0 +1,5 @@
+{
+    "source_url": "https://stac.core.eopf.eodc.eu/collections/sentinel-1-l1-grd/items/S1C_IW_GRDH_1SDV_20251008T163126_20251008T163151_004473_008DBA_9AB4",
+    "item_id": "S1C_IW_GRDH_20251008_test",
+    "collection": "sentinel-1-l1-grd-dp-test"
+}