text:
"""Abstract base class of Port-specific entry points for the layout tests
test infrastructure (the Port and Driver classes).
"""

import cgi
import collections
import difflib
import errno
import functools
import json
import logging
import optparse
import re
import sys

from webkitpy.common import find_files
from webkitpy.common import read_checksum_from_png
from webkitpy.common.memoized import memoized
from webkitpy.common.system.executive import ScriptError
from webkitpy.common.system.path import cygpath, abspath_to_uri
from webkitpy.common.webkit_finder import WebKitFinder
from webkitpy.layout_tests.layout_package.bot_test_expectations import BotTestExpectationsFactory
from webkitpy.layout_tests.models import test_run_results
from webkitpy.layout_tests.models.test_configuration import TestConfiguration
from webkitpy.layout_tests.models.test_expectations import SKIP
from webkitpy.layout_tests.port import driver
from webkitpy.layout_tests.port import server_process
from webkitpy.layout_tests.port.factory import PortFactory
from webkitpy.layout_tests.servers import apache_http
from webkitpy.layout_tests.servers import pywebsocket
from webkitpy.layout_tests.servers import wptserve

_log = logging.getLogger(__name__)

# FIXME: This class should merge with WebKitPort now that Chromium behaves mostly like other webkit ports.
class Port(object):
    """Abstract class for Port-specific hooks for the layout_test package."""

    # Subclasses override this. This should indicate the basic implementation
    # part of the port name, e.g., 'mac', 'win', 'gtk'; there is probably (?)
    # one unique value per class.

    # FIXME: We should probably rename this to something like 'implementation_name'.
    port_name = None

    # Test names resemble unix relative paths, and use '/' as a directory separator.
    TEST_PATH_SEPARATOR = '/'

    ALL_BUILD_TYPES = ('debug', 'release')

    CONTENT_SHELL_NAME = 'content_shell'

    ALL_SYSTEMS = (

        # FIXME: We treat Retina (High-DPI) devices as if they are running
        # a different operating system version. This isn't accurate, but will work until
        # we need to test and support baselines across multiple O/S versions.
        ('retina', 'x86'),

        ('mac10.9', 'x86'),
        ('mac10.10', 'x86'),
        ('mac10.11', 'x86'),
        ('win7', 'x86'),
        ('win10', 'x86'),
        ('trusty', 'x86_64'),
        # FIXME: Technically this should be 'arm', but adding a third architecture type breaks TestConfigurationConverter.
        # If we need this to be 'arm' in the future, then we first have to fix TestConfigurationConverter.
        ('icecreamsandwich', 'x86'),
    )

    CONFIGURATION_SPECIFIER_MACROS = {
        'mac': ['retina', 'mac10.9', 'mac10.10', 'mac10.11'],
        'win': ['win7', 'win10'],
        'linux': ['trusty'],
        'android': ['icecreamsandwich'],
    }

    DEFAULT_BUILD_DIRECTORIES = ('out',)

    # overridden in subclasses.
    FALLBACK_PATHS = {}

    SUPPORTED_VERSIONS = []

    # URL to the build requirements page.
    BUILD_REQUIREMENTS_URL = ''

    # Because this is an abstract base class, arguments to functions may be unused
    # in this class, but used by subclasses - pylint: disable=unused-argument

    @classmethod
    def latest_platform_fallback_path(cls):
        return cls.FALLBACK_PATHS[cls.SUPPORTED_VERSIONS[-1]]

    @classmethod
    def _static_build_path(cls, filesystem, build_directory, chromium_base, target, comps):
        if build_directory:
            return filesystem.join(build_directory, target, *comps)

        hits = []
        for directory in cls.DEFAULT_BUILD_DIRECTORIES:
            base_dir = filesystem.join(chromium_base, directory, target)
            path = filesystem.join(base_dir, *comps)
            if filesystem.exists(path):
                hits.append((filesystem.mtime(path), path))

        if hits:
            hits.sort(reverse=True)
            return hits[0][1]  # Return the newest file found.

        # We have to default to something, so pick the last one.
        return filesystem.join(base_dir, *comps)

    @classmethod
    def determine_full_port_name(cls, host, options, port_name):
        """Return a fully-specified port name that can be used to construct objects."""
        # Subclasses will usually override this.
        assert port_name.startswith(cls.port_name)
        return port_name

    def __init__(self, host, port_name, options=None, **kwargs):

        # This value may be different from cls.port_name by having version modifiers
        # and other fields appended to it (for example, 'qt-arm' or 'mac-wk2').
        self._name = port_name

        # These are default values that should be overridden in a subclasses.
        self._version = ''
        self._architecture = 'x86'

        # FIXME: Ideally we'd have a package-wide way to get a
        # well-formed options object that had all of the necessary
        # options defined on it.
        self._options = options or optparse.Values()

        self.host = host
        self._executive = host.executive
        self._filesystem = host.filesystem
        self._webkit_finder = WebKitFinder(host.filesystem)

        self._http_server = None
        self._websocket_server = None
        self._is_wptserve_enabled = getattr(options, 'enable_wptserve', False)
        self._wpt_server = None
        self._image_differ = None
        self.server_process_constructor = server_process.ServerProcess  # overridable for testing
        self._http_lock = None  # FIXME: Why does this live on the port object?
        self._dump_reader = None

        # FIXME: prettypatch.py knows this path, why is it copied here?
        self._pretty_patch_path = self.path_from_webkit_base("Tools", "Scripts", "webkitruby", "PrettyPatch", "prettify.rb")
        self._pretty_patch_available = None

        if not hasattr(options, 'configuration') or not options.configuration:
            self.set_option_default('configuration', self.default_configuration())
        if not hasattr(options, 'target') or not options.target:
            self.set_option_default('target', self._options.configuration)
        self._test_configuration = None
        self._reftest_list = {}
        self._results_directory = None
        self._virtual_test_suites = None

    def __str__(self):
        return "Port{name=%s, version=%s, architecture=%s, test_configuration=%s}" % (
            self._name, self._version, self._architecture, self._test_configuration)

    def buildbot_archives_baselines(self):
        return True

    def additional_driver_flag(self):
        if self.driver_name() == self.CONTENT_SHELL_NAME:
            return ['--run-layout-test']
        return []

    def supports_per_test_timeout(self):
        return False

    def default_pixel_tests(self):
        return True

    def default_smoke_test_only(self):
        return False

    def default_timeout_ms(self):
        timeout_ms = 6 * 1000
        if self.get_option('configuration') == 'Debug':
            # Debug is usually 2x-3x slower than Release.
            return 3 * timeout_ms
        return timeout_ms

    def driver_stop_timeout(self):
        """Returns the amount of time in seconds to wait before killing the process in driver.stop()."""
        # We want to wait for at least 3 seconds, but if we are really slow, we want to be slow on cleanup as
        # well (for things like ASAN, Valgrind, etc.)
        return 3.0 * float(self.get_option('time_out_ms', '0')) / self.default_timeout_ms()

    def pretty_patch_available(self):
        if self._pretty_patch_available is None:
            self._pretty_patch_available = self.check_pretty_patch(more_logging=False)
        return self._pretty_patch_available

    def default_batch_size(self):
        """Return the default batch size to use for this port."""
        if self.get_option('enable_sanitizer'):
            # ASAN/MSAN/TSAN use more memory than regular content_shell. Their
            # memory usage may also grow over time, up to a certain point.
            # Relaunching the driver periodically helps keep it under control.
            return 40
        # The default is infinite batch size.
        return None

    def default_child_processes(self):
        """Return the number of child processes to use for this port."""
        return self._executive.cpu_count()

    def max_drivers_per_process(self):
        """The maximum number of drivers a child process can use for this port."""
        return 2

    def default_max_locked_shards(self):
        """Return the number of "locked" shards to run in parallel (like the http tests)."""
        max_locked_shards = int(self.default_child_processes()) / 4
        if not max_locked_shards:
            return 1
        return max_locked_shards

    def baseline_platform_dir(self):
        """Return the absolute path to the default (version-independent) platform-specific results."""
        return self._filesystem.join(self.layout_tests_dir(), 'platform', self.port_name)

    def baseline_version_dir(self):
        """Return the absolute path to the platform-and-version-specific results."""
        baseline_search_paths = self.baseline_search_path()
        return baseline_search_paths[0]

    def virtual_baseline_search_path(self, test_name):
        suite = self.lookup_virtual_suite(test_name)
        if not suite:
            return None
        return [self._filesystem.join(path, suite.name) for path in self.default_baseline_search_path()]

    def baseline_search_path(self):
        return (self.get_option('additional_platform_directory', []) +
                self._flag_specific_baseline_search_path() +
                self._compare_baseline() +
                self.default_baseline_search_path())

    def default_baseline_search_path(self):
        """Return a list of absolute paths to directories to search under for
        baselines. The directories are searched in order.
        """
        return map(self._absolute_baseline_path, self.FALLBACK_PATHS[self.version()])

    @memoized
    def _compare_baseline(self):
        factory = PortFactory(self.host)
        target_port = self.get_option('compare_port')
        if target_port:
            return factory.get(target_port).default_baseline_search_path()
        return []

    def _check_file_exists(self, path_to_file, file_description,
                           override_step=None, more_logging=True):
        """Verify the file is present where expected or log an error.

        Args:
            file_name: The (human friendly) name or description of the file
                you're looking for (e.g., "HTTP Server"). Used for error logging.
            override_step: An optional string to be logged if the check fails.
            more_logging: Whether or not to log the error messages.

        Returns:
            True if the file exists, else False.
        """
        if not self._filesystem.exists(path_to_file):
            if more_logging:
                _log.error('Unable to find %s', file_description)
                _log.error('    at %s', path_to_file)
                if override_step:
                    _log.error('    %s', override_step)
                    _log.error('')
            return False
        return True

    def check_build(self, needs_http, printer):
        result = True

        dump_render_tree_binary_path = self._path_to_driver()
        result = self._check_file_exists(dump_render_tree_binary_path,
                                         'test driver') and result
        if not result and self.get_option('build'):
            result = self._check_driver_build_up_to_date(
                self.get_option('configuration'))
        else:
            _log.error('')

        if self.get_option('pixel_tests'):
            result = self.check_image_diff() and result

        # It's okay if pretty patch isn't available, but we will at least log messages.
        self._pretty_patch_available = self.check_pretty_patch()

        if self._dump_reader:
            result = self._dump_reader.check_is_functional() and result

        if needs_http:
            result = self.check_httpd() and result

        return test_run_results.OK_EXIT_STATUS if result else test_run_results.UNEXPECTED_ERROR_EXIT_STATUS

    def _check_driver(self):
        driver_path = self._path_to_driver()
        if not self._filesystem.exists(driver_path):
            _log.error("%s was not found at %s", self.driver_name(), driver_path)
            return False
        return True

    def _check_port_build(self):
        # Ports can override this method to do additional checks.
        return True

    def check_sys_deps(self, needs_http):
        """Checks whether the system is properly configured.

        If the port needs to do some runtime checks to ensure that the
        tests can be run successfully, it should override this routine.
        This step can be skipped with --nocheck-sys-deps.

        Returns:
            An exit status code.
        """
        cmd = [self._path_to_driver(), '--check-layout-test-sys-deps']

        local_error = ScriptError()

        def error_handler(script_error):
            local_error.exit_code = script_error.exit_code

        output = self._executive.run_command(cmd, error_handler=error_handler)
        if local_error.exit_code:
            _log.error('System dependencies check failed.')
            _log.error('To override, invoke with --nocheck-sys-deps')
            _log.error('')
            _log.error(output)
            if self.BUILD_REQUIREMENTS_URL is not '':
                _log.error('')
                _log.error('For complete build requirements, please see:')
                _log.error(self.BUILD_REQUIREMENTS_URL)
            return test_run_results.SYS_DEPS_EXIT_STATUS
        return test_run_results.OK_EXIT_STATUS

    def check_image_diff(self):
        """Checks whether image_diff binary exists."""
        image_diff_path = self._path_to_image_diff()
        if not self._filesystem.exists(image_diff_path):
            _log.error("image_diff was not found at %s", image_diff_path)
            return False
        return True

    def check_pretty_patch(self, more_logging=True):
        """Checks whether we can use the PrettyPatch ruby script."""
        try:
            _ = self._executive.run_command(['ruby', '--version'])
        except OSError as error:
            if error.errno in [errno.ENOENT, errno.EACCES, errno.ECHILD]:
                if more_logging:
                    _log.warning("Ruby is not installed; can't generate pretty patches.")
                    _log.warning('')
                return False

        if not self._filesystem.exists(self._pretty_patch_path):
            if more_logging:
                _log.warning("Unable to find %s; can't generate pretty patches.", self._pretty_patch_path)
                _log.warning('')
            return False

        return True

    def check_httpd(self):
        httpd_path = self.path_to_apache()
        if httpd_path:
            try:
                env = self.setup_environ_for_server()
                if self._executive.run_command([httpd_path, "-v"], env=env, return_exit_code=True) != 0:
                    _log.error("httpd seems broken. Cannot run http tests.")
                    return False
                return True
            except OSError:
                pass
        _log.error("No httpd found. Cannot run http tests.")
        return False

    def do_text_results_differ(self, expected_text, actual_text):
        return expected_text != actual_text

    def do_audio_results_differ(self, expected_audio, actual_audio):
        return expected_audio != actual_audio

    def diff_image(self, expected_contents, actual_contents):
        """Compare two images and return a tuple of an image diff, and an error string.

        If an error occurs (like image_diff isn't found, or crashes, we log an error and return True (for a diff).
        """
        # If only one of them exists, return that one.
        if not actual_contents and not expected_contents:
            return (None, None)
        if not actual_contents:
            return (expected_contents, None)
        if not expected_contents:
            return (actual_contents, None)

        tempdir = self._filesystem.mkdtemp()

        expected_filename = self._filesystem.join(str(tempdir), "expected.png")
        self._filesystem.write_binary_file(expected_filename, expected_contents)

        actual_filename = self._filesystem.join(str(tempdir), "actual.png")
        self._filesystem.write_binary_file(actual_filename, actual_contents)

        diff_filename = self._filesystem.join(str(tempdir), "diff.png")

        # image_diff needs native win paths as arguments, so we need to convert them if running under cygwin.
        native_expected_filename = self._convert_path(expected_filename)
        native_actual_filename = self._convert_path(actual_filename)
        native_diff_filename = self._convert_path(diff_filename)

        executable = self._path_to_image_diff()
        # Note that although we are handed 'old', 'new', image_diff wants 'new', 'old'.
        command = [executable, '--diff', native_actual_filename, native_expected_filename, native_diff_filename]

        result = None
        err_str = None
        try:
            exit_code = self._executive.run_command(command, return_exit_code=True)
            if exit_code == 0:
                # The images are the same.
                result = None
            elif exit_code == 1:
                result = self._filesystem.read_binary_file(native_diff_filename)
            else:
                err_str = "Image diff returned an exit code of %s. See http://crbug.com/278596" % exit_code
        except OSError as error:
            err_str = 'error running image diff: %s' % error
        finally:
            self._filesystem.rmtree(str(tempdir))

        return (result, err_str or None)

    def driver_name(self):
        if self.get_option('driver_name'):
            return self.get_option('driver_name')
        return self.CONTENT_SHELL_NAME

    def expected_baselines_by_extension(self, test_name):
        """Returns a dict mapping baseline suffix to relative path for each baseline in
        a test. For reftests, it returns ".==" or ".!=" instead of the suffix.
        """
        # FIXME: The name similarity between this and expected_baselines() below, is unfortunate.
        # We should probably rename them both.
        baseline_dict = {}
        reference_files = self.reference_files(test_name)
        if reference_files:
            # FIXME: How should this handle more than one type of reftest?
            baseline_dict['.' + reference_files[0][0]] = self.relative_test_filename(reference_files[0][1])

        for extension in self.baseline_extensions():
            path = self.expected_filename(test_name, extension, return_default=False)
            baseline_dict[extension] = self.relative_test_filename(path) if path else path

        return baseline_dict

    def baseline_extensions(self):
        """Returns a tuple of all of the non-reftest baseline extensions we use. The extensions include the leading '.'."""
        return ('.wav', '.txt', '.png')

    def expected_baselines(self, test_name, suffix, all_baselines=False):
        """Given a test name, finds where the baseline results are located.

        Args:
        test_name: name of test file (usually a relative path under LayoutTests/)
        suffix: file suffix of the expected results, including dot; e.g.
            '.txt' or '.png'.  This should not be None, but may be an empty
            string.
        all_baselines: If True, return an ordered list of all baseline paths
            for the given platform. If False, return only the first one.
        Returns
        a list of ( platform_dir, results_filename ), where
            platform_dir - abs path to the top of the results tree (or test
                tree)
            results_filename - relative path from top of tree to the results
                file
            (port.join() of the two gives you the full path to the file,
                unless None was returned.)
        Return values will be in the format appropriate for the current
        platform (e.g., "\\" for path separators on Windows). If the results
        file is not found, then None will be returned for the directory,
        but the expected relative pathname will still be returned.

        This routine is generic but lives here since it is used in
        conjunction with the other baseline and filename routines that are
        platform specific.
        """
        baseline_filename = self._filesystem.splitext(test_name)[0] + '-expected' + suffix
        baseline_search_path = self.baseline_search_path()

        baselines = []
        for platform_dir in baseline_search_path:
            if self._filesystem.exists(self._filesystem.join(platform_dir, baseline_filename)):
                baselines.append((platform_dir, baseline_filename))

            if not all_baselines and baselines:
                return baselines

        # If it wasn't found in a platform directory, return the expected
        # result in the test directory, even if no such file actually exists.
        platform_dir = self.layout_tests_dir()
        if self._filesystem.exists(self._filesystem.join(platform_dir, baseline_filename)):
            baselines.append((platform_dir, baseline_filename))

        if baselines:
            return baselines

        return [(None, baseline_filename)]

    def expected_filename(self, test_name, suffix, return_default=True):
        """Given a test name, returns an absolute path to its expected results.

        If no expected results are found in any of the searched directories,
        the directory in which the test itself is located will be returned.
        The return value is in the format appropriate for the platform
        (e.g., "\\" for path separators on windows).

        Args:
        test_name: name of test file (usually a relative path under LayoutTests/)
        suffix: file suffix of the expected results, including dot; e.g. '.txt'
            or '.png'.  This should not be None, but may be an empty string.
        platform: the most-specific directory name to use to build the
            search list of directories, e.g., 'win', or
            'chromium-cg-mac-leopard' (we follow the WebKit format)
        return_default: if True, returns the path to the generic expectation if nothing
            else is found; if False, returns None.

        This routine is generic but is implemented here to live alongside
        the other baseline and filename manipulation routines.
        """
        # FIXME: The [0] here is very mysterious, as is the destructured return.
        platform_dir, baseline_filename = self.expected_baselines(test_name, suffix)[0]
        if platform_dir:
            return self._filesystem.join(platform_dir, baseline_filename)

        actual_test_name = self.lookup_virtual_test_base(test_name)
        if actual_test_name:
            return self.expected_filename(actual_test_name, suffix)

        if return_default:
            return self._filesystem.join(self.layout_tests_dir(), baseline_filename)
        return None

    def expected_checksum(self, test_name):
        """Returns the checksum of the image we expect the test to produce, or None if it is a text-only test."""
        png_path = self.expected_filename(test_name, '.png')

        if self._filesystem.exists(png_path):
            with self._filesystem.open_binary_file_for_reading(png_path) as filehandle:
                return read_checksum_from_png.read_checksum(filehandle)

        return None

    def expected_image(self, test_name):
        """Returns the image we expect the test to produce."""
        baseline_path = self.expected_filename(test_name, '.png')
        if not self._filesystem.exists(baseline_path):
            return None
        return self._filesystem.read_binary_file(baseline_path)

    def expected_audio(self, test_name):
        baseline_path = self.expected_filename(test_name, '.wav')
        if not self._filesystem.exists(baseline_path):
            return None
        return self._filesystem.read_binary_file(baseline_path)

    def expected_text(self, test_name):
        """Returns the text output we expect the test to produce, or None
        if we don't expect there to be any text output.
        End-of-line characters are normalized to '\n'.
        """
        # FIXME: DRT output is actually utf-8, but since we don't decode the
        # output from DRT (instead treating it as a binary string), we read the
        # baselines as a binary string, too.
        baseline_path = self.expected_filename(test_name, '.txt')
        if not self._filesystem.exists(baseline_path):
            return None
        text = self._filesystem.read_binary_file(baseline_path)
        return text.replace("\r\n", "\n")

    def _get_reftest_list(self, test_name):
        dirname = self._filesystem.join(self.layout_tests_dir(), self._filesystem.dirname(test_name))
        if dirname not in self._reftest_list:
            self._reftest_list[dirname] = Port._parse_reftest_list(self._filesystem, dirname)
        return self._reftest_list[dirname]

    @staticmethod
    def _parse_reftest_list(filesystem, test_dirpath):
        reftest_list_path = filesystem.join(test_dirpath, 'reftest.list')
        if not filesystem.isfile(reftest_list_path):
            return None
        reftest_list_file = filesystem.read_text_file(reftest_list_path)

        parsed_list = {}
        for line in reftest_list_file.split('\n'):
            line = re.sub('#.+$', '', line)
            split_line = line.split()
            if len(split_line) == 4:
                # FIXME: Probably one of mozilla's extensions in the reftest.list format. Do we need to support this?
                _log.warning("unsupported reftest.list line '%s' in %s", line, reftest_list_path)
                continue
            if len(split_line) < 3:
                continue
            expectation_type, test_file, ref_file = split_line
            parsed_list.setdefault(filesystem.join(test_dirpath, test_file), []).append(
                (expectation_type, filesystem.join(test_dirpath, ref_file)))
        return parsed_list

    def reference_files(self, test_name):
        """Return a list of expectation (== or !=) and filename pairs"""

        reftest_list = self._get_reftest_list(test_name)
        if not reftest_list:
            reftest_list = []
            for expectation, prefix in (('==', ''), ('!=', '-mismatch')):
                for extension in Port._supported_file_extensions:
                    path = self.expected_filename(test_name, prefix + extension)
                    if self._filesystem.exists(path):
                        reftest_list.append((expectation, path))
            return reftest_list

        return reftest_list.get(self._filesystem.join(self.layout_tests_dir(), test_name), [])  # pylint: disable=E1103

    def tests(self, paths):
        """Return the list of tests found matching paths."""
        tests = self.real_tests(paths)

        suites = self.virtual_test_suites()
        if paths:
            tests.extend(self._virtual_tests_matching_paths(paths, suites))
        else:
            tests.extend(self._all_virtual_tests(suites))
        return tests

    def real_tests(self, paths):
        # When collecting test cases, skip these directories
        skipped_directories = set(['.svn', '_svn', 'platform', 'resources', 'support', 'script-tests', 'reference', 'reftest'])
        files = find_files.find(self._filesystem, self.layout_tests_dir(), paths,
                                skipped_directories, functools.partial(Port.is_test_file, self), self.test_key)
        return self._convert_wpt_file_paths_to_url_paths([self.relative_test_filename(f) for f in files])

    # When collecting test cases, we include any file with these extensions.
    _supported_file_extensions = set(['.html', '.xml', '.xhtml', '.xht', '.pl',
                                      '.htm', '.php', '.svg', '.mht', '.pdf'])

    @staticmethod
    # If any changes are made here be sure to update the isUsedInReftest method in old-run-webkit-tests as well.
    def is_reference_html_file(filesystem, dirname, filename):
        if filename.startswith('ref-') or filename.startswith('notref-'):
            return True
        filename_without_ext, _ = filesystem.splitext(filename)
        for suffix in ['-expected', '-expected-mismatch', '-ref', '-notref']:
            if filename_without_ext.endswith(suffix):
                return True
        return False

    @staticmethod
    def _has_supported_extension(filesystem, filename):
        """Return true if filename is one of the file extensions we want to run a test on."""
        extension = filesystem.splitext(filename)[1]
        return extension in Port._supported_file_extensions

    def is_test_file(self, filesystem, dirname, filename):
        match = re.search(r'[/\\]external[/\\]wpt([/\\].*)?$', dirname)
        if match:
            if match.group(1):
                path_in_wpt = match.group(1)[1:].replace('\\', '/') + '/' + filename
            else:
                path_in_wpt = filename
            return self._manifest_items_for_path(path_in_wpt) is not None
        if 'inspector-unit' in dirname:
            return filesystem.splitext(filename)[1] == '.js'
        return Port._has_supported_extension(
            filesystem, filename) and not Port.is_reference_html_file(filesystem, dirname, filename)

    def _convert_wpt_file_paths_to_url_paths(self, files):
        tests = []
        for file_path in files:
            # Path separators are normalized by relative_test_filename().
            match = re.search(r'external/wpt/(.*)$', file_path)
            if not match:
                tests.append(file_path)
                continue
            path_in_wpt = match.group(1)
            manifest_items = self._manifest_items_for_path(path_in_wpt)
            assert manifest_items is not None
            if len(manifest_items) != 1:
                continue
            url = manifest_items[0][0]
            if url[1:] != path_in_wpt:
                # TODO(tkent): foo.any.js and bar.worker.js should be accessed
                # as foo.any.html, foo.any.worker, and bar.worker with WPTServe.
                continue
            tests.append(file_path)
        return tests

    @memoized
    def _wpt_manifest(self):
        path = self._filesystem.join(self.layout_tests_dir(), 'external', 'wpt', 'MANIFEST.json')
        return json.loads(self._filesystem.read_text_file(path))

    def _manifest_items_for_path(self, path_in_wpt):
        """Returns a manifest item for the given WPT path, or None if not found.

        The format of a manifest item depends on
        https://github.com/w3c/wpt-tools/blob/master/manifest/item.py
        and is assumed to be a list of the format [url, extras],
        or [url, references, extras] for reftests, or None if not found.

        For most testharness tests, the returned manifest_items is expected
        to look like this:: [["/some/test/path.html", {}]]
        """
        items = self._wpt_manifest()['items']
        if path_in_wpt in items['manual']:
            return items['manual'][path_in_wpt]
        elif path_in_wpt in items['reftest']:
            return items['reftest'][path_in_wpt]
        elif path_in_wpt in items['testharness']:
            return items['testharness'][path_in_wpt]
        return None

    ALL_TEST_TYPES = ['audio', 'harness', 'pixel', 'ref', 'text', 'unknown']

    def test_type(self, test_name):
        fs = self._filesystem
        if fs.exists(self.expected_filename(test_name, '.png')):
            return 'pixel'
        if fs.exists(self.expected_filename(test_name, '.wav')):
            return 'audio'
        if self.reference_files(test_name):
            return 'ref'
        txt = self.expected_text(test_name)
        if txt:
            if 'layer at (0,0) size 800x600' in txt:
                return 'pixel'
            for line in txt.splitlines():
                if line.startswith('FAIL') or line.startswith('TIMEOUT') or line.startswith('PASS'):
                    return 'harness'
            return 'text'
        return 'unknown'

    def test_key(self, test_name):
        """Turns a test name into a pair of sublists: the natural sort key of the
        dirname, and the natural sort key of the basename.

        This can be used when sorting paths so that files in a directory.
        directory are kept together rather than being mixed in with files in
        subdirectories.
        """
        dirname, basename = self.split_test(test_name)
        return (
            self._natural_sort_key(dirname + self.TEST_PATH_SEPARATOR),
            self._natural_sort_key(basename)
        )

    def _natural_sort_key(self, string_to_split):
        """Turns a string into a list of string and number chunks, i.e. "z23a" -> ["z", 23, "a"]

        This can be used to implement "natural sort" order. See:
        http://www.codinghorror.com/blog/2007/12/sorting-for-humans-natural-sort-order.html
        http://nedbatchelder.com/blog/200712.html#e20071211T054956
        """
        def tryint(val):
            try:
                return int(val)
            except ValueError:
                return val

        return [tryint(chunk) for chunk in re.split(r'(\d+)', string_to_split)]

    def test_dirs(self):
        """Returns the list of top-level test directories."""
        layout_tests_dir = self.layout_tests_dir()
        fs = self._filesystem
        return [d for d in fs.listdir(layout_tests_dir) if fs.isdir(fs.join(layout_tests_dir, d))]

    @memoized
    def test_isfile(self, test_name):
        """Return True if the test name refers to a directory of tests."""
        # Used by test_expectations.py to apply rules to whole directories.
        if self._filesystem.isfile(self.abspath_for_test(test_name)):
            return True
        base = self.lookup_virtual_test_base(test_name)
        return base and self._filesystem.isfile(self.abspath_for_test(base))

    @memoized
    def test_isdir(self, test_name):
        """Return True if the test name refers to a directory of tests."""
        # Used by test_expectations.py to apply rules to whole directories.
        if self._filesystem.isdir(self.abspath_for_test(test_name)):
            return True
        base = self.lookup_virtual_test_base(test_name)
        return base and self._filesystem.isdir(self.abspath_for_test(base))

    @memoized
    def test_exists(self, test_name):
        """Return True if the test name refers to an existing test or baseline."""
        # Used by test_expectations.py to determine if an entry refers to a
        # valid test and by printing.py to determine if baselines exist.
        return self.test_isfile(test_name) or self.test_isdir(test_name)

    def split_test(self, test_name):
        """Splits a test name into the 'directory' part and the 'basename' part."""
        index = test_name.rfind(self.TEST_PATH_SEPARATOR)
        if index < 1:
            return ('', test_name)
        return (test_name[0:index], test_name[index:])

    def normalize_test_name(self, test_name):
        """Returns a normalized version of the test name or test directory."""
        if test_name.endswith('/'):
            return test_name
        if self.test_isdir(test_name):
            return test_name + '/'
        return test_name

    def driver_cmd_line(self):
        """Prints the DRT command line that will be used."""
        return self.create_driver(0).cmd_line(self.get_option('pixel_tests'), [])

    def update_baseline(self, baseline_path, data):
        """Updates the baseline for a test.

        Args:
            baseline_path: the actual path to use for baseline, not the path to
              the test. This function is used to update either generic or
              platform-specific baselines, but we can't infer which here.
            data: contents of the baseline.
        """
        self._filesystem.write_binary_file(baseline_path, data)

    # FIXME: update callers to create a finder and call it instead of these next five routines (which should be protected).
    def webkit_base(self):
        return self._webkit_finder.webkit_base()

    def path_from_webkit_base(self, *comps):
        return self._webkit_finder.path_from_webkit_base(*comps)

    def path_from_chromium_base(self, *comps):
        return self._webkit_finder.path_from_chromium_base(*comps)

    def path_to_script(self, script_name):
        return self._webkit_finder.path_to_script(script_name)

    def layout_tests_dir(self):
        custom_layout_tests_dir = self.get_option('layout_tests_directory')
        if custom_layout_tests_dir:
            return custom_layout_tests_dir
        return self._webkit_finder.layout_tests_dir()

    def perf_tests_dir(self):
        return self._webkit_finder.perf_tests_dir()

    def skipped_layout_tests(self, _):
        # TODO(qyearsley): Remove this method.
        return set()

    def skips_test(self, test, generic_expectations, full_expectations):
        """Checks whether the given test is skipped for this port.

        This should return True if the test is skipped because the port
        runs smoke tests only, or because the test is skipped in a file like
        NeverFixTests (but not TestExpectations).
        """
        fs = self.host.filesystem
        if self.default_smoke_test_only():
            smoke_test_filename = self.path_to_smoke_tests_file()
            if fs.exists(smoke_test_filename) and test not in fs.read_text_file(smoke_test_filename):
                return True

        # In general, Skip lines in the generic expectations file indicate
        # that the test is temporarily skipped, whereas if the test is skipped
        # in another file (e.g. WontFix in NeverFixTests), then the test may
        # always be skipped for this port.
        # TODO(qyearsley): Simplify this so that it doesn't rely on having
        # two copies of the test expectations.
        return (SKIP in full_expectations.get_expectations(test) and
                SKIP not in generic_expectations.get_expectations(test))

    def path_to_smoke_tests_file(self):
        return self.host.filesystem.join(self.layout_tests_dir(), 'SmokeTests')

    def _tests_from_skipped_file_contents(self, skipped_file_contents):
        tests_to_skip = []
        for line in skipped_file_contents.split('\n'):
            line = line.strip()
            line = line.rstrip('/')  # Best to normalize directory names to not include the trailing slash.
            if line.startswith('#') or not len(line):
                continue
            tests_to_skip.append(line)
        return tests_to_skip

    def _expectations_from_skipped_files(self, skipped_file_paths):
        tests_to_skip = []
        for search_path in skipped_file_paths:
            filename = self._filesystem.join(self._absolute_baseline_path(search_path), "Skipped")
            if not self._filesystem.exists(filename):
                _log.debug("Skipped does not exist: %s", filename)
                continue
            _log.debug("Using Skipped file: %s", filename)
            skipped_file_contents = self._filesystem.read_text_file(filename)
            tests_to_skip.extend(self._tests_from_skipped_file_contents(skipped_file_contents))
        return tests_to_skip

    @memoized
    def skipped_perf_tests(self):
        return self._expectations_from_skipped_files([self.perf_tests_dir()])

    def skips_perf_test(self, test_name):
        for test_or_category in self.skipped_perf_tests():
            if test_or_category == test_name:
                return True
            category = self._filesystem.join(self.perf_tests_dir(), test_or_category)
            if self._filesystem.isdir(category) and test_name.startswith(test_or_category):
                return True
        return False

    def name(self):
        """Returns a name that uniquely identifies this particular type of port
        (e.g., "mac-snowleopard" or "linux-trusty" and can be passed
        to factory.get() to instantiate the port.
        """
        return self._name

    def operating_system(self):
        # Subclasses should override this default implementation.
        return 'mac'

    def version(self):
        """Returns a string indicating the version of a given platform, e.g.
        'leopard' or 'win7'.

        This is used to help identify the exact port when parsing test
        expectations, determining search paths, and logging information.
        """
        return self._version

    def architecture(self):
        return self._architecture

    def get_option(self, name, default_value=None):
        return getattr(self._options, name, default_value)

    def set_option_default(self, name, default_value):
        return self._options.ensure_value(name, default_value)

    @memoized
    def path_to_generic_test_expectations_file(self):
        return self._filesystem.join(self.layout_tests_dir(), 'TestExpectations')

    def relative_test_filename(self, filename):
        """Returns a test_name a relative unix-style path for a filename under the LayoutTests
        directory. Ports may legitimately return abspaths here if no relpath makes sense.
        """
        # Ports that run on windows need to override this method to deal with
        # filenames with backslashes in them.
        if filename.startswith(self.layout_tests_dir()):
            return self.host.filesystem.relpath(filename, self.layout_tests_dir())
        else:
            return self.host.filesystem.abspath(filename)

    @memoized
    def abspath_for_test(self, test_name):
        """Returns the full path to the file for a given test name. This is the
        inverse of relative_test_filename().
        """
        return self._filesystem.join(self.layout_tests_dir(), test_name)

    def results_directory(self):
        """Absolute path to the place to store the test results (uses --results-directory)."""
        if not self._results_directory:
            option_val = self.get_option('results_directory') or self.default_results_directory()
            self._results_directory = self._filesystem.abspath(option_val)
        return self._results_directory

    def bot_test_times_path(self):
        return self._build_path('webkit_test_times', 'bot_times_ms.json')

    def perf_results_directory(self):
        return self._build_path()

    def inspector_build_directory(self):
        return self._build_path('resources', 'inspector')

    def apache_config_directory(self):
        return self.path_from_webkit_base('Tools', 'Scripts', 'apache_config')

    def default_results_directory(self):
        """Absolute path to the default place to store the test results."""
        return self._build_path('layout-test-results')

    def setup_test_run(self):
        """Perform port-specific work at the beginning of a test run."""
        # Delete the disk cache if any to ensure a clean test run.
        dump_render_tree_binary_path = self._path_to_driver()
        cachedir = self._filesystem.dirname(dump_render_tree_binary_path)
        cachedir = self._filesystem.join(cachedir, "cache")
        if self._filesystem.exists(cachedir):
            self._filesystem.rmtree(cachedir)

        if self._dump_reader:
            self._filesystem.maybe_make_directory(self._dump_reader.crash_dumps_directory())

    def num_workers(self, requested_num_workers):
        """Returns the number of available workers (possibly less than the number requested)."""
        return requested_num_workers

    def clean_up_test_run(self):
        """Perform port-specific work at the end of a test run."""
        if self._image_differ:
            self._image_differ.stop()
            self._image_differ = None

    def setup_environ_for_server(self):
        # We intentionally copy only a subset of the environment when
        # launching subprocesses to ensure consistent test results.
        clean_env = {}
        variables_to_copy = [
            'CHROME_DEVEL_SANDBOX',
            'CHROME_IPC_LOGGING',
            'ASAN_OPTIONS',
            'TSAN_OPTIONS',
            'MSAN_OPTIONS',
            'LSAN_OPTIONS',
            'UBSAN_OPTIONS',
            'VALGRIND_LIB',
            'VALGRIND_LIB_INNER',
        ]
        if self.host.platform.is_linux() or self.host.platform.is_freebsd():
            variables_to_copy += [
                'XAUTHORITY',
                'HOME',
                'LANG',
                'LD_LIBRARY_PATH',
                'DBUS_SESSION_BUS_ADDRESS',
                'XDG_DATA_DIRS',
            ]
            clean_env['DISPLAY'] = self.host.environ.get('DISPLAY', ':1')
        if self.host.platform.is_mac():
            clean_env['DYLD_LIBRARY_PATH'] = self._build_path()
            variables_to_copy += [
                'HOME',
            ]
        if self.host.platform.is_win():
            variables_to_copy += [
                'PATH',
                'GYP_DEFINES',  # Required to locate win sdk.
            ]
        if self.host.platform.is_cygwin():
            variables_to_copy += [
                'HOMEDRIVE',
                'HOMEPATH',
                '_NT_SYMBOL_PATH',
            ]

        for variable in variables_to_copy:
            if variable in self.host.environ:
                clean_env[variable] = self.host.environ[variable]

        for string_variable in self.get_option('additional_env_var', []):
            [name, value] = string_variable.split('=', 1)
            clean_env[name] = value

        return clean_env

    def show_results_html_file(self, results_filename):
        """This routine should display the HTML file pointed at by
        results_filename in a users' browser.
        """
        return self.host.user.open_url(abspath_to_uri(self.host.platform, results_filename))

    def create_driver(self, worker_number, no_timeout=False):
        """Return a newly created Driver subclass for starting/stopping the test driver."""
        return self._driver_class()(self, worker_number, pixel_tests=self.get_option('pixel_tests'), no_timeout=no_timeout)

    def requires_http_server(self):
        """Does the port require an HTTP server for running tests? This could
        be the case when the tests aren't run on the host platform.
        """
        return False

    def start_http_server(self, additional_dirs, number_of_drivers):
        """Start a web server. Raise an error if it can't start or is already running.

        Ports can stub this out if they don't need a web server to be running.
        """
        assert not self._http_server, 'Already running an http server.'

        server = apache_http.ApacheHTTP(self, self.results_directory(),
                                        additional_dirs=additional_dirs,
                                        number_of_servers=(number_of_drivers * 4))
        server.start()
        self._http_server = server

    def start_websocket_server(self):
        """Start a web server. Raise an error if it can't start or is already running.

        Ports can stub this out if they don't need a websocket server to be running.
        """
        assert not self._websocket_server, 'Already running a websocket server.'

        server = pywebsocket.PyWebSocket(self, self.results_directory())
        server.start()
        self._websocket_server = server

    def is_wptserve_enabled(self):
        """Used as feature flag for WPT Serve feature."""
        return self._is_wptserve_enabled

    @staticmethod
    def is_wptserve_test(test):
        """Whether wptserve should be used for a given test if enabled."""
        return test.startswith("external/wpt/")

    def should_use_wptserve(self, test):
        return self.is_wptserve_enabled() and self.is_wptserve_test(test)

    def start_wptserve(self):
        """Start a WPT web server. Raise an error if it can't start or is already running.

        Ports can stub this out if they don't need a WPT web server to be running.
        """
        assert not self._wpt_server, 'Already running an http server.'
        assert self.is_wptserve_enabled(), 'Cannot start server if WPT is not enabled.'

        # We currently don't support any output mechanism for the WPT server.
        server = wptserve.WPTServe(self, self.results_directory())
        server.start()
        self._wpt_server = server

    def stop_wptserve(self):
        """Shut down the WPT server if it is running. Do nothing if it isn't."""
        if self._wpt_server:
            self._wpt_server.stop()
            self._wpt_server = None

    def http_server_supports_ipv6(self):
        # Apache < 2.4 on win32 does not support IPv6, nor does cygwin apache.
        if self.host.platform.is_cygwin() or self.host.platform.is_win():
            return False
        return True

    def stop_http_server(self):
        """Shut down the http server if it is running. Do nothing if it isn't."""
        if self._http_server:
            self._http_server.stop()
            self._http_server = None

    def stop_websocket_server(self):
        """Shut down the websocket server if it is running. Do nothing if it isn't."""
        if self._websocket_server:
            self._websocket_server.stop()
            self._websocket_server = None

    #
    # TEST EXPECTATION-RELATED METHODS
    #

    def test_configuration(self):
        """Returns the current TestConfiguration for the port."""
        if not self._test_configuration:
            self._test_configuration = TestConfiguration(self._version, self._architecture, self._options.configuration.lower())
        return self._test_configuration

    # FIXME: Belongs on a Platform object.
    @memoized
    def all_test_configurations(self):
        """Returns a list of TestConfiguration instances, representing all available
        test configurations for this port.
        """
        return self._generate_all_test_configurations()

    # FIXME: Belongs on a Platform object.
    def configuration_specifier_macros(self):
        """Ports may provide a way to abbreviate configuration specifiers to conveniently
        refer to them as one term or alias specific values to more generic ones. For example:

        (vista, win7) -> win # Abbreviate all Windows versions into one namesake.
        (precise, trusty) -> linux  # Change specific name of Linux distro to a more generic term.

        Returns a dictionary, each key representing a macro term ('win', for example),
        and value being a list of valid configuration specifiers (such as ['vista', 'win7']).
        """
        return self.CONFIGURATION_SPECIFIER_MACROS

    def _generate_all_test_configurations(self):
        """Returns a sequence of the TestConfigurations the port supports."""
        # By default, we assume we want to test every graphics type in
        # every configuration on every system.
        test_configurations = []
        for version, architecture in self.ALL_SYSTEMS:
            for build_type in self.ALL_BUILD_TYPES:
                test_configurations.append(TestConfiguration(version, architecture, build_type))
        return test_configurations

    def warn_if_bug_missing_in_test_expectations(self):
        return True

    def _flag_specific_expectations_files(self):
        return [self._filesystem.join(self.layout_tests_dir(), 'FlagExpectations', flag.lstrip('-'))
                for flag in self.get_option('additional_driver_flag', [])]

    def _flag_specific_baseline_search_path(self):
        # TODO(skobes): Baselines specific to both flag and platform?
        return [self._filesystem.join(self.layout_tests_dir(), 'flag-specific', flag.lstrip('-'))
                for flag in self.get_option('additional_driver_flag', [])]

    def expectations_dict(self):
        """Returns an OrderedDict of name -> expectations strings.
        The names are expected to be (but not required to be) paths in the filesystem.
        If the name is a path, the file can be considered updatable for things like rebaselining,
        so don't use names that are paths if they're not paths.
        Generally speaking the ordering should be files in the filesystem in cascade order
        (TestExpectations followed by Skipped, if the port honors both formats),
        then any built-in expectations (e.g., from compile-time exclusions), then --additional-expectations options.
        """
        # FIXME: rename this to test_expectations() once all the callers are updated to know about the ordered dict.
        expectations = collections.OrderedDict()

        for path in self.expectations_files():
            if self._filesystem.exists(path):
                expectations[path] = self._filesystem.read_text_file(path)

        for path in self.get_option('additional_expectations', []):
            expanded_path = self._filesystem.expanduser(path)
            if self._filesystem.exists(expanded_path):
                _log.debug("reading additional_expectations from path '%s'", path)
                expectations[path] = self._filesystem.read_text_file(expanded_path)
            else:
                _log.warning("additional_expectations path '%s' does not exist", path)
        return expectations

    def all_expectations_dict(self):
        """Returns an OrderedDict of name -> expectations strings (see:
        |expectations_dict|), including all flag-specific expectation files."""
        expectations = self.expectations_dict()

        flag_path = self._filesystem.join(self.layout_tests_dir(), 'FlagExpectations')
        if not self._filesystem.exists(flag_path):
            return expectations

        for (_, _, filenames) in self._filesystem.walk(flag_path):
            if 'README.txt' in filenames:
                filenames.remove('README.txt')
            for filename in filenames:
                path = self._filesystem.join(flag_path, filename)
                expectations[path] = self._filesystem.read_text_file(path)

        return expectations

    def bot_expectations(self):
        if not self.get_option('ignore_flaky_tests'):
            return {}

        full_port_name = self.determine_full_port_name(self.host, self._options, self.port_name)
        builder_category = self.get_option('ignore_builder_category', 'layout')
        factory = BotTestExpectationsFactory(self.host.builders)
        # FIXME: This only grabs release builder's flakiness data. If we're running debug,
        # when we should grab the debug builder's data.
        expectations = factory.expectations_for_port(full_port_name, builder_category)

        if not expectations:
            return {}

        ignore_mode = self.get_option('ignore_flaky_tests')
        if ignore_mode == 'very-flaky' or ignore_mode == 'maybe-flaky':
            return expectations.flakes_by_path(ignore_mode == 'very-flaky')
        if ignore_mode == 'unexpected':
            return expectations.unexpected_results_by_path()
        _log.warning("Unexpected ignore mode: '%s'.", ignore_mode)
        return {}

    def expectations_files(self):
        paths = [
            self.path_to_generic_test_expectations_file(),
            self._filesystem.join(self.layout_tests_dir(), 'NeverFixTests'),
            self._filesystem.join(self.layout_tests_dir(), 'StaleTestExpectations'),
            self._filesystem.join(self.layout_tests_dir(), 'SlowTests'),
        ]
        if self.is_wptserve_enabled():
            paths.append(self._filesystem.join(self.layout_tests_dir(), 'WPTServeExpectations'))
        paths.extend(self._flag_specific_expectations_files())
        return paths

    def repository_path(self):
        """Returns the repository path for the chromium code base."""
        return self.path_from_chromium_base('build')

    # This is a class variable so we can test error output easily.
    _pretty_patch_error_html = "Failed to run PrettyPatch, see error log."

    def pretty_patch_text(self, diff_path):
        if self._pretty_patch_available is None:
            self._pretty_patch_available = self.check_pretty_patch(more_logging=False)
        if not self._pretty_patch_available:
            return self._pretty_patch_error_html
        command = ("ruby", "-I", self._filesystem.dirname(self._pretty_patch_path),
                   self._pretty_patch_path, diff_path)
        try:
            # Diffs are treated as binary (we pass decode_output=False) as they
            # may contain multiple files of conflicting encodings.
            return self._executive.run_command(command, decode_output=False)
        except OSError as error:
            # If the system is missing ruby log the error and stop trying.
            self._pretty_patch_available = False
            _log.error("Failed to run PrettyPatch (%s): %s", command, error)
            return self._pretty_patch_error_html
        except ScriptError as error:
            # If ruby failed to run for some reason, log the command
            # output and stop trying.
            self._pretty_patch_available = False
            _log.error("Failed to run PrettyPatch (%s):\n%s", command, error.message_with_output())
            return self._pretty_patch_error_html

    def default_configuration(self):
        return 'Release'

    def clobber_old_port_specific_results(self):
        pass

    # FIXME: This does not belong on the port object.
    @memoized
    def path_to_apache(self):
        """Returns the full path to the apache binary.

        This is needed only by ports that use the apache_http_server module.
        """
        raise NotImplementedError('Port.path_to_apache')

    def path_to_apache_config_file(self):
        """Returns the full path to the apache configuration file.

        If the WEBKIT_HTTP_SERVER_CONF_PATH environment variable is set, its
        contents will be used instead.

        This is needed only by ports that use the apache_http_server module.
        """
        config_file_from_env = self.host.environ.get('WEBKIT_HTTP_SERVER_CONF_PATH')
        if config_file_from_env:
            if not self._filesystem.exists(config_file_from_env):
                raise IOError('%s was not found on the system' % config_file_from_env)
            return config_file_from_env

        config_file_name = self._apache_config_file_name_for_platform()
        return self._filesystem.join(self.apache_config_directory(), config_file_name)

    #
    # PROTECTED ROUTINES
    #
    # The routines below should only be called by routines in this class
    # or any of its subclasses.
    #

    def _apache_version(self):
        config = self._executive.run_command([self.path_to_apache(), '-v'])
        return re.sub(r'(?:.|\n)*Server version: Apache/(\d+\.\d+)(?:.|\n)*', r'\1', config)

    def _apache_config_file_name_for_platform(self):
        if self.host.platform.is_cygwin():
            return 'cygwin-httpd.conf'  # CYGWIN is the only platform to still use Apache 1.3.
        if self.host.platform.is_linux():
            distribution = self.host.platform.linux_distribution()

            custom_configuration_distributions = ['arch', 'debian', 'redhat']
            if distribution in custom_configuration_distributions:
                return "%s-httpd-%s.conf" % (distribution, self._apache_version())

        return 'apache2-httpd-' + self._apache_version() + '.conf'

    def _path_to_driver(self, target=None):
        """Returns the full path to the test driver."""
        return self._build_path(target, self.driver_name())

    def _path_to_image_diff(self):
        """Returns the full path to the image_diff binary, or None if it is not available.

        This is likely used only by diff_image()
        """
        return self._build_path('image_diff')

    def _absolute_baseline_path(self, platform_dir):
        """Return the absolute path to the top of the baseline tree for a
        given platform directory.
        """
        return self._filesystem.join(self.layout_tests_dir(), 'platform', platform_dir)

    def _driver_class(self):
        """Returns the port's driver implementation."""
        return driver.Driver

    def output_contains_sanitizer_messages(self, output):
        if not output:
            return None
        if 'AddressSanitizer' in output:
            return 'AddressSanitizer'
        if 'MemorySanitizer' in output:
            return 'MemorySanitizer'
        return None

    def _get_crash_log(self, name, pid, stdout, stderr, newer_than):
        if self.output_contains_sanitizer_messages(stderr):
            # Running the symbolizer script can take a lot of memory, so we need to
            # serialize access to it across all the concurrently running drivers.

            llvm_symbolizer_path = self.path_from_chromium_base(
                'third_party', 'llvm-build', 'Release+Asserts', 'bin', 'llvm-symbolizer')
            if self._filesystem.exists(llvm_symbolizer_path):
                env = self.host.environ.copy()
                env['LLVM_SYMBOLIZER_PATH'] = llvm_symbolizer_path
            else:
                env = None
            sanitizer_filter_path = self.path_from_chromium_base('tools', 'valgrind', 'asan', 'asan_symbolize.py')
            sanitizer_strip_path_prefix = 'Release/../../'
            if self._filesystem.exists(sanitizer_filter_path):
                stderr = self._executive.run_command(
                    ['flock', sys.executable, sanitizer_filter_path, sanitizer_strip_path_prefix],
                    input=stderr, decode_output=False, env=env)

        name_str = name or '<unknown process name>'
        pid_str = str(pid or '<unknown>')

        # We require stdout and stderr to be bytestrings, not character strings.
        if stdout:
            assert isinstance(stdout, basestring)
            stdout_lines = stdout.decode('utf8', 'replace').splitlines()
        else:
            stdout_lines = [u'<empty>']
        if stderr:
            assert isinstance(stderr, basestring)
            stderr_lines = stderr.decode('utf8', 'replace').splitlines()
        else:
            stderr_lines = [u'<empty>']

        return (stderr, 'crash log for %s (pid %s):\n%s\n%s\n' % (name_str, pid_str,
                                                                  '\n'.join(('STDOUT: ' + l) for l in stdout_lines),
                                                                  '\n'.join(('STDERR: ' + l) for l in stderr_lines)))

    def look_for_new_crash_logs(self, crashed_processes, start_time):
        pass

    def look_for_new_samples(self, unresponsive_processes, start_time):
        pass

    def sample_process(self, name, pid):
        pass

    def physical_test_suites(self):
        return [
            # For example, to turn on force-compositing-mode in the svg/ directory:
            # PhysicalTestSuite('svg', ['--force-compositing-mode']),
            PhysicalTestSuite('fast/text', ["--enable-direct-write", "--enable-font-antialiasing"]),
        ]

    def virtual_test_suites(self):
        if self._virtual_test_suites is None:
            path_to_virtual_test_suites = self._filesystem.join(self.layout_tests_dir(), 'VirtualTestSuites')
            assert self._filesystem.exists(path_to_virtual_test_suites), 'LayoutTests/VirtualTestSuites not found'
            try:
                test_suite_json = json.loads(self._filesystem.read_text_file(path_to_virtual_test_suites))
                self._virtual_test_suites = [VirtualTestSuite(**d) for d in test_suite_json]
            except ValueError as error:
                raise ValueError("LayoutTests/VirtualTestSuites is not a valid JSON file: %s" % error)
        return self._virtual_test_suites

    def _all_virtual_tests(self, suites):
        tests = []
        for suite in suites:
            self._populate_virtual_suite(suite)
            tests.extend(suite.tests.keys())
        return tests

    def _virtual_tests_matching_paths(self, paths, suites):
        tests = []
        for suite in suites:
            if any(p.startswith(suite.name) for p in paths):
                self._populate_virtual_suite(suite)
            for test in suite.tests:
                if any(test.startswith(p) for p in paths):
                    tests.append(test)
        return tests

    def _populate_virtual_suite(self, suite):
        if not suite.tests:
            base_tests = self.real_tests([suite.base])
            suite.tests = {}
            for test in base_tests:
                suite.tests[test.replace(suite.base, suite.name, 1)] = test

    def is_virtual_test(self, test_name):
        return bool(self.lookup_virtual_suite(test_name))

    def lookup_virtual_suite(self, test_name):
        for suite in self.virtual_test_suites():
            if test_name.startswith(suite.name):
                return suite
        return None

    def lookup_virtual_test_base(self, test_name):
        suite = self.lookup_virtual_suite(test_name)
        if not suite:
            return None
        return test_name.replace(suite.name, suite.base, 1)

    def lookup_virtual_test_args(self, test_name):
        for suite in self.virtual_test_suites():
            if test_name.startswith(suite.name):
                return suite.args
        return []

    def lookup_virtual_reference_args(self, test_name):
        for suite in self.virtual_test_suites():
            if test_name.startswith(suite.name):
                return suite.reference_args
        return []

    def lookup_physical_test_args(self, test_name):
        for suite in self.physical_test_suites():
            if test_name.startswith(suite.name):
                return suite.args
        return []

    def lookup_physical_reference_args(self, test_name):
        for suite in self.physical_test_suites():
            if test_name.startswith(suite.name):
                return suite.reference_args
        return []

    def should_run_as_pixel_test(self, test_input):
        if not self._options.pixel_tests:
            return False
        if self._options.pixel_test_directories:
            return any(test_input.test_name.startswith(directory) for directory in self._options.pixel_test_directories)
        # TODO(burnik): Make sure this is the right way to do it.
        if self.should_use_wptserve(test_input.test_name):
            return False
        return True

    def _convert_path(self, path):
        """Handles filename conversion for subprocess command line args."""
        # See note above in diff_image() for why we need this.
        if sys.platform == 'cygwin':
            return cygpath(path)
        return path

    def _build_path(self, *comps):
        return self._build_path_with_target(self._options.target, *comps)

    def _build_path_with_target(self, target, *comps):
        # Note that we don't do the option caching that the
        # base class does, because finding the right directory is relatively
        # fast.
        target = target or self.get_option('target')
        return self._static_build_path(self._filesystem, self.get_option('build_directory'),
                                       self.path_from_chromium_base(), target, comps)

    def _check_driver_build_up_to_date(self, target):
        # We should probably get rid of this check altogether as it has
        # outlived its usefulness in a GN-based world, but for the moment
        # we will just check things if they are using the standard
        # Debug or Release target directories.
        if target not in ('Debug', 'Release'):
            return True

        try:
            debug_path = self._path_to_driver('Debug')
            release_path = self._path_to_driver('Release')

            debug_mtime = self._filesystem.mtime(debug_path)
            release_mtime = self._filesystem.mtime(release_path)

            if (debug_mtime > release_mtime and target == 'Release' or
                    release_mtime > debug_mtime and target == 'Debug'):
                most_recent_binary = 'Release' if target == 'Debug' else 'Debug'
                _log.warning('You are running the %s binary. However the %s binary appears to be more recent. '
                             'Please pass --%s.', target, most_recent_binary, most_recent_binary.lower())
                _log.warning('')
        # This will fail if we don't have both a debug and release binary.
        # That's fine because, in this case, we must already be running the
        # most up-to-date one.
        except OSError:
            pass
        return True

    def _chromium_baseline_path(self, platform):
        if platform is None:
            platform = self.name()
        return self.path_from_webkit_base('LayoutTests', 'platform', platform)

