kpet-schedule-test

#!/usr/bin/env python3

import math
import unittest
import pprint

def is_child_of(types, child, parent):
    if child == parent:
        return True
    for child_parent in types[child]:
        if is_child_of(types, child_parent, parent):
            return True
    return False


def host_types_parentage(host_types, parent, child):
    if child == parent:
        return 0
    min_parentage = math.inf
    for child_parent in host_types[child]:
        parentage = host_types_parentage(host_types, parent, child_parent)
        if parentage < min_parentage:
            min_parentage = parentage
    return min_parentage + 1


def host_types_specificity(host_types, parent, child):
    """
    Calculate specificity of a child host type in relation to a parent host
    type.

    Args:
        host_types: Host type map.
        parent:     The parent host type to calculate the specificity towards.
        child:      The child host type to calculate the specificity of.

    Returns:
        None if no relation was found.
        Zero, if the "parent" and the "child" were the same type.
        Positive number if the "child" was found to be descended from
        "parent", signifying the number of ancestors.
        Negative number if the "parent" was found to be descended from
        "child", signifying the number of ancestors.
    """
    parentage = host_types_parentage(host_types, parent, child)
    if parentage != math.inf:
        return parentage
    parentage = host_types_parentage(host_types, child, parent)
    if parentage != math.inf:
        return -parentage
    return None


def allocate_hosts(host_types, suites):
    hosts=[]

    for suite in suites:
        # Pick/allocate hosts for the suite
        available_hosts=hosts.copy()
        for task_host_type in suite["tasks"]:
            # Locate the most generic host satisfying the type requirements
            min_specificity = None
            min_i = None
            for i, available_host in enumerate(available_hosts):
                specificity = \
                    host_types_specificity(host_types, available_host['type'],
                                           task_host_type)
                print(f"host_types_specificity(host_types, "
                      f"{available_host['type']}, "
                      f"{task_host_type}) -> {specificity}")
                if specificity is not None and \
                   (min_i is None or \
                    specificity < min_specificity):
                    min_i = i
                    min_specificity = specificity

            # If we found no suitable host
            if min_i is None:
                print(f"add {task_host_type}")
                # Add one
                hosts.append({"type": task_host_type})
            else:
                # If the found host is more generic than we need
                if min_specificity > 0:
                    print(f"specialize {available_hosts[i]['type']} to {task_host_type}")
                    # Specialize it
                    available_hosts[i]["type"] = task_host_type
                # Pick it
                print(f"pick {available_hosts[i]['type']}")
                del available_hosts[i]

    return hosts

class AllocateTests(unittest.TestCase):
    def test_one_host_two_suites_one_task_each(self):
        self.assertEqual(
            allocate_hosts(
                dict(basic=[]),
                [dict(tasks=["basic"]), dict(tasks=["basic"])]
            ),
            [{"type": "basic"}]
        )

    def test_one_host_one_suite_two_tasks(self):
        self.assertEqual(
            allocate_hosts(
                dict(basic=[]),
                [dict(tasks=["basic", "basic"])]
            ),
            [{"type": "basic"}, {"type": "basic"}]
        )

    def test_specialize_type(self):
        self.assertEqual(
            allocate_hosts(
                dict(basic=[], specific=["basic"]),
                [dict(tasks=["basic"]), dict(tasks=["specific"])]
            ),
            [{"type": "specific"}]
        )

    def test_complicated(self):
        host_types = dict(
            basic=[],
            big_storage=["basic"],
            big_nas_storage=["big_storage"],
            big_local_storage=["big_storage"],
            big_us_nas_storage=["big_nas_storage"],
            sound_blaster=["basic"],
            sound_blaster_and_big_storage=["sound_blaster", "big_local_storage"],
            ne2000_nic=["basic"],
            ne2000_nic_big_local_storage=["ne2000_nic", "big_local_storage"],
        )

        suites = [
            dict(
                duration=200,
                tasks=["big_storage"]
            ),
            dict(
                duration=300,
                tasks=["ne2000_nic", "ne2000_nic_big_local_storage"]
            ),
            dict(
                duration=400,
                tasks=["basic", "basic", "big_local_storage"]
            ),
        ]

        self.assertEqual(
            [{'type': 'ne2000_nic_big_local_storage'},
             {'type': 'ne2000_nic'},
             {'type': 'basic'}],
            allocate_hosts(host_types, suites)
        )


unittest.main()