path: root/tests/gem5/fixture.py

# Copyright (c) 2019 ARM Limited
# All rights reserved
#
# The license below extends only to copyright in the software and shall
# not be construed as granting a license to any other intellectual
# property including but not limited to intellectual property relating
# to a hardware implementation of the functionality of the software
# licensed hereunder.  You may use the software subject to the license
# terms below provided that you ensure that this notice is replicated
# unmodified and in its entirety in all distributions of the software,
# modified or unmodified, in source code or in binary form.
#
# Copyright (c) 2017 Mark D. Hill and David A. Wood
# All rights reserved.
#
# Redistribution and use in source and binary forms, with or without
# modification, are permitted provided that the following conditions are
# met: redistributions of source code must retain the above copyright
# notice, this list of conditions and the following disclaimer;
# redistributions in binary form must reproduce the above copyright
# notice, this list of conditions and the following disclaimer in the
# documentation and/or other materials provided with the distribution;
# neither the name of the copyright holders nor the names of its
# contributors may be used to endorse or promote products derived from
# this software without specific prior written permission.
#
# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
# "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
# LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
# A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
# OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
# SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
# LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
# DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
# THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
# (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
# OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
#
# Authors: Sean Wilson
#          Nikos Nikoleris

import os
import tempfile
import shutil
import sys
import threading
import urllib
import urllib2

from testlib.fixture import Fixture
from testlib.config import config, constants
from testlib.helper import log_call, cacheresult, joinpath, absdirpath
import testlib.log as log


class VariableFixture(Fixture):
    def __init__(self, value=None, name=None):
        super(VariableFixture, self).__init__(name=name)
        self.value = value


class TempdirFixture(Fixture):
    def __init__(self):
        self.path = None
        super(TempdirFixture, self).__init__(
                name=constants.tempdir_fixture_name)

    def setup(self, testitem):
        self.path = tempfile.mkdtemp(prefix='gem5out')

    def teardown(self, testitem):
        if self.path is not None:
            shutil.rmtree(self.path)

    def skip_cleanup(self):
        # Set path to none so it's not deleted
        self.path = None

class UniqueFixture(Fixture):
    '''
    Base class for fixtures that generate a target in the
    filesystem. If the same fixture is used by more than one
    test/suite, rather than creating a copy of the fixture, it returns
    the same object and makes sure that setup is only executed
    once. Devired classses should override the _init and _setup
    functions.

    :param target: The absolute path of the target in the filesystem.

    '''
    fixtures = {}

    def __new__(cls, target):
        if target in cls.fixtures:
            obj = cls.fixtures[target]
        else:
            obj = super(UniqueFixture, cls).__new__(cls)
            obj.lock = threading.Lock()
            obj.target = target
            cls.fixtures[target] = obj
        return obj

    def __init__(self, *args, **kwargs):
        with self.lock:
            if hasattr(self, '_init_done'):
                return
            super(UniqueFixture, self).__init__(self, **kwargs)
            self._init(*args, **kwargs)
            self._init_done = True

    def setup(self, testitem):
        with self.lock:
            if hasattr(self, '_setup_done'):
                return
            self._setup_done = True
            self._setup(testitem)


class SConsFixture(UniqueFixture):
    '''
    Fixture will wait until all SCons targets are collected and tests are
    about to be ran, then will invocate a single instance of SCons for all
    targets.

    :param directory: The directory which scons will -C (cd) into before
        executing. If None is provided, will choose the config base_dir.
    '''

    def __new__(cls, target):
        obj = super(SConsFixture, cls).__new__(cls, target)
        return obj

    def _setup(self, testitem):
        if config.skip_build:
            return

        command = [
            'scons', '-C', self.directory,
            '-j', str(config.threads),
            '--ignore-style'
        ]

        if not self.targets:
            log.test_log.warn(
                'No SCons targets specified, this will'
                ' build the default all target.\n'
                'This is likely unintended, and you'
                ' may wish to kill testlib and reconfigure.')
        else:
            log.test_log.message(
                    'Building the following targets.'
                    ' This may take a while.')
            log.test_log.message('%s' % (', '.join(self.targets)))
            log.test_log.message(
                    "You may want to run with only a single ISA"
                    "(--isa=), use --skip-build, or use 'rerun'.")

        command.extend(self.targets)
        if self.options:
            command.extend(self.options)
        log_call(log.test_log, command, stderr=sys.stderr)

