8 min read1,799 words

Automating Connectivity Rule Validation in CI Pipelines for Utility Network GIS

Engineering Imperative

Manual connectivity rule reviews do not scale. A production domain network carries hundreds of connectivity rules spanning asset group, asset type, and terminal pairings, and every edit in ArcGIS Pro or via the schema API risks introducing drift between development, staging, and enterprise geodatabases. When that drift is caught only during a runtime topology validation pass — or worse, during an emergency trace — it forces a full topology rebuild, blocks deployments, and leaves a gap in the compliance audit trail. Folding deterministic connectivity rule validation into a continuous integration pipeline converts that production-blocking risk into a cheap pre-flight signal: the build fails fast, with the exact offending rule named, before a single feature is dirtied.

Environment Prerequisites

Reproducible validation depends on a pinned, version-controlled environment. Lock the following before wiring the gate into CI:

  • ArcGIS Pro 3.2+ (or matching ArcGIS Enterprise 11.2+ portal) for schema authoring and export consistency.
  • Utility Network version 6 or 7 schema — confirm the runner targets the same UN version as production to avoid false isAllowed/terminal-shape differences.
  • Python 3.9–3.11 on the CI runner. The validator below uses only the standard library (json, pathlib, logging, typing), so no arcpy license is consumed in the gate itself.
  • A current schema export (exportUtilityNetworkSchema output, or the JSON returned by the UN schema REST endpoint) committed or fetched as a pipeline artifact.
  • Topology in a validated state in the source workspace at export time — exporting mid-edit yields a matrix that does not reflect the built topology graph.

Schema-Aware Validation Protocol

Run these diagnostic steps in order. The most common failure cause — a stale or wrong-version export — comes first, because every downstream assertion is meaningless against the wrong document.

  1. Confirm the export is current. Compare the export timestamp against the last schema edit. A matrix exported before the latest rule change validates the old world and passes incorrectly.
  2. Verify the target domain network exists. A typo in the domain name (ElectricDistribution vs ElectricDist) makes the validator report zero rules and silently pass — treat a missing domain as a hard failure, not an empty success.
  3. Build asset-type and terminal indices first. Index every valid asset type ID and its declared terminal IDs before walking the rules, so each rule can be resolved in constant time and orphaned references surface deterministically.
  4. Assert three invariants per rule: asset existence (both fromAssetType and toAssetType resolve), terminal alignment (every referenced terminalId is declared on the source asset type), and flag consistency (isAllowed and isBidirectional cannot contradict).
  5. Emit structured failures. On any violation, output a machine-readable payload carrying the rule ID, the violating asset-type pair, and the expected-versus-actual state so engineers patch the rule rather than rolling back the build.

This protocol matters most when configuring connectivity rules for pipe and cable networks, where a single orphaned terminal mapping silently breaks hydraulic or electrical continuity and only manifests far downstream during a trace.

CI connectivity rule validation gate with pass and fail branches A schema export feeds a validator that runs three stacked invariant checks: asset existence, terminal alignment, and flag consistency. If every invariant holds the gate passes and the topology build proceeds to a trace-ready model; if any invariant fails the gate blocks the pull request and writes a structured diagnostics artifact identifying the offending rule. Schema export connectivityRules[] Validator gate 1 · Asset existence from/to assetType resolve 2 · Terminal alignment terminalId declared on type 3 · Flag consistency isAllowed ∧ isBidirectional all pass Topology build → trace-ready model any violation PR blocked (exit 1) diagnostics.json artifact

Minimal Reproducible Implementation

The validator below is a self-contained, copy-paste-ready routine for CI execution. It loads a schema export, builds fast-lookup indices, asserts the three invariants, and enforces strict exit codes so the pipeline halts on any violation. It leans on Python’s standard-library JSON parsing and returns a structured payload suitable for artifact collection.

#!/usr/bin/env python3
"""
CI-Ready Connectivity Rule Validator for Utility Network Schemas.
Validates asset references, terminal mappings, and bidirectional flags.
Exits 0 on success, 1 on validation failure, 2 on fatal error.
"""
import json
import sys
import logging
from pathlib import Path
from typing import Any, Dict, List

logging.basicConfig(level=logging.INFO, format="%(levelname)s: %(message)s")

def load_schema(schema_path: str) -> Dict[str, Any]:
    path = Path(schema_path)
    if not path.exists():
        raise FileNotFoundError(f"Schema export not found: {schema_path}")
    with open(path, "r", encoding="utf-8") as f:
        return json.load(f)

