Automating Connectivity Rule Validation in CI Pipelines for Utility Network GIS

Utility network deployments routinely fail during production topology validation because connectivity rule matrices drift across development, staging, and enterprise environments. When asset lifecycle automation relies on manual schema reviews or ad-hoc geodatabase checks, rule violations propagate silently into tracing configurations. This causes cascading failures in isolation, pressure, and fault propagation analyses. Integrating automated connectivity rule validation directly into continuous integration pipelines eliminates this compliance gap by enforcing deterministic schema verification before any topology build or trace configuration deployment. Establishing this guardrail aligns directly with enterprise-grade Topology & Tracing Workflows, where schema integrity dictates network trace accuracy and operational reliability.

Diagnostic Architecture & Schema Invariants

The core diagnostic challenge lies in translating declarative connectivity matrices into executable validation checks. A utility network connectivity rule defines permissible associations between junction and edge asset groups, asset types, and terminal configurations. When these rules are modified in ArcGIS Pro or via the UN schema API, the underlying JSON schema must remain strictly consistent with the enterprise geodatabase topology configuration. Schema drift typically manifests as orphaned terminal mappings, invalid containment hierarchies, or mismatched isAllowed flags across asset type pairs. Without automated pre-flight validation, these discrepancies surface only during runtime tracing or topology validation, triggering costly rollback cycles and compliance audit failures.

To isolate rule violations deterministically, pipeline diagnostics must parse the exported connectivity rule matrix and cross-reference it against the active enterprise schema. A minimal validation routine should extract the connectivityRules array from the UN schema export, iterate through each rule definition, and assert three critical invariants:

Asset Existence: Referenced asset groups and asset types must exist in the current domain network.
Terminal Alignment: Terminal configurations must match the physical asset model and defined terminal definitions.
Flag Consistency: Bidirectional connectivity flags and isAllowed states must align with engineering specifications.

When these assertions fail, the pipeline must emit structured error payloads containing the exact rule ID, violating asset type pair, and expected versus actual topology state. This diagnostic output enables infrastructure teams to pinpoint rule misconfigurations before they corrupt the topology graph. Properly structuring these assertions is particularly critical when Configuring Connectivity Rules for Pipe & Cable networks, where terminal mismatches can silently break hydraulic or electrical continuity.

Production-Grade Python Validator

The following implementation provides a schema-aware debugging routine designed for CI execution. It leverages Python’s native JSON parsing capabilities to load schema exports, validate invariants, and output machine-readable diagnostics. The script enforces strict exit codes to trigger pipeline halts on violation detection.

#!/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:
        errors.append({"type": "DOMAIN_MISSING", "message": f"Domain network '{domain_network}' not found in schema."})
        return errors

    # Build fast-lookup indices for asset validation
    valid_asset_types: Dict[int, str] = {}
    for ag in domain.get("assetGroups", []):
        for at in ag.get("assetTypes", []):
            valid_asset_types[at["id"]] = at["name"]

    # Extract valid terminal IDs per asset type
    valid_terminals: Dict[int, set] = {}
    for ag in domain.get("assetGroups", []):
        for at in ag.get("assetTypes", []):
            valid_terminals[at["id"]] = {t["terminalId"] for t in at.get("terminals", [])}

    connectivity_rules = domain.get("connectivityRules", [])
    for rule in connectivity_rules:
        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
        if terminals:
            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))
            # Output structured payload for CI artifact collection
            print(json.dumps(violations, indent=2), file=sys.stderr)
            sys.exit(1)
        else:
            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()

CI/CD Integration & Incident Response Workflow

Deploying this validator requires wiring it into your pipeline as a pre-deployment gate. In GitHub Actions or GitLab CI, execute the script immediately after schema export and before topology build jobs. Capture the sys.stderr JSON output as a pipeline artifact to enable rapid incident resolution. When a violation occurs, infrastructure teams should parse the structured payload to isolate the exact rule ID, cross-reference it against the Utility Network schema documentation, and apply targeted schema patches rather than rolling back entire topology builds.

For automated remediation, integrate the validator with a schema diffing step. If the pipeline detects drift, it should automatically generate a pull request containing the corrected connectivity matrix, tagged with the violating rule IDs. This closes the feedback loop between development and enterprise environments. Storing validation outputs as workflow artifacts ensures audit trails remain intact for compliance reviews.

Incident resolution accelerates when validation failures trigger immediate notifications to GIS technicians and asset engineers. By enforcing deterministic checks early, teams avoid the costly topology rebuild cycles that typically follow silent rule propagation. This approach is equally applicable to linear infrastructure, where Configuring Connectivity Rules for Pipe & Cable networks demand strict terminal-to-terminal continuity validation before hydraulic or electrical trace execution.

Conclusion

Automating connectivity rule validation transforms schema drift from a production-blocking risk into a pre-deployment diagnostic signal. By embedding invariant checks directly into CI pipelines, utility engineers and infrastructure teams gain deterministic control over topology integrity. The combination of structured JSON diagnostics, strict CI exit codes, and schema-aware debugging routines ensures that asset lifecycle automation remains resilient, auditable, and aligned with enterprise tracing requirements.