cacti/tests/tools/check_changed_php_coverage.py

#!/usr/bin/env python3
"""Require measured statement coverage for PHP lines added by a pull request.

Copyright (C) 2004-2026 The Cacti Group. GPL-2.0-or-later.
"""
import argparse
import hashlib
import json
import os
from pathlib import Path
import re
import subprocess
import sys
import xml.etree.ElementTree as ET


def added_lines(base, root):
    merge_base = subprocess.check_output(
        ['git', 'merge-base', base, 'HEAD'], cwd=root, text=True).strip()
    diff = subprocess.check_output(
        ['git', '-c', 'core.quotePath=false', 'diff', '--no-ext-diff',
         '--no-textconv', '--unified=0', merge_base, 'HEAD', '--', '*.php',
         ':(exclude)tests/**', ':(exclude)include/vendor/**'], cwd=root, text=True)
    changed = {}
    name = None
    line = 0
    for record in diff.splitlines():
        if record.startswith('diff --git '):
            name = None
        elif record.startswith('+++ b/'):
            name = record[6:]
            changed.setdefault(name, set())
        elif record.startswith('@@'):
            line = int(re.search(r'\+(\d+)', record).group(1))
        elif record.startswith('+') and name is not None:
            changed[name].add(line)
            line += 1
        elif not record.startswith(('-', '\\')):
            line += 1
    return {name: lines for name, lines in changed.items() if lines}


def prepare_config(config, output, changed):
    """Keep the test suite, but include every changed production PHP file."""
    tree = ET.parse(config)
    config_dir = Path(config).resolve().parent
    output_dir = Path(output).resolve().parent
    for element in tree.iter():
        for attribute in ('bootstrap', 'cacheDirectory', 'cacheResultFile',
                          'outputDirectory', 'outputFile', 'target'):
            value = element.get(attribute)
            if value and not Path(value).is_absolute():
                element.set(attribute, os.path.relpath(config_dir / value, output_dir))
        if element.tag in ('file', 'directory', 'exclude') and element.text:
            value = element.text
            if not Path(value).is_absolute():
                element.text = os.path.relpath(config_dir / value, output_dir)
    source = tree.getroot().find('source')
    if source is None:
        source = ET.SubElement(tree.getroot(), 'source')
    source.clear()
    include = ET.SubElement(source, 'include')
    for name in sorted(changed):
        ET.SubElement(include, 'file').text = os.path.relpath(Path.cwd() / name, output_dir)
    ET.indent(tree, space='    ')
    tree.write(output, encoding='utf-8', xml_declaration=True)


def validate_package_contract(request, path, root):
    contract = request.get('contract') or {}
    if not path.name.startswith('package-archive-') and contract.get('name') != 'package-archive':
        return
    producers = {'tests/fixtures/Core/PackageArchive.php', 'tests/Helpers/RequestCoverage.php'}
    if (contract.get('name') != 'package-archive' or contract.get('complete') is not True
            or type(contract.get('assertions')) is not int or contract['assertions'] != 2):
        raise ValueError('Package archive coverage requires both completed scenarios')
    if set(contract.get('producers', [])) != producers or set(contract.get('hashes', {})) != producers:
        raise ValueError('Package archive coverage is missing producer evidence')
    for name in producers:
        if hashlib.sha256((root / name).read_bytes()).hexdigest() != contract['hashes'][name]:
            raise ValueError(f'Package archive producer does not match {name}')


