Validating COG Structure in CI

To stop malformed rasters reaching an archive, express the required structure as assertions and run them over every produced file as part of the build:

import rasterio
from rio_cogeo.cogeo import cog_validate


def assert_publishable(path: str, *, min_overviews: int = 3, block: int = 512) -> None:
    ok, errors, _ = cog_validate(path)
    assert ok, f"{path}: invalid COG layout: {errors}"
    with rasterio.open(path) as src:
        assert src.block_shapes[0] == (block, block), f"{path}: wrong block shape"
        assert len(src.overviews(1)) >= min_overviews, f"{path}: too few overviews"
        assert src.nodata is not None, f"{path}: no declared nodata"

This turns the manual checks in Writing and Validating Cloud-Optimized GeoTIFFs into something a pipeline enforces on its own.

Why This Arises in Remote Sensing Workflows

Raster outputs are produced by code, and code drifts. A refactor drops tiled=True from a profile dictionary. A new output path forgets to build overviews. A dtype change makes a predictor inappropriate. None of these break any test that checks pixel values, because the pixels are fine — what changed is the property that makes the file cheap to read, and that property has no natural test.

The consequence is delayed and expensive. Malformed files enter an archive at production rate, and the symptom appears weeks later as a slow map or an unexplained egress bill. By then the archive holds thousands of files that all need rewriting, and the rewrite costs more compute than the original production run.

A structural gate closes that loop at the point of failure. It is cheap — every assertion above reads only the header, the same few kilobytes discussed in How to Read COG Headers Without Downloading Full Files — and it converts a slow, diffuse failure into a build error with a filename attached.

Where a structural regression is cheapest to catch Catching a bad profile in the build costs one failed test. Catching it at publish time costs re-running one job. Discovering it after ten thousand files have been written costs a full reprocessing campaign plus the egress already spent on slow reads. Same mistake, three moments to notice it in the build one assertion fails on a synthetic 2048 px file cost: seconds at publish time one job's outputs rejected before anything is uploaded cost: one re-run months later 10,000 files to rewrite plus the slow reads already paid cost: a campaign The assertions cost the same in all three columns. Only the blast radius changes. This is why the gate belongs in the build rather than in a review checklist. Header-only checks make "validate every output" affordable at production rate.

Environment & Setup

Package Version Why
rasterio ≥1.3.0 Header access for the structural assertions
rio-cogeo ≥5.0 cog_validate for layout and ordering
pytest ≥7.0 Test runner that most CI systems already understand
numpy ≥1.23 Generates the synthetic fixture 
pip install "rasterio>=1.3.0" "rio-cogeo>=5.0" "pytest>=7.0" "numpy>=1.23"

Complete Working Example

The gate has two halves: a reusable check that returns findings rather than raising, and a thin test that turns findings into a failure. Keeping them separate means the same code can run as a build gate, as a nightly audit over an existing archive, or as a report.

Policy fields and the failure each one prevents Each policy field maps to a specific downstream failure: wrong block shape makes windowed reads expensive, missing overviews make previews read full resolution, an undeclared nodata biases every statistic, and an unexpected dtype doubles archive size. What each assertion is actually protecting policy field typical value failure it prevents block shape 512 × 512 strip reads on every window overview levels ≥ 3 previews decimating full resolution nodata declared required fill values entering statistics compression deflate or zstd a consumer that cannot open the file dtype uint16 / float32 an archive four times larger than needed A failing row names the file and the property — which is what makes the CI output actionable. 
from dataclasses import dataclass, field

import rasterio
from rio_cogeo.cogeo import cog_validate


@dataclass
class CogPolicy:
    """The structural contract a published raster must satisfy."""
    block: int = 512
    min_overviews: int = 3
    require_nodata: bool = True
    allowed_compression: tuple[str, ...] = ("deflate", "zstd")
    allowed_dtypes: tuple[str, ...] = ("uint16", "int16", "float32", "uint8")


@dataclass
class Findings:
    path: str
    problems: list[str] = field(default_factory=list)

    @property
    def ok(self) -> bool:
        return not self.problems


def check_cog(path: str, policy: CogPolicy = CogPolicy()) -> Findings:
    """Check one raster against the policy. Never raises; collects every problem."""
    found = Findings(path)

    is_valid, errors, _warnings = cog_validate(path)
    if not is_valid:
        found.problems.extend(f"layout: {e}" for e in errors)

    with rasterio.open(path) as src:
        bx, by = src.block_shapes[0]
        if (bx, by) != (policy.block, policy.block):
            found.problems.append(f"block shape {bx}x{by}, expected {policy.block}²")

        levels = src.overviews(1)
        if len(levels) < policy.min_overviews:
            found.problems.append(f"{len(levels)} overview level(s), expected ≥ {policy.min_overviews}")

        if policy.require_nodata and src.nodata is None:
            found.problems.append("no declared nodata value")

        compress = (src.profile.get("compress") or "none").lower()
        if compress not in policy.allowed_compression:
            found.problems.append(f"compression {compress!r} not in {policy.allowed_compression}")

        dtype = src.dtypes[0]
        if dtype not in policy.allowed_dtypes:
            found.problems.append(f"dtype {dtype!r} not in {policy.allowed_dtypes}")

        if src.crs is None:
            found.problems.append("no CRS")

    return found

And the test that makes CI care about it:

from pathlib import Path

