# debugcommands.py - command processing for debug* commands

#

# Copyright 2005-2016 Olivia Mackall <olivia@selenic.com>

#

# This software may be used and distributed according to the terms of the

# GNU General Public License version 2 or any later version.

from __future__ import annotations

import binascii

import codecs

import collections

import contextlib

import difflib

import errno

import glob

import operator

import os

import platform

import random

import re

import socket

import ssl

import stat

import subprocess

import sys

import time

from .i18n import _

from .node import (

    bin,

    hex,

    nullrev,

    short,

)

from .pycompat import (

    open,

)

from . import (

    bundle2,

    bundlerepo,

    changegroup,

    cmdutil,

    color,

    context,

    copies,

    dagparser,

    dirstateutils,

    encoding,

    error,

    exchange,

    extensions,

    filelog,

    filemerge,

    filesetlang,

    formatter,

    hg,

    httppeer,

    localrepo,

    lock as lockmod,

    logcmdutil,

    manifest,

    mergestate as mergestatemod,

    metadata,

    obsolete,

    obsutil,

    pathutil,

    phases,

    policy,

    pvec,

    pycompat,

    registrar,

    repair,

    repoview,

    requirements,

    revlog,

    revset,

    revsetlang,

    scmutil,

    setdiscovery,

    simplemerge,

    sshpeer,

    sslutil,

    streamclone,

    strip,

    tags as tagsmod,

    templater,

    treediscovery,

    upgrade,

    url as urlmod,

    util,

    verify,

    vfs as vfsmod,

    wireprotoframing,

    wireprotoserver,

)

from .interfaces import repository

from .stabletailgraph import stabletailsort

from .utils import (

    cborutil,

    compression,

    dateutil,

    procutil,

    stringutil,

    urlutil,

)

from .revlogutils import (

    debug as revlog_debug,

    nodemap,

    rewrite,

    sidedata,

)

release = lockmod.release

table = {}

table.update(strip.command._table)

command = registrar.command(table)

@command(b'debugancestor', [], _(b'[INDEX] REV1 REV2'), optionalrepo=True)

def debugancestor(ui, repo, *args):

    """find the ancestor revision of two revisions in a given index"""

    if len(args) == 3:

        index, rev1, rev2 = args

        r = revlog.revlog(vfsmod.vfs(encoding.getcwd(), audit=False), index)

        lookup = r.lookup

    elif len(args) == 2:

        if not repo:

            raise error.Abort(

                _(b'there is no Mercurial repository here (.hg not found)')

            )

        rev1, rev2 = args

        r = repo.changelog

        lookup = repo.lookup

    else:

        raise error.Abort(_(b'either two or three arguments required'))

    a = r.ancestor(lookup(rev1), lookup(rev2))

    ui.write(b'%d:%s\n' % (r.rev(a), hex(a)))

@command(b'debugantivirusrunning', [])

def debugantivirusrunning(ui, repo):

    """attempt to trigger an antivirus scanner to see if one is active"""

    with repo.cachevfs.open(b'eicar-test-file.com', b'wb') as f:

        f.write(

            util.b85decode(

                # This is a base85-armored version of the EICAR test file. See

                # https://en.wikipedia.org/wiki/EICAR_test_file for details.

                b'ST#=}P$fV?P+K%yP+C|uG$>GBDK|qyDK~v2MM*<JQY}+dK~6+LQba95P'

                b'E<)&Nm5l)EmTEQR4qnHOhq9iNGnJx'

            )

        )

    # Give an AV engine time to scan the file.

    time.sleep(2)

    util.unlink(repo.cachevfs.join(b'eicar-test-file.com'))

@command(b'debugapplystreamclonebundle', [], b'FILE')

def debugapplystreamclonebundle(ui, repo, fname):

    """apply a stream clone bundle file"""

    f = hg.openpath(ui, fname)

    gen = exchange.readbundle(ui, f, fname)

    gen.apply(repo)

@command(

    b'debugbuilddag',

    [

        (

            b'm',

            b'mergeable-file',

            None,

            _(b'add single file mergeable changes'),

        ),

        (

            b'o',

            b'overwritten-file',

            None,

            _(b'add single file all revs overwrite'),

        ),

        (b'n', b'new-file', None, _(b'add new file at each rev')),

        (

            b'',

            b'from-existing',

            None,

            _(b'continue from a non-empty repository'),

        ),

    ],

    _(b'[OPTION]... [TEXT]'),

)