def validate_connectivity_rules(schema: Dict[str, Any], domain_network: str) -> List[Dict[str, Any]]:
    errors: List[Dict[str, Any]] = []
    domain = next((d for d in schema.get("domainNetworks", []) if d["name"] == domain_network), None)
    if not domain:
        # Step 2: a missing domain is a hard failure, never an empty success.
        errors.append({"type": "DOMAIN_MISSING", "message": f"Domain network '{domain_network}' not found in schema."})
        return errors

    # Step 3: build fast-lookup indices for asset and terminal validation.
    valid_asset_types: Dict[int, str] = {}
    valid_terminals: Dict[int, set] = {}
    for ag in domain.get("assetGroups", []):
        for at in ag.get("assetTypes", []):
            valid_asset_types[at["id"]] = at["name"]
            valid_terminals[at["id"]] = {t["terminalId"] for t in at.get("terminals", [])}

    for rule in domain.get("connectivityRules", []):
        rule_id = rule.get("id", "UNKNOWN")
        from_type_id = rule.get("fromAssetType")
        to_type_id = rule.get("toAssetType")
        terminals = rule.get("terminals", [])
        is_allowed = rule.get("isAllowed", True)
        is_bidirectional = rule.get("isBidirectional", False)

        # Invariant 1: asset-type existence.
        if from_type_id not in valid_asset_types or to_type_id not in valid_asset_types:
            errors.append({
                "type": "INVALID_ASSET_REFERENCE",
                "rule_id": rule_id,
                "message": f"Rule {rule_id} references non-existent asset types: from={from_type_id}, to={to_type_id}"
            })
            continue

        # Invariant 2: terminal-configuration alignment.
        for term in terminals:
            term_id = term.get("terminalId")
            if term_id is None or term_id not in valid_terminals.get(from_type_id, set()):
                errors.append({
                    "type": "ORPHANED_TERMINAL",
                    "rule_id": rule_id,
                    "message": f"Rule {rule_id} contains undefined or mismatched terminal ID: {term_id}"
                })

        # Invariant 3: bidirectional flag consistency.
        if not is_allowed and is_bidirectional:
            errors.append({
                "type": "FLAG_MISMATCH",
                "rule_id": rule_id,
                "message": f"Rule {rule_id} has conflicting isAllowed=False and isBidirectional=True flags."
            })

    return errors

def main() -> None:
    if len(sys.argv) < 2:
        logging.error("Usage: python validate_un_rules.py <schema.json> [domain_name]")
        sys.exit(2)

    schema_path = sys.argv[1]
    domain = sys.argv[2] if len(sys.argv) > 2 else "ElectricDistribution"

    try:
        schema = load_schema(schema_path)
        violations = validate_connectivity_rules(schema, domain)
        if violations:
            logging.error("Connectivity rule validation failed with %d violation(s).", len(violations))
            # Structured payload to stderr for CI artifact collection.
            print(json.dumps(violations, indent=2), file=sys.stderr)
            sys.exit(1)
        logging.info("Connectivity rule validation passed. Schema is compliant.")
        sys.exit(0)
    except Exception as e:
        logging.error("Fatal validation error: %s", e)
        sys.exit(2)

if __name__ == "__main__":
    main()

Production Deployment Pattern

Wire the validator in as a pre-build gate so no topology build or trace configuration ships against an unvalidated matrix. In GitHub Actions, run it immediately after the schema export step and before any build job, capturing the stderr JSON as a workflow artifact for the audit trail.

# .github/workflows/un-schema-gate.yml
name: UN Connectivity Rule Gate
on: [pull_request]
jobs:
  validate:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: "3.11"
      - name: Validate connectivity rules
        id: validate
        run: python validate_un_rules.py exports/schema.json ElectricDistribution 2> diagnostics.json
      - name: Publish diagnostics
        if: always()
        uses: actions/upload-artifact@v4
        with:
          name: connectivity-diagnostics
          path: diagnostics.json

For resilience, fetch the live schema export through a short retry/backoff loop before validation — the portal’s exportUtilityNetworkSchema endpoint can transiently 503 under load, and a blind single attempt produces a misleading FATAL exit. Pair the gate with a schema-diffing step: when drift is detected, have the pipeline open a pull request carrying the corrected matrix tagged with the violating rule IDs, and notify GIS technicians and asset engineers so remediation is targeted rather than a wholesale rollback. The same pattern feeds cleanly into batch topology processing with arcpy and geopandas once a matrix passes the gate and the build proceeds.

Conclusion

This automation moves connectivity rule integrity from a runtime gamble to a deterministic pre-deployment check. By indexing the schema, asserting asset existence, terminal alignment, and flag consistency, and enforcing strict CI exit codes, the gate names the offending rule before it can corrupt the topology graph — preserving an auditable trail for compliance reviews. The next logical step is to extend the same gate downstream into trace configuration, validating that the matrix the build consumes is the matrix upstream water valve tracing actually relies on.