class VirtualTestSuite(object):

    def __init__(self, prefix=None, base=None, args=None, references_use_default_args=False):
        assert base
        assert args
        assert '/' not in prefix, "Virtual test suites prefixes cannot contain /'s: %s" % prefix
        self.name = 'virtual/' + prefix + '/' + base
        self.base = base
        self.args = args
        self.reference_args = [] if references_use_default_args else args
        self.tests = {}

    def __repr__(self):
        return "VirtualTestSuite('%s', '%s', %s, %s)" % (self.name, self.base, self.args, self.reference_args)

class PhysicalTestSuite(object):

    def __init__(self, base, args, reference_args=None):
        self.name = base
        self.base = base
        self.args = args
        self.reference_args = args if reference_args is None else reference_args
        self.tests = set()

    def __repr__(self):
        return "PhysicalTestSuite('%s', '%s', %s, %s)" % (self.name, self.base, self.args, self.reference_args)


text:
/* ==========================================================
File:        LineStats.java
Description: Stores total lines in file and current cursor position.
Maintainer:  WakaTime <support@wakatime.com>
License:     BSD, see LICENSE for more details.
Website:     https://wakatime.com/
===========================================================*/

package com.wakatime.intellij.plugin;

public class LineStats {
    public Integer lineCount;
    public Integer lineNumber;
    public Integer cursorPosition;
}

