<commit_before>import os
import sys
import time
import logging

from junit_xml import TestCase, TestSuite

SILENT = False

class IntegrationTest(object):
    def __init__(self, name, path, package):
        self.name = name
        self.path = path
        self.package = package

    @classmethod
    def load(cls, path):
        from importlib import import_module
        return cls.from_module(import_module(path, ''))

    @classmethod
    def from_module(cls, m):
        return cls(m.name, m.__path__[0], m.package)

    @classmethod
    def discover(cls, path='.'):
        from importlib import import_module
        from operator import is_not
        from functools import partial

        dirs = [f for f in os.listdir(path) if os.path.isdir(f)]
        def to_module(f):
            try:
                return import_module(f, '')
            except ImportError:
                return None
        modules = filter(partial(is_not, None), map(to_module, dirs))
        modules = filter(lambda m: getattr(m, 'integration_test', False) == True, modules)
        tests = map(cls.from_module, modules)
        return list(tests)

    def run(self, unigornel_root):
        steps = [
            lambda: self.build(unigornel_root)
        ]
        cases = []
        for s in steps:
            c = s()
            cases.append(c)
            if c.is_failure():
                break

        return TestSuite(self.name, cases)

    def build(self, unigornel_root):
        def f():
            log("Building '{0}': package {1}".format(self.name, self.package))

            gopath = os.path.join(self.path, 'go')
            app_path = os.path.join(self.path, 'go', 'src', self.package)
            app = UnigornelApp(app_path, gopath, unigornel_root)
            output = app.build()
            log(output)
            return output, None
        return build_test_case('build', self.name, f)

    def execute(self):
        pass

    def clean(self):
        pass

class UnigornelApp(object):
    BUILD_CMD = './build.bash'

    def __init__(self, path, gopath, unigornel_root):
        self.path = path
        self.gopath = gopath
        self.unigornel_root = unigornel_root

    def build(self, out=None, build_all=True, verbose=True):
        from subprocess import Popen, PIPE, STDOUT

        argv = [self.BUILD_CMD, 'app', '--app', self.path]
        if out is not None: argv += ['-o', out]
        if build_all:       argv += ['-a']
        if verbose:         argv += ['-x']

        env = os.environ.copy()
        env['GOPATH'] = self.gopath

        with pushd(self.unigornel_root):
            log('[+]', ' '.join(argv))
            with Popen(argv, env=env, stdout=PIPE, stderr=STDOUT) as proc:
                b = proc.stdout.read()
                try:
                    stdout = b.decode('utf-8')
                except UnicodeDecodeError:
                    stdout = str(b)
                return stdout

                proc.wait()
                if proc.returncode != 0:
                    raise Exception('error: build.bash returned code {0}'.format(proc.returncode))

class Kernel(object):
    def __init__(self, kernel, memory, name, on_crash='preserve'):
        self.kernel = kernel
        self.memory = memory
        self.name = name
        self.on_crash = "preserve"

    def create_paused_guest(self):
        from tempfile import NamedTemporaryFile
        from subprocess import call
        with NamedTemporaryFile('w') as config:
            print('kernel = "{0}"'.format(self.kernel),     file=config)
            print('memory = {0}'.format(self.memory),       file=config)
            print('name = "{0}"'.format(self.name),         file=config)
            print('on_crash = "{0}"'.format(self.on_crash), file=config)
            config.flush()
            path = config.name

            log('Creating {0} with configuration {1}'.format(repr(self), path))
            call(['xl', 'create', '-p', '-f', path])

        try:
            guest = next(filter(lambda g: g.name == self.name, XenGuest.list()))
        except StopIteration:
            raise Exception('Failed to launch guest: could not find guest with matching name')

        assert(guest.state & guest.STATE_PAUSED)

        return guest

    def __repr__(self):
        f = [self.kernel, self.memory, self.name, self.on_crash]
        return 'Kernel(kernel={0}, memory={1}, name={2}, on_crash={3}'.format(*f)

