wpt-commit

#!/usr/bin/env python3

import csv
import os
import re
import subprocess
import sys
import yaml

progress_filepath=os.path.join(os.path.dirname(os.path.realpath(__file__)), 'progress.csv')

def read_classifiers(source):
    with open(source, 'r') as handle:
        return yaml.safe_load(handle)['features']

def compile_file_pattern(source):
    if source.startswith('!'):
        bias = False
        string = source[1:]
    else:
        bias = True
        string = source

    return bias, source, re.compile('^' + re.sub('\\*', '.*', string) + '$')

class Expansion():
    def __init__(self, directory, classifier):
        self.directory = directory
        self.feature_id = classifier['name']
        self.tests = []
        self.violations = []

    def violation(self, message):
        self.violations.append(
            f'{self.directory}{os.path.sep}WEB_FEATURES.yml ({self.feature_id}): {message}'
        )

def is_test(filepath):
    return not (
        '-ref.' in filepath or
        filepath.endswith('WEB_FEATURES.yml') or
        filepath.endswith('META.yml')
    )

def expand_classifier(directory, classifier):
    expansion = Expansion(directory, classifier)

    if classifier['files'] == '**':
        for dirpath, dirname, filenames in os.walk(directory):
            expansion.tests.extend(
                filter(
                    is_test,
                    map(lambda filename: os.path.join(dirpath, filename), filenames)
                )
            )
    else:
        patterns = [
            compile_file_pattern(file_pattern) for file_pattern in classifier['files']
        ]
        used = set()

        for direntry in os.scandir(directory):
            if not direntry.is_file() or not is_test(direntry.path):
                continue
            should_include = False

            for pattern_desc in patterns:
                bias, source, pattern = pattern_desc
                if pattern.search(direntry.name):
                    should_include = bias
                    used.add(source)

            if should_include:
                expansion.tests.append(direntry.path)

        unused = set(classifier['files']).difference(used)
        for pattern in unused:
            expansion.violation(f'Unused pattern: "{pattern}"')

    if len(expansion.tests) == 0:
        expansion.violation(f'Expected classifier to match at least one test file')

    return expansion

def name_from_id(feature_id):
    with open(progress_filepath, 'r') as progress_file:
        reader = csv.reader(progress_file)
        for row in reader:
            if row[0] == feature_id:
                return row[1]
        raise Exception(f'Unrecognized feature ID: "{feature_id}"')

def git(*args):
    return subprocess.run(
        ['git', *args],
        stdout=subprocess.PIPE,
        text=True,
        check=True
    )

def infer_feature_id(stdout):
    ids = set(re.findall('^\+\s*- name:\s*(.*)$', stdout, re.MULTILINE))
    assert len(ids) == 1, f'Expected exactly one feature ID, found {len(ids)}: {ids}'
    return ids.pop()

def main():
    feature_id = infer_feature_id(git('diff', '--staged', '--unified=0').stdout)
    violations = []
    newly_classified_tests = set()
    other_classified_tests = set()

    for dirpath, dirnames, filenames in os.walk('.'):
        if 'WEB_FEATURES.yml' not in filenames:
            continue

        classifiers = read_classifiers(os.path.join(dirpath, 'WEB_FEATURES.yml'))

        for classifier in classifiers:
            expansion = expand_classifier(dirpath, classifier)

            if classifier['name'] != feature_id:
                other_classified_tests.update(expansion.tests)
            else:
                newly_classified_tests.update(expansion.tests)
                violations.extend(expansion.violations)

    for test in newly_classified_tests.intersection(other_classified_tests):
        violations.append(f'Test already classified: "{test}"')

    if len(violations):
        raise AssertionError(
            f'{len(violations)} violations:\n' + '\n'.join(violations)
        )

    feature_name = name_from_id(feature_id)

    git('commit', '-m', f'Map "{feature_name}" to web-features')

    print(f'Successfully commited classifications for "{feature_name}" ({feature_id})')

if __name__ == '__main__':
    main()