redpajama_set_name: RedPajamaGithub


text:
require_relative '../../spec_helper'

describe '/v1/volumes' do

  let(:request_headers) do
    {
        'HTTP_AUTHORIZATION' => "Bearer #{valid_token.token_plain}"
    }
  end

  let! :grid do
    grid = Grid.create!(name: 'terminal-a')
  end

  let(:david) do
    user = User.create!(email: 'david@domain.com', external_id: '123456')
    grid.users << user

    user
  end

  let(:john) do
    User.create!(email: 'david@domain.com', external_id: '123456')
  end

  let(:johns_token) do
    AccessToken.create!(user: john, scopes: ['user'])
  end

  let(:valid_token) do
    AccessToken.create!(user: david, scopes: ['user'])
  end

  let! :stack do
    grid.stacks.create!(
      name: 'teststack',
    )
  end

  let! :volume do
    Volume.create!(
      grid: grid,
      name: 'external-vol',
      driver: 'some-driver',
      scope: 'node'
    )
  end

  let! :other_volume do
    Volume.create!(
      grid: grid,
      name: 'other-volume',
      driver: 'some-driver',
      scope: 'node'
    )
  end

  let! :node do
    grid.create_node!('node-1')
  end

  describe 'GET /' do
    it 'returns all volumes' do
      get "/v1/volumes/#{grid.name}", nil, request_headers
      expect(response.status).to eq(200)
      expect(json_response['volumes'].size).to eq(2)
    end
  end

  describe 'GET /:id' do
    it 'returns volume json' do
      volume.volume_instances.create!(name: 'foo', host_node: node)
      redis = GridServices::Create.run(
          grid: grid,
          image: 'redis:2.8',
          name: 'redis',
          stateful: false,
          volumes: [
            "#{volume.name}:/data:ro"
          ]
      ).result

      get "/v1/volumes/#{grid.name}/#{volume.name}", nil, request_headers
      expect(response.status).to eq(200)
      expect(json_response.keys.sort).to eq(%w(
        id created_at updated_at driver driver_opts name scope instances services
      ).sort)
      expect(json_response['id']).to eq("#{grid.name}/#{volume.name}")
      expect(json_response['driver']).to eq(volume.driver)
      expect(json_response['scope']).to eq(volume.scope)
      expect(json_response['instances'].first['node']).to eq(node.name)
      expect(json_response['instances'].first['name']).to eq('foo')
      expect(json_response['services']).to eq([{'id' => redis.to_path}])
    end

    it 'return 404 for non existing volume' do
      get "/v1/volumes/#{grid.name}/foo", nil, request_headers
      expect(response.status).to eq(404)
    end

    it 'return 404 for non existing grid' do
      get "/v1/volumes/foo/#{volume.name}", nil, request_headers
      expect(response.status).to eq(404)
    end

    it 'requires auth' do
      get "/v1/volumes/#{grid.name}/#{volume.name}", nil
      expect(response.status).to eq(403)
    end

    it 'returns 403 without grid access' do
      # John is not a user in this grid
      get "/v1/volumes/#{grid.name}/#{volume.name}", nil, { 'HTTP_AUTHORIZATION' => "Bearer #{johns_token.token_plain}"}
      expect(response.status).to eq(403)
    end
  end

  describe 'POST' do
    it 'creates new volume' do
      data = {
        name: 'foo',
        scope: 'instance',
        driver: 'local',
        driver_opts: {
          foo: 'bar'
        }
      }
      expect {
        post "/v1/volumes/#{grid.name}", data.to_json, request_headers
        expect(response.status).to eq(201)
      }.to change{ Volume.count }.by(1)
      expect(Volume.find_by(name: 'foo').driver_opts['foo']).to eq('bar')
    end
  end

  describe 'DELETE' do
    it 'deletes volume' do
      expect {
        delete "/v1/volumes/#{grid.name}/#{volume.name}", nil, request_headers
        expect(response.status).to eq(200)
      }.to change{Volume.count}. by (-1)

    end

    it 'returns 422 when volume still used by some service' do
      outcome = GridServices::Create.run(
          grid: grid,
          image: 'redis:2.8',
          name: 'redis',
          stateful: false,
          volumes: [
            "#{volume.name}:/data:ro"
          ]
      )

      expect {
        delete "/v1/volumes/#{grid.name}/#{volume.name}", nil, request_headers
        expect(response.status).to eq(422)
      }.not_to change{Volume.count}
    end
  end