def debugbuilddag(

    ui,

    repo,

    text=None,

    mergeable_file=False,

    overwritten_file=False,

    new_file=False,

    from_existing=False,

):

    """builds a repo with a given DAG from scratch in the current empty repo

    The description of the DAG is read from stdin if not given on the

    command line.

    Elements:

     - "+n" is a linear run of n nodes based on the current default parent

     - "." is a single node based on the current default parent

     - "$" resets the default parent to null (implied at the start);

           otherwise the default parent is always the last node created

     - "<p" sets the default parent to the backref p

     - "*p" is a fork at parent p, which is a backref

     - "*p1/p2" is a merge of parents p1 and p2, which are backrefs

     - "/p2" is a merge of the preceding node and p2

     - ":tag" defines a local tag for the preceding node

     - "@branch" sets the named branch for subsequent nodes

     - "#...\\n" is a comment up to the end of the line

    Whitespace between the above elements is ignored.

    A backref is either

     - a number n, which references the node curr-n, where curr is the current

       node, or

     - the name of a local tag you placed earlier using ":tag", or

     - empty to denote the default parent.

    All string valued-elements are either strictly alphanumeric, or must

    be enclosed in double quotes ("..."), with "\\" as escape character.

    """

    if text is None:

        ui.status(_(b"reading DAG from stdin\n"))

        text = ui.fin.read()

    cl = repo.changelog

    if len(cl) > 0 and not from_existing:

        raise error.Abort(_(b'repository is not empty'))

    # determine number of revs in DAG

    total = 0

    for type, data in dagparser.parsedag(text):

        if type == b'n':

            total += 1

    if mergeable_file:

        linesperrev = 2

        # make a file with k lines per rev

        initialmergedlines = [b'%d' % i for i in range(0, total * linesperrev)]

        initialmergedlines.append(b"")

    tags = []

    progress = ui.makeprogress(

        _(b'building'), unit=_(b'revisions'), total=total

    )

    with progress, repo.wlock(), repo.lock(), repo.transaction(b"builddag"):

        at = -1

        atbranch = b'default'

        nodeids = []

        id = 0

        progress.update(id)

        for type, data in dagparser.parsedag(text):

            if type == b'n':

                ui.note((b'node %s\n' % pycompat.bytestr(data)))

                id, ps = data

                files = []

                filecontent = {}

                p2 = None

                if mergeable_file:

                    fn = b"mf"

                    p1 = repo[ps[0]]

                    if len(ps) > 1:

                        p2 = repo[ps[1]]

                        pa = p1.ancestor(p2)

                        base, local, other = [

                            x[fn].data() for x in (pa, p1, p2)

                        ]

                        m3 = simplemerge.Merge3Text(base, local, other)

                        ml = [

                            l.strip()

                            for l in simplemerge.render_minimized(m3)[0]

                        ]

                        ml.append(b"")

                    elif at > 0:

                        ml = p1[fn].data().split(b"\n")

                    else:

                        ml = initialmergedlines

                    ml[id * linesperrev] += b" r%i" % id

                    mergedtext = b"\n".join(ml)

                    files.append(fn)

                    filecontent[fn] = mergedtext

                if overwritten_file:

                    fn = b"of"

                    files.append(fn)

                    filecontent[fn] = b"r%i\n" % id

                if new_file:

                    fn = b"nf%i" % id

                    files.append(fn)

                    filecontent[fn] = b"r%i\n" % id

                    if len(ps) > 1:

                        if not p2:

                            p2 = repo[ps[1]]

                        for fn in p2:

                            if fn.startswith(b"nf"):

                                files.append(fn)

                                filecontent[fn] = p2[fn].data()

                def fctxfn(repo, cx, path):

                    if path in filecontent:

                        return context.memfilectx(

                            repo, cx, path, filecontent[path]

                        )

                    return None

                if len(ps) == 0 or ps[0] < 0:

                    pars = [None, None]

                elif len(ps) == 1:

                    pars = [nodeids[ps[0]], None]

                else:

                    pars = [nodeids[p] for p in ps]

                cx = context.memctx(

                    repo,

                    pars,

                    b"r%i" % id,

                    files,

                    fctxfn,

                    date=(id, 0),

                    user=b"debugbuilddag",

                    extra={b'branch': atbranch},

                )

                nodeid = repo.commitctx(cx)

                nodeids.append(nodeid)

                at = id

            elif type == b'l':

                id, name = data

                ui.note((b'tag %s\n' % name))

                tags.append(b"%s %s\n" % (hex(repo.changelog.node(id)), name))

            elif type == b'a':

                ui.note((b'branch %s\n' % data))

                atbranch = data

            progress.update(id)

        if tags:

            repo.vfs.write(b"localtags", b"".join(tags))

