ci: generate source file/test db

Signed-off-by: Anas Nashif <[email protected]>

Author: nashif

scripts/ci/db/find_test_combo.py

@@ -0,0 +1,87 @@
+#!/usr/bin/env python3
+
+import json
+import os
+import re
+import sys
+from pathlib import Path
+from collections import defaultdict
+import argparse
+
+if "ZEPHYR_BASE" not in os.environ:
+    exit("$ZEPHYR_BASE environment variable undefined.")
+
+# These are globally used variables. They are assigned in __main__ and are visible in further methods
+# however, pylint complains that it doesn't recognize them when used (used-before-assignment).
+zephyr_base = Path(os.environ['ZEPHYR_BASE'])
+
+sys.path.insert(0, os.path.join(zephyr_base / "scripts"))
+from get_maintainer import Maintainers
+
+def load_database(database_path):
+    with open(database_path, 'r') as file:
+        return json.load(file)
+
+def find_areas(files):
+    maintf = zephyr_base / "MAINTAINERS.yml"
+    maintainer_file = Maintainers(maintf)
+
+    num_files = 0
+    all_areas = set()
+
+    for changed_file in files:
+        num_files += 1
+        print(f"file: {changed_file}")
+        areas = maintainer_file.path2areas(changed_file)
+
+        if not areas:
+            continue
+        all_areas.update(areas)
+    tests = []
+    for area in all_areas:
+        for suite in area.tests:
+            tests.append(f"{suite}.*")
+    return tests
+
+def find_best_coverage(database, changed_files, tests):
+    coverage = defaultdict(lambda: defaultdict(int))
+
+    for file in changed_files:
+        if file in database:
+            for entry in database[file]:
+                if not any(re.search(f"^{test}", entry["testsuite_id"]) for test in tests):
+                    print("skip")
+                    continue
+                testsuite_id = entry["testsuite_id"]
+                platform = entry["platform"]
+                coverage[testsuite_id][platform] += 1
+
+    best_coverage = []
+    for testsuite_id, platforms in coverage.items():
+        for platform, count in platforms.items():
+            best_coverage.append((testsuite_id, platform, count))
+
+    best_coverage.sort(key=lambda x: x[2], reverse=True)
+    return best_coverage
+
+def main(database_path, changed_files):
+    tests = find_areas(changed_files)
+    database = load_database(database_path)
+    best_coverage = find_best_coverage(database, changed_files, tests)
+
+    if best_coverage:
+        print("Best coverage testsuites and platforms:")
+        for testsuite_id, platform, count in best_coverage:
+            print(f"Testsuite: {testsuite_id}, Platform: {platform}, Coverage: {count}")
+    else:
+        print("No matching testsuites found for the provided files.")
+
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser(description="Find the best coverage testsuites and platforms for changed files.")
+    parser.add_argument("--database", help="Path to the testsuite database JSON file")
+    parser.add_argument("--changed-files", action="append", help="List of changed files")
+
+    args = parser.parse_args()
+
+    main(args.database, args.changed_files)

scripts/ci/db/gen_test_database.py

@@ -0,0 +1,40 @@
+#!/usr/bin/env python3
+import os
+import json
+import argparse
+
+def find_compile_commands_files(directory):
+    compile_commands_files = []
+    for root, _, files in os.walk(directory):
+        for file in files:
+            if file.startswith("compile_commands_") and file.endswith(".json"):
+                compile_commands_files.append(os.path.join(root, file))
+    return compile_commands_files
+
+def generate_testsuite_database(directory, output_file):
+    compile_commands_files = find_compile_commands_files(directory)
+    database = {}
+
+    for file_path in compile_commands_files:
+        with open(file_path, 'r') as file:
+            data = json.load(file)
+            testsuite_id = data.get("testsuite_id")
+            platform = data.get("platform")
+            files = data.get("files", [])
+
+            for file in files:
+                if file not in database:
+                    database[file] = []
+                database[file].append({"testsuite_id": testsuite_id, "platform": platform})
+
+    with open(output_file, 'w') as output:
+        json.dump(database, output, indent=4)
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser(description="Generate a testsuite database from compile_commands files.")
+    parser.add_argument("--directory", required=True, help="Directory to search for compile_commands files")
+    parser.add_argument("--output", required=True, help="Output file for the generated database")
+    args = parser.parse_args()
+
+    generate_testsuite_database(args.directory, args.output)
+    print(f"Database generated and saved to {args.output}")

scripts/pylib/twister/twisterlib/runner.py

@@ -15,6 +15,8 @@
 import subprocess
 import sys
 import time
+import json
+import hashlib
 import traceback
 from math import log10
 from multiprocessing import Lock, Process, Value
@@ -1013,6 +1015,33 @@ def process(self, pipeline, done, message, lock, results):
         elif op == "cmake":
             try:
                 ret = self.cmake()
+                compile_commands_path = os.path.join(self.instance.build_dir,
+                                                     'compile_commands.json')
+                if os.path.exists(compile_commands_path):
+                    with open(compile_commands_path, 'r') as f:
+                        compile_commands = json.load(f)
+
+                    # Extract only the file names from the compile commands
+                    file_names = [entry['file'] for entry in compile_commands]
+
+                    # Create a unique filename using a SHA hash based on the testsuite identifier and platform
+                    unique_id = f"{self.instance.testsuite.id}_{self.instance.platform.name}"
+                    sha_hash = hashlib.sha256(unique_id.encode()).hexdigest()
+                    instance_file = os.path.join(self.options.outdir, f'compile_commands_{sha_hash}.json')
+
+                    # Add testsuite id and platform to the data
+                    file_names_relative = list(set([
+                        os.path.relpath(file, start=ZEPHYR_BASE) for file in file_names
+                        if not file.startswith(self.options.outdir)
+                    ]))
+                    data = {
+                        "testsuite_id": self.instance.testsuite.id,
+                        "platform": self.instance.platform.name,
+                        "files": file_names_relative
+                    }
+
+                    with open(instance_file, 'w') as f:
+                        json.dump(data, f, indent=4)
                 if self.instance.status in [TwisterStatus.FAIL, TwisterStatus.ERROR]:
                     next_op = 'report'
                 elif self.options.cmake_only: