#!/usr/bin/env python
# Copyright 2022 The Chromium Authors
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
# -----------------------------FILE OVERVIEW----------------------------- #
# This script generates a list of tests that exist for given parameters.
#
# This script was created so that we could determine which tests are used
# by specific components, so that coverage can be generated for specific
# components.
#
# Use the built in help for instructions.
# ----------------------------------------------------------------------- #
import os
import sys
import argparse
import fnmatch
import re
import time
import json
from threading import Thread, Lock
sys.path.append(os.path.abspath("./third_party/blink/tools"))
sys.path.append(os.path.abspath("./"))
# Chromium check
NOT_CHROME_ERROR = (
'You must be in the chromium src directory to run this command.')
try:
file = open('./DIR_METADATA', 'r')
if not 'project: "chromium"' in file.read():
print(NOT_CHROME_ERROR)
exit()
except IOError:
print(NOT_CHROME_ERROR)
exit()
from third_party.blink.tools.blinkpy.w3c.directory_owners_extractor import DirectoryOwnersExtractor
from third_party.blink.tools.blinkpy.common.host import Host
# CONSTANTS
DIR_METADATA = 'DIR_METADATA'
COMMON_METADATA = 'COMMON_METADATA'
OWNERS = 'OWNERS'
TEST_NAME_REGEX = re.compile(
r"TEST(_[FP])?\(\s*'?([a-zA-Z][a-zA-Z0-9]*)'?,"
"\s*'?([a-zA-Z][a-zA-Z0-9_]*)'?", re.MULTILINE)
TEST_SUITE_REGEX = re.compile(
r'TEST_SUITE(_[FP])?\(\s*([a-zA-Z][a-zA-Z0-9]*),\s*([a-zA-Z][a-zA-Z0-9]*),',
re.MULTILINE)
parser = argparse.ArgumentParser(
"Determines test suites for given directories and components")
parser.add_argument(
'-c',
'--components',
nargs='+',
# TODO: Adjust to buganizer once the
# coverage team updates the tool
help='A monorail component to collect suites for',
default=[])
parser.add_argument(
'-d',
'--dirs',
nargs='*',
help='Directory to search through, at least one must be present',
default=['./'])
parser.add_argument('-i',
'--ignore_dir',
nargs='*',
help='Directories to ignore',
default=[])
parser.add_argument('-o',
'--out_file',
help='The file to write the suites to.',
required=True)
parser.add_argument('-j',
'--jobs',
required=False,
help='The number of threads to allow',
type=int,
default=0)
args = parser.parse_args()
class ThreadManager(int):
def __init__(self, max_threads):
self.num_threads = 0
self.max_threads = max_threads
self.lock = Lock()
def _thread_wrapper(self, func, args):
func(*args)
with self.lock:
self.num_threads -= 1
def run(self, func, args):
if self.num_threads < self.max_threads:
thread = Thread(target=self._thread_wrapper, args=(func, args))
with self.lock:
self.num_threads += 1
thread.start()
return True
else:
return False
def wait_until_threads_done(self):
while self.num_threads > 0:
time.sleep(0.1)
class TestFinder(argparse.Namespace):
def __init__(self, args):
self.host = Host()
self.component_map = dict()
self.dirs_with_tests = set()
self.disabled_tests = dict()
self.maybe_tests = dict()
self.total_disabled = 0
self.total_maybe = 0
self.total_working = 0
self.dirs = args.dirs
self.comps = args.components
self.dir_thread_manager = ThreadManager(args.jobs * .75)
self.read_thread_manager = ThreadManager(args.jobs / 4)
self.out_file = args.out_file
def is_hidden(self, path):
name = os.path.basename(os.path.abspath(path))
return name.startswith('.')
def is_output_dir(self, path):
name = os.path.basename(os.path.abspath(path))
return name.startswith('out')
def is_test_file(self, filepath):
if filepath.endswith('.py'):
return False
if filepath.endswith('.png'):
return False
noext = os.path.splitext(os.path.abspath(filepath))[0]
name = os.path.basename(noext)
return name.endswith('test')
def get_items_in_dir(self, path):
items_list = os.listdir(path)
files = list()
dirs = list()
for item in items_list:
item_path = os.path.abspath(os.path.join(path, item))
if os.path.isfile(item_path):
files.append(item_path)
else:
dirs.append(item_path)
return [files, dirs]
def get_component_for_dir(self, path, type):
# Python does something weird with hidden dirs.
# We don't care about those anyway.
if self.is_hidden(path):
return None
metadata_path = os.path.join(path, type)
extractor = DirectoryOwnersExtractor(Host())
try:
return extractor.extract_component(metadata_path)
except KeyError:
return None
def get_test_suites_for_file(self, filepath, suites):
try:
new_suites = set()
text = open(filepath, 'r').read()
relPath = os.path.relpath(filepath)
# Find regular tests
matches = re.findall(TEST_NAME_REGEX, text)
for m in matches:
new_suites.add(m[1])
# Check for disabled and maybe tests
if m[2] != None and m[2].startswith('DISABLED'):
disabled = self.disabled_tests.get(relPath)
self.total_disabled += 1
if disabled == None:
disabled = []
self.disabled_tests[relPath] = disabled
disabled.append(m[1] + "." + m[2])
elif m[2] != None and m[2].startswith('MAYBE'):
self.total_maybe += 1
maybe = self.maybe_tests.get(relPath)
if maybe == None:
maybe = []
self.maybe_tests[relPath] = maybe
maybe.append(m[1] + "." + m[2])
else:
self.total_working += 1
# Find suites
matches = re.findall(TEST_SUITE_REGEX, text)
for m in matches:
new_suites.add(m[2])
suites.extend(new_suites)
except IOError:
print('Failed to open ' + filepath)
return list()
def get_test_suites_for_dir(self, dir, suites):
files, _ = self.get_items_in_dir(dir)
files = list(filter(lambda file: self.is_test_file(file), files))
if files.__len__() > 0:
self.dirs_with_tests.add(os.path.relpath(dir))
for file in files:
if not self.read_thread_manager.run(self.get_test_suites_for_file,
(file, suites)):
self.get_test_suites_for_file(file, suites)
def start(self):
for dir in self.dirs:
self.build_component_map(dir, None)
self.dir_thread_manager.wait_until_threads_done()
self.save()
def save(self):
with open(self.out_file, 'w') as outfile:
data = {
"components": self.component_map,
"directories": [*self.dirs_with_tests],
"disabled_tests": self.disabled_tests,
"maybe_tests": self.maybe_tests,
"total_disabled": self.total_disabled,
"total_maybe": self.total_maybe,
"total_working": self.total_working,
"total_tests":
self.total_working + self.total_maybe + self.total_disabled
}
json.dump(data, outfile, indent=2)
def build_component_map(self, cur_dir, common_component):
# Normalize cur_dir
cur_dir = os.path.abspath(cur_dir)
# Get the component
component = self.get_component_for_dir(cur_dir, DIR_METADATA)
using_common_metadata = False
# If the component is in common metadata, it should pass through to sub dir.
if component == None and common_component == None:
component = self.get_component_for_dir(cur_dir, COMMON_METADATA)
using_common_metadata = True
if component != None:
allowed = self.comps.__len__() == 0
for comp_glob in self.comps:
result = fnmatch.filter([component], comp_glob)
allowed = result.__len__() > 0
if allowed:
break
if allowed:
tests_for_comp = self.component_map.get(component)
# If the component isn't in the map yet, add it
if tests_for_comp == None:
tests_for_comp = list()
self.component_map[component] = tests_for_comp
self.get_test_suites_for_dir(cur_dir, tests_for_comp)
# Even if the dir didn't have a component,
# we still need to check it's subdirs.
_, sub_dirs = self.get_items_in_dir(cur_dir)
for sub_dir in sub_dirs:
# skip hidden directories
if not self.is_hidden(sub_dir) and not self.is_output_dir(sub_dir):
common_component_to_pass = (component if using_common_metadata else
common_component)
if not self.dir_thread_manager.run(self.build_component_map,
(sub_dir, common_component_to_pass)):
self.build_component_map(sub_dir, common_component_to_pass)
try:
finder = TestFinder(args)
finder.start()
except KeyboardInterrupt:
print('\nUser Interrupted')
exit()
Codebase Browser
chromium