britney2-ubuntu/tests/__init__.py

# (C) 2015 Canonical Ltd.
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.

import os
import shutil
import subprocess
import tempfile
import unittest

PROJECT_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))

architectures = ['amd64', 'arm64', 'armhf', 'i386', 'powerpc', 'ppc64el', 'riscv64']


class TestData:

    def __init__(self):
        '''Construct local test package indexes.

        The archive is initially empty. You can create new packages with
        create_deb(). self.path contains the path of the archive, and
        self.apt_source provides an apt source "deb" line.

        It is kept in a temporary directory which gets removed when the Archive
        object gets deleted.
        '''
        self.path = tempfile.mkdtemp(prefix='testarchive.')
        self.apt_source = 'deb file://%s /' % self.path
        self.series = 'series'
        self.dirs = {False: os.path.join(self.path, 'data', self.series),
                     True: os.path.join(
                         self.path, 'data', '%s-proposed' % self.series)}
        os.makedirs(self.dirs[False])
        os.mkdir(self.dirs[True])
        self.added_sources = {False: set(), True: set()}
        self.added_binaries = {}
        for arch in architectures:
            self.added_binaries[arch] = {False: set(), True: set()}

        # pre-create all files for all architectures
        for arch in architectures:
            for dir in self.dirs.values():
                with open(os.path.join(dir, 'Packages_' + arch), 'w'):
                    pass
        for dir in self.dirs.values():
            for fname in ['Dates', 'Blocks']:
                with open(os.path.join(dir, fname), 'w'):
                    pass
            for dname in ['Hints']:
                os.mkdir(os.path.join(dir, dname))

        os.mkdir(os.path.join(self.path, 'output'))

        # create temporary home dir for proposed-migration autopktest status
        self.home = os.path.join(self.path, 'home')
        os.environ['HOME'] = self.home
        os.makedirs(os.path.join(self.home, 'proposed-migration',
                                 'autopkgtest', 'work'))

    def __del__(self):
        shutil.rmtree(self.path)

    def add(self, name, unstable, fields={}, add_src=True, testsuite=None, srcfields=None):
        '''Add a binary package to the index file.

        You need to specify at least the package name and in which list to put
        it (unstable==True for unstable/proposed, or False for
        testing/release). fields specifies all additional entries, e. g.
        {'Depends': 'foo, bar', 'Conflicts: baz'}. There are defaults for most
        fields.

        Unless add_src is set to False, this will also automatically create a
        source record, based on fields['Source'] and name. In that case, the
        "Testsuite:" field is set to the testsuite argument.
        '''
        fields.setdefault('Architecture', 'all')
        if fields['Architecture'] == 'all':
            arches = architectures
        else:
            arches = [fields['Architecture']]

        for a in arches:
            assert (name not in self.added_binaries[a][unstable])
            self.added_binaries[a][unstable].add(name)

        fields.setdefault('Version', '1')
        fields.setdefault('Priority', 'optional')
        fields.setdefault('Section', 'devel')
        fields.setdefault('Description', 'test pkg')
        if fields['Architecture'] == 'all':
            for a in architectures:
                self._append(name, unstable, 'Packages_' + a, fields)
        else:
            self._append(name, unstable, 'Packages_' + fields['Architecture'],
                         fields)

        if add_src:
            src = fields.get('Source', name)
            if src not in self.added_sources[unstable]:
                if srcfields is None:
                    srcfields = {}
                srcfields['Version'] = fields['Version']
                srcfields['Section'] = fields['Section']
                if testsuite:
                    srcfields['Testsuite'] = testsuite
                self.add_src(src, unstable, srcfields)

    def add_src(self, name, unstable, fields={}):
        '''Add a source package to the index file.

        You need to specify at least the package name and in which list to put
        it (unstable==True for unstable/proposed, or False for
        testing/release). fields specifies all additional entries, which can be
        Version (default: 1), Section (default: devel), Testsuite (default:
        none), and Extra-Source-Only.
        '''
        assert (name not in self.added_sources[unstable])
        self.added_sources[unstable].add(name)

        fields.setdefault('Version', '1')
        fields.setdefault('Section', 'devel')
        self._append(name, unstable, 'Sources', fields)

    def _append(self, name, unstable, file_name, fields):
        with open(os.path.join(self.dirs[unstable], file_name), 'a') as f:
            f.write('''Package: %s
Maintainer: Joe <joe@example.com>
''' % name)

            for k, v in fields.items():
                f.write('%s: %s\n' % (k, v))
            f.write('\n')

    def remove_all(self, unstable):
        '''Remove all added packages'''

        for arch in self.added_binaries:
            self.added_binaries[arch][unstable] = set()
        self.added_sources[unstable] = set()
        for a in architectures:
            open(os.path.join(self.dirs[unstable], 'Packages_' + a), 'w').close()
        open(os.path.join(self.dirs[unstable], 'Sources'), 'w').close()


class TestBase(unittest.TestCase):

    def setUp(self):
        super(TestBase, self).setUp()
        self.maxDiff = None
        self.data = TestData()
        self.britney = os.path.join(PROJECT_DIR, 'britney.py')
        # create temporary config so that tests can hack it
        self.britney_conf = os.path.join(self.data.path, 'britney.conf')
        shutil.copy(os.path.join(PROJECT_DIR, 'britney.conf'), self.britney_conf)
        assert os.path.exists(self.britney)

    def tearDown(self):
        del self.data

    def run_britney(self, args=[]):
        '''Run britney.

        Assert that it succeeds and does not produce anything on stderr.
        Return (excuses.yaml, excuses.html, britney_out).
        '''
        britney = subprocess.Popen([self.britney, '-v', '-c', self.britney_conf,
                                    '--distribution=ubuntu',
                                    '--series=%s' % self.data.series],
                                   stdout=subprocess.PIPE,
                                   stderr=subprocess.PIPE,
                                   cwd=self.data.path,
                                   universal_newlines=True)
        (out, err) = britney.communicate()
        self.assertEqual(britney.returncode, 0, out + err)
        self.assertEqual(err, '')

        with open(os.path.join(self.data.path, 'output', self.data.series,
                               'excuses.yaml')) as f:
            yaml = f.read()
        with open(os.path.join(self.data.path, 'output', self.data.series,
                               'excuses.html')) as f:
            html = f.read()

        return (yaml, html, out)

    def create_hint(self, username, content):
        '''Create a hint file for the given username and content'''

        hints_path = os.path.join(
            self.data.path, 'data', self.data.series + '-proposed', 'Hints', username)
        with open(hints_path, 'a') as fd:
            fd.write(content)
            fd.write('\n')