def _get_obsolete_pushkey_message()

in pylib/vcsreplicator/vcsreplicator/pushnotifications.py [0:0]

• 69 lines of code
• 19 McCabe index (conditional complexity)

def _get_obsolete_pushkey_message(local_path, public_url, rawdata):
    logger.warn("processing obsolete pushkey message for %s" % public_url)

    # ASSERTION: vcsreplicator extension loaded in system/user config.
    with hglib.open(local_path, encoding="utf-8") as hgclient:
        out = hgclient.rawcommand([b"debugbase85obsmarkers", pycompat.bytestr(rawdata)])
        markers = json.loads(out)
        logger.warn("processing %d obsolete markers" % len(markers))

        def rev_info(node):
            template = b"{node}\\0{desc}\\0{pushid}"
            args = hglib.util.cmdbuilder(b"log", hidden=True, r=node, template=template)
            # Mercurial will abort with "unknown revision" if you give it
            # 40 character hash that isn't known.
            try:
                out = hgclient.rawcommand(args)
                return out.split(b"\0")
            except hglib.error.CommandError as e:
                if b"unknown revision" in e.err:
                    return None
                else:
                    raise

        def node_payload(node):
            assert len(node) == 40
            if isinstance(node, pycompat.unicode):
                node = node.encode("latin1")

            rev = rev_info(node)

            # Determine if changeset is visible/hidden..
            if rev:
                args = hglib.util.cmdbuilder(b"log", r=node, template=b"{node}")
                try:
                    out = hgclient.rawcommand(args)
                    visible = bool(out.strip())
                except hglib.error.CommandError as e:
                    if b"hidden revision" in e.err:
                        visible = False
                    else:
                        visible = None
            else:
                visible = None

            # Obtain pushlog entry for this node.
            if rev and rev[2]:
                pushes = _get_pushlog_info(hgclient, public_url, [node])
                if pushes:
                    push = pushes[int(rev[2])]
                else:
                    push = None
            else:
                push = None

            return {
                "node": pycompat.sysstr(node),
                "known": bool(rev),
                "visible": visible,
                "desc": pycompat.sysstr(rev[1]) if rev else None,
                "push": push,
            }

        data = []

        for marker in markers:
            # We collect data about the new and old changesets (if available)
            # because the repo may not expose information on hidden
            # changesets to public consumers.
            precursor = node_payload(marker["precursor"])
            successors = [node_payload(node) for node in marker["successors"]]

            user = None
            for m in marker["metadata"]:
                if m[0] == "user":
                    user = pycompat.sysstr(m[1])

            data.append(
                {
                    "precursor": precursor,
                    "successors": successors,
                    "user": user,
                    "time": float(marker["date"][0]),
                }
            )

    return "obsolete.1", {
        "repo_url": public_url,
        "markers": data,
    }