class XenGuest(object):
    STATE_RUNNING       = 0x01
    STATE_BLOCKED       = 0x02
    STATE_PAUSED        = 0x04
    STATE_SHUTDOWN      = 0x08
    STATE_CRASHED       = 0x10
    STATE_DYING         = 0x20

    def __init__(self, name, id, mem, vcpus, raw_state, time):
        self.name = name
        self.id = id
        self.mem = mem
        self.vcpus = vcpus
        self.raw_state = raw_state
        self.time = time

    @property
    def state(self):
        return XenGuest.parse_state(self.raw_state)

    def destroy(self):
        from subprocess import call
        call(['xl', 'destroy', str(self.id)])

    @classmethod
    def from_xl_list_line(cls, line):
        import re
        name, id, mem, vcpus, state, time = re.split('\s+', line)
        return cls(name, int(id), int(mem), int(vcpus), state, float(time))

    @classmethod
    def list(cls):
        from subprocess import Popen, PIPE

        with Popen(['xl', 'list'], stdout=PIPE) as proc:
            proc.stdout.readline() # discard header
            m = map(lambda b: cls.from_xl_list_line(b.decode('utf-8').strip()), proc.stdout)
            guests = list(m)

            proc.wait()
            if proc.returncode != 0:
                raise Exception('error: xl list returned code {0}'.format(proc.returncode))

            return guests

    @staticmethod
    def parse_state(raw_state):<commit_msg>Implement running a Xen guest and reading its console<commit_after>import os
import sys
import time
import logging

from junit_xml import TestCase, TestSuite

SILENT = False

class IntegrationTest(object):
    def __init__(self, name, path, package):
        self.name = name
        self.path = path
        self.package = package

    @classmethod
    def load(cls, path):
        from importlib import import_module
        return cls.from_module(import_module(path, ''))

    @classmethod
    def from_module(cls, m):
        return cls(m.name, m.__path__[0], m.package)

    @classmethod
    def discover(cls, path='.'):
        from importlib import import_module
        from operator import is_not
        from functools import partial

        dirs = [f for f in os.listdir(path) if os.path.isdir(f)]
        def to_module(f):
            try:
                return import_module(f, '')
            except ImportError:
                return None
        modules = filter(partial(is_not, None), map(to_module, dirs))
        modules = filter(lambda m: getattr(m, 'integration_test', False) == True, modules)
        tests = map(cls.from_module, modules)
        return list(tests)

    def run(self, unigornel_root):
        steps = [
            lambda: self.build(unigornel_root)
        ]
        cases = []
        for s in steps:
            c = s()
            cases.append(c)
            if c.is_failure():
                break

        return TestSuite(self.name, cases)

    def build(self, unigornel_root):
        def f():
            log("Building '{0}': package {1}".format(self.name, self.package))

            gopath = os.path.join(self.path, 'go')
            app_path = os.path.join(self.path, 'go', 'src', self.package)
            app = UnigornelApp(app_path, gopath, unigornel_root)
            output = app.build()
            log(output)
            return output, None
        return build_test_case('build', self.name, f)

    def execute(self):
        pass

    def clean(self):
        pass

class UnigornelApp(object):
    BUILD_CMD = './build.bash'

    def __init__(self, path, gopath, unigornel_root):
        self.path = path
        self.gopath = gopath
        self.unigornel_root = unigornel_root

    def build(self, out=None, build_all=True, verbose=True):
        from subprocess import Popen, PIPE, STDOUT

        argv = [self.BUILD_CMD, 'app', '--app', self.path]
        if out is not None: argv += ['-o', out]
        if build_all:       argv += ['-a']
        if verbose:         argv += ['-x']

        env = os.environ.copy()
        env['GOPATH'] = self.gopath

        with pushd(self.unigornel_root):
            log('[+]', ' '.join(argv))
            with Popen(argv, env=env, stdout=PIPE, stderr=STDOUT) as proc:
                b = proc.stdout.read()
                try:
                    stdout = b.decode('utf-8')
                except UnicodeDecodeError:
                    stdout = str(b)
                return stdout

                proc.wait()
                if proc.returncode != 0:
                    raise Exception('error: build.bash returned code {0}'.format(proc.returncode))

