comparison tests/wireprotosimplecache.py @ 43076:2372284d9457

formatting: blacken the codebase

This is using my patch to black
(https://github.com/psf/black/pull/826) so we don't un-wrap collection
literals.

Done with:

  hg files 'set:**.py - mercurial/thirdparty/** - "contrib/python-zstandard/**"' | xargs black -S

# skip-blame mass-reformatting only

# no-check-commit reformats foo_bar functions

Differential Revision: https://phab.mercurial-scm.org/D6971

43075:57875cf423c9

)
from mercurial.interfaces import (
    repository,
    util as interfaceutil,
)
from mercurial.utils import (
    stringutil,
)

CACHE = None

configtable = {}
configitem = registrar.configitem(configtable)

configitem(b'simplecache', b'cacheapi',
           default=False)
configitem(b'simplecache', b'cacheobjects',
           default=False)
configitem(b'simplecache', b'redirectsfile',
           default=None)

# API handler that makes cached keys available.
def handlecacherequest(rctx, req, res, checkperm, urlparts):
    if rctx.repo.ui.configbool(b'simplecache', b'cacheobjects'):
        res.status = b'500 Internal Server Error'

    res.status = b'200 OK'
    res.headers[b'Content-Type'] = b'application/mercurial-cbor'
    res.setbodybytes(CACHE[key])

def cachedescriptor(req, repo):
    return {}

wireprotoserver.API_HANDLERS[b'simplecache'] = {
    b'config': (b'simplecache', b'cacheapi'),
    b'handler': handlecacherequest,
    b'apidescriptor': cachedescriptor,
}

@interfaceutil.implementer(repository.iwireprotocolcommandcacher)
class memorycacher(object):
    def __init__(self, ui, command, encodefn, redirecttargets, redirecthashes,
                 req):
        self.ui = ui
        self.encodefn = encodefn
        self.redirecttargets = redirecttargets
        self.redirecthashes = redirecthashes
        self.req = req

            paths.append(b'simplecache')
            paths.append(self.key)

            url = b'%s/%s' % (self.req.baseurl, b'/'.join(paths))

            #url = b'http://example.com/%s' % self.key
            self.ui.log(b'simplecache', b'sending content redirect for %s to '
                                        b'%s\n', self.key, url)
            response = wireprototypes.alternatelocationresponse(
                url=url,
                mediatype=b'application/mercurial-cbor')

            return {b'objs': [response]}

        if self.cacheobjects:
            return {

        else:
            CACHE[self.key] = b''.join(self.buffered)

        return []

def makeresponsecacher(orig, repo, proto, command, args, objencoderfn,
                       redirecttargets, redirecthashes):
    return memorycacher(repo.ui, command, objencoderfn, redirecttargets,
                        redirecthashes, proto._req)

def loadredirecttargets(ui):
    path = ui.config(b'simplecache', b'redirectsfile')
    if not path:
        return []

    with open(path, 'rb') as fh:
        s = fh.read()

    return stringutil.evalpythonliteral(s)

def getadvertisedredirecttargets(orig, repo, proto):
    return loadredirecttargets(repo.ui)

def extsetup(ui):
    global CACHE

    CACHE = util.lrucachedict(10000)

    extensions.wrapfunction(wireprotov2server, b'makeresponsecacher',
                            makeresponsecacher)
    extensions.wrapfunction(wireprotov2server, b'getadvertisedredirecttargets',
                            getadvertisedredirecttargets)

43076:2372284d9457

)
from mercurial.interfaces import (
    repository,
    util as interfaceutil,
)
from mercurial.utils import stringutil

CACHE = None

configtable = {}
configitem = registrar.configitem(configtable)

configitem(b'simplecache', b'cacheapi', default=False)
configitem(b'simplecache', b'cacheobjects', default=False)
configitem(b'simplecache', b'redirectsfile', default=None)

# API handler that makes cached keys available.
def handlecacherequest(rctx, req, res, checkperm, urlparts):
    if rctx.repo.ui.configbool(b'simplecache', b'cacheobjects'):
        res.status = b'500 Internal Server Error'

    res.status = b'200 OK'
    res.headers[b'Content-Type'] = b'application/mercurial-cbor'
    res.setbodybytes(CACHE[key])


def cachedescriptor(req, repo):
    return {}


wireprotoserver.API_HANDLERS[b'simplecache'] = {
    b'config': (b'simplecache', b'cacheapi'),
    b'handler': handlecacherequest,
    b'apidescriptor': cachedescriptor,
}


@interfaceutil.implementer(repository.iwireprotocolcommandcacher)
class memorycacher(object):
    def __init__(
        self, ui, command, encodefn, redirecttargets, redirecthashes, req
    ):
        self.ui = ui
        self.encodefn = encodefn
        self.redirecttargets = redirecttargets
        self.redirecthashes = redirecthashes
        self.req = req

            paths.append(b'simplecache')
            paths.append(self.key)

            url = b'%s/%s' % (self.req.baseurl, b'/'.join(paths))

            # url = b'http://example.com/%s' % self.key
            self.ui.log(
                b'simplecache',
                b'sending content redirect for %s to ' b'%s\n',
                self.key,
                url,
            )
            response = wireprototypes.alternatelocationresponse(
                url=url, mediatype=b'application/mercurial-cbor'
            )

            return {b'objs': [response]}

        if self.cacheobjects:
            return {

        else:
            CACHE[self.key] = b''.join(self.buffered)

        return []


def makeresponsecacher(
    orig,
    repo,
    proto,
    command,
    args,
    objencoderfn,
    redirecttargets,
    redirecthashes,
):
    return memorycacher(
        repo.ui,
        command,
        objencoderfn,
        redirecttargets,
        redirecthashes,
        proto._req,
    )


def loadredirecttargets(ui):
    path = ui.config(b'simplecache', b'redirectsfile')
    if not path:
        return []

    with open(path, 'rb') as fh:
        s = fh.read()

    return stringutil.evalpythonliteral(s)


def getadvertisedredirecttargets(orig, repo, proto):
    return loadredirecttargets(repo.ui)


def extsetup(ui):
    global CACHE

    CACHE = util.lrucachedict(10000)

    extensions.wrapfunction(
        wireprotov2server, b'makeresponsecacher', makeresponsecacher
    )
    extensions.wrapfunction(
        wireprotov2server,
        b'getadvertisedredirecttargets',
        getadvertisedredirecttargets,
    )