client/configuration.py

# Copyright (c) Meta Platforms, Inc. and affiliates.
#
# This source code is licensed under the MIT license found in the
# LICENSE file in the root directory of this source tree.

# pyre-unsafe

import abc
import dataclasses
import glob
import json
import logging
import multiprocessing
import os
import platform
import re
import shutil
import site
import subprocess
import sys
import textwrap
from dataclasses import dataclass, field
from logging import Logger
from pathlib import Path
from typing import (
    Any,
    Callable,
    ClassVar,
    Dict,
    Generic,
    Iterable,
    List,
    Optional,
    Sequence,
    Set,
    Type,
    TypeVar,
    Union,
)

from . import command_arguments, find_directories
from .exceptions import EnvironmentException
from .filesystem import assert_readable_directory, expand_relative_path
from .find_directories import (
    BINARY_NAME,
    CONFIGURATION_FILE,
    LOCAL_CONFIGURATION_FILE,
    LOG_DIRECTORY,
    get_relative_local_root,
)


LOG: Logger = logging.getLogger(__name__)
T = TypeVar("T")


def _expand_global_root(path: str, global_root: str) -> str:
    if path.startswith("//"):
        return expand_relative_path(global_root, path[2:])
    return path


def _expand_relative_root(path: str, relative_root: str) -> str:
    if not path.startswith("//"):
        return expand_relative_path(relative_root, path)
    return path


def _get_optional_value(source: Optional[T], default: T) -> T:
    return source if source is not None else default


def _expand_and_get_existent_ignore_all_errors_path(
    ignore_all_errors: Iterable[str], project_root: str
) -> List[str]:
    expanded_ignore_paths = []
    for path in ignore_all_errors:
        expanded = glob.glob(_expand_global_root(path, global_root=project_root))
        if not expanded:
            expanded_ignore_paths.append(path)
        else:
            expanded_ignore_paths.extend(expanded)

    paths = []
    for path in expanded_ignore_paths:
        if os.path.exists(path):
            paths.append(path)
        else:
            LOG.warning(f"Nonexistent paths passed in to `ignore_all_errors`: `{path}`")
    return paths


class InvalidConfiguration(Exception):
    def __init__(self, message: str) -> None:
        self.message = f"Invalid configuration: {message}"
        super().__init__(self.message)


class InvalidPythonVersion(InvalidConfiguration):
    def __init__(self, message: str) -> None:
        super().__init__(message)


class SearchPathElement(abc.ABC):
    @abc.abstractmethod
    def path(self) -> str:
        raise NotImplementedError

    @abc.abstractmethod
    def get_root(self) -> str:
        raise NotImplementedError

    @abc.abstractmethod
    def command_line_argument(self) -> str:
        raise NotImplementedError

    @abc.abstractmethod
    def expand_global_root(self, global_root: str) -> "SearchPathElement":
        raise NotImplementedError

    @abc.abstractmethod
    def expand_relative_root(self, relative_root: str) -> "SearchPathElement":
        raise NotImplementedError

    @abc.abstractmethod
    def expand_glob(self) -> List["SearchPathElement"]:
        raise NotImplementedError


@dataclasses.dataclass(frozen=True)
class SimpleSearchPathElement(SearchPathElement):
    root: str

    def path(self) -> str:
        return self.root

    def get_root(self) -> str:
        return self.root

    def command_line_argument(self) -> str:
        return self.root

    def expand_global_root(self, global_root: str) -> SearchPathElement:
        return SimpleSearchPathElement(
            _expand_global_root(self.root, global_root=global_root)
        )

    def expand_relative_root(self, relative_root: str) -> SearchPathElement:
        return SimpleSearchPathElement(
            _expand_relative_root(self.root, relative_root=relative_root)
        )

    def expand_glob(self) -> List[SearchPathElement]:
        expanded = sorted(glob.glob(self.get_root()))
        if expanded:
            return [SimpleSearchPathElement(path) for path in expanded]
        else:
            LOG.warning(f"'{self.path()}' does not match any paths.")
            return []


@dataclasses.dataclass(frozen=True)
class SubdirectorySearchPathElement(SearchPathElement):
    root: str
    subdirectory: str

    def path(self) -> str:
        return os.path.join(self.root, self.subdirectory)

    def get_root(self) -> str:
        return self.root

    def command_line_argument(self) -> str:
        return self.root + "$" + self.subdirectory

    def expand_global_root(self, global_root: str) -> SearchPathElement:
        return SubdirectorySearchPathElement(
            root=_expand_global_root(self.root, global_root=global_root),
            subdirectory=self.subdirectory,
        )

    def expand_relative_root(self, relative_root: str) -> SearchPathElement:
        return SubdirectorySearchPathElement(
            root=_expand_relative_root(self.root, relative_root=relative_root),
            subdirectory=self.subdirectory,
        )

    def expand_glob(self) -> List["SearchPathElement"]:
        return [self]


@dataclasses.dataclass(frozen=True)
class SitePackageSearchPathElement(SearchPathElement):
    site_root: str
    package_name: str
    is_toplevel_module: bool = False

    def package_path(self) -> str:
        module_suffix = ".py" if self.is_toplevel_module else ""
        return self.package_name + module_suffix

    def path(self) -> str:
        return os.path.join(self.site_root, self.package_path())

    def get_root(self) -> str:
        return self.site_root

    def command_line_argument(self) -> str:
        return self.site_root + "$" + self.package_path()

    def expand_global_root(self, global_root: str) -> SearchPathElement:
        # Site package does not participate in root expansion.
        return self

    def expand_relative_root(self, relative_root: str) -> SearchPathElement:
        # Site package does not participate in root expansion.
        return self

    def expand_glob(self) -> List["SearchPathElement"]:
        return [self]


@dataclasses.dataclass
class ExtensionElement:
    suffix: str
    include_suffix_in_module_qualifier: bool

    def command_line_argument(self) -> str:
        options = ""
        if self.include_suffix_in_module_qualifier:
            options = "$" + "include_suffix_in_module_qualifier"
        return self.suffix + options

    @staticmethod
    def from_json(json: Union[str, Dict[str, Union[str, bool]]]) -> "ExtensionElement":
        if isinstance(json, str):
            return ExtensionElement(
                suffix=json, include_suffix_in_module_qualifier=False
            )
        elif isinstance(json, dict):
            include_suffix_in_module_qualifier = False
            if "include_suffix_in_module_qualifier" in json:
                value = json["include_suffix_in_module_qualifier"]
                if isinstance(value, bool):
                    include_suffix_in_module_qualifier = value
            if "suffix" in json:
                suffix = json["suffix"]
                if isinstance(suffix, str):
                    return ExtensionElement(
                        suffix=suffix,
                        include_suffix_in_module_qualifier=include_suffix_in_module_qualifier,
                    )

        raise InvalidConfiguration(f"Invalid extension element: {json}")


