Validation by Policy¶

Policy assertions can be expressed according to the Rego Policy language directly inside in the original CWL Workflow as a Hint

In [1]:

Copied!





module = """
package workflow

# --- BBOX sanity checks ---

deny_bbox[msg] {
  b := input.aoi.bbox
  not valid_bbox(b)
  msg := sprintf("invalid aoi.bbox: %v", [b])
}

valid_bbox(b) {
  count(b) == 4
  b[0] >= -180
  b[2] <= 180
  b[1] >= -90
  b[3] <= 90
  b[0] < b[2]
  b[1] < b[3]
}

# --- Bands sanity checks ("green" must be first; second is "nir" or "nir08") ---

deny_bands[msg] {
  b := input.bands
  not valid_bands(b)
  msg := sprintf("invalid bands: %v", [b])
}

valid_bands(b) {
  count(b) == 2
  b[0] == "green"
  allowed_nir(b[1])
}

# helper predicate for the allowed second band (disjunction via multiple rules)
allowed_nir(x) { x == "nir" }
allowed_nir(x) { x == "nir08" }
"""

# List the deny queries you want to collect messages from
deny_queries = [
    "data.workflow.deny_bbox[_]",
    "data.workflow.deny_bands[_]",
]
module = """
package workflow

# --- BBOX sanity checks ---

deny_bbox[msg] {
  b := input.aoi.bbox
  not valid_bbox(b)
  msg := sprintf("invalid aoi.bbox: %v", [b])
}

valid_bbox(b) {
  count(b) == 4
  b[0] >= -180
  b[2] <= 180
  b[1] >= -90
  b[3] <= 90
  b[0] < b[2]
  b[1] < b[3]
}

# --- Bands sanity checks ("green" must be first; second is "nir" or "nir08") ---

deny_bands[msg] {
  b := input.bands
  not valid_bands(b)
  msg := sprintf("invalid bands: %v", [b])
}

valid_bands(b) {
  count(b) == 2
  b[0] == "green"
  allowed_nir(b[1])
}

# helper predicate for the allowed second band (disjunction via multiple rules)
allowed_nir(x) { x == "nir" }
allowed_nir(x) { x == "nir08" }
"""

# List the deny queries you want to collect messages from
deny_queries = [
    "data.workflow.deny_bbox[_]",
    "data.workflow.deny_bands[_]",
]

1. Parse the Rego module¶

In [2]:

Copied!

from assertions_mate import RegoPolicyHint

validator = RegoPolicyHint(module=module, queries=deny_queries).validator()
from assertions_mate import RegoPolicyHint

validator = RegoPolicyHint(module=module, queries=deny_queries).validator()

---------------------------------------------------------------------------
ModuleNotFoundError                       Traceback (most recent call last)
Cell In[2], line 1
----> 1 from assertions_mate import RegoPolicyHint
      2 
      3 validator = RegoPolicyHint(module=module, queries=deny_queries).validator()

ModuleNotFoundError: No module named 'assertions_mate'

2. Define your inputs set¶

In [3]:

Copied!





inputs = {
    "aoi": {"bbox": [-1181.985, 38.432, -118.183, 38.938], "crs": "CRS84"},
    "filesB": "EPSG:4326",
    "bands": ["green", "nir089"],
    "item": "https://planetarycomputer.microsoft.com/api/stac/v1/collections/landsat-c2-l2/items/LC08_L2SP_042033_20231007_02_T1",
}
inputs = {
    "aoi": {"bbox": [-1181.985, 38.432, -118.183, 38.938], "crs": "CRS84"},
    "filesB": "EPSG:4326",
    "bands": ["green", "nir089"],
    "item": "https://planetarycomputer.microsoft.com/api/stac/v1/collections/landsat-c2-l2/items/LC08_L2SP_042033_20231007_02_T1",
}

3. Validate the inputs and report the errors¶

In [4]:

Copied!





error = validator.validate_inputs(inputs)

if error:
    print(error.model_dump_json(indent=2, exclude_none=True))
else:
    # No violations — choose what to output. A simple success marker is fine:
    print("Go ahead on submitting the request to the next level validation")
error = validator.validate_inputs(inputs)

if error:
    print(error.model_dump_json(indent=2, exclude_none=True))
else:
    # No violations — choose what to output. A simple success marker is fine:
    print("Go ahead on submitting the request to the next level validation")

---------------------------------------------------------------------------
NameError                                 Traceback (most recent call last)
Cell In[4], line 1
----> 1 error = validator.validate_inputs(inputs)
      2 
      3 if error:
      4     print(error.model_dump_json(indent=2, exclude_none=True))

NameError: name 'validator' is not defined

In [ ]: