mini_buildd.misc module

mini_buildd.misc.open_utf8(path, mode='r', **kwargs)
mini_buildd.misc.check_multiprocessing()

Multiprocessing needs shared memory. This may be use to check for misconfigured shm early for better error handling.

class mini_buildd.misc.API

Bases: object

Helper class to implement an API check.

Inheriting classes must define an __API__ class attribute that should be increased on incompatible changes, and may then check via api_check() method.

api_check()
class mini_buildd.misc.Status(stati)

Bases: object

Helper class to implement an internal status.

Inheriting classes must give a stati dict to init.

status
status_desc
set_status(status, desc='')

Set status with optional description.

get_status()

Get raw (integer) status.

mini_buildd.misc.skip_if_keep_in_debug(func, *args, **kwargs)
class mini_buildd.misc.TmpDir(tmpdir=None)

Bases: object

Use with contextlib.closing() to guarantee tmpdir is purged afterwards.

close()
tmpdir
classmethod file_dir(file_name)
class mini_buildd.misc.ConfFile(file_path, snippet='', comment='#')

Bases: object

ConfFile generation helper.

>>> ConfFile("/tmp/mini_buildd_test_conf_file", "my_option=7").add("my_2nd_option=42").save()
add(snippet)
save()
class mini_buildd.misc.BlockQueue(maxsize)

Bases: queue.Queue

Wrapper around Queue to get put() block until <= maxsize tasks are actually done. In Queue.Queue, task_done() is only used together with join().

This way can use the Queue directly to limit the number of actually worked-on items for incoming and builds.

load
put(item, **kwargs)

Put an item into the queue.

If optional args ‘block’ is true and ‘timeout’ is None (the default), block if necessary until a free slot is available. If ‘timeout’ is a non-negative number, it blocks at most ‘timeout’ seconds and raises the Full exception if no free slot was available within that time. Otherwise (‘block’ is false), put an item on the queue if a free slot is immediately available, else raise the Full exception (‘timeout’ is ignored in that case).

task_done()

Indicate that a formerly enqueued task is complete.

Used by Queue consumer threads. For each get() used to fetch a task, a subsequent call to task_done() tells the queue that the processing on the task is complete.

If a join() is currently blocking, it will resume when all items have been processed (meaning that a task_done() call was received for every item that had been put() into the queue).

Raises a ValueError if called more times than there were items placed in the queue.

class mini_buildd.misc.HoPo(bind)

Bases: object

Convenience class to parse bind string “hostname:port”

test_bind()

Check that we can bind to the first found addrinfo (not already bound, permissions).

mini_buildd.misc.nop(*_args, **_kwargs)
mini_buildd.misc.dont_care_run(func, *args, **kwargs)
mini_buildd.misc.codename_has_lintian_suppress(codename)

Test if the distribution (identified by the codename) has a recent lintian with the ‘–suppress-tags’ option.

class mini_buildd.misc.Distribution(dist)

Bases: object

A mini-buildd distribution string.

Normal distribution:

>>> d = Distribution("squeeze-test-stable")
>>> d.codename, d.repository, d.suite
('squeeze', 'test', 'stable')
>>> d.get()
'squeeze-test-stable'

Rollback distribution:

>>> d = Distribution("squeeze-test-stable-rollback5")
>>> d.is_rollback
True
>>> d.codename, d.repository, d.suite, d.rollback
('squeeze', 'test', 'stable', 'rollback5')
>>> d.get()
'squeeze-test-stable-rollback5'
>>> d.rollback_no
5

Malformed distributions:

>>> Distribution("-squeeze-stable")
Traceback (most recent call last):
...
Exception: Malformed distribution '-squeeze-stable': Must be '<codename>-<repoid>-<suite>[-rollback<n>]'
>>> Distribution("squeeze--stable")
Traceback (most recent call last):
...
Exception: Malformed distribution 'squeeze--stable': Must be '<codename>-<repoid>-<suite>[-rollback<n>]'
>>> Distribution("squeeze-test-stable-")
Traceback (most recent call last):
...
Exception: Malformed distribution 'squeeze-test-stable-': Must be '<codename>-<repoid>-<suite>[-rollback<n>]'
>>> Distribution("squeeze-test-stable-rollback")
Traceback (most recent call last):
...
Exception: Malformed distribution 'squeeze-test-stable-rollback': Must be '<codename>-<repoid>-<suite>[-rollback<n>]'
>>> Distribution("squeeze-test-stable-rolback0")
Traceback (most recent call last):
...
Exception: Malformed distribution 'squeeze-test-stable-rolback0': Must be '<codename>-<repoid>-<suite>[-rollback<n>]'
get(rollback=True)
codename
repository
suite
is_rollback
rollback
rollback_no

Rollback (int) number: ‘rollback0’ -> 0

has_lintian_suppress()
mini_buildd.misc.strip_epoch(version)

Strip the epoch from a version string.

mini_buildd.misc.guess_codeversion(release)

Guess the ‘codeversion’ aka the first two digits of a Debian release version; for releases without version, this falls back to the uppercase codename.

In Debian,
  • point release <= sarge had the ‘M.PrN’ syntax (with 3.1 being a major release).
  • point release in squeeze used ‘M.0.N’ syntax.
  • point releases for >= wheezy have the ‘M.N’ syntax (with 7.1 being a point release).
  • testing and unstable do not gave a version in Release and fall back to uppercase codename

Ubuntu just uses YY.MM which we can use as-is.

>>> guess_codeversion({"Origin": "Debian", "Version": "3.1r8", "Codename": "sarge"})
'31'
>>> guess_codeversion({"Origin": "Debian", "Version": "4.0r9", "Codename": "etch"})
'40'
>>> guess_codeversion({"Origin": "Debian", "Version": "6.0.6", "Codename": "squeeze"})
'60'
>>> guess_codeversion({"Origin": "Debian", "Version": "7.0", "Codename": "wheezy"})
'7'
>>> guess_codeversion({"Origin": "Debian", "Version": "7.1", "Codename": "wheezy"})
'7'
>>> guess_codeversion({"Origin": "Debian", "Codename": "jessie"})
'~JESSIE'
>>> guess_codeversion({"Origin": "Debian", "Codename": "sid"})
'~SID'
>>> guess_codeversion({"Origin": "Ubuntu", "Version": "12.10", "Codename": "quantal"})
'1210'
mini_buildd.misc.guess_default_dirchroot_backend(overlay, aufs)
mini_buildd.misc.chroot_libdir_path(codename, architecture)
mini_buildd.misc.pkg_fmt(status, distribution, package, version, extra=None, message=None)

Generate a package status line.

class mini_buildd.misc.PkgLog(repository, installed, package, version)

Bases: object

classmethod get_path(repository, installed, package, version=None, architecture=None, relative=False)
classmethod make_relative(path)
mini_buildd.misc.subst_placeholders(template, placeholders)

Substitue placeholders in string from a dict.

>>> subst_placeholders("Repoversionstring: %IDENTITY%%CODEVERSION%", { "IDENTITY": "test", "CODEVERSION": "60" })
'Repoversionstring: test60'
mini_buildd.misc.fromdos(string)
mini_buildd.misc.run_as_thread(thread_func, name, daemon=False, **kwargs)
mini_buildd.misc.hash_of_file(file_name, hash_type='md5')

Helper to get any hash from file contents.

mini_buildd.misc.md5_of_file(file_name)
mini_buildd.misc.sha1_of_file(file_name)
mini_buildd.misc.u2b64(unicode_string)

Convert unicode string to base46.

>>> b64 = u2b64("Ünicode strüng")
>>> b64.__class__.__name__
'bytes'
>>> b64
b'w5xuaWNvZGUgc3Ryw7xuZw=='
mini_buildd.misc.b642u(base64_bytestream)

Convert base64 string (in mini_buildd.setup.CHAR_ENCODING) to unicode.

>>> u = b642u('w5xuaWNvZGUgc3Ryw7xuZw==')
>>> u.__class__.__name__
'str'
>>> print(u)
Ünicode strüng
mini_buildd.misc.get_cpus()
mini_buildd.misc.list_get(list_, index, default=None)
mini_buildd.misc.rmdirs(path)

Remove path recursively. Succeed even if it does not exist in the first place.

class mini_buildd.misc.UserURL(url, username=None)

Bases: object

URL with a username attached.

>>> U = UserURL("http://admin@localhost:8066")
>>> (U.username, U.plain, U.full)
('admin', 'http://localhost:8066', 'http://admin@localhost:8066')
>>> U = UserURL("http://example.org:8066", "admin")
>>> (U.username, U.plain, U.full)
('admin', 'http://example.org:8066', 'http://admin@example.org:8066')
>>> UserURL("http://localhost:8066")
Traceback (most recent call last):
  ...
Exception: UserURL: No username given
>>> UserURL("http://admin@localhost:8066", "root")
Traceback (most recent call last):
  ...
Exception: UserURL: Username given in twice, in URL and parameter
username
plain

URL string without username.

full

URL string with username.

mini_buildd.misc.qualname(obj)
class mini_buildd.misc.Keyring(service)

Bases: object

reset_save_policy()
set(key, password)
get(host, user='')
mini_buildd.misc.urlopen_ca_certificates(url)

Wrapper for urlib2.urlopen, optionally using certificates from ca-certificates package, when installed. (See https://bugs.debian.org/cgi-bin/bugreport.cgi?bug=832350).

mini_buildd.misc.canonize_url(url)

Poor man’s URL canonizer: Always include the port (currently only works for ‘http’ and ‘https’ default ports).

mini_buildd.misc.web_login(host, user, credentials, proto='http', login_loc='/accounts/login/', next_loc='/mini_buildd/')
mini_buildd.misc.clone_log(dst, src='mini_buildd')

Setup logger named ‘dst’ with the same handlers and loglevel as the logger named ‘src’.

mini_buildd.misc.setup_console_logging(level=10)