def get_site_roots() -> List[str]:
    try:
        return site.getsitepackages() + [site.getusersitepackages()]
    except AttributeError:
        # There are a few Python versions that ship with a broken venv,
        # where `getsitepackages` is not available.
        LOG.warning(
            "Either `site.getusersitepackages()` or `site.getsitepackages()` "
            + "is not available in your virtualenv. This is a known virtualenv "
            + 'bug and as a workaround please avoid using `"site-package"` in '
            + "your search path configuration."
        )
        return []


def create_search_paths(
    json: Union[str, Dict[str, Union[str, bool]]], site_roots: Iterable[str]
) -> List[SearchPathElement]:
    if isinstance(json, str):
        return [SimpleSearchPathElement(json)]
    elif isinstance(json, dict):
        if "root" in json and "subdirectory" in json:
            return [
                SubdirectorySearchPathElement(
                    root=str(json["root"]), subdirectory=str(json["subdirectory"])
                )
            ]
        if "import_root" in json and "source" in json:
            return [
                SubdirectorySearchPathElement(
                    root=str(json["import_root"]), subdirectory=str(json["source"])
                )
            ]
        elif "site-package" in json:
            is_toplevel_module = (
                "is_toplevel_module" in json and json["is_toplevel_module"]
            )
            return [
                SitePackageSearchPathElement(
                    site_root=root,
                    package_name=str(json["site-package"]),
                    is_toplevel_module=bool(is_toplevel_module),
                )
                for root in site_roots
            ]

    raise InvalidConfiguration(f"Invalid search path element: {json}")


def assert_readable_directory_in_configuration(
    directory: str, field_name: str = ""
) -> None:
    try:
        assert_readable_directory(directory, error_message_prefix=f"{field_name} ")
    except EnvironmentException as error:
        raise InvalidConfiguration(str(error))


def _in_virtual_environment(override: Optional[bool] = None) -> bool:
    if override is not None:
        return override

    return sys.prefix != sys.base_prefix


def _expand_and_get_existent_paths(
    paths: Sequence[SearchPathElement],
) -> List[SearchPathElement]:
    expanded_search_paths = [
        expanded_path
        for search_path_element in paths
        for expanded_path in search_path_element.expand_glob()
    ]
    existent_paths = []
    for search_path_element in expanded_search_paths:
        search_path = search_path_element.path()
        if os.path.exists(search_path):
            existent_paths.append(search_path_element)
        else:
            LOG.warning(f"Path does not exist: {search_path}")
    return existent_paths


PLATFORM_MAPPING = {
    "Darwin": "macos",
    "Windows": "windows",
    "Linux": "linux",
    "default": "default",
}


@dataclass(frozen=True)
class PlatformAware(Generic[T]):
    default: T
    windows: Optional[T] = None
    macos: Optional[T] = None
    linux: Optional[T] = None

    @staticmethod
    def from_json(
        value: Union[T, Dict[str, T]], field_name: str
    ) -> Optional["PlatformAware"]:
        if value is None:
            return None
        elif isinstance(value, Dict):
            if len(value) == 0:
                return None

            invalid_keys = value.keys() - PLATFORM_MAPPING.values()
            if not len(invalid_keys) == 0:
                raise InvalidConfiguration(
                    f"Configuration `{field_name}` only supports platforms: "
                    f"{PLATFORM_MAPPING.values()} but got: `{invalid_keys}`."
                )

            default = (
                value["default"]
                if "default" in value
                else value[sorted(value.keys())[0]]
            )
            return PlatformAware(
                default=default,
                windows=value["windows"] if "windows" in value else None,
                macos=value["macos"] if "macos" in value else None,
                linux=value["linux"] if "linux" in value else None,
            )
        else:
            return PlatformAware(default=value)

    @staticmethod
    def merge(
        base: Optional["PlatformAware"], override: Optional["PlatformAware"]
    ) -> Optional["PlatformAware"]:
        if base is None:
            return override
        elif override is None:
            return base
        else:
            return PlatformAware(
                default=override.default,
                windows=override.windows
                if override.windows is not None
                else base.windows,
                macos=override.macos if override.macos is not None else base.macos,
                linux=override.linux if override.linux is not None else base.linux,
            )

    def get(self, key: Optional[str] = None) -> T:
        if key is None:
            key = PLATFORM_MAPPING[platform.system()]
        value: T = self.__getattribute__(key)
        return value if value is not None else self.default

    def validate_values(self, check: Callable[[T], bool]):
        return (
            check(self.default)
            and (check(self.windows) if self.windows is not None else True)
            and (check(self.macos) if self.macos is not None else True)
            and (check(self.linux) if self.linux is not None else True)
        )

    def to_json(self):
        result = {"default": self.default}
        if self.windows is not None:
            result["windows"] = self.windows
        if self.linux is not None:
            result["linux"] = self.linux
        if self.macos is not None:
            result["macos"] = self.macos
        return result


@dataclass(frozen=True)
class PythonVersion:
    major: int
    minor: int = 0
    micro: int = 0

    @staticmethod
    def from_string(input: str) -> "PythonVersion":
        try:
            splits = input.split(".")
            if len(splits) == 1:
                return PythonVersion(major=int(splits[0]))
            elif len(splits) == 2:
                return PythonVersion(major=int(splits[0]), minor=int(splits[1]))
            elif len(splits) == 3:
                return PythonVersion(
                    major=int(splits[0]), minor=int(splits[1]), micro=int(splits[2])
                )
            raise InvalidPythonVersion(
                "Version string is expected to have the form of 'X.Y.Z' but got "
                + f"'{input}'"
            )
        except ValueError as error:
            raise InvalidPythonVersion(str(error))

    def to_string(self) -> str:
        return f"{self.major}.{self.minor}.{self.micro}"


@dataclass(frozen=True)
class SharedMemory:
    heap_size: Optional[int] = None
    dependency_table_power: Optional[int] = None
    hash_table_power: Optional[int] = None

    def to_json(self) -> Dict[str, int]:
        heap_size = self.heap_size
        dependency_table_power = self.dependency_table_power
        hash_table_power = self.hash_table_power
        return {
            **({"heap_size": heap_size} if heap_size is not None else {}),
            **(
                {"dependency_table_power": dependency_table_power}
                if dependency_table_power is not None
                else {}
            ),
            **(
                {"hash_table_power": hash_table_power}
                if hash_table_power is not None
                else {}
            ),
        }


@dataclass(frozen=True)
class IdeFeatures:
    hover_enabled: Optional[bool] = None
    DEFAULT_HOVER_ENABLED: ClassVar[bool] = False
    go_to_definition_enabled: Optional[bool] = None
    DEFAULT_GO_TO_DEFINITION_ENABLED: ClassVar[bool] = False

    def to_json(self) -> Dict[str, int]:
        return {
            **(
                {"hover_enabled": self.hover_enabled}
                if self.hover_enabled is not None
                else {}
            ),
            **(
                {"go_to_definition_enabled": self.go_to_definition_enabled}
                if self.go_to_definition_enabled is not None
                else {}
            ),
        }

    def is_hover_enabled(self) -> bool:
        return (
            self.hover_enabled
            if self.hover_enabled is not None
            else self.DEFAULT_HOVER_ENABLED
        )

    def is_go_to_definition_enabled(self) -> bool:
        return (
            self.go_to_definition_enabled
            if self.go_to_definition_enabled is not None
            else self.DEFAULT_GO_TO_DEFINITION_ENABLED
        )


@dataclass(frozen=True)
class UnwatchedFiles:
    root: str
    checksum_path: str

    @staticmethod
    def from_json(json: Dict[str, object]) -> "UnwatchedFiles":
        root = json.get("root", None)
        if root is None:
            raise InvalidConfiguration("Missing `root` field in UnwatchedFiles")
        if not isinstance(root, str):
            raise InvalidConfiguration(
                "`root` field in UnwatchedFiles must be a string"
            )

        checksum_path = json.get("checksum_path", None)
        if checksum_path is None:
            raise InvalidConfiguration(
                "Missing `checksum_path` field in UnwatchedFiles"
            )
        if not isinstance(checksum_path, str):
            raise InvalidConfiguration(
                "`checksum_path` field in UnwatchedFiles must be a string"
            )

        return UnwatchedFiles(root=root, checksum_path=checksum_path)

    def to_json(self) -> Dict[str, str]:
        return {
            "root": self.root,
            "checksum_path": self.checksum_path,
        }


@dataclass(frozen=True)
class UnwatchedDependency:
    change_indicator: str
    files: UnwatchedFiles

    @staticmethod
    def from_json(json: Dict[str, object]) -> "UnwatchedDependency":
        change_indicator = json.get("change_indicator", None)
        if change_indicator is None:
            raise InvalidConfiguration(
                "Missing `change_indicator` field in UnwatchedDependency"
            )
        if not isinstance(change_indicator, str):
            raise InvalidConfiguration(
                "`change_indicator` field in UnwatchedDependency must be a string"
            )

        files_json = json.get("files", None)
        if files_json is None:
            raise InvalidConfiguration("Missing `files` field in UnwatchedDependency")
        if not isinstance(files_json, dict):
            raise InvalidConfiguration(
                "`files` field in UnwatchedDependency must be a dict"
            )

        return UnwatchedDependency(
            change_indicator=change_indicator,
            files=UnwatchedFiles.from_json(files_json),
        )

    def to_json(self) -> Dict[str, object]:
        return {
            "change_indicator": str(self.change_indicator),
            "files": self.files.to_json(),
        }


