astro py:docs

- Index
- Modules

### Navigation

- asdf v »
- Module code »

# Source code for asdf.testing.helpers

    """
    Helpers for writing unit tests of ASDF support.
    """

    from io import BytesIO

    import asdf
    from asdf.versioning import AsdfSpec

    def format_tag(organization, standard, version, tag_name):
        """
        Format a YAML tag.
        """
        tag = f"tag:{organization}:{standard}/{tag_name}"

        if version is None:
            return tag

        if isinstance(version, AsdfSpec):
            version = str(version.spec)

        return f"{tag}-{version}"

    def roundtrip_object(obj, version=None):
        """
        Add the specified object to an AsdfFile's tree, write the file to
        a buffer, then read it back in and return the deserialized object.

        Parameters
        ----------
        obj : object
            Object to serialize.
        version : str or None.
            ASDF Standard version.  If None, use the library's default version.

        Returns
        -------
        object
            The deserialized object.
        """
        buff = BytesIO()
        with asdf.AsdfFile(version=version) as af:
            af["obj"] = obj
            af.write_to(buff)

        buff.seek(0)
        with asdf.open(buff, lazy_load=False, memmap=False) as af:
            return af["obj"]

    def yaml_to_asdf(yaml_content, version=None):
        """
        Given a string of YAML content, adds the extra pre-
        and post-amble to make it an ASDF file.

        Parameters
        ----------
        yaml_content : string or bytes
            YAML content.
        version : str or None.
            ASDF Standard version.  If None, use the library's default version.

        Returns
        -------
        io.BytesIO
            A file-like object containing the ASDF file.
        """
        if isinstance(yaml_content, str):
            yaml_content = yaml_content.encode("utf-8")

        buff = BytesIO()
        with asdf.AsdfFile(version=version) as af:
            af["$REPLACE"] = "ME"
            af.write_to(buff)

        buff.seek(0)
        asdf_content = buff.read().replace(b"$REPLACE: ME", yaml_content)

        return BytesIO(asdf_content)