Search-Based Test Generation - Part 2.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "# Search-Based Test Generation - Part 2"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "slideshow": {
     "slide_type": "skip"
    }
   },
   "outputs": [],
   "source": [
    "import random\n",
    "import matplotlib.pyplot as plt\n",
    "import numpy as np\n",
    "from inspect import signature\n",
    "import inspect\n",
    "import ast\n",
    "import astor\n",
    "import sys"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "skip"
    }
   },
   "source": [
    "We continue testing the triangle example program."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "slideshow": {
     "slide_type": "fragment"
    }
   },
   "outputs": [],
   "source": [
    "def triangle(a, b, c):\n",
    "    if a <= 0 or b <= 0 or c <= 0:\n",
    "        return 4 # invalid\n",
    "    \n",
    "    if a + b <= c or a + c <= b or b + c <= a:\n",
    "        return 4 # invalid\n",
    "    \n",
    "    if a == b and b == c:\n",
    "        return 1 # equilateral\n",
    "    \n",
    "    if a == b or b == c or a == c:\n",
    "        return 2 # isosceles\n",
    "    \n",
    "    return 3 # scalene"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "## Whole Test Suite Optimisation"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "skip"
    }
   },
   "source": [
    "In Whole Test Suite Optimisation, our goal is to optimise a test suite so that it achieves the highest possible code coverage."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "slideshow": {
     "slide_type": "fragment"
    }
   },
   "outputs": [],
   "source": [
    "sig = signature(triangle)\n",
    "num_parameters = len(sig.parameters)\n",
    "num_parameters"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "slideshow": {
     "slide_type": "fragment"
    }
   },
   "outputs": [],
   "source": [
    "num_tests = 30\n",
    "MAX = 1000\n",
    "MIN = -MAX"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "slideshow": {
     "slide_type": "fragment"
    }
   },
   "outputs": [],
   "source": [
    "def get_random_individual():\n",
    "    num = random.randint(1, num_tests)\n",
    "    return [[random.randint(MIN, MAX) for _ in range(num_parameters)] for _ in range(num)]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "slideshow": {
     "slide_type": "fragment"
    }
   },
   "outputs": [],
   "source": [
    "get_random_individual()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "skip"
    }
   },
   "source": [
    "When applying mutation, we need to be able to modify individual tests as before. To keep things challenging, we will not use our optimised mutation that copies parameters, but aim to achieve the entire optimisation just using small steps:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "outputs": [],
   "source": [
    "def mutate_test(solution):\n",
    "    P_mutate = 1/len(solution)\n",
    "    mutated = solution[:]\n",
    "    for position in range(len(solution)):\n",
    "        if random.random() < P_mutate:\n",
    "            mutated[position] = min(MAX, max(MIN, int(random.gauss(mutated[position], MAX*0.01))))\n",
    "            \n",
    "    return mutated"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "skip"
    }
   },
   "source": [
    "However, modifying tests is only one of the things we can do when mutating our actual individuals, which consist of multiple tests. Besides modifying existing tests, we could also delete or add tests, for example like this."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "slideshow": {
     "slide_type": "fragment"
    }
   },
   "outputs": [],
   "source": [
    "def mutate_set(solution):\n",
    "    P_mutate = 1/len(solution)\n",
    "    mutated = []\n",
    "    for position in range(len(solution)):\n",
    "        if random.random() >= P_mutate:\n",
    "            mutated.append(solution[position][:])\n",
    "            \n",
    "    if not mutated:\n",
    "        mutated = solution[:]\n",
    "    for position in range(len(mutated)):\n",
    "        if random.random() < P_mutate:\n",
    "            mutated[position] = mutate_test(mutated[position])\n",
    " \n",
    "    ALPHA = 1/3\n",
    "    count = 1\n",
    "    while random.random() < ALPHA ** count:\n",
    "        count += 1\n",
    "        mutated.append([random.randint(MIN, MAX) for _ in range(num_parameters)])\n",
    "    \n",
    "    return mutated"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "skip"
    }
   },
   "source": [
    "With a certain probability, each of the tests can be removed from a test suite; similarly, each remaining test may be mutated like we mutated tests previously. Finally, with a probability `ALPHA` we insert a new test; if we do so, we insert another one with probability `ALPHA`$^2$, and so on."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "skip"
    }
   },
   "source": [
    "When crossing over two individuals, they might have different length, which makes choosing a crossover point difficult. For example, we might pick a crossover point that is longer than one of the parent chromosomes, and then what do we do? A simple solution would be to pick two different crossover points."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "outputs": [],
   "source": [
    "def variable_crossover(parent1, parent2):\n",
    "    pos1 = random.randint(1, len(parent1))\n",
    "    pos2 = random.randint(1, len(parent2))\n",
    "    offspring1 = parent1[:pos1] + parent2[pos2:]\n",
    "    offspring2 = parent2[:pos2] + parent1[pos1:]\n",
    "    return (offspring1, offspring2)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "skip"
    }
   },
   "source": [
    "Besides mutation and crossover we also require selection and elitism operators, but these are independent of the representation and therefore no different from the ones we used previously."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "slideshow": {
     "slide_type": "fragment"
    }
   },
   "outputs": [],
   "source": [
    "tournament_size = 3\n",
    "def tournament_selection(population):\n",
    "    candidates = random.sample(population, tournament_size)        \n",
    "    winner = min(candidates, key = lambda x: get_fitness(x))\n",
    "    return winner"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "slideshow": {
     "slide_type": "fragment"
    }
   },
   "outputs": [],
   "source": [
    "elite_size = 2\n",
    "def elitism_standard(population):\n",
    "    population.sort(key = lambda k: get_fitness(k))\n",
    "    return population[:elite_size]"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "skip"
    }
   },
   "source": [
    "To see this works, we need to define the fitness function. Since we want to cover _everything_ we simply need to make sure that every single branch is covered at least once in a test suite. A branch is covered if its minimum branch distance is 0; thus, if everything is covered, then the sum of minimal branch distances should be 0. In order to calculate the fitness value, we need to collect information about which branches were executed. For each condition we keep track of the branch distance for reaching the true branch and the false branch."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "outputs": [],
   "source": [
    "distances_true = {}\n",
    "distances_false = {}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "skip"
    }
   },
   "source": [
    "There is one special case: If an if-statement is executed only once, then optimising the true/false distance may lead to a suboptimal, oscillising evolution. We therefore also count how often each if-condition was executed. If it was only executed once, then the fitness value for that branch needs to be higher than if it was executed twice. We define a helper function `update_maps` that updates our global variables with the values observed."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "slideshow": {
     "slide_type": "fragment"
    }
   },
   "outputs": [],
   "source": [
    "condition_count = {}"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "slideshow": {
     "slide_type": "fragment"
    }
   },
   "outputs": [],
   "source": [
    "def update_maps(condition_num, d_true, d_false):\n",
    "    global distances_true, distances_false, condition_count\n",
    "\n",
    "    if condition_num in condition_count.keys():\n",
    "        condition_count[condition_num] = condition_count[condition_num] + 1\n",
    "    else:\n",
    "        condition_count[condition_num] = 1\n",
    "        \n",
    "    if condition_num in distances_true.keys():\n",
    "        distances_true[condition_num] = min(distances_true[condition_num], d_true)\n",
    "    else:\n",
    "        distances_true[condition_num] = d_true\n",
    "\n",
    "    if condition_num in distances_false.keys():\n",
    "        distances_false[condition_num] = min(distances_false[condition_num], d_false)\n",
    "    else:\n",
    "        distances_false[condition_num] = d_false"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "skip"
    }
   },
   "source": [
    "Now we need to finish implementing the `evaluate_condition` function. We add yet another parameter to denote the ID of the branch we are instrumenting:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "outputs": [],
   "source": [
    "def evaluate_condition(num, op, lhs, rhs):\n",
    "    distance_true = 0\n",
    "    distance_false = 0\n",
    "\n",
    "    # Make sure the distance can be calculated on number and character\n",
    "    # comparisons (needed for cgi_decode later)\n",
    "    if isinstance(lhs, str):\n",
    "        lhs = ord(lhs)\n",
    "    if isinstance(rhs, str):\n",
    "        rhs = ord(rhs)\n",
    "\n",
    "    if op == \"Eq\":\n",
    "        if lhs == rhs:\n",
    "            distance_false = 1\n",
    "        else:\n",
    "            distance_true = abs(lhs - rhs)\n",
    "\n",
    "    elif op == \"Gt\":\n",
    "        if lhs > rhs:\n",
    "            distance_false = lhs - rhs\n",
    "        else:\n",
    "            distance_true = rhs - lhs + 1\n",
    "    elif op == \"Lt\":\n",
    "        if lhs < rhs:\n",
    "            distance_false = rhs - lhs\n",
    "        else:\n",
    "            distance_true = lhs - rhs + 1\n",
    "    elif op == \"LtE\":\n",
    "        if lhs <= rhs:\n",
    "            distance_false = rhs - lhs + 1\n",
    "        else:\n",
    "            distance_true = lhs - rhs\n",
    "    # ...\n",
    "    # handle other comparison operators\n",
    "    # ...\n",
    "\n",
    "    elif op == \"In\":\n",
    "        minimum = sys.maxsize\n",
    "        for elem in rhs.keys():\n",
    "            distance = abs(lhs - ord(elem))\n",
    "            if distance < minimum:\n",
    "                minimum = distance\n",
    "\n",
    "        distance_true = minimum\n",
    "        if distance_true == 0:\n",
    "            distance_false = 1\n",
    "    else:\n",
    "        assert False\n",
    "\n",
    "    update_maps(num, normalise(distance_true), normalise(distance_false))\n",
    "\n",
    "    if distance_true == 0:\n",
    "        return True\n",
    "    else:\n",
    "        return False"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "skip"
    }
   },
   "source": [
    "We need to normalise branch distances since different comparisons will be on different scales, and this would bias the search. We will use the normalisaction function defined in the previous chapter:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "outputs": [],
   "source": [
    "def normalise(x):\n",
    "    return x / (1.0 + x)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "skip"
    }
   },
   "source": [
    "We also need to extend our instrumentation function to take care of all comparisons, and not just equality comparisons:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "outputs": [],
   "source": [
    "import ast\n",
    "class BranchTransformer(ast.NodeTransformer):\n",
    "\n",
    "    condition_num = 0\n",
    "\n",
    "    def visit_FunctionDef(self, node):\n",
    "        node.name = node.name + \"_instrumented\"\n",
    "        return self.generic_visit(node)\n",
    "\n",
    "    def visit_Compare(self, node):\n",
    "        if node.ops[0] in [ast.Is, ast.IsNot, ast.In, ast.NotIn]:\n",
    "            return node\n",
    "\n",
    "        self.condition_num += 1\n",
    "        return ast.Call(func=ast.Name(\"evaluate_condition\", ast.Load()),\n",
    "                        args=[ast.Num(self.condition_num - 1),\n",
    "                              ast.Str(node.ops[0].__class__.__name__),\n",
    "                              node.left,\n",
    "                              node.comparators[0]],\n",
    "                        keywords=[],\n",
    "                        starargs=None,\n",
    "                        kwargs=None)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "outputs": [],
   "source": [
    "source = inspect.getsource(triangle)\n",
    "node = ast.parse(source)\n",
    "transformer = BranchTransformer()\n",
    "transformer.visit(node)\n",
    "\n",
    "# Make sure the line numbers are ok before printing\n",
    "node = ast.fix_missing_locations(node)\n",
    "num_conditions = transformer.condition_num\n",
    "\n",
    "print(astor.to_source(node))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "skip"
    }
   },
   "source": [
    "To define an executable version of the instrumented triangle function, we introduce the`create_instrumented_function` function:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "outputs": [],
   "source": [
    "def create_instrumented_function(f):\n",
    "    source = inspect.getsource(f)\n",
    "    node = ast.parse(source)\n",
    "    node = BranchTransformer().visit(node)\n",
    "\n",
    "    # Make sure the line numbers are ok so that it compiles\n",
    "    node = ast.fix_missing_locations(node)\n",
    "\n",
    "    # Compile and add the instrumented function to the current module\n",
    "    current_module = sys.modules[__name__]\n",
    "    code = compile(node, filename=\"<ast>\", mode=\"exec\")\n",
    "    exec(code, current_module.__dict__)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "outputs": [],
   "source": [
    "create_instrumented_function(triangle)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "slideshow": {
     "slide_type": "fragment"
    }
   },
   "outputs": [],
   "source": [
    "triangle(4, 4, 4)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "slideshow": {
     "slide_type": "fragment"
    }
   },
   "outputs": [],
   "source": [
    "triangle_instrumented(4, 4, 4)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "slideshow": {
     "slide_type": "fragment"
    }
   },
   "outputs": [],
   "source": [
    "distances_true"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "slideshow": {
     "slide_type": "fragment"
    }
   },
   "outputs": [],
   "source": [
    "distances_false"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "slideshow": {
     "slide_type": "fragment"
    }
   },
   "outputs": [],
   "source": [
    "condition_count"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "skip"
    }
   },
   "source": [
    "The actual fitness function now is the sum of minimal distances after all tests have been executed. If an if-condition was not executed at all, then the true distance and the false distance will be 1, resulting in a sum of 2 for the if-condition. If the condition was covered only once, we set the fitness to exactly 1. If the condition was executed more than once, then at least either the true or false distance has to be 0, such that in sum, true and false distances will be less than 0."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "outputs": [],
   "source": [
    "def get_fitness(x):\n",
    "    # Reset any distance values from previous executions\n",
    "    global distances_true, distances_false, condition_count\n",
    "    distances_true =  {x: 1.0 for x in range(num_conditions)}\n",
    "    distances_false = {x: 1.0 for x in range(num_conditions)}\n",
    "    condition_count = {x:   0 for x in range(num_conditions)}\n",
    "\n",
    "    # Run the function under test\n",
    "    for test in x:\n",
    "        triangle_instrumented(*test)\n",
    "\n",
    "    # Sum up branch distances\n",
    "    fitness = 0.0\n",
    "    for condition in range(num_conditions):\n",
    "        if condition_count[condition] == 1:\n",
    "            fitness += 1\n",
    "        else:\n",
    "            fitness += distances_true[condition]\n",
    "            fitness += distances_false[condition]\n",
    "\n",
    "    return fitness"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "skip"
    }
   },
   "source": [
    "Before we run some experiments on this, let's make a small addition to our genetic algorithm: Since the size of individuals is variable it will be interesting to observe how this evolves. We'll captures the average population size in a separate list. Since the costs of evaluating fitness are no longer constant per individual but depend on the number of tests executed, we will also change our stopping criterion to the number of executed tests."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "outputs": [],
   "source": [
    "from statistics import mean\n",
    "\n",
    "length_values = []\n",
    "max_executions = 10000\n",
    "\n",
    "def ga():\n",
    "    population = [get_random_individual() for _ in range(population_size)]\n",
    "    best_fitness = sys.maxsize\n",
    "    tests_executed = 0\n",
    "    for p in population:\n",
    "        fitness = get_fitness(p)\n",
    "        tests_executed += len(p)\n",
    "        if fitness < best_fitness:\n",
    "            best_fitness = fitness\n",
    "            best_solution = p\n",
    "    while tests_executed < max_executions:\n",
    "        fitness_values.append(best_fitness)\n",
    "        length_values.append(mean([len(x) for x in population]))\n",
    "        new_population = elitism(population)\n",
    "        while len(new_population) < len(population):\n",
    "            parent1 = selection(population)\n",
    "            parent2 = selection(population)\n",
    "\n",
    "            if random.random() < P_xover:\n",
    "                offspring1, offspring2 = crossover(parent1, parent2)\n",
    "            else:\n",
    "                offspring1, offspring2 = parent1, parent2\n",
    "\n",
    "            offspring1 = mutate(offspring1)\n",
    "            offspring2 = mutate(offspring2)\n",
    "            \n",
    "            new_population.append(offspring1)\n",
    "            new_population.append(offspring2)\n",
    "            tests_executed += len(offspring1) + len(offspring2)\n",
    "\n",
    "        population = new_population\n",
    "        for p in population:\n",
    "            fitness = get_fitness(p)\n",
    "            if fitness < best_fitness:\n",
    "                best_fitness = fitness\n",
    "                best_solution = p\n",
    "    print(f\"Best fitness: {best_fitness}, size {len(best_solution)}\")\n",
    "\n",
    "    return best_solution"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "skip"
    }
   },
   "source": [
    "Since we now have all the operators we need in place, let's run a first experiment aiming to achieve 100% coverage on the triangle example."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "outputs": [],
   "source": [
    "max_executions = 500000\n",
    "MAX = 1000\n",
    "MIN = -MAX\n",
    "P_xover = 0.7\n",
    "crossover = variable_crossover\n",
    "selection = tournament_selection\n",
    "elitism   = elitism_standard \n",
    "mutate    = mutate_set\n",
    "tournament_size = 4\n",
    "population_size = 50\n",
    "fitness_values = []\n",
    "length_values = []\n",
    "ga()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true,
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "outputs": [],
   "source": [
    "plt.plot(fitness_values)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "skip"
    }
   },
   "source": [
    "The plot shows iterations of the genetic algorithm on the x-axis. Very likely, the result likely isn't great. But why? Let's look at the average population length."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true,
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "outputs": [],
   "source": [
    "plt.plot(length_values)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "skip"
    }
   },
   "source": [
    "What you can see here is a phenomenon called _bloat_: Individuals grow in size through mutation and crossover, and the search has no incentive to reduce their size (adding a test can never decrease coverage; removing a test can). As a result the individuals just keep growing, and quickly eat up all the available resources for the search."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "skip"
    }
   },
   "source": [
    "We need to make sure that our individuals don't grow without bounds, so we adapt the other operators accordingly. When applying crossover, we pick a relative position and then cut the individuals at their relative position, such that no offspring becomes longer than the longest parent:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "outputs": [],
   "source": [
    "def relative_crossover(parent1, parent2):\n",
    "    pos = random.random()\n",
    "    pos1 = int(len(parent1) * pos)\n",
    "    pos2 = int(len(parent2) * pos)\n",
    "    offspring1 = parent1[:pos1] + parent2[pos2:]\n",
    "    offspring2 = parent2[:pos2] + parent1[pos1:]\n",
    "    return (offspring1, offspring2)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "skip"
    }
   },
   "source": [
    "What's the influence of relative crossover?"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "outputs": [],
   "source": [
    "max_executions = 500000\n",
    "MAX = 1000\n",
    "MIN = -MAX\n",
    "crossover = relative_crossover\n",
    "selection = tournament_selection\n",
    "elitism   = elitism_standard \n",
    "mutate    = mutate_set\n",
    "tournament_size = 4\n",
    "population_size = 50\n",
    "fitness_values = []\n",
    "length_values = []\n",
    "ga()\n",
    "plt.plot(length_values);"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "skip"
    }
   },
   "source": [
    "When selecting individuals, we apply a secondary criterion: For any two individuals with equal fitness function, we prefer the shorter one:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "outputs": [],
   "source": [
    "def tournament_selection_length(population, n = tournament_size):\n",
    "    candidates = random.sample(population, n)        \n",
    "    winner = min(candidates, key = lambda x: (get_fitness(x), len(x)))    \n",
    "                \n",
    "    return winner"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "skip"
    }
   },
   "source": [
    "What is the influence of length-based tournament selection?"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true,
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "outputs": [],
   "source": [
    "max_executions = 500000\n",
    "MAX = 1000\n",
    "MIN = -MAX\n",
    "crossover = variable_crossover\n",
    "selection = tournament_selection_length\n",
    "elitism   = elitism_standard \n",
    "mutate    = mutate_set\n",
    "tournament_size = 4\n",
    "population_size = 50\n",
    "fitness_values = []\n",
    "length_values = []\n",
    "ga()\n",
    "plt.plot(length_values);"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "skip"
    }
   },
   "source": [
    "The criterion also has to hold when we apply elitism -- we don't only want the test suites with the best fitness, but we also want the smallest test suites:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "outputs": [],
   "source": [
    "def elitism_length(population):\n",
    "    population.sort(key = lambda k: (get_fitness(k), len(k)))\n",
    "    return population[:elite_size]"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "skip"
    }
   },
   "source": [
    "What is the influence of length-based elitism?"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true,
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "outputs": [],
   "source": [
    "max_executions = 500000\n",
    "MAX = 1000\n",
    "MIN = -MAX\n",
    "crossover = variable_crossover\n",
    "selection = tournament_selection\n",
    "elitism   = elitism_length \n",
    "mutate    = mutate_set\n",
    "tournament_size = 4\n",
    "population_size = 50\n",
    "fitness_values = []\n",
    "length_values = []\n",
    "ga()\n",
    "plt.plot(length_values);"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "skip"
    }
   },
   "source": [
    "When adding new tests during mutation, we can also ensure that the maximum size (which we set to 30 earlier) is not exceeded."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "outputs": [],
   "source": [
    "def mutate_bounded(solution):\n",
    "    P_mutate = 1/len(solution)\n",
    "    mutated = []\n",
    "    for position in range(len(solution)):\n",
    "        if random.random() >= P_mutate:\n",
    "            mutated.append(solution[position][:])\n",
    "            \n",
    "    if not mutated:\n",
    "        mutated = solution[:]\n",
    "    for position in range(len(mutated)):\n",
    "        if random.random() < P_mutate:\n",
    "            mutated[position] = mutate_test(mutated[position])\n",
    " \n",
    "    ALPHA = 1/3\n",
    "    count = 1\n",
    "    while random.random() < ALPHA ** count and len(mutated) < num_tests:\n",
    "        count += 1\n",
    "        mutated.append([random.randint(MIN, MAX) for _ in range(num_parameters)])\n",
    "    \n",
    "    return mutated"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "skip"
    }
   },
   "source": [
    "What is the influence of bounded mutation?"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "outputs": [],
   "source": [
    "max_executions = 500000\n",
    "MAX = 1000\n",
    "MIN = -MAX\n",
    "crossover = variable_crossover\n",
    "selection = tournament_selection\n",
    "elitism   = elitism_standard \n",
    "mutate    = mutate_bounded\n",
    "tournament_size = 4\n",
    "population_size = 50\n",
    "fitness_values = []\n",
    "length_values = []\n",
    "ga()\n",
    "plt.plot(length_values);"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "skip"
    }
   },
   "source": [
    "Now let's put everything together."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true,
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "outputs": [],
   "source": [
    "max_executions = 500000\n",
    "MAX = 1000\n",
    "MIN = -MAX\n",
    "crossover = relative_crossover\n",
    "selection = tournament_selection_length\n",
    "elitism   = elitism_length \n",
    "mutate    = mutate_bounded\n",
    "tournament_size = 4\n",
    "population_size = 50\n",
    "fitness_values = []\n",
    "length_values = []\n",
    "ga()\n",
    "plt.plot(fitness_values);"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "skip"
    }
   },
   "source": [
    "The search might not always achieve 100% coverage, but note that the number of generations of the algorithm is substantially larger than in the earlier plots, and we can also see that we have contained the problem of bloat."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true,
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "outputs": [],
   "source": [
    "plt.plot(length_values)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "skip"
    }
   },
   "source": [
    "There are some drawbacks to this approach: Once we have a test for a coverage goal, we are actually no longer interested in optimising tests for it (except for the case of minimising the overall test suite size). However, since we are optimising test suites, we always keep around all tests for all goals in the population. Usually, when you run whole test suite optimisation on our triangle example it will converge to cover most goals quickly except for the case of an equilateral triangle. Once it has reached that point the search needs to pick a test case to mutate that is not needed to satisfy another coverage goal, and then by chance mutate it such that the result is three equal sides. The chances of this happening are low, and the algorithm will not always produce 100% coverage unless waiting very long. We will therefore consider an improved approach next."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "## Many-Objective Optimisation"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "skip"
    }
   },
   "source": [
    "Since we are attempting to evolve tests for multiple coverage goals, it would be tempting to use multi-objective optimisation algorithms. However, the MOEAs we have considered previously are all based on the central concept of domination. With increasing number of objectives, however, dominance becomes increasingly rare, and there are too many non-dominated individuals. For example, in a population of size 20 with NSGA-II, after combining the parent and offspring populations, if there are more than 20 non-dominated individuals, then Pareto sorting has no effect on parent selection, and there exists no selective pressure towards the Pareto front. Therefore, traditional MOEAs are usually only applied to problems with 2-3 objectives. Optimisation with more than 3 objectives are known as _many-objective_ optimisation problems. There are extensions of the MOEAs we have considered to the many objective scenario such as NSGA-III or Two Archives2:\n",
    "\n",
    "- Deb, K., & Jain, H. (2013). An evolutionary many-objective optimization algorithm using reference-point-based nondominated sorting approach, part I: solving problems with box constraints. IEEE transactions on evolutionary computation, 18(4), 577-601.\n",
    "\n",
    "- Wang, H., Jiao, L., & Yao, X. (2014). Two_Arch2: An improved two-archive algorithm for many-objective optimization. IEEE Transactions on Evolutionary Computation, 19(4), 524-541.\n",
    "\n",
    "However, even such many objective optimisation algorithms can only cope with only up to ~20 objectives, whereas in test generation there can be hundreds of coverage goals to optimise for."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "skip"
    }
   },
   "source": [
    "### MOSA"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "skip"
    }
   },
   "source": [
    "The Many-Objective Sorting Algorithm (MOSA) is an extension of the NSGA-II algorithm specifically tailored for test generation.\n",
    "\n",
    "Panichella, A., Kifetew, F. M., & Tonella, P. (2015, April). Reformulating branch coverage as a many-objective optimization problem. In 2015 IEEE 8th international conference on software testing, verification and validation (ICST) (pp. 1-10). IEEE.\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "skip"
    }
   },
   "source": [
    "MOSA is built on NSGA-II, and includes a preference criterion: Given a branch $b$, a test case $x$ is preferred over another test case $y$ if and only if the values of the distance function $f$ for $b$ satisfy the condition $f(x) < f(y)$. This preference criterion is used to distinguish between test cases within a set of non-dominated individuals."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "skip"
    }
   },
   "source": [
    "In order to more conveniently handle the many-objective setting let's capture the notion of coverage goals explicitly (in the previous examples this was implicitly done in the fitness function, where we only considered an individual goal or the sum of all goals):"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "slideshow": {
     "slide_type": "fragment"
    }
   },
   "outputs": [],
   "source": [
    "class Goal:\n",
    "    def __init__(self, condition, value):\n",
    "        self.condition = condition\n",
    "        self.value = value\n",
    "    \n",
    "    def get_distance(self, test):\n",
    "        return test.distances[self]\n",
    "    \n",
    "    def is_coveredby(self, test):\n",
    "        return test.distances[self] == 0\n",
    "    \n",
    "    def __str__(self):\n",
    "        return f\"Condition {self.condition}/{self.value}\""
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "skip"
    }
   },
   "source": [
    "Our instrumentation gives us a number of branches, but technically, these are conditions and each condition can evaluate to true and to false, thus our coverage goals are as follows:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "slideshow": {
     "slide_type": "fragment"
    }
   },
   "outputs": [],
   "source": [
    "goals = [Goal(num, val) for num in range(num_conditions) for val in [True, False]]"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "len(goals)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "skip"
    }
   },
   "source": [
    "In order to avoid redundant fitness calculations, we will also use our wrapper class to cache fitness values:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "outputs": [],
   "source": [
    "class L(list):\n",
    "    \"\"\"\n",
    "    A subclass of list that can accept additional attributes.\n",
    "    Should be able to be used just like a regular list.\n",
    "    \"\"\"\n",
    "    def __new__(self, *args, **kwargs):\n",
    "        return super(L, self).__new__(self, args, kwargs)\n",
    "\n",
    "    def __init__(self, *args, **kwargs):\n",
    "        if len(args) == 1 and hasattr(args[0], '__iter__'):\n",
    "            list.__init__(self, args[0])\n",
    "        else:\n",
    "            list.__init__(self, args)\n",
    "        self.__dict__.update(kwargs)\n",
    "\n",
    "    def __call__(self, **kwargs):\n",
    "        self.__dict__.update(kwargs)\n",
    "        return self"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "skip"
    }
   },
   "source": [
    "In the many objective scenario, we are back to optimising _test cases_ rather than test suites:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "outputs": [],
   "source": [
    "def get_random_individual():\n",
    "    individual = L([random.randint(MIN, MAX) for _ in range(num_parameters)])\n",
    "    evaluate(individual)\n",
    "    return individual"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "slideshow": {
     "slide_type": "fragment"
    }
   },
   "outputs": [],
   "source": [
    "def mutate(solution):\n",
    "    P_mutate = 1/len(solution)\n",
    "    mutated = L(solution[:])\n",
    "    for position in range(len(solution)):\n",
    "        if random.random() < P_mutate:\n",
    "            if random.random() < 0.9:\n",
    "                mutated[position] = min(MAX, max(MIN, int(random.gauss(mutated[position], MAX*0.01))))\n",
    "            else:\n",
    "                mutated[position] = random.randint(MIN, MAX)\n",
    "            \n",
    "    evaluate(mutated)\n",
    "    return mutated"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "skip"
    }
   },
   "source": [
    "Let's also return to the regular single point crossover, since we are no longer optimising test suites, but test cases."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "outputs": [],
   "source": [
    "def singlepoint_crossover(parent1, parent2):\n",
    "    pos = random.randint(0, len(parent1))\n",
    "    offspring1 = L(parent1[:pos] + parent2[pos:])\n",
    "    offspring2 = L(parent2[:pos] + parent1[pos:])\n",
    "    return (offspring1, offspring2)\n",
    "\n",
    "crossover = singlepoint_crossover"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "skip"
    }
   },
   "source": [
    "We will simply use the branch distance for each of the goals again (technically, this could be refined to also include the approach level). Thus, to calculate all fitness values we need to execute a test, and store the according branch distances in a map:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "outputs": [],
   "source": [
    "def evaluate(individual):\n",
    "    global distances_true, distances_false\n",
    "    distances_true =  {x: 1.0 for x in range(num_conditions)}\n",
    "    distances_false = {x: 1.0 for x in range(num_conditions)}\n",
    "\n",
    "    triangle_instrumented(*individual)\n",
    "\n",
    "    individual.distances = {}\n",
    "    for goal in goals:\n",
    "        if goal.value:\n",
    "            individual.distances[goal] = distances_true[goal.condition]\n",
    "        else:\n",
    "            individual.distances[goal] = distances_false[goal.condition]"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "skip"
    }
   },
   "source": [
    "There are two major differences to NSGA-II now: Rather than just using the non-dominated sorted fronts directly, MOSA first applies the preference criterion to find the tests that are preferrable for the goals not yet covered. In order to keep track of which goals have already been satisfied, MOSA uses an archive to store satisfied coverage goals with the corresponding tests."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "outputs": [],
   "source": [
    "def update_archive(archive, population):\n",
    "    \n",
    "    # Ignores secondary criterion because length is fixed here\n",
    "    for test in population:\n",
    "        for goal in goals:\n",
    "            if goal in archive:\n",
    "                continue\n",
    "            if goal.is_coveredby(test):\n",
    "                archive[goal] = L(test[:])\n",
    "                print(f\"Covered goal {goal}: {test}\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "skip"
    }
   },
   "source": [
    "The MOSA algorithm is a variation of NSGA-II. We thus need to define the regular notion of dominance, adapted to our setting with `Goals`. In contrast to NSGA-II, however, the dominance relation only looks at uncovered goals. For this, we use our archive:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "outputs": [],
   "source": [
    "def dominates(solution1, solution2, archive):\n",
    "    for goal in goals:\n",
    "        if goal in archive:\n",
    "            continue # Already covered\n",
    "            \n",
    "        if solution1.distances[goal] > solution2.distances[goal]:\n",
    "            return False\n",
    "\n",
    "    for goal in goals:\n",
    "        if goal in archive:\n",
    "            continue # Already covered\n",
    "\n",
    "        if solution1.distances[goal] < solution2.distances[goal]:\n",
    "            return True\n",
    "\n",
    "    return False"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "skip"
    }
   },
   "source": [
    "Using the dominance relation, we can apply the fast non-dominated sort algorithm we already know from NSGA-II in order to sort a population into fronts. Again we only consider uncovered goals as defined by the archive:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "outputs": [],
   "source": [
    "def fast_non_dominated_sort(solutions, archive):\n",
    "    front = [[]]\n",
    "\n",
    "    S = [[] for _ in range(len(solutions))]\n",
    "    n = [0 for _ in range(len(solutions))]\n",
    "\n",
    "    for p in range(len(solutions)):\n",
    "        S[p] = []\n",
    "        n[p] = 0\n",
    "        for q in range(len(solutions)):\n",
    "            if dominates(solutions[p], solutions[q], archive):\n",
    "                S[p].append(q)\n",
    "            elif dominates(solutions[q], solutions[p], archive):\n",
    "                n[p] = n[p] + 1\n",
    "\n",
    "        if n[p] == 0:\n",
    "            front[0].append(p)\n",
    "            solutions[p].rank = 0\n",
    "\n",
    "    i = 0\n",
    "    while front[i]:\n",
    "        Q = []\n",
    "        for p in front[i]:\n",
    "            for q in S[p]:\n",
    "                n[q] = n[q] - 1\n",
    "                if n[q] == 0:\n",
    "                    Q.append(q)\n",
    "                    solutions[q].rank = i + 1\n",
    "        i = i + 1\n",
    "        front.append(Q)\n",
    "\n",
    "    del front[len(front) - 1]\n",
    "    return front"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "skip"
    }
   },
   "source": [
    "As the archive can tell us which goals are covered and which are not, the prefernce sorting can focus on those individuals that are not covered yet. For each uncovered goal, the best individual is put into a special first front. All other individuals are sorted according to regular non-dominated sorting:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "outputs": [],
   "source": [
    "def preference_sorting(population, archive):\n",
    "    # First front consists of best test per goal\n",
    "    \n",
    "    fronts = []\n",
    "    f0 = []\n",
    "\n",
    "    # Only consider uncovered goals\n",
    "    for goal in goals:\n",
    "        if goal not in archive:\n",
    "            best = min(population, key=lambda t: goal.get_distance(t))\n",
    "            if best not in f0:\n",
    "                f0.append(best)\n",
    "\n",
    "    for t in f0:\n",
    "        t.rank = 0\n",
    "    \n",
    "    fronts.append(f0)\n",
    "    \n",
    "    remaining = [x for x in population if not x in f0]\n",
    "    \n",
    "    # Remaining tests: \n",
    "    if remaining:\n",
    "        # Calculate non-dominance only using uncovered branches\n",
    "        remaining_fronts = fast_non_dominated_sort(remaining, archive)\n",
    "        for i in range(len(remaining_fronts)):\n",
    "            front = [remaining[index] for index in remaining_fronts[i]]\n",
    "            for t in front:\n",
    "                t.rank = i + 1\n",
    "            fronts.append(front)\n",
    "    \n",
    "    return fronts"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "skip"
    }
   },
   "source": [
    "The sorting of individuals within a front based on their crowding distance is the same we used previously, and only has to be adapted to consider our `Goal` implementation:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "outputs": [],
   "source": [
    "def calculate_crowding_distance_and_sort(front, archive):\n",
    "\n",
    "    if not front:\n",
    "        return\n",
    "    distance = [0 for _ in range(0,len(front))]\n",
    "\n",
    "    for goal in goals:\n",
    "        if goal in archive:\n",
    "            continue # Already covered\n",
    "\n",
    "        data = [(x, front[x].distances[goal]) for x in range(len(front))]\n",
    "        sorted_front = sorted(data, key=lambda tup: tup[1])\n",
    "        range_fitness = sorted_front[-1][1] - sorted_front[0][1]\n",
    "        if range_fitness > 0.0:\n",
    "            distance[sorted_front[0][0]] = sys.maxsize\n",
    "            distance[sorted_front[-1][0]] = sys.maxsize\n",
    "\n",
    "            for k in range(1,len(front)-1):\n",
    "                index = sorted_front[k][0]\n",
    "                distance[index] = distance[index] + (sorted_front[k+1][1] - sorted_front[k-1][1]) / range_fitness\n",
    "\n",
    "    for k in range(0, len(front)):\n",
    "        front[k].distance = distance[k]\n",
    "\n",
    "    front.sort(key = lambda i: i.distance, reverse=True)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "skip"
    }
   },
   "source": [
    "Since we now have ranks based on preference sorting and we have a crowding distance, we can use the binary rank tournament selection operator we already know from NSGA-II:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "outputs": [],
   "source": [
    "def binary_rank_tournament(population):\n",
    "    individual1 = random.choice(population)\n",
    "    individual2 = random.choice(population)\n",
    "\n",
    "    if individual1.rank < individual2.rank:\n",
    "        return individual1\n",
    "    elif individual1.rank > individual2.rank:\n",
    "        return individual2\n",
    "    else:\n",
    "        return max([individual1, individual2], key = lambda i: i.distance)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "skip"
    }
   },
   "source": [
    "Using the selection operator, we can also breed new generations:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "outputs": [],
   "source": [
    "def generate_offspring(population):\n",
    "    offspring_population = []\n",
    "    while len(offspring_population) < len(population):\n",
    "        parent1 = binary_rank_tournament(population)\n",
    "        parent2 = binary_rank_tournament(population)\n",
    "        if random.random() < P_xover:\n",
    "            offspring1, offspring2 = crossover(parent1, parent2)\n",
    "        else:\n",
    "            offspring1, offspring2 = parent1, parent2\n",
    "            \n",
    "        offspring1 = mutate(offspring1)\n",
    "        offspring2 = mutate(offspring2)\n",
    "\n",
    "        offspring_population.append(offspring1)\n",
    "        offspring_population.append(offspring2)\n",
    "\n",
    "    return offspring_population"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "skip"
    }
   },
   "source": [
    "We also need to derive the initial population:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "outputs": [],
   "source": [
    "def get_initial_population(archive):\n",
    "    population = [get_random_individual() for _ in range(population_size)]\n",
    "    update_archive(archive, population)\n",
    "    fronts = preference_sorting(population, archive)\n",
    "    for front in fronts:\n",
    "        calculate_crowding_distance_and_sort(front, archive)\n",
    "\n",
    "    return population"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "skip"
    }
   },
   "source": [
    "Now we have all components in place and can implement the actual algorithm, which only slightly differs from NSGA-II. The final solution is represented by the contents of the archive."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "outputs": [],
   "source": [
    "def mosa():\n",
    "    archive = {}\n",
    "    population = get_initial_population(archive)\n",
    "\n",
    "    for iteration in range(max_gen):\n",
    "        uncovered = sum([1 for x in goals if x not in archive])\n",
    "        if uncovered > 0:\n",
    "            print(f\"Iteration {iteration}, missing goals: {uncovered}\")\n",
    "\n",
    "        offspring_population = generate_offspring(population)\n",
    "        combined = population + offspring_population\n",
    "        \n",
    "        # TODO: Explain\n",
    "        update_archive(archive, offspring_population)\n",
    "        \n",
    "        fronts = preference_sorting(combined, archive)\n",
    "        population = []\n",
    "\n",
    "        for front in fronts:\n",
    "            calculate_crowding_distance_and_sort(front, archive)\n",
    "\n",
    "            for t in front:\n",
    "                population.append(t)\n",
    "                if len(population) == population_size:\n",
    "                    break\n",
    "            if len(population) == population_size:\n",
    "                break\n",
    "\n",
    "\n",
    "    return archive"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true,
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "outputs": [],
   "source": [
    "max_gen=500\n",
    "MAX=10000\n",
    "MIN=-MAX\n",
    "result = mosa()\n",
    "print(f\"Covered {len(result)} out of {len(goals)} goals\")\n",
    "for goal, test in result.items():\n",
    "    print(f\"Goal {goal}: {test}\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "### MIO"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "skip"
    }
   },
   "source": [
    "The Many Independent Objective search algorithm (MIO) addresses the problem that in test generation the number of objectives can be very large, and even with MOSA it will likely not be possible to achieve 100% coverage. Instead, the search tries to focus on those goals that are most promising, given the resources available.\n",
    "\n",
    "Arcuri, A. (2017, September). Many independent objective (MIO) algorithm for test suite generation. In International Symposium on Search Based Software Engineering (pp. 3-17). Springer, Cham.\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "skip"
    }
   },
   "source": [
    "The MIO algorithm is based on a simple (1+1)EA and maintains an archive of tests. In the archive, for each coverage goal there is a population of $n$ tests. Thus, for $z$ coverage goals there can be up to $n×z$ tests in the archive\n",
    "at the same time. The archive is initially empty, and the first individual is generated randomly. Once there are tests in the archive, MIO probabilistically either samples a new random test, or chooses a test from the archive and mutates a copy of it. For each new/mutated test the fitness is calculated, and the test is added to the archive if needed."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "skip"
    }
   },
   "source": [
    "Updating the archive with a test is based a heuristic function that is calculated for each coverage goal. If a goal is covered by a new test, then the archive for that test is replaced with that individual test. If the goal is already covered, then the existing covering test is only replaced if it is better in some other dimension (e.g. size, although that does not apply in our current scenario of the triangle program). If the heuristic value is the worst possible value (e.g., the relevant branch predicate is not executed at all and there is no branch distance), then that test is not a candidate for the archive of that goal; else, the test is added to the archive for that goal if the maximum number of tests in that archive has not exceeded a maximum size. If the maximum size is reached, then the new test replaces the worst test in the archive, if it has a better heuristic value."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "outputs": [],
   "source": [
    "archive_size = 10\n",
    "def update_mio_archive(archive, t):\n",
    "    for goal in goals:\n",
    "        if goal in archive and len(archive[goal]) == 1 and goal.is_coveredby(archive[goal][0]):\n",
    "            # Goal is already covered\n",
    "            continue\n",
    "        h = goal.get_distance(t)\n",
    "        if h == 0:\n",
    "            # Keep only that test, remove all others            \n",
    "            print(f\"Goal {goal} covered by {t}\")\n",
    "            archive[goal] = [t]\n",
    "\n",
    "        elif h < 1:\n",
    "            if goal not in archive:\n",
    "                archive[goal] = [t]\n",
    "            elif len(archive[goal]) < archive_size and not t in archive[goal]:\n",
    "                archive[goal].append(t)\n",
    "            elif h < goal.get_distance(archive[goal][-1]):\n",
    "                # Replace worst, if this one is better\n",
    "                del archive[goal][-1]\n",
    "                archive[goal].append(t)\n",
    "                \n",
    "            archive[goal].sort(key = lambda t: goal.get_distance(t))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "skip"
    }
   },
   "source": [
    "When sampling a test from the archive in order to produce a new offspring, the algorithm first randomly selects an uncovered goal, and then randomly selects a test for that goal from the archive. If the archive contains no tests for uncovered goals, then it randomly selects a covered goal and uses that test."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "outputs": [],
   "source": [
    "def sample_archive(archive):\n",
    "    uncovered_goals = []\n",
    "    for goal, tests in archive.items():\n",
    "        if len(tests) == 1 and goal.is_coveredby(tests[0]):\n",
    "            continue\n",
    "        elif len(tests) > 0:\n",
    "            uncovered_goals.append(goal)\n",
    "            \n",
    "    if not uncovered_goals:\n",
    "        for goal, tests in archive.items():\n",
    "            if len(tests) > 0:\n",
    "                uncovered_goals.append(goal)\n",
    "                \n",
    "    goal = random.choice(uncovered_goals)\n",
    "    \n",
    "    # Now choose one test randomly for \"goal\" in archive\n",
    "    return random.choice(archive[goal])"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "skip"
    }
   },
   "source": [
    "As a further improvement of the sampling process, MIO implements feedback directed sampling: The idea is that goals for which we recently achieved an improvement should be preferred, since there is potential to reach a test there, whereas goals for which no improvement has recently been achieved are either difficult or infeasible, and should thus be postponed until the easier targets have been covered. To achieve this, we add a `sample_counter` which counts how often sampling was applied to each goal. Whenever a new test for a goal is added to the archive, the counter is reset to 0. Then, we can rank the goals by their counter, and choose the goal with the lowest counter. To integrate the counter, we need to update our `update_archive` function. While making this change, we also keep track if the archive has been changed, which we will need later:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "outputs": [],
   "source": [
    "sample_counter = {}"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "slideshow": {
     "slide_type": "fragment"
    }
   },
   "outputs": [],
   "source": [
    "archive_size = 10\n",
    "def update_mio_archive(archive, t):\n",
    "    updated = False\n",
    "    for goal in goals:\n",
    "        if goal in archive and len(archive[goal]) == 1 and goal.is_coveredby(archive[goal][0]):\n",
    "            # Goal is already covered\n",
    "            continue\n",
    "        h = goal.get_distance(t)\n",
    "        if h == 0:\n",
    "            # Keep only that test, remove all others            \n",
    "            print(f\"Goal {goal} covered by {t}\")\n",
    "            sample_counter[goal] = 0\n",
    "            archive[goal] = [t]\n",
    "            updated = True\n",
    "\n",
    "        elif h < 1:\n",
    "            if goal not in archive:\n",
    "                sample_counter[goal] = 0\n",
    "                archive[goal] = [t]\n",
    "                updated = True\n",
    "            elif len(archive[goal]) < archive_size and not t in archive[goal]:\n",
    "                sample_counter[goal] = 0\n",
    "                updated = True\n",
    "                archive[goal].append(t)\n",
    "            elif h < goal.get_distance(archive[goal][-1]):\n",
    "                # Replace worst, if this one is better\n",
    "                del archive[goal][-1]\n",
    "                sample_counter[goal] = 0\n",
    "                archive[goal].append(t)\n",
    "                updated = True\n",
    "                \n",
    "            archive[goal].sort(key = lambda t: goal.get_distance(t))\n",
    "            \n",
    "    return updated"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "skip"
    }
   },
   "source": [
    "Accordingly, we can now update the sampling method to consider the counter:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "outputs": [],
   "source": [
    "def sample_archive(archive):\n",
    "    uncovered_goals = []    \n",
    "    for goal, tests in archive.items():\n",
    "        if len(tests) == 1 and goal.is_coveredby(tests[0]):\n",
    "            continue\n",
    "        elif len(tests) > 0:\n",
    "            uncovered_goals.append(goal)\n",
    "            \n",
    "    if not uncovered_goals:        \n",
    "        for goal, tests in archive.items():            \n",
    "            if len(tests) > 0:\n",
    "                uncovered_goals.append(goal)\n",
    "    \n",
    "    # Shuffle first such that in the case of ties\n",
    "    # we still have some variation\n",
    "    random.shuffle(uncovered_goals)\n",
    "    uncovered_goals.sort(key = lambda g: sample_counter[g])\n",
    "\n",
    "    goal = uncovered_goals[0]\n",
    "    sample_counter[goal] += 1\n",
    "    \n",
    "    # Now choose one test randomly for \"goal\" in archive\n",
    "    return random.choice(archive[goal])"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "skip"
    }
   },
   "source": [
    "At the end of the search, for each goal which has exactly one test in the archive and that test covers the goal, this test is part of the resulting test suite:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "outputs": [],
   "source": [
    "def get_solutions(archive):\n",
    "    test_suite = {}\n",
    "    for goal, tests in archive.items():\n",
    "        if len(tests) == 1 and goal.is_coveredby(tests[0]):\n",
    "            test_suite[goal] = tests[0]\n",
    "    return test_suite"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "skip"
    }
   },
   "source": [
    "The algorithm itself is now very simple:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "outputs": [],
   "source": [
    "P_sample = 0.5\n",
    "\n",
    "def mio():\n",
    "    archive = {}\n",
    "    \n",
    "    for step in range(max_gen * population_size):\n",
    "        if random.random() < P_sample or len(archive) == 0:\n",
    "            current = get_random_individual()\n",
    "        else:\n",
    "            current = sample_archive(archive)\n",
    "            current = mutate(current)\n",
    "\n",
    "        update_mio_archive(archive, current)\n",
    "        \n",
    "    return get_solutions(archive)\n",
    "    "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true,
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "outputs": [],
   "source": [
    "sample_counter = {}\n",
    "result = mio()\n",
    "print(f\"Covered {len(result)}/{len(goals)} goals\")\n",
    "for goal, test in result.items():\n",
    "    print(f\"Goal {goal}: {test}\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "skip"
    }
   },
   "source": [
    "As a further refinement, MIO can take into account that, during initial phases of the search more exploration is generally needed, while later phases should focus on exploitation. More exploration means that the probability of sampling a new, random test should be higher, while focusing on exploitation would reduce that probability and focus on improving fewer existing individuals, thus also reducing the archive size. MIO therefore defines two sets of parameters: The probability of sampling new tests, the size of the archive, and the number of mutations applied to the same individual with their initial values, and their values once the focused phase begins:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "outputs": [],
   "source": [
    "P_sample_focused = 0\n",
    "P_sample_first = 0.5\n",
    "\n",
    "archive_size_focused = 1\n",
    "archive_size_first = 10\n",
    "\n",
    "mutations_focused = 10\n",
    "mutations_first = 1\n",
    "\n",
    "# Percentage of search budget spent when focused phase should begin\n",
    "F = 0.5"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "skip"
    }
   },
   "source": [
    "From the initial phase to the beginning of the focused phase, the parameters are gradually moved towards their focused values:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "outputs": [],
   "source": [
    "focused = False\n",
    "def update_parameters(steps, archive):\n",
    "    global focused, P_sample, archive_size, num_mutations, F\n",
    "    progress = steps / (max_gen * population_size)\n",
    "    progress_until_focused = progress / F;\n",
    "    if progress > F:\n",
    "        P_sample = P_sample_focused\n",
    "        archive_size = archive_size_focused\n",
    "        num_mutations = mutations_focused\n",
    "        if not focused:\n",
    "            print(f\"Entering focused phase\")\n",
    "            focused = True\n",
    "    else:\n",
    "        pps = P_sample\n",
    "        pas = archive_size\n",
    "        pnm = num_mutations\n",
    "        \n",
    "        P_sample = P_sample_first + (P_sample_focused - P_sample_first) * progress_until_focused\n",
    "        archive_size = int(archive_size_first + (archive_size_focused - archive_size_first) * progress_until_focused)\n",
    "        num_mutations = int(mutations_first + (mutations_focused - mutations_first) * progress_until_focused)\n",
    "        \n",
    "        if pas != archive_size:\n",
    "            # Crop archives if necessary\n",
    "            for goal in goals:\n",
    "                if goal in archive:\n",
    "                    while len(archive[goal]) > archive_size:\n",
    "                        # Delete worst\n",
    "                        del archive[goal][-1]"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "skip"
    }
   },
   "source": [
    "The updated MIO algorithm needs to update the parameters, and also consider the case that the current individual can be mutated multiple times:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "outputs": [],
   "source": [
    "import copy\n",
    "P_sample = 0.7\n",
    "archive_size = 10\n",
    "num_mutations = 1\n",
    "F = 0.5\n",
    "def mio():\n",
    "    sample_counter = {}\n",
    "    archive = {}\n",
    "    update_parameters(0, archive)\n",
    "    current_mutations = 0\n",
    "    current = None\n",
    "    \n",
    "    for step in range(max_gen * population_size):\n",
    "        if current and current_mutations < num_mutations:\n",
    "            offspring = mutate(current)\n",
    "            current_mutations += 1\n",
    "        elif random.random() < P_sample or len(archive) == 0:\n",
    "            offspring = get_random_individual()\n",
    "            current_mutations = 1\n",
    "        else:\n",
    "            offspring = sample_archive(archive)\n",
    "            offspring = mutate(offspring)\n",
    "            current_mutations = 1\n",
    "\n",
    "        update_parameters(step, archive)\n",
    "        result = update_mio_archive(archive, offspring)\n",
    "        if result:\n",
    "            current = copy.deepcopy(offspring)\n",
    "        \n",
    "    return get_solutions(archive)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": false,
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "outputs": [],
   "source": [
    "sample_counter = {}\n",
    "focused = False\n",
    "result = mio()\n",
    "print(f\"Covered {len(result)}/{len(goals)} goals\")\n",
    "for goal, test in result.items():\n",
    "    print(f\"Goal {goal}: {test}\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "### Evaluation"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "skip"
    }
   },
   "source": [
    "The hypervolume is less suitable to evaluate the results of these algorithms -- in MIO we don't have an explicit Pareto front, and even though in MOSA we do have a first front, test cases that do not quite cover a coverage goal are of no practical use. The metric we therefore consider is the number of coverage goals satisfied. For our implementations of MIO and MOSA we can see this number from the number of tests in a test suite."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "outputs": [],
   "source": [
    "from IPython.utils import io\n",
    "\n",
    "def run_times(algorithm, repetitions):\n",
    "    global fitness_values, diversity_values, ims\n",
    "    result = []\n",
    "    for i in range(repetitions):\n",
    "        with io.capture_output() as captured: \n",
    "            testsuite = algorithm()\n",
    "        result.append(len(testsuite) / len(goals))\n",
    "    return result"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "skip"
    }
   },
   "source": [
    "Let's increase the difficulty of the search problem somewhat so we can actually see some differences:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "slideshow": {
     "slide_type": "fragment"
    }
   },
   "outputs": [],
   "source": [
    "MAX=1000000\n",
    "MIN=-MAX"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "outputs": [],
   "source": [
    "results = {\"MIO\"  : run_times(mio, 10), \n",
    "           \"MOSA\" : run_times(mosa, 10)}\n",
    "fig, ax = plt.subplots()\n",
    "ax.boxplot(results.values())\n",
    "ax.set_xticklabels(results.keys())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "### Variable Length Tests"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "skip"
    }
   },
   "source": [
    "The `cgi_decode()` function originates in the book _Pezzè, M., & Young, M. (2008). Software testing and analysis: process, principles, and techniques. John Wiley & Sons._, and we will use the Python implementation used in the [Fuzzingbook](https://www.fuzzingbook.org/). It takes as input a string using the CGI encoding common for HTTP-Get queries, and returns a decoded string."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "slideshow": {
     "slide_type": "fragment"
    }
   },
   "outputs": [],
   "source": [
    "def cgi_decode(s):\n",
    "    \"\"\"Decode the CGI-encoded string `s`:\n",
    "       * replace \"+\" by \" \"\n",
    "       * replace \"%xx\" by the character with hex number xx.\n",
    "       Return the decoded string.  Raise `ValueError` for invalid inputs.\"\"\"\n",
    "\n",
    "    # Mapping of hex digits to their integer values\n",
    "    hex_values = {\n",
    "        '0': 0, '1': 1, '2': 2, '3': 3, '4': 4,\n",
    "        '5': 5, '6': 6, '7': 7, '8': 8, '9': 9,\n",
    "        'a': 10, 'b': 11, 'c': 12, 'd': 13, 'e': 14, 'f': 15,\n",
    "        'A': 10, 'B': 11, 'C': 12, 'D': 13, 'E': 14, 'F': 15,\n",
    "    }\n",
    "\n",
    "    t = \"\"\n",
    "    i = 0\n",
    "    while i < len(s):\n",
    "        c = s[i]\n",
    "        if c == '+':\n",
    "            t += ' '\n",
    "        elif c == '%':\n",
    "            digit_high, digit_low = s[i + 1], s[i + 2]\n",
    "            i += 2\n",
    "            if digit_high in hex_values and digit_low in hex_values:\n",
    "                v = hex_values[digit_high] * 16 + hex_values[digit_low]\n",
    "                t += chr(v)\n",
    "            else:\n",
    "                raise ValueError(\"Invalid encoding\")\n",
    "        else:\n",
    "            t += c\n",
    "        i += 1\n",
    "    return t"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "outputs": [],
   "source": [
    "cgi_decode(\"Hello\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "slideshow": {
     "slide_type": "fragment"
    }
   },
   "outputs": [],
   "source": [
    "cgi_decode(\"Hello+World\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "slideshow": {
     "slide_type": "fragment"
    }
   },
   "outputs": [],
   "source": [
    "cgi_decode(\"Hello+World%21\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "skip"
    }
   },
   "source": [
    "As usual, we can apply the source transformation to calculate our branch distances and determine the total number of goals:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "outputs": [],
   "source": [
    "source = inspect.getsource(cgi_decode)\n",
    "node = ast.parse(source)\n",
    "transformer = BranchTransformer()\n",
    "transformer.visit(node)\n",
    "node = ast.fix_missing_locations(node)\n",
    "num_conditions = transformer.condition_num\n",
    "print(astor.to_source(node))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "outputs": [],
   "source": [
    "create_instrumented_function(cgi_decode)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "slideshow": {
     "slide_type": "fragment"
    }
   },
   "outputs": [],
   "source": [
    "goals = [Goal(num, val) for num in range(num_conditions) for val in [True, False]]\n",
    "len(goals)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "skip"
    }
   },
   "source": [
    "In order to apply search to this problem, we first need to update our `evaluate` function so that it calls `cgi_decoded_instrumented`. We also need to take into account that calling the function may throw an exception:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "outputs": [],
   "source": [
    "def evaluate(individual):\n",
    "    global distances_true, distances_false\n",
    "    distances_true =  {x: 1.0 for x in range(num_conditions)}\n",
    "    distances_false = {x: 1.0 for x in range(num_conditions)}\n",
    "\n",
    "    try:\n",
    "        cgi_decode_instrumented(''.join(individual))\n",
    "    except:\n",
    "        pass\n",
    "\n",
    "    individual.distances = {}\n",
    "    for goal in goals:\n",
    "        if goal.value:\n",
    "            individual.distances[goal] = distances_true[goal.condition]\n",
    "        else:\n",
    "            individual.distances[goal] = distances_false[goal.condition]"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "skip"
    }
   },
   "source": [
    "Now we need to decide on a representation. A simple way to encode strings is as a list of characters; this way, we don't need to make many changes to our existing operators. As another modification, we restrict the range to printable characters. However, what length should we use for our tests? If we fix the length, then we might end up with redundantly long strings. Therefore, let's use a variable size representation:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "outputs": [],
   "source": [
    "string_length = 10\n",
    "def get_random_individual():\n",
    "    s = []\n",
    "    length = random.randint(1, string_length)\n",
    "    for i in range(string_length):\n",
    "        s.append(chr(random.randrange(32, 127)))\n",
    "    individual = L(s)\n",
    "\n",
    "    evaluate(individual)\n",
    "    return individual"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "outputs": [],
   "source": [
    "def mutate(solution):\n",
    "    P_mutate = 1/len(solution) if solution else 0\n",
    "    result = []\n",
    "    for position in range(len(solution)):\n",
    "        if random.random() > P_mutate:\n",
    "            result.append(solution[position])\n",
    "    for position in range(len(result)):\n",
    "        if random.random() < P_mutate:\n",
    "            result[position] = chr(int(random.gauss(ord(result[position]), 100) % 65536))\n",
    "    ALPHA = 0.1\n",
    "    count = 1\n",
    "    while random.random() < ALPHA ** count and len(result) < string_length:\n",
    "        count += 1\n",
    "        pos = random.randint(0, len(result))\n",
    "        result.insert(pos, chr(random.randrange(32, 127)))\n",
    "            \n",
    "    mutated = L(result)\n",
    "    evaluate(mutated)\n",
    "    return mutated"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "skip"
    }
   },
   "source": [
    "MOSA and MIO both allow for secondary optimisation criteria: If we have already covered a test goal but have now found a new test that is shorter, we can replace the existing test in the archive. We thus need to update the archive updating functions of both algorithms to take this into account:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "outputs": [],
   "source": [
    "def update_archive(archive, population):\n",
    "    \n",
    "    for test in population:\n",
    "        for goal in goals:\n",
    "            if goal.is_coveredby(test):\n",
    "                if not goal in archive or len(test) < len(archive[goal]):\n",
    "                    archive[goal] = L(test[:])\n",
    "                    print(f\"Covered goal {goal}: {test}\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "outputs": [],
   "source": [
    "def update_mio_archive(archive, t):\n",
    "    updated = False\n",
    "    for goal in goals:\n",
    "        h = goal.get_distance(t)\n",
    "        if h == 0:\n",
    "            if not goal in archive or not goal.is_coveredby(archive[goal][0]):\n",
    "                print(f\"Goal {goal} covered by {t}: {goal in archive}\")\n",
    "                sample_counter[goal] = 0\n",
    "                archive[goal] = [t]\n",
    "                updated = True\n",
    "            elif goal.is_coveredby(archive[goal][0]):\n",
    "                # Goal is already covered\n",
    "                if len(t) < len(archive[goal][0]):\n",
    "                    sample_counter[goal] = 0\n",
    "                    archive[goal] = [t]\n",
    "                    updated = True\n",
    "\n",
    "        elif h < 1:\n",
    "            if goal in archive and len(archive[goal]) == 1 and goal.is_coveredby(archive[goal][0]):\n",
    "                # Goal is already covered\n",
    "                continue\n",
    "            if goal not in archive:\n",
    "                sample_counter[goal] = 0\n",
    "                archive[goal] = [t]\n",
    "                updated = True\n",
    "            elif len(archive[goal]) < archive_size and not t in archive[goal]:\n",
    "                sample_counter[goal] = 0\n",
    "                updated = True\n",
    "                archive[goal].append(t)\n",
    "            elif h < goal.get_distance(archive[goal][-1]):\n",
    "                # Replace worst, if this one is better\n",
    "                del archive[goal][-1]\n",
    "                sample_counter[goal] = 0\n",
    "                archive[goal].append(t)\n",
    "                updated = True\n",
    "                \n",
    "            archive[goal].sort(key = lambda t: goal.get_distance(t))\n",
    "            \n",
    "    return updated"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "skip"
    }
   },
   "source": [
    "Without any further changes, we can now apply MOSA and MIO on this test generation problem:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true,
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "outputs": [],
   "source": [
    "result = mio()\n",
    "print(len(result))\n",
    "for goal, test in result.items():\n",
    "    print(f\"Goal {goal}: {''.join(test)}\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": true,
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "outputs": [],
   "source": [
    "result = mosa()\n",
    "print(len(result))\n",
    "for goal, test in result.items():\n",
    "    print(f\"Goal {goal}: {''.join(test)}\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "skip"
    }
   },
   "source": [
    "In the final test suites, there should be tests containing the character `+`, there should be invalid hex codes, valid hexcodes, and some non-escaped characters. There will be some redundancy in the overall test suite: The same test may cover multiple goals. Minimising the final test suite further, however, is another optimisation problem."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "outputs": [],
   "source": [
    "results = {\"MIO\"  : run_times(mio, 10), \n",
    "           \"MOSA\" : run_times(mosa, 10)}\n",
    "fig, ax = plt.subplots()\n",
    "ax.boxplot(results.values())\n",
    "ax.set_xticklabels(results.keys())"
   ]
  }
 ],
 "metadata": {
  "celltoolbar": "Slideshow",
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.10.5"
  },
  "rise": {
   "enable_chalkboard": true,
   "scroll": true
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}