@dataclass(frozen=True)
class PartialConfiguration:
    binary: Optional[str] = None
    buck_mode: Optional[PlatformAware[str]] = None
    disabled: Optional[bool] = None
    do_not_ignore_errors_in: Sequence[str] = field(default_factory=list)
    dot_pyre_directory: Optional[Path] = None
    excludes: Sequence[str] = field(default_factory=list)
    extensions: Sequence[ExtensionElement] = field(default_factory=list)
    ide_features: Optional[IdeFeatures] = None
    ignore_all_errors: Sequence[str] = field(default_factory=list)
    ignore_infer: Sequence[str] = field(default_factory=list)
    isolation_prefix: Optional[str] = None
    logger: Optional[str] = None
    number_of_workers: Optional[int] = None
    oncall: Optional[str] = None
    other_critical_files: Sequence[str] = field(default_factory=list)
    pysa_version_hash: Optional[str] = None
    python_version: Optional[PythonVersion] = None
    shared_memory: SharedMemory = SharedMemory()
    search_path: Sequence[SearchPathElement] = field(default_factory=list)
    source_directories: Optional[Sequence[SearchPathElement]] = None
    strict: Optional[bool] = None
    taint_models_path: Sequence[str] = field(default_factory=list)
    targets: Optional[Sequence[str]] = None
    typeshed: Optional[str] = None
    unwatched_dependency: Optional[UnwatchedDependency] = None
    use_buck2: Optional[bool] = None
    version_hash: Optional[str] = None

    @staticmethod
    def _get_depreacted_map() -> Dict[str, str]:
        return {"do_not_check": "ignore_all_errors"}

    @staticmethod
    def _get_extra_keys() -> Set[str]:
        return {
            "create_open_source_configuration",
            "saved_state",
            "stable_client",
            "taint_models_path",
            "unstable_client",
        }

    @staticmethod
    def from_command_arguments(
        arguments: command_arguments.CommandArguments,
    ) -> "PartialConfiguration":
        strict: Optional[bool] = True if arguments.strict else None
        source_directories = [
            SimpleSearchPathElement(element) for element in arguments.source_directories
        ] or None
        targets: Optional[List[str]] = (
            arguments.targets if len(arguments.targets) > 0 else None
        )
        python_version_string = arguments.python_version
        ide_features = (
            IdeFeatures(
                hover_enabled=arguments.enable_hover,
                go_to_definition_enabled=arguments.enable_go_to_definition,
            )
            if arguments.enable_hover is not None
            or arguments.enable_go_to_definition is not None
            else None
        )
        return PartialConfiguration(
            binary=arguments.binary,
            buck_mode=PlatformAware.from_json(arguments.buck_mode, "buck_mode"),
            disabled=None,
            do_not_ignore_errors_in=arguments.do_not_ignore_errors_in,
            dot_pyre_directory=arguments.dot_pyre_directory,
            excludes=arguments.exclude,
            extensions=[],
            ide_features=ide_features,
            ignore_all_errors=[],
            ignore_infer=[],
            isolation_prefix=arguments.isolation_prefix,
            logger=arguments.logger,
            number_of_workers=arguments.number_of_workers,
            oncall=None,
            other_critical_files=[],
            pysa_version_hash=None,
            python_version=(
                PythonVersion.from_string(python_version_string)
                if python_version_string is not None
                else None
            ),
            shared_memory=SharedMemory(
                heap_size=arguments.shared_memory_heap_size,
                dependency_table_power=arguments.shared_memory_dependency_table_power,
                hash_table_power=arguments.shared_memory_hash_table_power,
            ),
            search_path=[
                SimpleSearchPathElement(element) for element in arguments.search_path
            ],
            source_directories=source_directories,
            strict=strict,
            taint_models_path=[],
            targets=targets,
            typeshed=arguments.typeshed,
            unwatched_dependency=None,
            use_buck2=arguments.use_buck2,
            version_hash=None,
        )

    @staticmethod
    def from_string(contents: str) -> "PartialConfiguration":
        def is_list_of_string(elements: object) -> bool:
            return isinstance(elements, list) and all(
                isinstance(element, str) for element in elements
            )

        def ensure_option_type(
            json: Dict[str, Any], name: str, expected_type: Type[T]
        ) -> Optional[T]:
            result = json.pop(name, None)
            if result is None:
                return None
            elif isinstance(result, expected_type):
                return result
            raise InvalidConfiguration(
                f"Configuration `{name}` is expected to have type "
                f"{expected_type} but got: `{json}`."
            )

        def ensure_optional_string_or_string_dict(
            json: Dict[str, Any], name: str
        ) -> Optional[Union[Dict[str, str], str]]:
            result = json.pop(name, None)
            if result is None:
                return None
            elif isinstance(result, str):
                return result
            elif isinstance(result, Dict):
                for value in result.values():
                    if not isinstance(value, str):
                        raise InvalidConfiguration(
                            f"Configuration `{name}` is expected to be a "
                            + f"dict of strings but got `{json}`."
                        )
                return result
            raise InvalidConfiguration(
                f"Configuration `{name}` is expected to be a string or a "
                + f"dict of strings but got `{json}`."
            )

        def ensure_optional_string_list(
            json: Dict[str, Any], name: str
        ) -> Optional[List[str]]:
            result = json.pop(name, None)
            if result is None:
                return None
            elif is_list_of_string(result):
                return result
            raise InvalidConfiguration(
                f"Configuration `{name}` is expected to be a list of "
                + f"strings but got `{json}`."
            )

        def ensure_string_list(
            json: Dict[str, Any], name: str, allow_single_string: bool = False
        ) -> List[str]:
            result = json.pop(name, [])
            if allow_single_string and isinstance(result, str):
                result = [result]
            if is_list_of_string(result):
                return result
            raise InvalidConfiguration(
                f"Configuration `{name}` is expected to be a list of "
                + f"strings but got `{json}`."
            )

        def ensure_list(json: Dict[str, Any], name: str) -> List[Any]:
            result = json.pop(name, [])
            if isinstance(result, list):
                return result
            raise InvalidConfiguration(
                f"Configuration `{name}` is expected to be a list but got `{json}`."
            )

        try:
            configuration_json = json.loads(contents)

            dot_pyre_directory = ensure_option_type(
                configuration_json, "dot_pyre_directory", str
            )

            search_path_json = configuration_json.pop("search_path", [])
            if isinstance(search_path_json, list):
                search_path = [
                    element
                    for json in search_path_json
                    for element in create_search_paths(
                        json, site_roots=get_site_roots()
                    )
                ]
            else:
                search_path = create_search_paths(
                    search_path_json, site_roots=get_site_roots()
                )

            python_version_json = configuration_json.pop("python_version", None)
            if python_version_json is None:
                python_version = None
            elif isinstance(python_version_json, str):
                python_version = PythonVersion.from_string(python_version_json)
            else:
                raise InvalidConfiguration(
                    "Expect python version to be a string but got"
                    + f"'{python_version_json}'"
                )

            shared_memory_json = ensure_option_type(
                configuration_json, "shared_memory", dict
            )
            if shared_memory_json is None:
                shared_memory = SharedMemory()
            else:
                shared_memory = SharedMemory(
                    heap_size=ensure_option_type(shared_memory_json, "heap_size", int),
                    dependency_table_power=ensure_option_type(
                        shared_memory_json, "dependency_table_power", int
                    ),
                    hash_table_power=ensure_option_type(
                        shared_memory_json, "hash_table_power", int
                    ),
                )
                for unrecognized_key in shared_memory_json:
                    LOG.warning(f"Unrecognized configuration item: {unrecognized_key}")

            source_directories_json = ensure_option_type(
                configuration_json, "source_directories", list
            )
            if isinstance(source_directories_json, list):
                source_directories = [
                    element
                    for json in source_directories_json
                    for element in create_search_paths(
                        json, site_roots=get_site_roots()
                    )
                ]
            else:
                source_directories = None

            ide_features_json = ensure_option_type(
                configuration_json, "ide_features", dict
            )
            if ide_features_json is None:
                ide_features = None
            else:
                ide_features = IdeFeatures(
                    hover_enabled=ensure_option_type(
                        ide_features_json, "hover_enabled", bool
                    ),
                    go_to_definition_enabled=ensure_option_type(
                        ide_features_json, "go_to_definition_enabled", bool
                    ),
                )
                for unrecognized_key in ide_features_json:
                    LOG.warning(f"Unrecognized configuration item: {unrecognized_key}")

            unwatched_dependency_json = ensure_option_type(
                configuration_json, "unwatched_dependency", dict
            )
            if unwatched_dependency_json is None:
                unwatched_dependency = None
            else:
                unwatched_dependency = UnwatchedDependency.from_json(
                    unwatched_dependency_json
                )

            partial_configuration = PartialConfiguration(
                binary=ensure_option_type(configuration_json, "binary", str),
                buck_mode=PlatformAware.from_json(
                    ensure_optional_string_or_string_dict(
                        configuration_json, "buck_mode"
                    ),
                    "buck_mode",
                ),
                disabled=ensure_option_type(configuration_json, "disabled", bool),
                do_not_ignore_errors_in=ensure_string_list(
                    configuration_json, "do_not_ignore_errors_in"
                ),
                dot_pyre_directory=Path(dot_pyre_directory)
                if dot_pyre_directory is not None
                else None,
                excludes=ensure_string_list(
                    configuration_json, "exclude", allow_single_string=True
                ),
                extensions=[
                    ExtensionElement.from_json(json)
                    for json in ensure_list(configuration_json, "extensions")
                ],
                ide_features=ide_features,
                ignore_all_errors=ensure_string_list(
                    configuration_json, "ignore_all_errors"
                ),
                ignore_infer=ensure_string_list(configuration_json, "ignore_infer"),
                isolation_prefix=ensure_option_type(
                    configuration_json, "isolation_prefix", str
                ),
                logger=ensure_option_type(configuration_json, "logger", str),
                number_of_workers=ensure_option_type(
                    configuration_json, "workers", int
                ),
                oncall=ensure_option_type(configuration_json, "oncall", str),
                other_critical_files=ensure_string_list(
                    configuration_json, "critical_files"
                ),
                pysa_version_hash=ensure_option_type(
                    configuration_json, "pysa_version", str
                ),
                python_version=python_version,
                shared_memory=shared_memory,
                search_path=search_path,
                source_directories=source_directories,
                strict=ensure_option_type(configuration_json, "strict", bool),
                taint_models_path=ensure_string_list(
                    configuration_json, "taint_models_path", allow_single_string=True
                ),
                targets=ensure_optional_string_list(configuration_json, "targets"),
                typeshed=ensure_option_type(configuration_json, "typeshed", str),
                unwatched_dependency=unwatched_dependency,
                use_buck2=ensure_option_type(configuration_json, "use_buck2", bool),
                version_hash=ensure_option_type(configuration_json, "version", str),
            )

            # Check for deprecated and unused keys
            for (
                deprecated_key,
                replacement_key,
            ) in PartialConfiguration._get_depreacted_map().items():
                if deprecated_key in configuration_json:
                    configuration_json.pop(deprecated_key)
                    LOG.warning(
                        f"Configuration file uses deprecated item `{deprecated_key}`. "
                        f"Please migrate to its replacement `{replacement_key}`"
                    )
            extra_keys = PartialConfiguration._get_extra_keys()
            for unrecognized_key in configuration_json:
                if unrecognized_key not in extra_keys:
                    LOG.warning(f"Unrecognized configuration item: {unrecognized_key}")

            return partial_configuration
        except json.JSONDecodeError as error:
            raise InvalidConfiguration(f"Invalid JSON file: {error}")

    @staticmethod
    def from_file(path: Path) -> "PartialConfiguration":
        try:
            contents = path.read_text(encoding="utf-8")
            return PartialConfiguration.from_string(contents)
        except OSError as error:
            raise InvalidConfiguration(f"Error when reading {path}: {error}")

    def expand_relative_paths(self, root: str) -> "PartialConfiguration":
        binary = self.binary
        if binary is not None:
            binary = expand_relative_path(root, binary)
        logger = self.logger
        if logger is not None:
            logger = expand_relative_path(root, logger)
        source_directories = self.source_directories
        if source_directories is not None:
            source_directories = [
                path.expand_relative_root(root) for path in source_directories
            ]
        typeshed = self.typeshed
        if typeshed is not None:
            typeshed = expand_relative_path(root, typeshed)
        unwatched_dependency = self.unwatched_dependency
        if unwatched_dependency is not None:
            files = unwatched_dependency.files
            unwatched_dependency = UnwatchedDependency(
                change_indicator=unwatched_dependency.change_indicator,
                files=UnwatchedFiles(
                    root=expand_relative_path(root, files.root),
                    checksum_path=files.checksum_path,
                ),
            )
        return PartialConfiguration(
            binary=binary,
            buck_mode=self.buck_mode,
            disabled=self.disabled,
            do_not_ignore_errors_in=[
                expand_relative_path(root, path)
                for path in self.do_not_ignore_errors_in
            ],
            dot_pyre_directory=self.dot_pyre_directory,
            excludes=self.excludes,
            extensions=self.extensions,
            ide_features=self.ide_features,
            ignore_all_errors=[
                expand_relative_path(root, path) for path in self.ignore_all_errors
            ],
            ignore_infer=[
                expand_relative_path(root, path) for path in self.ignore_infer
            ],
            isolation_prefix=self.isolation_prefix,
            logger=logger,
            number_of_workers=self.number_of_workers,
            oncall=self.oncall,
            other_critical_files=[
                expand_relative_path(root, path) for path in self.other_critical_files
            ],
            pysa_version_hash=self.pysa_version_hash,
            python_version=self.python_version,
            shared_memory=self.shared_memory,
            search_path=[path.expand_relative_root(root) for path in self.search_path],
            source_directories=source_directories,
            strict=self.strict,
            taint_models_path=[
                expand_relative_path(root, path) for path in self.taint_models_path
            ],
            targets=self.targets,
            typeshed=typeshed,
            unwatched_dependency=unwatched_dependency,
            use_buck2=self.use_buck2,
            version_hash=self.version_hash,
        )