end


text:
package org.endeavourhealth.enterprise.core.database.models;

import javax.persistence.Column;
import javax.persistence.Id;
import java.io.Serializable;
import java.util.Objects;

public class EmisCsvCodeMapEntityPK implements Serializable {
    private byte medication;
    private long codeId;

    @Column(name = "medication")
    @Id
    public byte getMedication() {
        return medication;
    }

    public void setMedication(byte medication) {
        this.medication = medication;
    }

    @Column(name = "code_id")
    @Id
    public long getCodeId() {
        return codeId;
    }

    public void setCodeId(long codeId) {
        this.codeId = codeId;
    }

    @Override
    public boolean equals(Object o) {
        if (this == o) return true;
        if (o == null || getClass() != o.getClass()) return false;
        EmisCsvCodeMapEntityPK that = (EmisCsvCodeMapEntityPK) o;
        return medication == that.medication &&
                codeId == that.codeId;
    }

    @Override
    public int hashCode() {

        return Objects.hash(medication, codeId);
    }
}


text:
package freegeoip

import (
	"encoding/csv"
	"encoding/json"
	"encoding/xml"
	"io"
	"math"
	"net"
	"net/http"
	"strconv"
	"strings"

	// otto is used for testing the JSONP encoder. It's imported here
	// to make `go get` download it before `go test` fails.
	_ "github.com/robertkrimen/otto"
)

