# Copyright 2021 The Emscripten Authors. All rights reserved.

# Emscripten is available under two separate licenses, the MIT license and the

# University of Illinois/NCSA Open Source License. Both these licenses can be

# found in the LICENSE file.

from enum import Enum

from functools import wraps

from pathlib import Path

from subprocess import PIPE, STDOUT

from typing import Dict, Tuple

from urllib.parse import unquote, unquote_plus

from http.server import HTTPServer, SimpleHTTPRequestHandler

import contextlib

import difflib

import hashlib

import logging

import multiprocessing

import os

import re

import shlex

import shutil

import stat

import string

import subprocess

import sys

import tempfile

import time

import webbrowser

import unittest

import clang_native

import jsrun

from tools.shared import EMCC, EMXX, DEBUG, EMCONFIGURE, EMCMAKE

from tools.shared import get_canonical_temp_dir, path_from_root

from tools.utils import MACOS, WINDOWS, read_file, read_binary, write_file, write_binary, exit_with_error

from tools import shared, line_endings, building, config, utils

logger = logging.getLogger('common')

# User can specify an environment variable EMTEST_BROWSER to force the browser

# test suite to run using another browser command line than the default system

# browser.

# There are two special value that can be used here if running in an actual

# browser is not desired:

# EMTEST_BROWSER=0 : This will disable the actual running of the test and simply

# verify that it compiles and links.

# EMTEST_BROWSER=node : This will attempt to run the browser test under node.

# For most browser tests this does not work, but it can

# be useful for running pthread tests under node.

EMTEST_BROWSER = None

EMTEST_DETECT_TEMPFILE_LEAKS = None

EMTEST_SAVE_DIR = None

# generally js engines are equivalent, testing 1 is enough. set this

# to force testing on all js engines, good to find js engine bugs

EMTEST_ALL_ENGINES = None

EMTEST_SKIP_SLOW = None

EMTEST_LACKS_NATIVE_CLANG = None

EMTEST_VERBOSE = None

EMTEST_REBASELINE = None

EMTEST_FORCE64 = None

# Verbosity level control for subprocess calls to configure + make.

# 0: disabled.

# 1: Log stderr of configure/make.

# 2: Log stdout and stderr configure/make. Print out subprocess commands that were executed.

# 3: Log stdout and stderr, and pass VERBOSE=1 to CMake/configure/make steps.

EMTEST_BUILD_VERBOSE = int(os.getenv('EMTEST_BUILD_VERBOSE', '0'))

if 'EM_BUILD_VERBOSE' in os.environ:

exit_with_error('EM_BUILD_VERBOSE has been renamed to EMTEST_BUILD_VERBOSE')

# Special value for passing to assert_returncode which means we expect that program

# to fail with non-zero return code, but we don't care about specifically which one.

NON_ZERO = -1

TEST_ROOT = path_from_root('test')

WEBIDL_BINDER = shared.bat_suffix(path_from_root('tools/webidl_binder'))

EMBUILDER = shared.bat_suffix(path_from_root('embuilder'))

EMMAKE = shared.bat_suffix(path_from_root('emmake'))

EMRUN = shared.bat_suffix(shared.path_from_root('emrun'))

WASM_DIS = Path(building.get_binaryen_bin(), 'wasm-dis')

LLVM_OBJDUMP = os.path.expanduser(shared.build_llvm_tool_path(shared.exe_suffix('llvm-objdump')))

PYTHON = sys.executable

def test_file(*path_components):

"""Construct a path relative to the emscripten "tests" directory."""

return str(Path(TEST_ROOT, *path_components))

# checks if browser testing is enabled

def has_browser():

return EMTEST_BROWSER != '0'

def compiler_for(filename, force_c=False):

if shared.suffix(filename) in ('.cc', '.cxx', '.cpp') and not force_c:

return EMXX

else:

return EMCC

# Generic decorator that calls a function named 'condition' on the test class and

# skips the test if that function returns true

def skip_if(func, condition, explanation='', negate=False):

assert callable(func)

explanation_str = ' : %s' % explanation if explanation else ''

@wraps(func)

def decorated(self, *args, **kwargs):

choice = self.__getattribute__(condition)()

if negate:

