#!/usr/bin/env python3

# Copyright (C) 2024 Apple Inc. All rights reserved.

#

# Redistribution and use in source and binary forms, with or without

# modification, are permitted provided that the following conditions

# are met:

# 1. Redistributions of source code must retain the above copyright

# notice, this list of conditions and the following disclaimer.

# 2. Redistributions in binary form must reproduce the above copyright

# notice, this list of conditions and the following disclaimer in the

# documentation and/or other materials provided with the distribution.

#

# THIS SOFTWARE IS PROVIDED BY APPLE INC. AND ITS CONTRIBUTORS "AS IS" AND

# ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED

# WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE

# DISCLAIMED. IN NO EVENT SHALL APPLE INC. OR ITS CONTRIBUTORS BE LIABLE FOR

# ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL

# DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR

# SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER

# CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,

# OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE

# OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.

import os

import argparse

import sys

import re

import time

# This prints the progress every interval_seconds to avoid that the step gets killed

# by buildbot because "command timed out: XXX seconds without output running YYY"

class KeepAliveStdoutProgress:

def __init__(self, interval_seconds=300, check_every_n_lines=100):

self.line_count = 0

self.last_progress_time = time.time()

self.interval_seconds = interval_seconds

self.check_every_n_lines = check_every_n_lines

def maybe_update_progress(self):

self.line_count += 1

# Only check time every N lines to avoid expensive system calls

if self.line_count % self.check_every_n_lines == 0:

current_time = time.time()

if current_time - self.last_progress_time >= self.interval_seconds:

print(f'filter-test-logs progress: {self.line_count} lines processed', flush=True)

self.last_progress_time = current_time

def parser():

parser = argparse.ArgumentParser(description='filter test output')

parser.add_argument(

'filter_type',

choices=['jsc', 'layout', 'scan-build', 'webdriver', 'test262', 'minibrowser', 'swift', 'api'],

help='Choose which test logs to filter'

)

parser.add_argument(

'--output',

help='output directory for unfiltered log',

default='logs.txt'

)

return parser.parse_args()

def filter_jsc_tests(log_file_name):

log_file = os.path.abspath(f'{log_file_name}')

keep_alive_stdout = KeepAliveStdoutProgress()

with open(log_file, 'w') as f:

print_all_lines = False

for line in sys.stdin:

f.write(line)

if print_all_lines:

print(line, end='')

elif line.startswith('** The following JSC') or line.startswith('Results'):

print_all_lines = True

print(line, end='')

else:

keep_alive_stdout.maybe_update_progress()

def filter_layout_tests(log_file_name):

log_file = os.path.abspath(f'{log_file_name}')

keep_alive_stdout = KeepAliveStdoutProgress()

with open(log_file, 'w') as f:

print_all_lines = False

for line in sys.stdin:

match = re.search(r'[\uD800-\uDFFF]', line)

if not match:

f.write(line)

if print_all_lines:

print(line, end='')

elif line.startswith('=> Results') or 'Exiting early' in line or 'leaks found' in line:

print_all_lines = True

print(line, end='')

else:

keep_alive_stdout.maybe_update_progress()

def filter_scan_build(log_file_name):

log_file = os.path.abspath(f'{log_file_name}')

keep_alive_stdout = KeepAliveStdoutProgress()

with open(log_file, 'w') as f:

print_all_lines = False

for line in sys.stdin:

f.write(line)

if line.startswith('=====') or print_all_lines:

print(line, end='')

# A success or failure is always preceeded by **

elif line.startswith('**'):

print_all_lines = True

print(line, end='')

elif ': error:' in line:

print(line, end='')

else:

keep_alive_stdout.maybe_update_progress()

def filter_webdriver_tests(log_file_name):

log_file = os.path.abspath(f'{log_file_name}')

keep_alive_stdout = KeepAliveStdoutProgress()

with open(log_file, 'w') as f:

print_all_lines = False

for line in sys.stdin:

f.write(line)

if print_all_lines:

print(line, end='')

elif 'tests ran as expected' in line:

print_all_lines = True

print(line, end='')

else:

keep_alive_stdout.maybe_update_progress()

def filter_test262(log_file_name):

log_file = os.path.abspath(f'{log_file_name}')

keep_alive_stdout = KeepAliveStdoutProgress()

with open(log_file, 'w') as f:

print_all_lines = False

for line in sys.stdin:

f.write(line)

if print_all_lines or line.startswith('! NEW FAIL'):

print(line, end='')

elif 'TESTS SUMMARY---' in line:

print_all_lines = True

print(line, end='')

else:

keep_alive_stdout.maybe_update_progress()

def filter_minibrowser_tests(log_file_name):

log_file = os.path.abspath(f'{log_file_name}')

keep_alive_stdout = KeepAliveStdoutProgress()

with open(log_file, 'w') as f:

print_all_lines = False

for line in sys.stdin:

f.write(line)

if print_all_lines:

print(line, end='')

elif 'Bundle passed tests on all this' in line:

print_all_lines = True

print(line, end='')

else:

keep_alive_stdout.maybe_update_progress()

def filter_swift(log_file_name):

log_file = os.path.abspath(f'{log_file_name}')

keep_alive_stdout = KeepAliveStdoutProgress()

with open(log_file, 'w') as f:

print_all_lines = False

for line in sys.stdin:

f.write(line)

if print_all_lines:

print(line, end='')

elif line.startswith('ERROR:') or line.startswith('FAILURE:') or '--- Build Script Analyzer ---' in line:

print_all_lines = True

print(line, end='')

elif 'error:' in line.lower():

print(line, end='')

else:

keep_alive_stdout.maybe_update_progress()

def filter_api_tests(log_file_name):

log_file = os.path.abspath(f'{log_file_name}')

keep_alive_stdout = KeepAliveStdoutProgress()

with open(log_file, 'w') as f:

print_all_lines = False

for line in sys.stdin:

f.write(line)

if print_all_lines:

print(line, end='')

elif re.match(r'.*Ran \d+ tests of \d+', line):

print_all_lines = True

print(line, end='')

else:

keep_alive_stdout.maybe_update_progress()

def main():

args = parser()

if args.filter_type == 'jsc':

filter_jsc_tests(args.output)

elif args.filter_type == 'layout':

filter_layout_tests(args.output)

elif args.filter_type == 'scan-build':

filter_scan_build(args.output)

elif args.filter_type == 'webdriver':

filter_webdriver_tests(args.output)

elif args.filter_type == 'test262':

filter_test262(args.output)

elif args.filter_type == 'minibrowser':

filter_minibrowser_tests(args.output)

elif args.filter_type == 'swift':

filter_swift(args.output)

elif args.filter_type == 'api':

filter_api_tests(args.output)

if __name__ == '__main__':

main()