testing: code improvements

Signed-off-by: IonutMuthi <[email protected]>

Author: IonutMuthi

tools/testing/file_filters.py

@@ -0,0 +1,82 @@
+#!/usr/bin/env python3
+"""
+File Filters Module
+Handles file and content filtering logic for test environment setup.
+
+This module provides functionality for:
+- Component-based file filtering
+- RBP-based content filtering
+- New test detection and filtering
+- Helper functions for filtering decisions
+"""
+
+from rst_parser import parse_rst_structure, rebuild_rst_file
+
+
+def _should_include_file(test_file, component_filter, new_uids):
+    """Determine if file should be included based on filters."""
+    # Include files from specified components
+    if test_file['component'] in component_filter:
+        return True
+
+    # Include files containing new tests (bypass component filter)
+    if new_uids and any(test['uid'] in new_uids for test in test_file['tests']):
+        return True
+
+    # Include only top-level index files, not component-specific ones
+    if 'index.rst' in test_file['relative_path']:
+        path_parts = test_file['relative_path'].split('/')
+        # Include plugins/index.rst, general/index.rst but not plugins/m2k/index.rst
+        return len(path_parts) == 2 and path_parts[1] == 'index.rst'
+
+    return False
+
+
+def filter_by_component(test_files, component_filter, new_uids=None):
+    """Simple file-level filtering by component, with new-since support."""
+    if not component_filter:
+        return test_files
+
+    return [tf for tf in test_files
+            if _should_include_file(tf, component_filter, new_uids)]
+
+
+def _should_include_test(test, rbp_filter, new_uids):
+    """Determine if test should be included based on filters."""
+    # Include if it's a new test (always)
+    if new_uids and test['uid'] in new_uids:
+        return True
+
+    # Include if matches RBP filter
+    if rbp_filter and test['rbp'] in rbp_filter:
+        return True
+
+    # Include if no filters specified
+    if not rbp_filter and not new_uids:
+        return True
+
+    return False
+
+
+def _extract_all_uids(test_files):
+    """Extract all UIDs from test files efficiently."""
+    return {test['uid'] for file_info in test_files
+            for test in file_info['tests']
+            if test['uid'] != 'MISSING'}
+
+
+def filter_rst_content(file_path, rbp_filter, new_uids=None):
+    """Parse RST file and keep only tests matching RBP filter or new tests."""
+
+    with open(file_path, 'r', encoding='utf-8') as f:
+        content = f.read()
+
+    # Split into header + test sections
+    structure = parse_rst_structure(content)
+
+    # Filter tests by RBP or new-since
+    filtered_tests = [test for test in structure['tests']
+                     if _should_include_test(test, rbp_filter, new_uids)]
+
+    # Reconstruct file
+    return rebuild_rst_file(structure['header'], filtered_tests)

tools/testing/file_operations.py