choice = not choice

if choice:

self.skipTest(condition + explanation_str)

func(self, *args, **kwargs)

return decorated

def needs_dylink(func):

assert callable(func)

@wraps(func)

def decorated(self, *args, **kwargs):

self.check_dylink()

return func(self, *args, **kwargs)

return decorated

def is_slow_test(func):

assert callable(func)

@wraps(func)

def decorated(self, *args, **kwargs):

if EMTEST_SKIP_SLOW:

return self.skipTest('skipping slow tests')

return func(self, *args, **kwargs)

return decorated

def disabled(note=''):

assert not callable(note)

return unittest.skip(note)

def no_mac(note=''):

assert not callable(note)

if MACOS:

return unittest.skip(note)

return lambda f: f

def no_windows(note=''):

assert not callable(note)

if WINDOWS:

return unittest.skip(note)

return lambda f: f

def only_windows(note=''):

assert not callable(note)

if not WINDOWS:

return unittest.skip(note)

return lambda f: f

def requires_native_clang(func):

assert callable(func)

@wraps(func)

def decorated(self, *args, **kwargs):

if EMTEST_LACKS_NATIVE_CLANG:

return self.skipTest('native clang tests are disabled')

return func(self, *args, **kwargs)

return decorated

def requires_node(func):

assert callable(func)

def decorated(self, *args, **kwargs):

self.require_node()

return func(self, *args, **kwargs)

return decorated

def requires_wasm64(func):

assert callable(func)

def decorated(self, *args, **kwargs):

self.require_wasm64()

return func(self, *args, **kwargs)

return decorated

def requires_wasm_eh(func):

assert callable(func)

def decorated(self, *args, **kwargs):

self.require_wasm_eh()

return func(self, *args, **kwargs)

return decorated

def requires_v8(func):

assert callable(func)

def decorated(self, *args, **kwargs):

self.require_v8()

return func(self, *args, **kwargs)

return decorated

def node_pthreads(f):

@wraps(f)

def decorated(self, *args, **kwargs):

self.setup_node_pthreads()

f(self, *args, **kwargs)

return decorated

def crossplatform(f):

f.is_crossplatform_test = True

return f

@contextlib.contextmanager

def env_modify(updates):

"""A context manager that updates os.environ."""

# This could also be done with mock.patch.dict() but taking a dependency

# on the mock library is probably not worth the benefit.

old_env = os.environ.copy()

print("env_modify: " + str(updates))

# Seting a value to None means clear the environment variable

clears = [key for key, value in updates.items() if value is None]

updates = {key: value for key, value in updates.items() if value is not None}

os.environ.update(updates)

for key in clears:

if key in os.environ:

del os.environ[key]

try:

yield

finally:

os.environ.clear()

os.environ.update(old_env)

# Decorator version of env_modify

def with_env_modify(updates):

assert not callable(updates)

def decorated(f):

def modified(self, *args, **kwargs):

with env_modify(updates):

return f(self, *args, **kwargs)

return modified

return decorated

def also_with_minimal_runtime(f):

assert callable(f)

def metafunc(self, with_minimal_runtime):

assert self.get_setting('MINIMAL_RUNTIME') is None

if with_minimal_runtime:

self.set_setting('MINIMAL_RUNTIME', 1)

f(self)

metafunc._parameterize = {'': (False,),

'minimal_runtime': (True,)}

return metafunc

def also_with_wasm_bigint(f):

assert callable(f)

def metafunc(self, with_bigint):

if with_bigint:

if not self.is_wasm():

self.skipTest('wasm2js does not support WASM_BIGINT')

if self.get_setting('WASM_BIGINT') is not None:

self.skipTest('redundant in bigint test config')

self.set_setting('WASM_BIGINT')

self.node_args += shared.node_bigint_flags()

f(self)

else:

f(self)

metafunc._parameterize = {'': (False,),

'bigint': (True,)}

return metafunc

# This works just like `with_both_eh_sjlj` above but doesn't enable exceptions.

# Use this for tests that use setjmp/longjmp but not exceptions handling.

def with_both_sjlj(f):

assert callable(f)

def metafunc(self, is_native):

if is_native:

if not self.is_wasm():

