author Jérémy Bobbio <>
Wed, 10 Jul 2019 14:58:41 +0200
changeset 12690 d1b5fd6129bd
parent 12291 b640ef6ad633
child 12302 0d474f888f4a
permissions -rw-r--r--
[pkg] Add new autopkgtest for skeleton packaging When running `cubicweb-ctl newcube`, the generated template provides debian packaging. So let's add automated tests for that using autopkgtest. The new test will generated a new cube, build a source tarball, build the package, try to install it, see if the Python 2 and 3 modules are available and even run autopkgtest against the newly created packages. Along the way it will also print information about the generated package: buildinfo, package content, and lintian report.

# copyright 2015 LOGILAB S.A. (Paris, FRANCE), all rights reserved.
# contact --
# This file is part of CubicWeb.
# CubicWeb is free software: you can redistribute it and/or modify it under the
# terms of the GNU Lesser General Public License as published by the Free
# Software Foundation, either version 2.1 of the License, or (at your option)
# any later version.
# CubicWeb is distributed in the hope that it will be useful, but WITHOUT
# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
# FOR A PARTICULAR PURPOSE.  See the GNU Lesser General Public License for more
# details.
# You should have received a copy of the GNU Lesser General Public License along
# with CubicWeb.  If not, see <>.

"""Simple statsd_ logger for cubicweb.

This module is meant to be configured by setting a couple of global variables:

- ``bucket`` global variable will be used as statsd bucket in every
statsd_ UDP sent packet.

`- `address`` is a pair (IP, port) specifying the address of the
statsd_ server

There are 3 kinds of statds_ message::

- ``statsd_c(context, n)`` is a simple function to send statsd_
  counter-type of messages like::


- ``statsd_g(context, value)`` to send statsd_ gauge-type of messages


- ``statsd_t(context, ms)`` to send statsd_ time-type of messages


There is also a decorator (``statsd_timeit``) that may be used to
measure and send to the statsd_ server the time passed in a function
or a method and the number of calls. It will send a message like::


.. _statsd:


import time
import socket

_bucket = 'cubicweb'
_address = None
_socket = None

def setup(bucket, address):
    """Configure the statsd endpoint

    :param bucket: the name of the statsd bucket that will be used to
                   build messages.

    :param address: the UDP endpoint of the statsd server. Must a
                    couple (ip, port).
    global _bucket, _address, _socket
    packed = None
    for family in (socket.AF_INET6, socket.AF_INET):
            packed = socket.inet_pton(family, address[0])
        except socket.error:
    if packed is None:
    _bucket, _address = bucket, address
    _socket = socket.socket(family, socket.SOCK_DGRAM)

def statsd_c(context, n=1):
    if _address is not None:
        _socket.sendto('{0}.{1}:{2}|c\n'.format(_bucket, context, n), _address)

def statsd_g(context, value):
    if _address is not None:
        _socket.sendto('{0}.{1}:{2}|g\n'.format(_bucket, context, value), _address)

def statsd_t(context, value):
    if _address is not None:
        _socket.sendto('{0}.{1}:{2:.4f}|ms\n'.format(_bucket, context, value), _address)

class statsd_timeit(object):
    __slots__ = ('callable',)

    def __init__(self, callableobj):
        self.callable = callableobj

    def __doc__(self):
        return self.callable.__doc__

    def __name__(self):
        return self.callable.__name__

    def __call__(self, *args, **kw):
        if _address is None:
            return self.callable(*args, **kw)
        t0 = time.time()
            return self.callable(*args, **kw)
            dt = 1000 * (time.time() - t0)
            msg = '{0}.{1}:{2:.4f}|ms\n{0}.{1}:1|c\n'.format(
                _bucket, self.__name__, dt)
            _socket.sendto(msg, _address)

    def __get__(self, obj, objtype):
        """Support instance methods."""
        if obj is None:  # class method or some already wrapped method
            return self
        import functools
        return functools.partial(self.__call__, obj)