Commit a29e50ac authored by Luke Mewburn's avatar Luke Mewburn
Browse files

compat_asn1: refactor to a class 

Implement class CompatCheck with most of the behaviour
encapsulated, and deprecate Context.
Refactor compare_files() to return the list of error messages.
parent 0fef4c9e

testing/asn_process.py

+7 −7
Original line number Diff line number Diff line
@@ -148,7 +148,7 @@ def compileAllTargets(asnFiles): 
    return results





def compatCheck(asnFiles):

def compatCheckFiles(asnFiles):

    """

    Attempts to check backwards compatibility using the pycrate ASN.1 tools.
@@ -163,6 +163,7 @@ def compatCheck(asnFiles): 


    """

    results = {}

    compatCheck = compat_asn1.CompatCheck()

    for target, config in asnFiles.items():

        logging.info(f"Compatibility checking {target}")

        try:
@@ -173,14 +174,13 @@ def compatCheck(asnFiles): 
                targetDeps = getDependencies(asnFiles, target)

                for compatName, compatExceptions in compatFiles.items():

                    compatDeps = getDependencies(asnFiles, compatName)

                    ctx = compat_asn1.Context()

                    compat_asn1.compare_files(

                        ctx=ctx, a_files=compatDeps, b_files=targetDeps

                    compatErrors = compatCheck.compare_files(

                        a_files=compatDeps, b_files=targetDeps

                    )

                    errors.extend(

                        [

                            {"message": error}

                            for error in ctx.errors

                            for error in compatErrors

                            if error not in compatExceptions

                        ]

                    )
@@ -188,7 +188,7 @@ def compatCheck(asnFiles): 
                        [

                            {"message": f"Missing error: {error}"}

                            for error in compatExceptions

                            if error not in ctx.errors

                            if error not in compatErrors

                        ]

                    )

            results[str(target)] = {
@@ -293,7 +293,7 @@ def main(): 
        retval = 1



    logging.info("Checking ASN.1 backwards compatibility")

    compatResults = compatCheck(asnFiles)

    compatResults = compatCheckFiles(asnFiles)

    if processResults(compatResults, "Compatibility checking") > 0:

        retval = 1

testing/check_compat_asn1.py

+5 −5
Original line number Diff line number Diff line
@@ -11,20 +11,20 @@ import compat_asn1 


def compare_releases(files):

    total_errors = 0

    compatCheck = compat_asn1.CompatCheck()

    for idx in range(len(files) - 1):

        a_files = files[idx]

        b_files = files[idx + 1]

        ctx = compat_asn1.Context()

        compat_asn1.compare_files(ctx=ctx, a_files=a_files, b_files=b_files)

        if ctx.errors:

        compatErrors = compatCheck.compare_files(a_files=a_files, b_files=b_files)

        if compatErrors:

            print("-----------------------------")

            print(f"File 1: {a_files[0]}")

            print(f"File 2: {b_files[0]}")

            print("Errors:")

            for error in ctx.errors:

            for error in compatErrors:

                print(f"  {error}")

            print("-----------------------------")

        total_errors += len(ctx.errors)

        total_errors += len(compatErrors)

    return total_errors

testing/compat_asn1.py

+240 −250
Original line number Diff line number Diff line
@@ -5,7 +5,6 @@ Module to check backwards compatibility between two ASN.1 files. 
"""



import copy

import dataclasses

import logging

import os

import pathlib
@@ -15,7 +14,7 @@ import pycrate_asn1c.asnobj as pa_asnobj 
import pycrate_asn1c.asnproc





# TODO: require python >= 3.7

# TODO: require python >= 3.10





# pylint: disable=logging-fstring-interpolation
@@ -23,20 +22,6 @@ import pycrate_asn1c.asnproc 
TYPE_CONTAINER_OF = (pa_asnobj.TYPE_SEQ_OF, pa_asnobj.TYPE_SET_OF)





@dataclasses.dataclass

class Context:

    errors: list[str] = dataclasses.field(default_factory=list)

    fa_mod: pa_asnobj.ASN1Dict | None = None

    fb_mod: pa_asnobj.ASN1Dict | None = None

    seen: set[str] = dataclasses.field(default_factory=set)





def process_error(ctx: Context, message: str) -> None:

    """Log `message` and append `message` to `errors`."""

    logging.info(f"Test Failure: {message}")

    ctx.errors.append(message)





def dict_swap(dictionary: dict) -> dict:

    """Swap keys and values in `dictionary`.
@@ -45,19 +30,6 @@ def dict_swap(dictionary: dict) -> dict: 
    return {value: key for key, value in dictionary.items()}





def parse_asn1(asn_files: list[str]) -> pa_asnobj.ASN1Dict:

    """Parse `asn_files` as ASN.1 using pycrate, returning the module object."""

    logging.info(f"Parsing {' '.join(asn_files)}")

    asn_text = []

    for filename in asn_files:

        filepath = pathlib.Path(filename)

        asn_text.append(filepath.read_text())

    pycrate_asn1c.asnproc.GLOBAL.clear()

    pycrate_asn1c.asnproc.compile_text(asn_text, filenames=asn_files)

    res = copy.deepcopy(pycrate_asn1c.asnproc.GLOBAL.MOD)

    return res





def repr_modname(asnobj: pa_asnobj.ASN1Obj) -> str:

    """Return name of `asnobj` as MODULE.NAME or NAME."""

    result = asnobj._name
@@ -149,38 +121,64 @@ def repr_type_tagpath(asnobj: pa_asnobj.ASN1Obj) -> str: 
    return result





def container_to_tag_dict(ctx: Context, container: pa_asnobj.ASN1Dict) -> dict:

class CompatCheck:

    """Check backwards compatibility between two ASN.1 modules."""



    def __init__(self):

        self._clear()



    def _clear(self) -> None:

        """Reset to a fresh state."""

        self._errors = list()

        self._fa_mod = None

        self._fb_mod = None

        self._seen = set()



    def _process_error(self, message: str) -> None:

        """Log `message` and append `message` to `self._errors`."""

        logging.info(f"Test Failure: {message}")

        self._errors.append(message)



    def _parse_asn1(self, asn_files: list[str]) -> pa_asnobj.ASN1Dict:

        """Parse `asn_files` as ASN.1 using pycrate, returning the module object."""

        logging.info(f"Parsing {' '.join(asn_files)}")

        asn_text = []

        for filename in asn_files:

            filepath = pathlib.Path(filename)

            asn_text.append(filepath.read_text())

        pycrate_asn1c.asnproc.GLOBAL.clear()

        pycrate_asn1c.asnproc.compile_text(asn_text, filenames=asn_files)

        res = copy.deepcopy(pycrate_asn1c.asnproc.GLOBAL.MOD)

        return res



    def _container_to_tag_dict(self, container: pa_asnobj.ASN1Dict) -> dict:

        """Convert `container` to dict indexed on tag, value is type."""

        result = {}

        for name, asntype in container.items():

            tagtuple = asntype._tag

            if tagtuple is None:

            process_error(

                ctx, f"{repr_type_name(container)} field '{name}' doesn't have a tag"

                self._process_error(

                    f"{repr_type_name(container)} field '{name}' doesn't have a tag",

                )

                continue

            result[tagtuple[0]] = asntype

        return result





def types_already_seen(

    ctx: Context, fa_type: pa_asnobj.ASN1Obj, fb_type: pa_asnobj.ASN1Obj

    def _types_already_seen(

        self, fa_type: pa_asnobj.ASN1Obj, fb_type: pa_asnobj.ASN1Obj

    ) -> bool:

        """Return True if fa_type and fb_type have already been seen."""

        fa_repr = repr_type_tagpath(fa_type)

        fb_repr = repr_type_tagpath(fb_type)

        seenkey = f"{fa_repr} and {fb_repr}"

        logging.info(f"Seen check {seenkey}")

    if seenkey in ctx.seen:

        if seenkey in self._seen:

            logging.info(f"Skipping, already seen {seenkey}")

            return True

    ctx.seen.add(seenkey)

        self._seen.add(seenkey)

        return False





def compare_enum(

    ctx: Context, fa_enum: pa_asnobj.ENUM, fb_enum: pa_asnobj.ENUM

) -> None:

    def _compare_enum(self, fa_enum: pa_asnobj.ENUM, fb_enum: pa_asnobj.ENUM) -> None:

        """Compare ASN.1 ENUMERATED `fa_enum` and `fb_enum`."""

        assert fa_enum.TYPE == pa_asnobj.TYPE_ENUM

        assert fb_enum.TYPE == pa_asnobj.TYPE_ENUM
@@ -189,7 +187,7 @@ def compare_enum( 
        fb_desc = repr_type_name(fb_enum)

        logging.info(f"Comparing enum {fa_desc} with {fb_desc}")



    if types_already_seen(ctx, fa_enum, fb_enum):

        if self._types_already_seen(fa_enum, fb_enum):

            return



        fa_tags = dict_swap(fa_enum.get_cont())
@@ -198,21 +196,18 @@ def compare_enum( 
        for tag, ca_name in fa_tags.items():

            cb_name = fb_tags.get(tag, None)

            if cb_name is None:

            process_error(

                ctx,

                self._process_error(

                    f"{fa_desc} missing enum '{ca_name}({tag})'",

                )

                continue  # tag not in fb_enum, stop checking



            if ca_name != cb_name:

            process_error(

                ctx,

                self._process_error(

                    f"{fa_desc}.{ca_name}({tag}) renamed to '{cb_name}({tag})'",

                )





def compare_container(

    ctx: Context, fa_type: pa_asnobj.ASN1Obj, fb_type: pa_asnobj.ASN1Obj

    def _compare_container(

        self, fa_type: pa_asnobj.ASN1Obj, fb_type: pa_asnobj.ASN1Obj

    ) -> None:

        """Compare ASN.1 containers `fa_type` and `fb_type`."""

        assert fa_type.TYPE in pa_asnobj.TYPE_CONSTRUCT
@@ -222,11 +217,11 @@ def compare_container( 
        fb_desc = repr_type_name(fb_type)

        logging.info(f"Comparing container {fa_desc} with {fb_desc}")



    if types_already_seen(ctx, fa_type, fb_type):

        if self._types_already_seen(fa_type, fb_type):

            return



    fa_tags = container_to_tag_dict(ctx=ctx, container=fa_type.get_cont())

    fb_tags = container_to_tag_dict(ctx=ctx, container=fb_type.get_cont())

        fa_tags = self._container_to_tag_dict(container=fa_type.get_cont())

        fb_tags = self._container_to_tag_dict(container=fb_type.get_cont())



        for tag, ca_child in fa_tags.items():

            ca_name = ca_child._name
@@ -234,8 +229,7 @@ def compare_container( 


            cb_child = fb_tags.get(tag, None)

            if cb_child is None:

            process_error(

                ctx,

                self._process_error(

                    f"{fa_desc} missing field '{ca_name}[{tag}]'",

                )

                continue  # tag not in fb_type, stop checking
@@ -244,18 +238,16 @@ def compare_container( 
            cb_typerepr = repr_type(cb_child)



            if ca_name != cb_name:

            process_error(

                ctx,

                self._process_error(

                    f"{fa_desc}.{ca_name}[{tag}] renamed to '{cb_name}[{tag}]'",

                )



            if ca_typerepr != cb_typerepr:

            process_error(

                ctx,

                self._process_error(

                    f"{fa_desc}.{ca_name}[{tag}] type '{ca_typerepr}' changed to '{cb_typerepr}'",

                )



        compare_asntype(ctx=ctx, fa_type=ca_child, fb_type=cb_child)

            self._compare_asntype(fa_type=ca_child, fb_type=cb_child)



            # check new SEQUENCE and SET tags are OPTIONAL

        if fb_type.TYPE not in (pa_asnobj.TYPE_SEQ, pa_asnobj.TYPE_SET):
@@ -266,14 +258,12 @@ def compare_container( 
            logging.info(f"Container {fb_desc} has new tags {set(fb_newtags.keys())}")

        for tag, child in fb_newtags.items():

            if child._flag is None or pa_asnobj.FLAG_OPT not in child._flag:

            process_error(

                ctx,

                self._process_error(

                    f"{fb_desc} new field '{child._name}[{tag}]' is not OPTIONAL",

                )





def compare_container_of(

    ctx: Context, fa_type: pa_asnobj.ASN1Obj, fb_type: pa_asnobj.ASN1Obj

    def _compare_container_of(

        self, fa_type: pa_asnobj.ASN1Obj, fb_type: pa_asnobj.ASN1Obj

    ) -> None:

        """Compare ASN.1 containers SEQ OF or SET OF `fa_type` and `fb_type`."""

        assert fa_type.TYPE in TYPE_CONTAINER_OF
@@ -285,17 +275,16 @@ def compare_container_of( 


        # TODO: .name may end with "._item_"; removesuffix?



    if types_already_seen(ctx, fa_type, fb_type):

        if self._types_already_seen(fa_type, fb_type):

            return



        fa_of = fa_type.get_cont()

        fb_of = fa_type.get_cont()



    compare_asntype(ctx=ctx, fa_type=fa_of, fb_type=fb_of)



        self._compare_asntype(fa_type=fa_of, fb_type=fb_of)



def compare_asntype(

    ctx: Context, fa_type: pa_asnobj.ASN1Obj, fb_type: pa_asnobj.ASN1Obj

    def _compare_asntype(

        self, fa_type: pa_asnobj.ASN1Obj, fb_type: pa_asnobj.ASN1Obj

    ) -> None:

        """Compare ASN.1 types `fa_type` and `fb_type`."""

        fa_desc = repr_type_name(fa_type)
@@ -303,8 +292,7 @@ def compare_asntype( 
        logging.info(f"Comparing {fa_desc} with {fb_desc}")



        if fa_type.TYPE != fb_type.TYPE:

        process_error(

            ctx,

            self._process_error(

                f"{fa_desc} type '{fa_type.TYPE}' mismatch with {fb_desc} '{fb_type.TYPE}'",

            )

            # No further processing of this type can be performed
@@ -317,22 +305,20 @@ def compare_asntype( 
            fb_const = [c["text"] for c in fb_type.get_const()]



            if fa_const != fb_const:

            process_error(

                ctx,

                self._process_error(

                    f"{fa_desc} constraint '{','.join(fa_const)}' changed to '{','.join(fb_const)}'",

                )



        if fa_type.TYPE is pa_asnobj.TYPE_ENUM:

        compare_enum(ctx=ctx, fa_enum=fa_type, fb_enum=fb_type)

            self._compare_enum(fa_enum=fa_type, fb_enum=fb_type)

        elif fa_type.TYPE in TYPE_CONTAINER_OF:

        compare_container_of(ctx=ctx, fa_type=fa_type, fb_type=fb_type)

            self._compare_container_of(fa_type=fa_type, fb_type=fb_type)

        elif fa_type.TYPE in pa_asnobj.TYPE_CONSTRUCT:

        compare_container(ctx=ctx, fa_type=fa_type, fb_type=fb_type)

            self._compare_container(fa_type=fa_type, fb_type=fb_type)

        else:

            logging.debug(f"Unimplemented check of type '{fa_type.TYPE}'")





def compare_files(ctx: Context, a_files: list[str], b_files: list[str]) -> None:

    def compare_files(self, a_files: list[str], b_files: list[str]) -> list(str):

        """Compare ASN.1 in files `a_files` with files `b_files` for

        backwards compatibility issues:

        - Module is the same
@@ -344,34 +330,38 @@ def compare_files(ctx: Context, a_files: list[str], b_files: list[str]) -> None: 
               - Same inner type at the tag.

               - New tags in SEQUENCE and SET are OPTIONAL.



    Updates ctx.errors with a list of error messages (if any).

        Returns a list of error messages (if any).

        """

        logging.info(f"Comparing file {a_files[0]} with {b_files[0]}")



    ctx.fa_mod = parse_asn1(a_files)

    ctx.fb_mod = parse_asn1(b_files)

        self._clear()



    for module_name, fa_module in ctx.fa_mod.items():

        self._fa_mod = self._parse_asn1(a_files)

        self._fb_mod = self._parse_asn1(b_files)



        for module_name, fa_module in self._fa_mod.items():

            if module_name.startswith("_"):

                # Skip pycrate internal keys

                continue



            logging.info(f"Checking module {module_name}")

        if module_name not in ctx.fb_mod:

            process_error(ctx, f"Module {module_name} not present")

            if module_name not in self._fb_mod:

                self._process_error(f"Module {module_name} not present")

                continue

        fb_module = ctx.fb_mod[module_name]

            fb_module = self._fb_mod[module_name]



            for type_name in fa_module["_type_"]:

                fa_type = fa_module[type_name]

                logging.info(f"Checking type {repr_type_name(fa_type)}")



                if type_name not in fb_module:

                process_error(ctx, f"Type {type_name} not present")

                    self._process_error(f"Type {type_name} not present")

                    continue

                fb_type = fb_module[type_name]



            compare_asntype(ctx=ctx, fa_type=fa_type, fb_type=fb_type)

                self._compare_asntype(fa_type=fa_type, fb_type=fb_type)



        return self._errors





def main():
@@ -387,15 +377,15 @@ def main(): 
    a_files.extend(sys.argv[3:])

    b_files.extend(sys.argv[3:])



    ctx = Context()

    compare_files(ctx=ctx, a_files=a_files, b_files=b_files)

    if ctx.errors:

    compatCheck = CompatCheck()

    compatErrors = compatCheck.compare_files(a_files=a_files, b_files=b_files)

    if compatErrors:

        print("-----------------------------")

        print(f"File 1:       {a_files[0]}")

        print(f"File 2:       {b_files[0]}")

        print(f"Dependencies: {' '.join(a_files[1:])}")

        print("Errors:")

        for error in ctx.errors:

        for error in compatErrors:

            print(f"  {error}")

        print("-----------------------------")

        sys.exit(1)