// A Query object is used to query the IP database.
//
// Currently the only database supported is MaxMind, and the query is a
// data structure with tags that are used by the maxminddb.Lookup function.
type Query interface{}

// An Encoder that can provide a query specification to be used for
// querying the IP database, and later encode the results of that
// query in a specific format.
type Encoder interface {
	// NewQuery returns a query specification that is used to query
	// the IP database. It should be a data structure with tags
	// associated to its fields describing what fields to query in
	// the IP database, such as country and city.
	//
	// See the maxminddb package documentation for details on
	// fields available for the MaxMind database.
	NewQuery() Query

	// Encode writes data to the response of an http request
	// using the results of a query to the IP database.
	//
	// It encodes the query object into a specific format such
	// as XML or JSON and writes to the response.
	//
	// The IP passed to the encoder may be the result of a DNS
	// lookup, and if there are multiple IPs associated to the
	// hostname this will be a random one from the list.
	Encode(w http.ResponseWriter, r *http.Request, q Query, ip net.IP) error
}

// JSONEncoder encodes the results of an IP lookup as JSON.
type JSONEncoder struct {
	Indent bool
}

// NewQuery implements the Encoder interface.
func (f *JSONEncoder) NewQuery() Query {
	return &maxmindQuery{}
}

// Encode implements the Encoder interface.
func (f *JSONEncoder) Encode(w http.ResponseWriter, r *http.Request, q Query, ip net.IP) error {
	record := newResponse(q.(*maxmindQuery), ip, requestLang(r))
	callback := r.FormValue("callback")
	if len(callback) > 0 {
		return f.P(w, r, record, callback)
	}
	w.Header().Set("Content-Type", "application/json")
	if f.Indent {
	}
	return json.NewEncoder(w).Encode(record)
}

