def local_validation()

in mozilla_schema_generator/validate_bigquery.py [0:0]

• 48 lines of code
• 16 McCabe index (conditional complexity)

def local_validation(head, base, repository, artifact, incompatibility_allowlist):
    """Validate schemas using a heuristic from the compact schemas."""
    head_path, base_path = checkout_copy_schemas_revisions(
        head, base, repository, artifact
    )
    is_error = 0

    # look at the compact schemas
    head_files = (head_path).glob("*.txt")
    base_files = (base_path).glob("*.txt")

    # also look at the exceptions
    allowed_incompatibility_base_files = []
    if incompatibility_allowlist:
        for glob in parse_incompatibility_allowlist(Path(incompatibility_allowlist)):
            allowed_incompatibility_base_files += list((base_path).glob(f"{glob}.txt"))

    a = set([p.name for p in base_files])
    b = set([p.name for p in head_files])
    allowed_incompatibility = set([p.name for p in allowed_incompatibility_base_files])

    # Check that we're not removing any schemas. If there are exceptions, we
    # remove this from the base set before checking for evolution.
    if allowed_incompatibility:
        print("allowing incompatible changes in the following documents:")
        print("\n".join([f"\t{x}" for x in allowed_incompatibility]))
    is_error |= check_evolution((a - allowed_incompatibility), b, verbose=True)

    for schema_name in a & b:
        base = base_path / schema_name
        head = head_path / schema_name
        base_data = base.read_text().split("\n")
        head_data = head.read_text().split("\n")
        diff = "\n".join(
            # control lines contain a newline at the end
            [
                line.strip()
                for line in difflib.unified_diff(
                    base_data,
                    head_data,
                    fromfile=base.as_posix(),
                    tofile=head.as_posix(),
                    n=1,
                )
            ]
        )
        if not diff:
            # no difference detected
            continue
        # check if this is an error condition
        print(diff + "\n")
        err_code = check_evolution(base_data, head_data)
        if err_code and schema_name in allowed_incompatibility:
            print("found incompatible changes, but continuing")
            continue
        is_error |= err_code

    if not is_error:
        click.echo("no incompatible changes detected")
    else:
        click.echo("found incompatible changes")

    sys.exit(is_error)