class Gem5Fixture(SConsFixture):
    def __new__(cls, isa, variant, protocol=None):
        target_dir = joinpath(config.build_dir, isa.upper())
        if protocol:
            target_dir += '_' + protocol
        target = joinpath(target_dir, 'gem5.%s' % variant)
        obj = super(Gem5Fixture, cls).__new__(cls, target)
        return obj

    def _init(self, isa, variant, protocol=None):
        self.name = constants.gem5_binary_fixture_name

        self.targets = [self.target]
        self.path = self.target
        self.directory = config.base_dir

        self.options = []
        if protocol:
            self.options = [ '--default=' + isa.upper(),
                             'PROTOCOL=' + protocol ]
        self.set_global()

class MakeFixture(Fixture):
    def __init__(self, directory, *args, **kwargs):
        name = 'make -C %s' % directory
        super(MakeFixture, self).__init__(build_once=True, lazy_init=False,
                                          name=name,
                                          *args, **kwargs)
        self.targets = []
        self.directory = directory

    def setup(self):
        super(MakeFixture, self).setup()
        targets = set(self.required_by)
        command = ['make', '-C', self.directory]
        command.extend([target.target for target in targets])
        log_call(command)


class MakeTarget(Fixture):
    def __init__(self, target, make_fixture=None, *args, **kwargs):
        '''
        :param make_fixture: The make invocation we will be attached to.
        Since we don't have a single global instance of make in gem5 like we do
        scons we need to know what invocation to attach to. If none given,
        creates its own.
        '''
        super(MakeTarget, self).__init__(name=target, *args, **kwargs)
        self.target = self.name

        if make_fixture is None:
            make_fixture = MakeFixture(
                    absdirpath(target),
                    lazy_init=True,
                    build_once=False)

        self.make_fixture = make_fixture

        # Add our self to the required targets of the main MakeFixture
        self.require(self.make_fixture)

    def setup(self, testitem):
        super(MakeTarget, self).setup()
        self.make_fixture.setup()
        return self

class TestProgram(MakeTarget):
    def __init__(self, program, isa, os, recompile=False):
        make_dir = joinpath('test-progs', program)
        make_fixture = MakeFixture(make_dir)
        target = joinpath('bin', isa, os, program)
        super(TestProgram, self).__init__(target, make_fixture)
        self.path = joinpath(make_dir, target)
        self.recompile = recompile

    def setup(self, testitem):
        # Check if the program exists if it does then only compile if
        # recompile was given.
        if self.recompile:
            super(MakeTarget, self).setup()
        elif not os.path.exists(self.path):
            super(MakeTarget, self).setup()

class DownloadedProgram(UniqueFixture):
    """ Like TestProgram, but checks the version in the gem5 binary repository
        and downloads an updated version if it is needed.
    """

    def __new__(cls, url, path, filename):
        target = joinpath(path, filename)
        return super(DownloadedProgram, cls).__new__(cls, target)

    def _init(self, url, path, filename, **kwargs):
        """
        url: string
            The url of the archive
        path: string
            The absolute path of the directory containing the archive
        filename: string
            The name of the archive
        """

        self.url = url
        self.path = path
        self.filename = joinpath(path, filename)
        self.name = "Downloaded:" + self.filename

    def _download(self):
        import errno
        log.test_log.debug("Downloading " + self.url + " to " + self.path)
        if not os.path.exists(self.path):
            try:
                os.makedirs(self.path)
            except OSError as e:
                if e.errno != errno.EEXIST:
                    raise
        urllib.urlretrieve(self.url, self.filename)

    def _getremotetime(self):
        import datetime, time
        import _strptime # Needed for python threading bug

        u = urllib2.urlopen(self.url)
        return time.mktime(datetime.datetime.strptime( \
                    u.info().getheaders("Last-Modified")[0],
                    "%a, %d %b %Y %X GMT").timetuple())

    def _setup(self, testitem):
        # Check to see if there is a file downloaded
        if not os.path.exists(self.filename):
            self._download()
        else:
            try:
                t = self._getremotetime()
            except urllib2.URLError:
                # Problem checking the server, use the old files.
                log.test_log.debug("Could not contact server. Binaries may be old.")
                return
            # If the server version is more recent, download it
            if t > os.path.getmtime(self.filename):
                self._download()

class DownloadedArchive(DownloadedProgram):
    """ Like TestProgram, but checks the version in the gem5 binary repository
        and downloads an updated version if it is needed.
    """

    def _extract(self):
        import tarfile
        with tarfile.open(self.filename) as tf:
            tf.extractall(self.path)

    def _setup(self, testitem):
        # Check to see if there is a file downloaded
        if not os.path.exists(self.filename):
            self._download()
            self._extract()
        else:
            try:
                t = self._getremotetime()
            except urllib2.URLError:
                # Problem checking the server, use the old files.
                log.test_log.debug("Could not contact server. "
                                   "Binaries may be old.")
                return
            # If the server version is more recent, download it
            if t > os.path.getmtime(self.filename):
                self._download()
                self._extract()