// P writes JSONP to an http response.
func (f *JSONEncoder) P(w http.ResponseWriter, r *http.Request, record *responseRecord, callback string) error {
	w.Header().Set("Content-Type", "application/javascript")
	_, err := io.WriteString(w, callback+"(")
	if err != nil {
		return err
	}
	err = json.NewEncoder(w).Encode(record)
	if err != nil {
		return err
	}
	_, err = io.WriteString(w, ");")
	return err
}

// XMLEncoder encodes the results of an IP lookup as XML.
type XMLEncoder struct {
	Indent bool
}

// NewQuery implements the Encoder interface.
func (f *XMLEncoder) NewQuery() Query {
	return &maxmindQuery{}
}

// Encode implements the Encoder interface.
func (f *XMLEncoder) Encode(w http.ResponseWriter, r *http.Request, q Query, ip net.IP) error {
	record := newResponse(q.(*maxmindQuery), ip, requestLang(r))
	w.Header().Set("Content-Type", "application/xml")
	_, err := io.WriteString(w, xml.Header)
	if err != nil {
		return err
	}
	if f.Indent {
		enc := xml.NewEncoder(w)
		enc.Indent("", "\t")
		err := enc.Encode(record)
		if err != nil {
			return err
		}
		_, err = w.Write([]byte("\n"))
		return err
	}
	return xml.NewEncoder(w).Encode(record)
}