self.skipTest('wasm2js does not support wasm SjLj')

self.require_wasm_eh()

# FIXME Temporarily disabled. Enable this later when the bug is fixed.

if '-fsanitize=address' in self.emcc_args:

self.skipTest('Wasm EH does not work with asan yet')

self.set_setting('SUPPORT_LONGJMP', 'wasm')

f(self)

else:

self.set_setting('SUPPORT_LONGJMP', 'emscripten')

f(self)

metafunc._parameterize = {'': (False,),

'wasm_sjlj': (True,)}

return metafunc

def ensure_dir(dirname):

dirname = Path(dirname)

dirname.mkdir(parents=True, exist_ok=True)

def limit_size(string):

maxbytes = 800000 * 20

if sys.stdout.isatty():

maxlines = 500

max_line = 500

else:

max_line = 5000

maxlines = 100000

lines = string.splitlines()

for i, line in enumerate(lines):

if len(line) > max_line:

lines[i] = line[:max_line] + '[..]'

if len(lines) > maxlines:

lines = lines[0:maxlines // 2] + ['[..]'] + lines[-maxlines // 2:]

lines.append('(not all output shown. See `limit_size`)')

string = '\n'.join(lines) + '\n'

if len(string) > maxbytes:

string = string[0:maxbytes // 2] + '\n[..]\n' + string[-maxbytes // 2:]

return string

def create_file(name, contents, binary=False):

name = Path(name)

assert not name.is_absolute()

if binary:

name.write_bytes(contents)

else:

name.write_text(contents, encoding='utf-8')

def make_executable(name):

Path(name).chmod(stat.S_IREAD | stat.S_IWRITE | stat.S_IEXEC)

def make_dir_writeable(dirname):

# Ensure all files are readable and writable by the current user.

permission_bits = stat.S_IWRITE | stat.S_IREAD

def is_writable(path):

return (os.stat(path).st_mode & permission_bits) != permission_bits

def make_writable(path):

new_mode = os.stat(path).st_mode | permission_bits

os.chmod(path, new_mode)

# Some tests make files and subdirectories read-only, so rmtree/unlink will not delete

# them. Force-make everything writable in the subdirectory to make it

# removable and re-attempt.

if not is_writable(dirname):

make_writable(dirname)

for directory, subdirs, files in os.walk(dirname):

for item in files + subdirs:

i = os.path.join(directory, item)

if not os.path.islink(i):

make_writable(i)

def force_delete_dir(dirname):

make_dir_writeable(dirname)

utils.delete_dir(dirname)

def parameterized(parameters):

"""

Mark a test as parameterized.

Usage:

@parameterized({

'subtest1': (1, 2, 3),

'subtest2': (4, 5, 6),

})

def test_something(self, a, b, c):

... # actual test body

This is equivalent to defining two tests:

def test_something_subtest1(self):

# runs test_something(1, 2, 3)

def test_something_subtest2(self):

# runs test_something(4, 5, 6)

"""

def decorator(func):

func._parameterize = parameters

return func

return decorator

class RunnerMeta(type):

@classmethod

def make_test(mcs, name, func, suffix, args):

"""

This is a helper function to create new test functions for each parameterized form.

:param name: the original name of the function

:param func: the original function that we are parameterizing

:param suffix: the suffix to append to the name of the function for this parameterization

:param args: the positional arguments to pass to the original function for this parameterization

:returns: a tuple of (new_function_name, new_function_object)

"""

# Create the new test function. It calls the original function with the specified args.

# We use @functools.wraps to copy over all the function attributes.

@wraps(func)

def resulting_test(self):

return func(self, *args)

# Add suffix to the function name so that it displays correctly.

if suffix:

resulting_test.__name__ = f'{name}_{suffix}'

else:

resulting_test.__name__ = name

# On python 3, functions have __qualname__ as well. This is a full dot-separated path to the

# function. We add the suffix to it as well.

resulting_test.__qualname__ = f'{func.__qualname__}_{suffix}'

return resulting_test.__name__, resulting_test

def __new__(mcs, name, bases, attrs):

# This metaclass expands parameterized methods from `attrs` into separate ones in `new_attrs`.

new_attrs = {}

for attr_name, value in attrs.items():

# Check if a member of the new class has _parameterize, the tag inserted by @parameterized.

if hasattr(value, '_parameterize'):

# If it does, we extract the parameterization information, build new test functions.

for suffix, args in value._parameterize.items():

new_name, func = mcs.make_test(attr_name, value, suffix, args)

assert new_name not in new_attrs, 'Duplicate attribute name generated when parameterizing %s' % attr_name

new_attrs[new_name] = func

else:

# If not, we just copy it over to new_attrs verbatim.

assert attr_name not in new_attrs, '%s collided with an attribute from parameterization' % attr_name

new_attrs[attr_name] = value

# We invoke type, the default metaclass, to actually create the new class, with new_attrs.

return type.__new__(mcs, name, bases, new_attrs)

class RunnerCore(unittest.TestCase, metaclass=RunnerMeta):

# default temporary directory settings. set_temp_dir may be called later to

# override these

temp_dir = shared.TEMP_DIR

canonical_temp_dir = get_canonical_temp_dir(shared.TEMP_DIR)

# This avoids cluttering the test runner output, which is stderr too, with compiler warnings etc.

# Change this to None to get stderr reporting, for debugging purposes

stderr_redirect = STDOUT

def is_wasm(self):

return self.get_setting('WASM') != 0

def check_dylink(self):

if self.get_setting('ALLOW_MEMORY_GROWTH') == 1 and not self.is_wasm():

self.skipTest('no dynamic linking with memory growth (without wasm)')

if not self.is_wasm():

self.skipTest('no dynamic linking support in wasm2js yet')

if '-fsanitize=undefined' in self.emcc_args:

self.skipTest('no dynamic linking support in UBSan yet')

def require_v8(self):

if not config.V8_ENGINE or config.V8_ENGINE not in config.JS_ENGINES:

if 'EMTEST_SKIP_V8' in os.environ:

self.skipTest('test requires v8 and EMTEST_SKIP_V8 is set')

else:

self.fail('d8 required to run this test. Use EMTEST_SKIP_V8 to skip')

self.require_engine(config.V8_ENGINE)

self.emcc_args.append('-sENVIRONMENT=shell')

def require_node(self):

if not config.NODE_JS or config.NODE_JS not in config.JS_ENGINES:

if 'EMTEST_SKIP_NODE' in os.environ:

self.skipTest('test requires node and EMTEST_SKIP_NODE is set')

else:

self.fail('node required to run this test. Use EMTEST_SKIP_NODE to skip')

self.require_engine(config.NODE_JS)

def require_engine(self, engine):

logger.debug(f'require_engine: {engine}')

if self.required_engine and self.required_engine != engine:

self.skipTest(f'Skipping test that requires `{engine}` when `{self.required_engine}` was previously required')

self.required_engine = engine

self.js_engines = [engine]

self.wasm_engines = []

def require_wasm64(self):

if config.NODE_JS and config.NODE_JS in self.js_engines:

version = shared.check_node_version()

if version >= (16, 0, 0):

self.js_engines = [config.NODE_JS]

self.node_args += shared.node_memory64_flags()

return

if config.V8_ENGINE and config.V8_ENGINE in self.js_engines:

self.emcc_args.append('-sENVIRONMENT=shell')

self.js_engines = [config.V8_ENGINE]

self.v8_args.append('--experimental-wasm-memory64')

return

if 'EMTEST_SKIP_WASM64' in os.environ:

self.skipTest('test requires node >= 16 or d8 (and EMTEST_SKIP_WASM64 is set)')

else:

self.fail('either d8 or node >= 16 required to run wasm64 tests. Use EMTEST_SKIP_WASM64 to skip')

def require_simd(self):

if config.NODE_JS and config.NODE_JS in self.js_engines:

version = shared.check_node_version()

if version >= (16, 0, 0):

self.js_engines = [config.NODE_JS]

return

if config.V8_ENGINE and config.V8_ENGINE in self.js_engines:

self.emcc_args.append('-sENVIRONMENT=shell')

self.js_engines = [config.V8_ENGINE]

return

if 'EMTEST_SKIP_SIMD' in os.environ:

self.skipTest('test requires node >= 16 or d8 (and EMTEST_SKIP_SIMD is set)')

else:

self.fail('either d8 or node >= 16 required to run wasm64 tests. Use EMTEST_SKIP_SIMD to skip')

def require_wasm_eh(self):

if config.NODE_JS and config.NODE_JS in self.js_engines:

version = shared.check_node_version()

if version >= (17, 0, 0):

self.js_engines = [config.NODE_JS]

return

if config.V8_ENGINE and config.V8_ENGINE in self.js_engines:

self.emcc_args.append('-sENVIRONMENT=shell')

self.js_engines = [config.V8_ENGINE]

self.v8_args.append('--experimental-wasm-eh')

return

if 'EMTEST_SKIP_EH' in os.environ:

self.skipTest('test requires node >= 16 or d8 (and EMTEST_SKIP_EH is set)')

else:

self.fail('either d8 or node >= 16 required to run wasm-eh tests. Use EMTEST_SKIP_EH to skip')

def setup_node_pthreads(self):

self.require_node()

self.emcc_args += ['-Wno-pthreads-mem-growth', '-pthread']

if self.get_setting('MINIMAL_RUNTIME'):

self.skipTest('node pthreads not yet supported with MINIMAL_RUNTIME')

self.js_engines = [config.NODE_JS]

self.node_args += shared.node_pthread_flags()

def uses_memory_init_file(self):

if self.get_setting('SIDE_MODULE') or (self.is_wasm() and not self.get_setting('WASM2JS')):

return False

elif '--memory-init-file' in self.emcc_args:

return int(self.emcc_args[self.emcc_args.index('--memory-init-file') + 1])

else:

# side modules handle memory differently; binaryen puts the memory in the wasm module

opt_supports = any(opt in self.emcc_args for opt in ('-O2', '-O3', '-Os', '-Oz'))

return opt_supports

def set_temp_dir(self, temp_dir):

self.temp_dir = temp_dir

self.canonical_temp_dir = get_canonical_temp_dir(self.temp_dir)

# Explicitly set dedicated temporary directory for parallel tests

os.environ['EMCC_TEMP_DIR'] = self.temp_dir

@classmethod

def setUpClass(cls):

super().setUpClass()

print('(checking sanity from test runner)') # do this after we set env stuff

shared.check_sanity(force=True)

def setUp(self):

super().setUp()

self.settings_mods = {}

self.emcc_args = ['-Wclosure', '-Werror', '-Wno-limited-postlink-optimizations']

self.ldflags = []

# Increate stack trace limit to maximise usefulness of test failure reports

self.node_args = ['--stack-trace-limit=50']

node_version = shared.check_node_version()

if node_version:

if node_version < (11, 0, 0):

self.node_args.append('--unhandled-rejections=strict')

self.node_args.append('--experimental-wasm-se')

else:

# Include backtrace for all uncuaght exceptions (not just Error).

self.node_args.append('--trace-uncaught')

if node_version < (15, 0, 0):

# Opt in to node v15 default behaviour:

# https://nodejs.org/api/cli.html#cli_unhandled_rejections_mode

self.node_args.append('--unhandled-rejections=throw')

# If the version we are running tests in is lower than the version that

# emcc targets then we need to tell emcc to target that older version.

emcc_min_node_version_str = str(shared.settings.MIN_NODE_VERSION)

emcc_min_node_version = (

int(emcc_min_node_version_str[0:2]),

int(emcc_min_node_version_str[2:4]),

int(emcc_min_node_version_str[4:6])

)

if node_version < emcc_min_node_version:

self.emcc_args += building.get_emcc_node_flags(node_version)

self.v8_args = ['--wasm-staging']

self.env = {}

self.temp_files_before_run = []

self.uses_es6 = False

self.js_engines = config.JS_ENGINES.copy()

self.required_engine = None

self.wasm_engines = config.WASM_ENGINES.copy()

self.banned_js_engines = []

self.use_all_engines = EMTEST_ALL_ENGINES

if EMTEST_DETECT_TEMPFILE_LEAKS:

for root, dirnames, filenames in os.walk(self.temp_dir):

for dirname in dirnames:

self.temp_files_before_run.append(os.path.normpath(os.path.join(root, dirname)))

for filename in filenames:

self.temp_files_before_run.append(os.path.normpath(os.path.join(root, filename)))

if EMTEST_SAVE_DIR:

self.working_dir = path_from_root('out/test')

if os.path.exists(self.working_dir):

if EMTEST_SAVE_DIR == 2:

print('Not clearing existing test directory')

else:

print('Clearing existing test directory')

# Even when --save-dir is used we still try to start with an empty directory as many tests

# expect this. --no-clean can be used to keep the old contents for the new test

# run. This can be useful when iterating on a given test with extra files you want to keep

# around in the output directory.

utils.delete_contents(self.working_dir)

else:

print('Creating new test output directory')

ensure_dir(self.working_dir)

else:

self.working_dir = tempfile.mkdtemp(prefix='emscripten_test_' + self.__class__.__name__ + '_', dir=self.temp_dir)

os.chdir(self.working_dir)

if not EMTEST_SAVE_DIR:

self.has_prev_ll = False

for temp_file in os.listdir(shared.TEMP_DIR):

if temp_file.endswith('.ll'):

self.has_prev_ll = True

def tearDown(self):

if not EMTEST_SAVE_DIR:

# rmtree() fails on Windows if the current working directory is inside the tree.

os.chdir(os.path.dirname(self.get_dir()))

force_delete_dir(self.get_dir())

if EMTEST_DETECT_TEMPFILE_LEAKS and not DEBUG:

temp_files_after_run = []

for root, dirnames, filenames in os.walk(self.temp_dir):

for dirname in dirnames:

temp_files_after_run.append(os.path.normpath(os.path.join(root, dirname)))

for filename in filenames:

temp_files_after_run.append(os.path.normpath(os.path.join(root, filename)))

# Our leak detection will pick up *any* new temp files in the temp dir.

# They may not be due to us, but e.g. the browser when running browser

# tests. Until we figure out a proper solution, ignore some temp file

# names that we see on our CI infrastructure.

ignorable_file_prefixes = [

'/tmp/tmpaddon',

'/tmp/circleci-no-output-timeout',

'/tmp/wasmer'

]

left_over_files = set(temp_files_after_run) - set(self.temp_files_before_run)

left_over_files = [f for f in left_over_files if not any([f.startswith(prefix) for prefix in ignorable_file_prefixes])]

if len(left_over_files):

print('ERROR: After running test, there are ' + str(len(left_over_files)) + ' new temporary files/directories left behind:', file=sys.stderr)

for f in left_over_files:

print('leaked file: ' + f, file=sys.stderr)

self.fail('Test leaked ' + str(len(left_over_files)) + ' temporary files!')

def get_setting(self, key, default=None):

return self.settings_mods.get(key, default)

def set_setting(self, key, value=1):

if value is None:

self.clear_setting(key)

if type(value) == bool:

value = int(value)

self.settings_mods[key] = value

def has_changed_setting(self, key):

return key in self.settings_mods

def clear_setting(self, key):

self.settings_mods.pop(key, None)

def serialize_settings(self):

ret = []

for key, value in self.settings_mods.items():

if value == 1:

ret.append(f'-s{key}')

elif type(value) == list:

ret.append(f'-s{key}={",".join(value)}')

else:

ret.append(f'-s{key}={value}')

return ret

def get_dir(self):

return self.working_dir

def in_dir(self, *pathelems):

return os.path.join(self.get_dir(), *pathelems)

def add_pre_run(self, code):

assert not self.get_setting('MINIMAL_RUNTIME')

create_file('prerun.js', 'Module.preRun = function() { %s }\n' % code)

self.emcc_args += ['--pre-js', 'prerun.js']

def add_post_run(self, code):

assert not self.get_setting('MINIMAL_RUNTIME')

create_file('postrun.js', 'Module.postRun = function() { %s }\n' % code)

self.emcc_args += ['--pre-js', 'postrun.js']

def add_on_exit(self, code):

assert not self.get_setting('MINIMAL_RUNTIME')

create_file('onexit.js', 'Module.onExit = function() { %s }\n' % code)

self.emcc_args += ['--pre-js', 'onexit.js']

# returns the full list of arguments to pass to emcc

# param @main_file whether this is the main file of the test. some arguments

# (like --pre-js) do not need to be passed when building

# libraries, for example

def get_emcc_args(self, main_file=False, ldflags=True):

def is_ldflag(f):

return any(f.startswith(s) for s in ['-sENVIRONMENT=', '--pre-js=', '--post-js='])

args = self.serialize_settings() + self.emcc_args

if ldflags:

args += self.ldflags

else:

args = [a for a in args if not is_ldflag(a)]

if not main_file:

for i, arg in enumerate(args):

if arg in ('--pre-js', '--post-js'):

args[i] = None

args[i + 1] = None

args = [arg for arg in args if arg is not None]

return args

def verify_es5(self, filename):

es_check = shared.get_npm_cmd('es-check')

# use --quiet once its available

# See: https://github.com/dollarshaveclub/es-check/pull/126/

es_check_env = os.environ.copy()

es_check_env['PATH'] = os.path.dirname(config.NODE_JS[0]) + os.pathsep + es_check_env['PATH']

inputfile = os.path.abspath(filename)

# For some reason es-check requires unix paths, even on windows

if WINDOWS:

inputfile = inputfile.replace('\\', '/')

try:

# es-check prints the details of the errors to stdout, but it also prints

# stuff in the case there are no errors:

# ES-Check: there were no ES version matching errors!

# pipe stdout and stderr so that we can choose if/when to print this

# output and avoid spamming stdout when tests are successful.

shared.run_process(es_check + ['es5', inputfile], stdout=PIPE, stderr=STDOUT, env=es_check_env)

except subprocess.CalledProcessError as e:

print(e.stdout)

self.fail('es-check failed to verify ES5 output compliance')

# Build JavaScript code from source code

def build(self, filename, libraries=None, includes=None, force_c=False, js_outfile=True, emcc_args=None, output_basename=None):

suffix = '.js' if js_outfile else '.wasm'

compiler = [compiler_for(filename, force_c)]

if compiler[0] == EMCC:

# TODO(https://github.com/emscripten-core/emscripten/issues/11121)

# For historical reasons emcc compiles and links as C++ by default.

# However we want to run our tests in a more strict manner. We can

# remove this if the issue above is ever fixed.

compiler.append('-sNO_DEFAULT_TO_CXX')

if force_c:

assert shared.suffix(filename) != '.c', 'force_c is not needed for source files ending in .c'

compiler.append('-xc')

if output_basename:

output = output_basename + suffix

else:

basename = os.path.basename(filename)

output = shared.unsuffixed(basename) + suffix

cmd = compiler + [filename, '-o', output] + self.get_emcc_args(main_file=True)

if emcc_args:

cmd += emcc_args

if libraries:

cmd += libraries

if includes:

cmd += ['-I' + str(include) for include in includes]

self.run_process(cmd, stderr=self.stderr_redirect if not DEBUG else None)

self.assertExists(output)

if js_outfile and self.uses_memory_init_file():

src = read_file(output)

# side memory init file, or an empty one in the js

assert ('/* memory initializer */' not in src) or ('/* memory initializer */ allocate([]' in src)

return output

def get_func(self, src, name):

start = src.index('function ' + name + '(')

t = start

n = 0

while True:

if src[t] == '{':

n += 1

elif src[t] == '}':

n -= 1

if n == 0:

return src[start:t + 1]

t += 1

assert t < len(src)

def count_funcs(self, javascript_file):

num_funcs = 0

start_tok = "// EMSCRIPTEN_START_FUNCS"

end_tok = "// EMSCRIPTEN_END_FUNCS"

start_off = 0

end_off = 0

js = read_file(javascript_file)

blob = "".join(js.splitlines())

start_off = blob.find(start_tok) + len(start_tok)

end_off = blob.find(end_tok)

asm_chunk = blob[start_off:end_off]

num_funcs = asm_chunk.count('function ')

return num_funcs

def count_wasm_contents(self, wasm_binary, what):

out = self.run_process([os.path.join(building.get_binaryen_bin(), 'wasm-opt'), wasm_binary, '--metrics'], stdout=PIPE).stdout

# output is something like

# [?] : 125

for line in out.splitlines():

if '[' + what + ']' in line:

ret = line.split(':')[1].strip()

return int(ret)

self.fail('Failed to find [%s] in wasm-opt output' % what)

def get_wasm_text(self, wasm_binary):

return self.run_process([WASM_DIS, wasm_binary], stdout=PIPE).stdout

def is_exported_in_wasm(self, name, wasm):

wat = self.get_wasm_text(wasm)

return ('(export "%s"' % name) in wat

def measure_wasm_code_lines(self, wasm):

wat_lines = self.get_wasm_text(wasm).splitlines()

non_data_lines = [line for line in wat_lines if '(data ' not in line]

return len(non_data_lines)

def clean_js_output(self, output):

"""Cleaup the JS output prior to running verification steps on it.

Due to minification, when we get a crash report from JS it can sometimes

contains the entire program in the output (since the entire program is

on a single line). In this case we can sometimes get false positives

when checking for strings in the output. To avoid these false positives

and the make the output easier to read in such cases we attempt to remove

such lines from the JS output.

"""

lines = output.splitlines()

long_lines = []

def cleanup(line):

if len(line) > 2048:

# Sanity check that this is really the emscripten program/module on

# a single line.

assert line.startswith('var Module=typeof Module!="undefined"')

long_lines.append(line)

line = '<REPLACED ENTIRE PROGRAM ON SINGLE LINE>'

return line

lines = [cleanup(l) for l in lines]

if not long_lines:

# No long lines found just return the unmodified output

return output

# Sanity check that we only a single long line.

assert len(long_lines) == 1

return '\n'.join(lines)

def run_js(self, filename, engine=None, args=None,

output_nicerizer=None,

assert_returncode=0,

interleaved_output=True):

# use files, as PIPE can get too full and hang us

stdout_file = self.in_dir('stdout')

stderr_file = None

if interleaved_output:

stderr = STDOUT

else:

stderr_file = self.in_dir('stderr')

stderr = open(stderr_file, 'w')

stdout = open(stdout_file, 'w')

error = None

timeout_error = None

if not engine:

engine = self.js_engines[0]

if engine == config.NODE_JS:

engine = engine + self.node_args

if engine == config.V8_ENGINE:

engine = engine + self.v8_args

try:

jsrun.run_js(filename, engine, args,

stdout=stdout,

stderr=stderr,

assert_returncode=assert_returncode)

except subprocess.TimeoutExpired as e:

timeout_error = e

except subprocess.CalledProcessError as e:

error = e

finally:

stdout.close()

if stderr != STDOUT:

stderr.close()

# Make sure that we produced proper line endings to the .js file we are about to run.

if not filename.endswith('.wasm'):

self.assertEqual(line_endings.check_line_endings(filename), 0)

ret = read_file(stdout_file)

if not interleaved_output:

ret += read_file(stderr_file)

if output_nicerizer:

ret = output_nicerizer(ret)

if assert_returncode != 0:

ret = self.clean_js_output(ret)

if error or timeout_error or EMTEST_VERBOSE:

print('-- begin program output --')

print(limit_size(read_file(stdout_file)), end='')

print('-- end program output --')

if not interleaved_output:

print('-- begin program stderr --')

print(limit_size(read_file(stderr_file)), end='')

print('-- end program stderr --')

if timeout_error:

raise timeout_error

if error:

ret = limit_size(ret)

if assert_returncode == NON_ZERO:

self.fail('JS subprocess unexpectedly succeeded (%s): Output:\n%s' % (error.cmd, ret))

else:

self.fail('JS subprocess failed (%s): %s. Output:\n%s' % (error.cmd, error.returncode, ret))

# We should pass all strict mode checks

self.assertNotContained('strict warning:', ret)

return ret

def assertExists(self, filename, msg=None):

if not msg:

msg = f'Expected file not found: {filename}'

self.assertTrue(os.path.exists(filename), msg)

def assertNotExists(self, filename, msg=None):

if not msg:

msg = 'Unexpected file exists: ' + filename

self.assertFalse(os.path.exists(filename), msg)

# Tests that the given two paths are identical, modulo path delimiters. E.g. "C:/foo" is equal to "C:\foo".

def assertPathsIdentical(self, path1, path2):

path1 = path1.replace('\\', '/')