import numpy as np
import pytest
import rasterio
from rasterio.shutil import copy as rio_copy

from cogcheck import CogPolicy, check_cog     # the module above


@pytest.fixture(scope="session")
def sample_cog(tmp_path_factory) -> str:
    """A small synthetic COG — no binary fixtures in the repository."""
    tmp = tmp_path_factory.mktemp("cog")
    raw, out = tmp / "raw.tif", tmp / "sample_cog.tif"

    data = (np.indices((2048, 2048)).sum(axis=0) % 4096).astype("uint16")
    profile = dict(driver="GTiff", height=2048, width=2048, count=1, dtype="uint16",
                   crs="EPSG:32636", transform=rasterio.transform.from_origin(5e5, 1e7, 10, 10),
                   nodata=0, tiled=True, blockxsize=512, blockysize=512, compress="deflate")
    with rasterio.open(raw, "w", **profile) as dst:
        dst.write(data, 1)

    rio_copy(raw, out, driver="COG", compress="DEFLATE", blocksize=512,
             overview_resampling="average")
    return str(out)


def test_sample_cog_meets_policy(sample_cog):
    findings = check_cog(sample_cog, CogPolicy(min_overviews=2))
    assert findings.ok, findings.problems


@pytest.mark.parametrize("path", sorted(str(p) for p in Path("build/rasters").glob("*.tif")))
def test_every_output_meets_policy(path):
    findings = check_cog(path)
    assert findings.ok, f"{path}: " + "; ".join(findings.problems)

The parametrize line is what makes the failure useful: each output becomes its own test case, so the CI report names the offending file instead of failing one opaque test called test_outputs.

Variant Patterns

1. A remote read check for published objects

Structure is only half the contract. The other half is what an open costs over the network, which depends on layout and on the reader’s configuration together.

import rasterio


def open_request_count(url: str) -> int:
    """Count HTTP requests GDAL issues to open a remote raster and read its profile."""
    cfg = {"GDAL_DISABLE_READDIR_ON_OPEN": "EMPTY_DIR", "CPL_CURL_VERBOSE": "YES"}
    with rasterio.Env(**cfg) as env:
        with rasterio.open(url) as src:
            _ = src.profile, src.overviews(1)
        # GDAL reports transfer counters through the environment's stats where available
        return int(env.options.get("__request_count__", 0))

Counting requests precisely depends on the GDAL build, so in practice most teams assert on elapsed time and transferred bytes instead, sampled over a handful of representative objects. Either way, run it against a real published URL, not a local path, because the settings that dominate — sibling listing, ingest size, region — only exist remotely. Those settings are catalogued in How to Read COG Headers Without Downloading Full Files.

2. Auditing an existing archive with the same code

The check function takes a path, so it works equally well over an inventory. Run it as a scheduled job across the archive and write the findings to a table; the pattern is the metadata sweep from Automating Metadata Extraction for Batch Raster Jobs, with check_cog in place of the profile collector.

from concurrent.futures import ThreadPoolExecutor

def audit_archive(urls: list[str], workers: int = 16) -> list[Findings]:
    with ThreadPoolExecutor(max_workers=workers) as pool:
        results = list(pool.map(check_cog, urls))
    return [r for r in results if not r.ok]

Threads rather than processes, because each check is a header read and therefore network-bound.

3. Machine-readable output for the build system

What the gate should print Per-file findings are aggregated into a count per problem type, so a run that fails on four hundred files shows four distinct causes rather than four hundred lines. Each cause names one file as an example, which is what makes the failure actionable. Aggregate by cause, not by file problem files example block shape 256x256, expected 512² 412 T36NYF_20230615_ndvi.tif 2 overview level(s), expected ≥ 3 37 T37NBA_20230702_ndvi.tif no declared nodata value 9 T36NYG_20230620_mask.tif layout: overviews after pixel data 1 T36NYF_20230615_lst.tif Four causes, 459 files: one profile bug, one overview bug, one write path, one stale conversion.

Emit the summary as JSON alongside the human-readable table. A build system can then track the counts over time, and a regression shows up as a number that went from zero to four hundred rather than as a wall of red text.

Common Errors

The gate passes locally and fails in CI

The CI image has a different GDAL build, and cog_validate is stricter or laxer about layout between versions. Pin the GDAL and rio-cogeo versions in the same place you pin rasterio; reproducibility here is the same problem discussed in Containerizing Geospatial Python Environments.

Every file fails on nodata

The policy is wrong, not the files: some products legitimately have no fill value because the footprint is a full rectangle. Make require_nodata a per-collection setting rather than a global one.

The test suite takes minutes on a large output directory

The structural checks are header-only and fast; something else is slow. Usually it is cog_validate on very large files over a network mount — copy locally first, or sample.

Frequently Asked Questions

Q: Is cog_validate enough on its own? It checks structure and layout, which is the hard part, but it does not know your conventions. Overview depth, block size, declared nodata and dtype are project decisions, and they need explicit assertions alongside the validator.

Q: Should the gate run on every file or a sample? Every file for the structural assertions, which take milliseconds from the header alone. Sample for the remote read check, which needs network round trips and only varies per publishing configuration.

Q: How do I test this without large fixtures? Generate a small synthetic raster in the test itself. A 2048 by 2048 array is enough to exercise tiling and two overview levels, and it keeps the repository free of binary fixtures.