Stand alone delta debugging command

perun/cli.py

@@ -76,6 +76,7 @@
 import perun.profile.helpers as profiles
 import perun.view
 import perun.view_diff
+import perun.deltadebugging.factory as delta
 
 
 DEV_MODE = False
@@ -1268,6 +1269,38 @@ def fuzz_cmd(cmd: str, **kwargs: Any) -> None:
     fuzz.run_fuzzing_for_command(**kwargs)
 
 
+@cli.command()
+@click.argument("cmd", nargs=1, required=True)
+@click.argument(
+    "input-sample",
+    nargs=1,
+    required=True,
+)
+@click.argument(
+    "output-dir",
+    nargs=1,
+    required=True,
+    type=click.Path(exists=True, writable=True),
+    metavar="<path>",
+)
+@click.option(
+    "--timeout",
+    "-t",
+    nargs=1,
+    required=False,
+    default=1.0,
+    type=click.FloatRange(0.001, None, False),
+    metavar="<float>",
+    help="Time limit for delta debugging (in seconds). Default value is 1s.",
+)
+def deltadebugging(cmd: str, input_sample: str, output_dir: str, **kwargs: Any) -> None:
+    """Performs delta debugging on this command."""
+    kwargs["executable"] = Executable(cmd)
+    kwargs["input_sample"] = input_sample
+    kwargs["output_dir"] = output_dir
+    delta.run_delta_debugging_for_command(**kwargs)
+
+
 def init_unit_commands(lazy_init: bool = True) -> None:
     """Runs initializations for all of subcommands (shows, collectors, postprocessors)
 

perun/deltadebugging/__init__.py

@@ -0,0 +1,5 @@
+"""Collective package for delta debugging
+
+Contains the core delta debugging algorithm for various types of inputs,
+the debugging loop, and strategies/heuristics for minimizing failure-inducing
+inputs to their simplest forms."""

perun/deltadebugging/factory.py

@@ -0,0 +1,77 @@
+"""Collection of global methods for delta debugging.
+https://www.debuggingbook.org/html/DeltaDebugger.html
+"""
+
+from __future__ import annotations
+
+import subprocess
+from perun.utils.external import commands as external_commands
+from perun.utils import log
+from typing import Any, TYPE_CHECKING
+from pathlib import Path
+from perun.utils.common import common_kit
+
+if TYPE_CHECKING:
+    from perun.utils.structs import Executable
+
+
+def run_delta_debugging_for_command(
+    executable: Executable,
+    input_sample: str,
+    **kwargs: Any,
+) -> None:
+
+    timeout: float = kwargs.get("timeout", 1.0)
+    output_dir = Path(kwargs["output_dir"]).resolve()
+    read_input(input_sample)
+    inp = read_input(input_sample)
+    n = 2  # granularity
+
+    while len(inp) >= 2:
+        start = 0
+        subset_length = int(len(inp) / n)
+        program_fails = False
+
+        while start < len(inp):
+            complement = inp[: int(start)] + inp[int(start + subset_length) :]
+            try:
+                full_cmd = f"{executable} {complement}"
+                external_commands.run_safely_external_command(full_cmd, True, True, timeout)
+
+            except subprocess.TimeoutExpired:
+                inp = complement
+                n = max(n - 1, 2)
+                program_fails = True
+                break
+            start += subset_length
+
+        if not program_fails:
+            if n == len(inp):
+                break
+            n = min(n * 2, len(inp))
+
+    log.minor_info("shortest failing input = " + inp)
+    return create_debugging_file(output_dir, input_sample, inp)
+
+
+def read_input(input_file: str) -> str:
+    input_path = Path(input_file)
+    if input_path.is_file():
+        with open(input_path, "r") as file:
+            input_value = file.read()
+    else:
+        input_value = input_file
+
+    return input_value
+
+
+def create_debugging_file(output_dir: Path, file_name: str, input_data: str) -> None:
+    output_dir = output_dir.resolve()
+    dir_name = "delta_debugging"
+    full_dir_path = output_dir / dir_name
+    file_path = Path(file_name)
+    common_kit.touch_dir(str(full_dir_path))
+    file_path = full_dir_path / file_path.name
+
+    with open(file_path, "w") as file:
+        file.write(input_data)

perun/deltadebugging/meson.build

@@ -0,0 +1,11 @@
+perun_deltadebugging_dir = perun_dir / 'deltadebugging'
+
+perun_delta_files = files(
+    '__init__.py',
+    'factory.py',
+)
+
+py3.install_sources(
+    perun_delta_files,
+    subdir: perun_deltadebugging_dir,
+)

perun/meson.build

@@ -12,6 +12,7 @@ subdir('check')
 subdir('cli_groups')
 subdir('collect')
 subdir('fuzz')
+subdir('deltadebugging')
 subdir('logic')
 subdir('postprocess')
 subdir('profile')

tests/sources/delta_debugging_examples/dd-minimal/Makefile

@@ -0,0 +1,10 @@
+CC=gcc
+PROG=dd-minimal
+
+all: $(PROG)
+
+dd-minimal: main.c
+	$(CC) $< -g --coverage -o $@
+
+clean:
+	rm -rf $(PROG) *.gc*

tests/sources/delta_debugging_examples/dd-minimal/main.c

@@ -0,0 +1,34 @@
+#include <stdio.h>
+#include <stdlib.h>
+#include <unistd.h>
+#include <string.h>
+
+#define MAGIC_NUMBER 100
+
+void magicLoop() {
+    for (int i = 0; i < MAGIC_NUMBER; ++i) {
+        usleep(500000);
+    }
+}
+
+void checkInputString(const char* inputString) {
+    int count = 0;
+    for (int i = 0; inputString[i] != '\0'; ++i) {
+        char character = inputString[i];
+        if (character == '-') {
+            ++count;
+            if (count == 3) {
+                magicLoop();
+            }
+        }
+    }
+}
+
+int main(int argc, char* argv[]) {
+    if (argc != 2) {
+        return 1;
+    }
+
+    checkInputString(argv[1]);
+    return 0;
+}

tests/sources/delta_debugging_examples/samples/txt/simple.txt

@@ -0,0 +1 @@
+H---ello-Wo--rld

tests/test_delta_debugging.py

@@ -0,0 +1,41 @@
+from __future__ import annotations
+
+# Standard Imports
+from pathlib import Path
+import subprocess
+
+# Third-Party Imports
+import pytest
+from click.testing import CliRunner
+
+# Perun Imports
+from perun import cli
+from perun.testing import asserts
+
+
+@pytest.mark.usefixtures("cleandir")
+def test_delta_debugging_correct():
+    """Runs basic tests for delta debugging CLI"""
+    runner = CliRunner()
+    examples = Path(__file__).parent / "sources" / "delta_debugging_examples"
+    num_workload = examples / "samples" / "txt" / "simple.txt"
+
+    # 08. Testing for delta debugging minimal test
+    process = subprocess.Popen(["make", "-C", examples / "dd-minimal"])
+    process.communicate()
+    process.wait()
+
+    delta_debugging_test = examples / "dd-minimal" / "dd-minimal"
+    result = runner.invoke(
+        cli.deltadebugging,
+        [
+            str(delta_debugging_test),
+            str(num_workload),
+            str(examples),
+            "-t",
+            "2",
+        ],
+    )
+
+    asserts.predicate_from_cli(result, result.exit_code == 0)
+    asserts.predicate_from_cli(result, "---" in result.output)