def measure(changed, report, root, request_reports=None):
    # Deletion-only changes leave no source filter, so PHPUnit writes no report.
    if not changed and not Path(report).exists():
        return {'covered': 0, 'executable': 0, 'percent': None, 'files': {}}
    evidence = {}
    for file in ET.parse(report).iter('file'):
        try:
            name = str(Path(file.attrib['name']).resolve().relative_to(root.resolve()))
        except ValueError:
            continue
        statements = evidence.setdefault(name, {})
        for item in file.findall('line'):
            if item.attrib.get('type') == 'stmt':
                number = int(item.attrib['num'])
                statements[number] = max(statements.get(number, 0), int(item.attrib['count']))
    if request_reports:
        reports = list(request_reports.glob('*.json'))
        if not reports:
            raise ValueError('Request coverage directory contains no reports')
        for path in reports:
            request = json.loads(path.read_text())
            validate_package_contract(request, path, root)
            for file, lines in (request['coverage'] or {}).items():
                name = str(Path(file).relative_to(request['root']))
                if name not in changed:
                    continue
                if hashlib.sha256((root / name).read_bytes()).hexdigest() != request['hashes'].get(file):
                    raise ValueError(f'Request coverage source does not match {name}')
                statements = evidence.get(name, {})
                for number, hits in (lines or {}).items():
                    number = int(number)
                    # The application bootstrap runs before PHPUnit starts its driver.
                    # Its native request instructions must still enter the denominator.
                    if name == 'include/global.php' and number in changed[name] and hits != -2:
                        statements.setdefault(number, 0)
                    if number in statements and hits > 0:
                        statements[number] = max(statements[number], hits)
                for start, end in request.get('html_spans', {}).get(file, []):
                    if (lines or {}).get(str(end), 0) > 0:
                        for number in changed[name] & statements.keys():
                            if start <= number <= end:
                                statements[number] = max(statements[number], 1)
    files = {}
    for name, added in changed.items():
        if name not in evidence:
            files[name] = {'missing': True}
            continue
        statements = evidence[name]
        if name == 'include/global.php':
            source_lines = (root / name).read_text().splitlines()
            for number in added:
                if re.match(r'^\s*(?:\$|(?:return|throw|if|elseif|foreach|while|for|print|echo|require|require_once|include|include_once)\b|[A-Za-z_\\][\w\\]*\()', source_lines[number - 1]):
                    statements.setdefault(number, 0)
        executable = added & statements.keys()
        covered = {number for number in executable if statements[number] > 0}
        files[name] = {
            'covered': len(covered), 'executable': len(executable),
            'uncovered': sorted(executable - covered),
        }
    covered = sum(item.get('covered', 0) for item in files.values())
    total = sum(item.get('executable', 0) for item in files.values())
    return {
        'covered': covered, 'executable': total,
        'percent': covered / total * 100 if total else None, 'files': files,
    }


def passes(result, minimum):
    if any(item.get('missing') for item in result['files'].values()):
        return False
    return result['executable'] == 0 or result['covered'] * 100 >= minimum * result['executable']


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument('--base', required=True)
    parser.add_argument('--prepare-config', type=Path)
    parser.add_argument('--output-config', type=Path)
    parser.add_argument('--report', type=Path)
    parser.add_argument('--request-reports', type=Path)
    parser.add_argument('--minimum', type=float, default=80)
    parser.add_argument('--json-output', type=Path)
    args = parser.parse_args()
    if not 0 <= args.minimum <= 100:
        parser.error('--minimum must be between 0 and 100')
    root = Path.cwd()
    changed = added_lines(args.base, root)
    if args.prepare_config:
        if not args.output_config:
            parser.error('--prepare-config requires --output-config')
        prepare_config(args.prepare_config, args.output_config, changed)
        print(f'Coverage scope includes {len(changed)} changed production PHP files.')
        return 0
    if not args.report:
        parser.error('--report is required when measuring coverage')
    result = measure(changed, args.report, root, args.request_reports)
    output = json.dumps(result, indent=2)
    print(output)
    if args.json_output:
        args.json_output.write_text(output + '\n')
    return 0 if passes(result, args.minimum) else 1


if __name__ == '__main__':
    try:
        sys.exit(main())
    except (OSError, ValueError, KeyError, ET.ParseError, subprocess.CalledProcessError) as error:
        print(f'Changed PHP coverage could not be verified: {error}', file=sys.stderr)
        sys.exit(1)