def merge_partial_configurations(
    base: PartialConfiguration, override: PartialConfiguration
) -> PartialConfiguration:
    def overwrite_base(base: Optional[T], override: Optional[T]) -> Optional[T]:
        return base if override is None else override

    def overwrite_base_ide_features(
        base: Optional[IdeFeatures], override: Optional[IdeFeatures]
    ) -> Optional[IdeFeatures]:
        if override is None:
            return base
        if base is None:
            return override
        return IdeFeatures(
            hover_enabled=overwrite_base(base.hover_enabled, override.hover_enabled),
            go_to_definition_enabled=overwrite_base(
                base.go_to_definition_enabled, override.go_to_definition_enabled
            ),
        )

    def prepend_base(base: Sequence[T], override: Sequence[T]) -> Sequence[T]:
        return list(override) + list(base)

    def raise_when_overridden(
        base: Optional[T], override: Optional[T], name: str
    ) -> Optional[T]:
        if base is None:
            return override
        elif override is None:
            return base
        else:
            raise InvalidConfiguration(
                f"Configuration option `{name}` cannot be overridden."
            )

    return PartialConfiguration(
        binary=overwrite_base(base.binary, override.binary),
        buck_mode=PlatformAware.merge(base.buck_mode, override.buck_mode),
        disabled=overwrite_base(base.disabled, override.disabled),
        do_not_ignore_errors_in=prepend_base(
            base.do_not_ignore_errors_in, override.do_not_ignore_errors_in
        ),
        dot_pyre_directory=overwrite_base(
            base.dot_pyre_directory, override.dot_pyre_directory
        ),
        excludes=prepend_base(base.excludes, override.excludes),
        extensions=prepend_base(base.extensions, override.extensions),
        ide_features=overwrite_base_ide_features(
            base.ide_features, override.ide_features
        ),
        ignore_all_errors=prepend_base(
            base.ignore_all_errors, override.ignore_all_errors
        ),
        ignore_infer=prepend_base(base.ignore_infer, override=override.ignore_infer),
        isolation_prefix=overwrite_base(
            base.isolation_prefix, override.isolation_prefix
        ),
        logger=overwrite_base(base.logger, override.logger),
        number_of_workers=overwrite_base(
            base.number_of_workers, override.number_of_workers
        ),
        oncall=overwrite_base(base.oncall, override.oncall),
        other_critical_files=prepend_base(
            base.other_critical_files, override.other_critical_files
        ),
        pysa_version_hash=overwrite_base(
            base.pysa_version_hash, override.pysa_version_hash
        ),
        python_version=overwrite_base(base.python_version, override.python_version),
        shared_memory=SharedMemory(
            heap_size=overwrite_base(
                base.shared_memory.heap_size, override.shared_memory.heap_size
            ),
            dependency_table_power=overwrite_base(
                base.shared_memory.dependency_table_power,
                override.shared_memory.dependency_table_power,
            ),
            hash_table_power=overwrite_base(
                base.shared_memory.hash_table_power,
                override.shared_memory.hash_table_power,
            ),
        ),
        search_path=prepend_base(base.search_path, override.search_path),
        source_directories=raise_when_overridden(
            base.source_directories,
            override.source_directories,
            name="source_directories",
        ),
        strict=overwrite_base(base.strict, override.strict),
        taint_models_path=prepend_base(
            base.taint_models_path, override.taint_models_path
        ),
        targets=raise_when_overridden(base.targets, override.targets, name="targets"),
        typeshed=overwrite_base(base.typeshed, override.typeshed),
        unwatched_dependency=overwrite_base(
            base.unwatched_dependency, override.unwatched_dependency
        ),
        use_buck2=overwrite_base(base.use_buck2, override.use_buck2),
        version_hash=overwrite_base(base.version_hash, override.version_hash),
    )


@dataclass(frozen=True)
class Configuration:
    project_root: str
    dot_pyre_directory: Path

    binary: Optional[str] = None
    buck_mode: Optional[PlatformAware[str]] = None
    disabled: bool = False
    do_not_ignore_errors_in: Sequence[str] = field(default_factory=list)
    excludes: Sequence[str] = field(default_factory=list)
    extensions: Sequence[ExtensionElement] = field(default_factory=list)
    ide_features: Optional[IdeFeatures] = None
    ignore_all_errors: Sequence[str] = field(default_factory=list)
    ignore_infer: Sequence[str] = field(default_factory=list)
    isolation_prefix: Optional[str] = None
    logger: Optional[str] = None
    number_of_workers: Optional[int] = None
    oncall: Optional[str] = None
    other_critical_files: Sequence[str] = field(default_factory=list)
    pysa_version_hash: Optional[str] = None
    python_version: Optional[PythonVersion] = None
    shared_memory: SharedMemory = SharedMemory()
    relative_local_root: Optional[str] = None
    search_path: Sequence[SearchPathElement] = field(default_factory=list)
    source_directories: Optional[Sequence[SearchPathElement]] = None
    strict: bool = False
    taint_models_path: Sequence[str] = field(default_factory=list)
    targets: Optional[Sequence[str]] = None
    typeshed: Optional[str] = None
    unwatched_dependency: Optional[UnwatchedDependency] = None
    use_buck2: bool = False
    version_hash: Optional[str] = None

    @staticmethod
    def from_partial_configuration(
        project_root: Path,
        relative_local_root: Optional[str],
        partial_configuration: PartialConfiguration,
        in_virtual_environment: Optional[bool] = None,
    ) -> "Configuration":
        search_path = partial_configuration.search_path
        if len(search_path) == 0 and _in_virtual_environment(in_virtual_environment):
            LOG.warning("Using virtual environment site-packages in search path...")
            search_path = [SimpleSearchPathElement(root) for root in get_site_roots()]

        return Configuration(
            project_root=str(project_root),
            dot_pyre_directory=_get_optional_value(
                partial_configuration.dot_pyre_directory, project_root / LOG_DIRECTORY
            ),
            binary=partial_configuration.binary,
            buck_mode=partial_configuration.buck_mode,
            disabled=_get_optional_value(partial_configuration.disabled, default=False),
            do_not_ignore_errors_in=partial_configuration.do_not_ignore_errors_in,
            excludes=partial_configuration.excludes,
            extensions=partial_configuration.extensions,
            ide_features=partial_configuration.ide_features,
            ignore_all_errors=partial_configuration.ignore_all_errors,
            ignore_infer=partial_configuration.ignore_infer,
            isolation_prefix=partial_configuration.isolation_prefix,
            logger=partial_configuration.logger,
            number_of_workers=partial_configuration.number_of_workers,
            oncall=partial_configuration.oncall,
            other_critical_files=partial_configuration.other_critical_files,
            pysa_version_hash=partial_configuration.pysa_version_hash,
            python_version=partial_configuration.python_version,
            shared_memory=partial_configuration.shared_memory,
            relative_local_root=relative_local_root,
            search_path=[
                path.expand_global_root(str(project_root)) for path in search_path
            ],
            source_directories=partial_configuration.source_directories,
            strict=_get_optional_value(partial_configuration.strict, default=False),
            taint_models_path=partial_configuration.taint_models_path,
            targets=partial_configuration.targets,
            typeshed=partial_configuration.typeshed,
            unwatched_dependency=partial_configuration.unwatched_dependency,
            use_buck2=_get_optional_value(
                partial_configuration.use_buck2, default=False
            ),
            version_hash=partial_configuration.version_hash,
        )

    @property
    def log_directory(self) -> str:
        if self.relative_local_root is None:
            return str(self.dot_pyre_directory)
        return str(self.dot_pyre_directory / self.relative_local_root)

    @property
    def local_root(self) -> Optional[str]:
        if self.relative_local_root is None:
            return None
        return os.path.join(self.project_root, self.relative_local_root)

    def to_json(self) -> Dict[str, object]:
        """
        This method is for display purpose only. Do *NOT* expect this method
        to produce JSONs that can be de-serialized back into configurations.
        """
        binary = self.binary
        buck_mode = self.buck_mode
        isolation_prefix = self.isolation_prefix
        logger = self.logger
        number_of_workers = self.number_of_workers
        oncall = self.oncall
        pysa_version_hash = self.pysa_version_hash
        python_version = self.python_version
        relative_local_root = self.relative_local_root
        source_directories = self.source_directories
        targets = self.targets
        typeshed = self.typeshed
        unwatched_dependency = self.unwatched_dependency
        version_hash = self.version_hash
        return {
            "global_root": self.project_root,
            "dot_pyre_directory": str(self.dot_pyre_directory),
            **({"binary": binary} if binary is not None else {}),
            **({"buck_mode": buck_mode.to_json()} if buck_mode is not None else {}),
            "disabled": self.disabled,
            "do_not_ignore_errors_in": list(self.do_not_ignore_errors_in),
            "excludes": list(self.excludes),
            "extensions": list(self.extensions),
            "ignore_all_errors": list(self.ignore_all_errors),
            "ignore_infer": list(self.ignore_infer),
            **(
                {"isolation_prefix": isolation_prefix}
                if isolation_prefix is not None
                else {}
            ),
            **({"logger": logger} if logger is not None else {}),
            **({"oncall": oncall} if oncall is not None else {}),
            **({"workers": number_of_workers} if number_of_workers is not None else {}),
            "other_critical_files": list(self.other_critical_files),
            **(
                {"pysa_version_hash": pysa_version_hash}
                if pysa_version_hash is not None
                else {}
            ),
            **(
                {"python_version": python_version.to_string()}
                if python_version is not None
                else {}
            ),
            **(
                {"shared_memory": self.shared_memory.to_json()}
                if self.shared_memory != SharedMemory()
                else {}
            ),
            **(
                {"relative_local_root": relative_local_root}
                if relative_local_root is not None
                else {}
            ),
            "search_path": [path.path() for path in self.search_path],
            **(
                {"source_directories": [path.path() for path in source_directories]}
                if source_directories is not None
                else {}
            ),
            "strict": self.strict,
            "taint_models_path": list(self.taint_models_path),
            **({"targets": list(targets)} if targets is not None else {}),
            **({"typeshed": typeshed} if typeshed is not None else {}),
            **(
                {"unwatched_dependency": unwatched_dependency.to_json()}
                if unwatched_dependency is not None
                else {}
            ),
            "use_buck2": self.use_buck2,
            **({"version_hash": version_hash} if version_hash is not None else {}),
        }

    def get_source_directories(self) -> List[SearchPathElement]:
        return list(self.source_directories or [])

    def get_existent_unwatched_dependency(
        self,
    ) -> Optional[UnwatchedDependency]:
        unwatched_dependency = self.unwatched_dependency
        if unwatched_dependency is None:
            return None
        unwatched_root = Path(unwatched_dependency.files.root)
        try:
            if not unwatched_root.is_dir():
                LOG.warning(
                    "Nonexistent directory passed in to `unwatched_dependency`: "
                    f"`{unwatched_root}`"
                )
                return None
            checksum_path = unwatched_root / unwatched_dependency.files.checksum_path
            if not checksum_path.is_file():
                LOG.warning(
                    "Nonexistent file passed in to `unwatched_dependency`: "
                    f"`{checksum_path}`"
                )
                return None
            return self.unwatched_dependency
        except PermissionError as error:
            LOG.warning(str(error))
            return None

    # Expansion and validation of search paths cannot happen at Configuration creation
    # because link trees need to be built first.
    def expand_and_get_existent_search_paths(self) -> List[SearchPathElement]:
        existent_paths = _expand_and_get_existent_paths(self.search_path)

        typeshed_root = self.get_typeshed_respecting_override()
        typeshed_paths = (
            []
            if typeshed_root is None
            else [
                SimpleSearchPathElement(str(element))
                for element in find_directories.find_typeshed_search_paths(
                    Path(typeshed_root)
                )
            ]
        )

        # pyre-ignore: Unsupported operand [58]: `+` is not supported for
        # operand types `List[SearchPathElement]` and `Union[List[typing.Any],
        # List[SimpleSearchPathElement]]`
        return existent_paths + typeshed_paths

    def expand_and_filter_nonexistent_paths(self) -> "Configuration":
        source_directories = self.source_directories

        return Configuration(
            project_root=self.project_root,
            dot_pyre_directory=self.dot_pyre_directory,
            binary=self.binary,
            buck_mode=self.buck_mode,
            disabled=self.disabled,
            do_not_ignore_errors_in=self.do_not_ignore_errors_in,
            excludes=self.excludes,
            extensions=self.extensions,
            ide_features=self.ide_features,
            ignore_all_errors=self.ignore_all_errors,
            ignore_infer=self.ignore_infer,
            isolation_prefix=self.isolation_prefix,
            logger=self.logger,
            number_of_workers=self.number_of_workers,
            oncall=self.oncall,
            other_critical_files=self.other_critical_files,
            pysa_version_hash=self.pysa_version_hash,
            python_version=self.python_version,
            shared_memory=self.shared_memory,
            relative_local_root=self.relative_local_root,
            search_path=self.search_path,
            source_directories=_expand_and_get_existent_paths(source_directories)
            if source_directories
            else None,
            strict=self.strict,
            taint_models_path=self.taint_models_path,
            targets=self.targets,
            typeshed=self.typeshed,
            unwatched_dependency=self.unwatched_dependency,
            use_buck2=self.use_buck2,
            version_hash=self.version_hash,
        )

    def get_existent_ignore_infer_paths(self) -> List[str]:
        existent_paths = []
        for path in self.ignore_infer:
            if os.path.exists(path):
                existent_paths.append(path)
            else:
                LOG.warn(f"Filtering out nonexistent path in `ignore_infer`: {path}")
        return existent_paths

    def get_existent_do_not_ignore_errors_in_paths(self) -> List[str]:
        """
        This is a separate method because we want to check for existing files
        at the time this is called, not when the configuration is
        constructed.
        """
        ignore_paths = [
            _expand_global_root(path, global_root=self.project_root)
            for path in self.do_not_ignore_errors_in
        ]
        paths = []
        for path in ignore_paths:
            if os.path.exists(path):
                paths.append(path)
            else:
                LOG.debug(
                    "Filtering out nonexistent paths in `do_not_ignore_errors_in`: "
                    f"{path}"
                )
        return paths

    def get_existent_ignore_all_errors_paths(self) -> List[str]:
        """
        This is a separate method because we want to check for existing files
        at the time this is called, not when the configuration is
        constructed.
        """
        return _expand_and_get_existent_ignore_all_errors_path(
            self.ignore_all_errors, self.project_root
        )

    def get_binary_respecting_override(self) -> Optional[str]:
        binary = self.binary
        if binary is not None:
            return binary

        LOG.info(f"No binary specified, looking for `{BINARY_NAME}` in PATH")
        binary_candidate = shutil.which(BINARY_NAME)
        if binary_candidate is None:
            binary_candidate_name = os.path.join(
                os.path.dirname(sys.argv[0]), BINARY_NAME
            )
            binary_candidate = shutil.which(binary_candidate_name)
        if binary_candidate is not None:
            return binary_candidate
        return None

    def get_typeshed_respecting_override(self) -> Optional[str]:
        typeshed = self.typeshed
        if typeshed is not None:
            return typeshed

        LOG.info("No typeshed specified, looking for it...")
        auto_determined_typeshed = find_directories.find_typeshed()
        if auto_determined_typeshed is None:
            LOG.warning(
                "Could not find a suitable typeshed. Types for Python builtins "
                "and standard libraries may be missing!"
            )
            return None
        else:
            LOG.info(f"Found: `{auto_determined_typeshed}`")
            return str(auto_determined_typeshed)

    def get_version_hash_respecting_override(self) -> Optional[str]:
        overriding_version_hash = os.getenv("PYRE_VERSION_HASH")
        if overriding_version_hash:
            LOG.warning(f"Version hash overridden with `{overriding_version_hash}`")
            return overriding_version_hash
        return self.version_hash

    def get_binary_version(self) -> Optional[str]:
        binary = self.get_binary_respecting_override()
        if binary is None:
            return None
        status = subprocess.run(
            [binary, "-version"], stdout=subprocess.PIPE, universal_newlines=True
        )
        return status.stdout.strip() if status.returncode == 0 else None

    def get_number_of_workers(self) -> int:
        number_of_workers = self.number_of_workers
        if number_of_workers is not None and number_of_workers > 0:
            return number_of_workers

        try:
            default_number_of_workers = max(multiprocessing.cpu_count() - 4, 1)
        except NotImplementedError:
            default_number_of_workers = 4

        LOG.info(
            "Could not determine the number of Pyre workers from configuration. "
            f"Auto-set the value to {default_number_of_workers}."
        )
        return default_number_of_workers

    def is_hover_enabled(self) -> bool:
        if self.ide_features is None:
            return IdeFeatures.DEFAULT_HOVER_ENABLED
        return self.ide_features.is_hover_enabled()

    def is_go_to_definition_enabled(self) -> bool:
        if self.ide_features is None:
            return IdeFeatures.DEFAULT_GO_TO_DEFINITION_ENABLED
        return self.ide_features.is_go_to_definition_enabled()

    def get_valid_extension_suffixes(self) -> List[str]:
        vaild_extensions = []
        for extension in self.extensions:
            if not extension.suffix.startswith("."):
                LOG.warning(
                    "Filtering out extension which does not start with `.`: "
                    f"`{extension.suffix}`"
                )
            else:
                vaild_extensions.append(extension.command_line_argument())
        return vaild_extensions

    def get_isolation_prefix_respecting_override(self) -> Optional[str]:
        """We need this to disable an isolation prefix set in a configuration.
        Merely omitting the CLI flag would not disable the isolation prefix
        because we would just fall back to the configuration value.

        With this, we can pass `--isolation-prefix ''` as a CLI argument or
        override `isolation_prefix` as `""` in a local configuration."""
        return None if self.isolation_prefix == "" else self.isolation_prefix

    def get_python_version(self) -> PythonVersion:
        python_version = self.python_version
        if python_version is not None:
            return python_version
        else:
            version_info = sys.version_info
            return PythonVersion(
                major=version_info.major,
                minor=version_info.minor,
                micro=version_info.micro,
            )


def create_configuration(
    arguments: command_arguments.CommandArguments, base_directory: Path
) -> Configuration:
    local_root_argument = arguments.local_configuration
    search_base = (
        base_directory
        if local_root_argument is None
        else base_directory / local_root_argument
    )
    found_root = find_directories.find_global_and_local_root(search_base)

    # If the local root was explicitly specified but does not exist, return an
    # error instead of falling back to current directory.
    if local_root_argument is not None:
        if found_root is None:
            raise InvalidConfiguration(
                "A local configuration path was explicitly specified, but no"
                + f" {CONFIGURATION_FILE} file was found in {search_base}"
                + " or its parents."
            )
        elif found_root.local_root is None:
            raise InvalidConfiguration(
                "A local configuration path was explicitly specified, but no"
                + f" {LOCAL_CONFIGURATION_FILE} file was found in {search_base}"
                + " or its parents."
            )

    command_argument_configuration = PartialConfiguration.from_command_arguments(
        arguments
    ).expand_relative_paths(str(Path.cwd()))
    if found_root is None:
        project_root = Path.cwd()
        relative_local_root = None
        partial_configuration = command_argument_configuration
    else:
        project_root = found_root.global_root
        relative_local_root = None
        partial_configuration = PartialConfiguration.from_file(
            project_root / CONFIGURATION_FILE
        ).expand_relative_paths(str(project_root))
        local_root = found_root.local_root
        if local_root is not None:
            relative_local_root = get_relative_local_root(project_root, local_root)
            partial_configuration = merge_partial_configurations(
                base=partial_configuration,
                override=PartialConfiguration.from_file(
                    local_root / LOCAL_CONFIGURATION_FILE
                ).expand_relative_paths(str(local_root)),
            )
        partial_configuration = merge_partial_configurations(
            base=partial_configuration,
            override=command_argument_configuration,
        )

    configuration = Configuration.from_partial_configuration(
        project_root, relative_local_root, partial_configuration
    )
    return configuration.expand_and_filter_nonexistent_paths()


def check_nested_local_configuration(configuration: Configuration) -> None:
    """
    Raises `InvalidConfiguration` if the check fails.
    """
    local_root = configuration.local_root
    if local_root is None:
        return

    def is_subdirectory(child: Path, parent: Path) -> bool:
        return parent == child or parent in child.parents

    # We search from the parent of the local root, looking for another local
    # configuration file that lives above the current one
    local_root_path = Path(local_root).resolve()
    current_directory = local_root_path.parent
    while True:
        found_root = find_directories.find_global_and_local_root(current_directory)
        if found_root is None:
            break

        nesting_local_root = found_root.local_root
        if nesting_local_root is None:
            break

        nesting_configuration = PartialConfiguration.from_file(
            nesting_local_root / LOCAL_CONFIGURATION_FILE
        ).expand_relative_paths(str(nesting_local_root))
        nesting_ignored_all_errors_path = (
            _expand_and_get_existent_ignore_all_errors_path(
                nesting_configuration.ignore_all_errors, str(found_root.global_root)
            )
        )
        if not any(
            is_subdirectory(child=local_root_path, parent=Path(path))
            for path in nesting_ignored_all_errors_path
        ):
            error_message = (
                "Local configuration is nested under another local configuration at "
                f"`{nesting_local_root}`.\nPlease add `{local_root_path}` to the "
                "`ignore_all_errors` field of the parent, or combine the sources "
                "into a single configuration, or split the parent configuration to "
                "avoid inconsistent errors."
            )
            raise InvalidConfiguration(error_message)
        current_directory = nesting_local_root.parent


def check_open_source_version(configuration: Configuration) -> None:
    """
    Check if version specified in configuration matches running version and warn
    if it does not.
    """
    expected_version = configuration.version_hash
    if expected_version is None or not re.match(r"\d+\.\d+\.\d+", expected_version):
        return

    try:
        # pyre-ignore[21]: dynamic import
        from pyre_check import __version__ as actual_version

        if expected_version != actual_version:
            LOG.warning(
                textwrap.dedent(
                    f"""\
                    Your running version does not match the configured version for this
                    project (running {actual_version}, expected {expected_version})."""
                )
            )
    except ImportError:
        pass