def _debugchangegroup(ui, gen, all=None, indent=0, **opts):

    indent_string = b' ' * indent

    if all:

        ui.writenoi18n(

            b"%sformat: id, p1, p2, cset, delta base, len(delta)\n"

            % indent_string

        )

        def showchunks(named):

            ui.write(b"\n%s%s\n" % (indent_string, named))

            for deltadata in gen.deltaiter():

                node, p1, p2, cs, deltabase, delta, flags, sidedata = deltadata

                ui.write(

                    b"%s%s %s %s %s %s %d\n"

                    % (

                        indent_string,

                        hex(node),

                        hex(p1),

                        hex(p2),

                        hex(cs),

                        hex(deltabase),

                        len(delta),

                    )

                )

        gen.changelogheader()

        showchunks(b"changelog")

        gen.manifestheader()

        showchunks(b"manifest")

        for chunkdata in iter(gen.filelogheader, {}):

            fname = chunkdata[b'filename']

            showchunks(fname)

    else:

        if isinstance(gen, bundle2.unbundle20):

            raise error.Abort(_(b'use debugbundle2 for this file'))

        gen.changelogheader()

        for deltadata in gen.deltaiter():

            node, p1, p2, cs, deltabase, delta, flags, sidedata = deltadata

            ui.write(b"%s%s\n" % (indent_string, hex(node)))

def _debugobsmarkers(ui, part, indent=0, **opts):

    """display version and markers contained in 'data'"""

    data = part.read()

    indent_string = b' ' * indent

    try:

        version, markers = obsolete._readmarkers(data)

    except error.UnknownVersion as exc:

        msg = b"%sunsupported version: %s (%d bytes)\n"

        msg %= indent_string, exc.version, len(data)

        ui.write(msg)

    else:

        msg = b"%sversion: %d (%d bytes)\n"

        msg %= indent_string, version, len(data)

        ui.write(msg)

        fm = ui.formatter(b'debugobsolete', pycompat.byteskwargs(opts))

        for rawmarker in sorted(markers):

            m = obsutil.marker(None, rawmarker)

            fm.startitem()

            fm.plain(indent_string)

            cmdutil.showmarker(fm, m)

        fm.end()

def _debugphaseheads(ui, data, indent=0):

    """display version and markers contained in 'data'"""

    indent_string = b' ' * indent

    headsbyphase = phases.binarydecode(data)

    for phase in phases.allphases:

        for head in headsbyphase[phase]:

            ui.write(indent_string)

            ui.write(b'%s %s\n' % (hex(head), phases.phasenames[phase]))

def _quasirepr(thing):

    if isinstance(thing, (dict, util.sortdict, collections.OrderedDict)):

        return b'{%s}' % (

            b', '.join(b'%s: %s' % (k, thing[k]) for k in sorted(thing))

        )

    return pycompat.bytestr(repr(thing))

def _debugbundle2(ui, gen, all=None, **opts):

    """lists the contents of a bundle2"""

    if not isinstance(gen, bundle2.unbundle20):

        raise error.Abort(_(b'not a bundle2 file'))

    ui.write((b'Stream params: %s\n' % _quasirepr(gen.params)))

    parttypes = opts.get('part_type', [])

    for part in gen.iterparts():

        if parttypes and part.type not in parttypes:

            continue

        msg = b'%s -- %s (mandatory: %r)\n'

        ui.write((msg % (part.type, _quasirepr(part.params), part.mandatory)))

        if part.type == b'changegroup':

            version = part.params.get(b'version', b'01')

            cg = changegroup.getunbundler(version, part, b'UN')

            if not ui.quiet:

                _debugchangegroup(ui, cg, all=all, indent=4, **opts)

        if part.type == b'obsmarkers':

            if not ui.quiet:

                _debugobsmarkers(ui, part, indent=4, **opts)

        if part.type == b'phase-heads':

            if not ui.quiet:

                _debugphaseheads(ui, part, indent=4)

@command(

    b'debugbundle',

    [

        (b'a', b'all', None, _(b'show all details')),

        (b'', b'part-type', [], _(b'show only the named part type')),

        (b'', b'spec', None, _(b'print the bundlespec of the bundle')),

    ],

    _(b'FILE'),

    norepo=True,

)

def debugbundle(ui, bundlepath, all=None, spec=None, **opts):

    """lists the contents of a bundle"""

    with hg.openpath(ui, bundlepath) as f:

        if spec:

            spec = exchange.getbundlespec(ui, f)

            ui.write(b'%s\n' % spec)

            return

        gen = exchange.readbundle(ui, f, bundlepath)

        if isinstance(gen, bundle2.unbundle20):

            return _debugbundle2(ui, gen, all=all, **opts)