# Copyright 2019 Google LLC

#

# Licensed under the Apache License, Version 2.0 (the "License");

# you may not use this file except in compliance with the License.

# You may obtain a copy of the License at

#

# http://www.apache.org/licenses/LICENSE-2.0

#

# Unless required by applicable law or agreed to in writing, software

# distributed under the License is distributed on an "AS IS" BASIS,

# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

# See the License for the specific language governing permissions and

# limitations under the License.

import os

import subprocess

from lib import command_trace

from lib import source

def find_changed_or_files(all, rev_or_files, patterns):

"""Finds files.

Args:

all: Force finding all files.

rev_or_files: A single revision, a list of files, or empty.

patterns: A list of git matching patterns

Returns:

Files that match.

If rev_or_files is a single revision, the result is all files that match

the patterns that have changed since the revision.

If rev_or_files is a list of files, the result is all the files that match

that list of files. The files can be patterns.

If rev_or_files is empty, the result is all the files that match patterns.

"""

if all:

return find_files(patterns)

if not rev_or_files:

return find_changed('origin/main', patterns)

if len(rev_or_files) == 1 and is_revision(rev_or_files[0]):

return find_changed(rev_or_files[0], patterns)

else:

return find_files(rev_or_files)

def is_revision(word):

"""Returns true if the given word is a revision name according to git."""

command = ['git', 'rev-parse', word, '--']

with open(os.devnull, 'w') as dev_null:

command_trace.log(command)

rc = subprocess.call(command, stdout=dev_null, stderr=dev_null)

return rc == 0

def find_changed(revision, patterns):

"""Finds files changed since a revision."""

# Always include -- indicate that revision is known to be a revision, even

# if no patterns follow.

command = ['git', 'diff', '-z', '--name-only', '--diff-filter=ACMR',

revision, '--']

command.extend(patterns)

command.extend(standard_exclusions())

return _null_split_output(command)

def find_files(patterns=None):

"""Finds files matching the given patterns using git ls-files."""

command = ['git', 'ls-files', '-z', '--']

if patterns:

command.extend(patterns)

command.extend(standard_exclusions())

return _null_split_output(command)

def find_lines_matching(pattern, sources=None):

command = [

'git', 'grep',

'-n', # show line numbers

'-I', # exclude binary files

pattern,

'--'

]

if sources:

command.extend(sources)

command.extend(standard_exclusions())

command_trace.log(command)

bufsize = 4096

proc = subprocess.Popen(command, bufsize=bufsize, stdout=subprocess.PIPE)

result = []

try:

while proc.poll() is None:

result.append(proc.stdout.read(bufsize))

except KeyboardInterrupt:

proc.terminate()

proc.wait()

return b''.join(result).decode('utf8', errors='replace')

def make_patterns(dirs):

"""Returns a list of git match patterns for the given directories."""

return ['%s/**' % d for d in dirs]

def make_exclusions(dirs):

return [':(exclude)' + d for d in dirs]

def standard_exclusions():

result = make_exclusions(source.IGNORE)

result.append(':(exclude)**/third_party/**')

return result

def is_within_repo():

"""Returns whether the current working directory is within a git repo."""

try:

subprocess.check_output(['git', 'status'])

return True

except subprocess.CalledProcessError:

return False

def get_repo_root():

"""Returns the absolute path to the root of the current git repo."""

command = ['git', 'rev-parse', '--show-toplevel']

return subprocess.check_output(command, text=True, errors='replace').rstrip()

def _null_split_output(command):

"""Runs the given command and splits its output on the null byte."""

command_trace.log(command)

result = subprocess.check_output(command, text=True, errors='replace')

return [name for name in result.rstrip().split('\0') if name]