class Kernel(object):
    def __init__(self, kernel, memory, name, on_crash='preserve'):
        self.kernel = kernel
        self.memory = memory
        self.name = name
        self.on_crash = "preserve"

    def create_paused_guest(self):
        from tempfile import NamedTemporaryFile
        from subprocess import call
        with NamedTemporaryFile('w') as config:
            print('kernel = "{0}"'.format(self.kernel),     file=config)
            print('memory = {0}'.format(self.memory),       file=config)
            print('name = "{0}"'.format(self.name),         file=config)
            print('on_crash = "{0}"'.format(self.on_crash), file=config)
            config.flush()
            path = config.name

            log('Creating {0} with configuration {1}'.format(repr(self), path))
            call(['xl', 'create', '-p', '-f', path])

        try:
            guest = next(filter(lambda g: g.name == self.name, XenGuest.list()))
        except StopIteration:
            raise Exception('Failed to launch guest: could not find guest with matching name')

        assert(guest.state & guest.STATE_PAUSED)

        return guest

    def __repr__(self):
        f = [self.kernel, self.memory, self.name, self.on_crash]
        return 'Kernel(kernel={0}, memory={1}, name={2}, on_crash={3}'.format(*f)

class XenGuestState(object):
    def __init__(self, final_state, did_timeout, console):
        self.final_state = final_state
        self.did_timeout = did_timeout
        self.console = console

class XenGuest(object):
    STATE_UNKNOWN       = 0x00
    STATE_RUNNING       = 0x01
    STATE_BLOCKED       = 0x02
    STATE_PAUSED        = 0x04
    STATE_SHUTDOWN      = 0x08
    STATE_CRASHED       = 0x10
    STATE_DYING         = 0x20

    def __init__(self, name, id, mem, vcpus, raw_state, time):
        self.name = name
        self.id = id
        self.mem = mem
        self.vcpus = vcpus
        self.raw_state = raw_state
        self.time = time

    @property
    def state(self):
        return XenGuest.parse_state(self.raw_state)

    def is_shutdown(self): return bool(self.state & self.STATE_SHUTDOWN)
    def is_crashed(self):  return bool(self.state & self.STATE_CRASHED)

    def current(self):
        return next(filter(lambda g: g.id == self.id, XenGuest.list()), None)

    def destroy(self):
        from subprocess import call
        call(['xl', 'destroy', str(self.id)])

    def unpause_and_collect(self, timeout):
        from subprocess import Popen, PIPE, STDOUT, call
        with Popen(['xl', 'console', str(self.id)], stdout=PIPE, stderr=STDOUT, stdin=PIPE) as console:
            call(['xl', 'unpause', str(self.id)])

            # Wait until the unikernel is shutdown or the timeout is reached
            deadline = time.monotonic() + timeout
            did_timeout = False
            while True:
                current = self.current()
                if not current:
                    final_state = self.STATE_UNKNOWN
                    break
                elif current.is_shutdown() or current.is_crashed():
                    final_state = current.state
                    break
                elif time.monotonic() > deadline:
                    final_state = current.state
                    did_timeout = True
                    break
                time.sleep(1.0)

            console.kill()
            console.wait(timeout=1)

            b = console.stdout.read()
            try:
                stdout = b.decode('utf-8')
            except UnicodeDecodeError:
                stdout = str(b)
            return XenGuestState(final_state, did_timeout, stdout)

    @classmethod
    def from_xl_list_line(cls, line):
        import re
        name, id, mem, vcpus, state, time = re.split('\s+', line)
        return cls(name, int(id), int(mem), int(vcpus), state, float(time))

    @classmethod
    def list(cls):
        from subprocess import Popen, PIPE

        with Popen(['xl', 'list'], stdout=PIPE) as proc:
            proc.stdout.readline() # discard header
            m = map(lambda b: cls.from_xl_list_line(b.decode('utf-8').strip()), proc.stdout)
            guests = list(m)

            proc.wait()
            if proc.returncode != 0:
                raise Exception('error: xl list returned code {0}'.format(proc.returncode))

            return guests

    @staticmethod
    def parse_state(raw_state):
{'fraction_non_alphanumeric': 0.11450951137635211, 'fraction_numerical': 0.006042521447221186, 'mean_word_length': 1.9711879432624113, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 4, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 2}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '630812', 'n_tokens_mistral': 4162, 'n_tokens_neox': 3973, 'n_words': 1000}