@@ -0,0 +1,127 @@
+#!/usr/bin/env python3
+"""
+File Operations Module
+Handles file copying operations and index file management.
+
+This module provides functionality for:
+- Copying test files with or without modification
+- Managing index file toctree entries
+- Component tracking for copied files
+- RST content-level filtering and file operations
+"""
+
+import os
+import shutil
+import re
+from file_filters import filter_rst_content
+
+
+def get_copied_components(filtered_files):
+    """Track which components were actually copied."""
+    copied_components = set()
+    for tf in filtered_files:
+        if tf['component'] not in ['other', 'general']:
+            copied_components.add(tf['component'])
+    return copied_components
+
+
+def extract_component_from_toctree_entry(entry):
+    """Extract component name from toctree entry."""
+    # "   adc/index" → "adc"
+    # "   core/hp_tests" → "core"
+    entry = entry.strip()
+    if '/' in entry:
+        return entry.split('/')[0]
+    return entry
+
+
+def filter_index_content(file_path, copied_components):
+    """Filter toctree entries in index files based on copied components."""
+
+    with open(file_path, 'r', encoding='utf-8') as f:
+        content = f.read()
+
+    # Find toctree section
+    toctree_pattern = r'(\.\. toctree::\n(?:\s+:[^\n]+\n)*\s*\n)(.*?)(\n\n|\.\.|$)'
+    match = re.search(toctree_pattern, content, re.DOTALL)
+
+    if not match:
+        return content  # No toctree found
+
+    header = match.group(1)
+    entries = match.group(2)
+    footer = content[match.end():]
+
+    # Filter toctree entries
+    filtered_entries = []
+    for line in entries.split('\n'):
+        if line.strip():
+            # Extract component from entry like "adc/index" → "adc"
+            component = extract_component_from_toctree_entry(line)
+            if component in copied_components:
+                filtered_entries.append(line)
+
+    # Reconstruct content
+    before_toctree = content[:match.start()]
+    new_entries = '\n'.join(filtered_entries)
+
+    return f"{before_toctree}{header}{new_entries}\n{footer}"
+
+
+def copy_files_unchanged(test_files, dest_dir, component_filter=None):
+    """Copy files without content modification."""
+    copied_count = 0
+    copied_components = get_copied_components(test_files) if component_filter else None
+
+    for test_file in test_files:
+        src_path = test_file['file_path']
+        dest_path = os.path.join(dest_dir, test_file['relative_path'])
+
+        # Create destination directory
+        os.makedirs(os.path.dirname(dest_path), exist_ok=True)
+
+        # Filter index files when component filtering is active
+        if component_filter and 'index.rst' in test_file['relative_path']:
+            filtered_content = filter_index_content(src_path, copied_components)
+            with open(dest_path, 'w', encoding='utf-8') as f:
+                f.write(filtered_content)
+        else:
+            # Copy file unchanged
+            shutil.copy2(src_path, dest_path)
+
+        copied_count += 1
+
+    return copied_count
+
+
+def filter_by_rbp_content(test_files, rbp_filter, dest_dir, component_filter=None, new_uids=None):
+    """Content-level filtering: modify files to keep only specified RBP tests or new tests."""
+    if not rbp_filter and not new_uids:
+        # No filters - copy files as-is
+        return copy_files_unchanged(test_files, dest_dir, component_filter)
+
+    copied_count = 0
+    copied_components = get_copied_components(test_files) if component_filter else None
+
+    # Filter active - parse and modify content
+    for test_file in test_files:
+        dest_path = os.path.join(dest_dir, test_file['relative_path'])
+
+        # Create destination directory
+        os.makedirs(os.path.dirname(dest_path), exist_ok=True)
+
+        if test_file['tests']:  # File contains tests
+            filtered_content = filter_rst_content(test_file['file_path'], rbp_filter, new_uids)
+            with open(dest_path, 'w', encoding='utf-8') as f:
+                f.write(filtered_content)
+        elif component_filter and 'index.rst' in test_file['relative_path']:
+            # Filter index files when component filtering is active
+            filtered_content = filter_index_content(test_file['file_path'], copied_components)
+            with open(dest_path, 'w', encoding='utf-8') as f:
+                f.write(filtered_content)
+        else:  # File has no tests (index, docs, etc.)
+            shutil.copy2(test_file['file_path'], dest_path)
+
+        copied_count += 1
+
+    return copied_count

tools/testing/git_operations.py

@@ -0,0 +1,83 @@
+#!/usr/bin/env python3
+"""
+Git Operations Module
+Handles git repository operations and new test detection.
+
+This module provides functionality for:
+- Git repository validation and operations
+- Extracting test files from specific git versions
+- Detecting new tests between versions
+"""
+
+import os
+import subprocess
+import tempfile
+import shutil
+from rst_parser import scan_test_files
+
+
+def get_git_root():
+    """Get the git repository root directory."""
+    try:
+        result = subprocess.run(['git', 'rev-parse', '--show-toplevel'],
+                              capture_output=True, check=True, text=True)
+        return result.stdout.strip()
+    except subprocess.CalledProcessError:
+        return None
+
+
+def validate_git_repository():
+    """Ensure we're in a git repository."""
+    return get_git_root() is not None
+
+
+def git_tag_exists(tag_name):
+    """Check if a git tag exists."""
+    git_root = get_git_root()
+    if not git_root:
+        return False
+
+    try:
+        result = subprocess.run(['git', 'tag', '-l', tag_name],
+                              cwd=git_root, capture_output=True, check=True, text=True)
+        return tag_name in result.stdout.strip().split('\n')
+    except subprocess.CalledProcessError:
+        return False
+
+
+def detect_new_tests(baseline_version):
+    """Detect new test UIDs since baseline version."""
+    git_root = get_git_root()
+    if not git_root:
+        raise RuntimeError("Not in a git repository")
+
+    temp_dir = tempfile.mkdtemp()
+
+    try:
+        # Extract docs/tests directory at baseline version from git root
+        cmd = ['git', 'archive', baseline_version, 'docs/tests']
+        with subprocess.Popen(cmd, cwd=git_root, stdout=subprocess.PIPE) as git_proc:
+            subprocess.run(['tar', '-x', '-C', temp_dir],
+                         stdin=git_proc.stdout, check=True)
+
+        # Scan extracted files for UIDs
+        baseline_tests_dir = os.path.join(temp_dir, 'docs', 'tests')
+        if not os.path.exists(baseline_tests_dir):
+            raise FileNotFoundError(f"docs/tests not found in {baseline_version}")
+
+        baseline_files = scan_test_files(baseline_tests_dir)
+
+        baseline_uids = set()
+        for file_info in baseline_files:
+            for test in file_info['tests']:
+                if test['uid'] != 'MISSING':
+                    baseline_uids.add(test['uid'])
+
+        print(f"Baseline version {baseline_version}: {len(baseline_uids)} tests")
+        return baseline_uids
+
+    except Exception as e:
+        print(f"Error extracting baseline UIDs: {e}")
+        raise
+    finally:
+        shutil.rmtree(temp_dir)