// CSVEncoder encodes the results of an IP lookup as CSV.
type CSVEncoder struct {
	UseCRLF bool
}

// NewQuery implements the Encoder interface.
func (f *CSVEncoder) NewQuery() Query {
	return &maxmindQuery{}
}

// Encode implements the Encoder interface.
func (f *CSVEncoder) Encode(w http.ResponseWriter, r *http.Request, q Query, ip net.IP) error {
	record := newResponse(q.(*maxmindQuery), ip, requestLang(r))
	w.Header().Set("Content-Type", "text/csv")
	cw := csv.NewWriter(w)
	cw.UseCRLF = f.UseCRLF
	err := cw.Write([]string{
		ip.String(),
		record.CountryCode,
		record.CountryName,
		record.RegionCode,
		record.RegionName,
		record.City,
		record.ZipCode,
		record.TimeZone,
		strconv.FormatFloat(record.Latitude, 'f', 2, 64),
		strconv.FormatFloat(record.Longitude, 'f', 2, 64),
		strconv.Itoa(int(record.MetroCode)),
	})
	if err != nil {
		return err
	}
	cw.Flush()
	return nil
}

// maxmindQuery is the object used to query the maxmind database.
//
// See the maxminddb package documentation for details.
type maxmindQuery struct {
	Country struct {
		ISOCode string            `maxminddb:"iso_code"`
		Names   map[string]string `maxminddb:"names"`
	} `maxminddb:"country"`
	Region []struct {
		ISOCode string            `maxminddb:"iso_code"`
		Names   map[string]string `maxminddb:"names"`
	} `maxminddb:"subdivisions"`
	City struct {
		Names map[string]string `maxminddb:"names"`
	} `maxminddb:"city"`
	Location struct {
		Latitude  float64 `maxminddb:"latitude"`
		Longitude float64 `maxminddb:"longitude"`
		MetroCode uint    `maxminddb:"metro_code"`
		TimeZone  string  `maxminddb:"time_zone"`
	} `maxminddb:"location"`
	Postal struct {
		Code string `maxminddb:"code"`
	} `maxminddb:"postal"`
}

// responseRecord is the object that gets encoded as the response of an
// IP lookup request. It is encoded to formats such as xml and json.
type responseRecord struct {
	XMLName     xml.Name `xml:"Response" json:"-"`
	IP          string   `json:"ip"`
	CountryCode string   `json:"country_code"`
	CountryName string   `json:"country_name"`
	RegionCode  string   `json:"region_code"`
	RegionName  string   `json:"region_name"`
	City        string   `json:"city"`
	ZipCode     string   `json:"zip_code"`
	TimeZone    string   `json:"time_zone"`
	Latitude    float64  `json:"latitude"`
	Longitude   float64  `json:"longitude"`
	MetroCode   uint     `json:"metro_code"`
}

// newResponse translates a maxmindQuery into a responseRecord, setting
// the country, region and city names to their localized name according
// to the given lang.
//
// See the maxminddb documentation for supported languages.
func newResponse(query *maxmindQuery, ip net.IP, lang []string) *responseRecord {

	record := &responseRecord{
		IP:          ip.String(),
		CountryCode: query.Country.ISOCode,
		CountryName: localizedName(query.Country.Names, lang),
		City:        localizedName(query.City.Names, lang),
		ZipCode:     query.Postal.Code,
		TimeZone:    query.Location.TimeZone,
		Latitude:    roundFloat(query.Location.Latitude, .5, 4),
		Longitude:   roundFloat(query.Location.Longitude, .5, 4),
		MetroCode:   query.Location.MetroCode,
	}
	if len(query.Region) > 0 {
		record.RegionCode = query.Region[0].ISOCode
		record.RegionName = localizedName(query.Region[0].Names, lang)
	}
	return record
}

func requestLang(r *http.Request) (list []string) {
	// TODO: Check Accept-Charset, sort languages by qvalue.
	l := r.Header.Get("Accept-Language")
	if len(l) == 0 {
		return nil
	}
	accpt := strings.Split(l, ",")
	if len(accpt) == 0 {
		return nil
	}
	for n, name := range accpt {
		accpt[n] = strings.Trim(strings.SplitN(name, ";", 2)[0], " ")
	}
	return accpt
}

func localizedName(field map[string]string, accept []string) (name string) {
	if accept != nil {
		var f string
		var ok bool
		for _, l := range accept {
			f, ok = field[l]
			if ok {
				return f
			}
		}
	}
	return field["en"]
}

func roundFloat(val float64, roundOn float64, places int) (newVal float64) {
	var round float64
	pow := math.Pow(10, float64(places))
	digit := pow * val
	_, div := math.Modf(digit)
	if div >= roundOn {
		round = math.Ceil(digit)
	} else {
		round = math.Floor(digit)
	}
	return round / pow
}


text:
require 'helper'

class TestPaginationHelpers < Test::Unit::TestCase
  should "be autoloadable" do
    Pagination::Helpers
  end
end