db = odoo.tools.config['db_name']

# If the database name is not provided on the command-line,

# use the one on the thread (which means if it is provided on

# the command-line, this will break when installing another

# database from XML-RPC).

if not db and hasattr(threading.current_thread(), 'dbname'):

return threading.current_thread().dbname

""" Decorator for standalone test functions. This is somewhat dedicated to

tests that install, upgrade or uninstall some modules, which is currently

forbidden in regular test cases. The function is registered under the given

``tags`` and the corresponding Odoo module name.

"""

def register(func):

# register func by odoo module name

if func.__module__.startswith('odoo.addons.'):

module = func.__module__.split('.')[2]

standalone_tests[module].append(func)

# register func with aribitrary name, if any

for tag in tags:

standalone_tests[tag].append(func)

standalone_tests['all'].append(func)

return func

""" Helper function to create a new test user. It allows to quickly create

users given its login and groups (being a comma separated list of xml ids).

Kwargs are directly propagated to the create to further customize the

created user.

User creation uses a potentially customized environment using the context

parameter allowing to specify a custom context. It can be used to force a

specific behavior and/or simplify record creation. An example is to use

mail-related context keys in mail tests to speedup record creation.

Some specific fields are automatically filled to avoid issues

* groups_id: it is filled using groups function parameter;

* name: "login (groups)" by default as it is required;

* email: it is either the login (if it is a valid email) or a generated

string 'x.x@example.com' (x being the first login letter). This is due

to email being required for most odoo operations;

"""

if not login:

raise ValueError('New users require at least a login')

if not groups:

raise ValueError('New users require at least user groups')

if context is None:

context = {}

groups_id = [(6, 0, [env.ref(g.strip()).id for g in groups.split(',')])]

create_values = dict(kwargs, login=login, groups_id=groups_id)

# automatically generate a name as "Login (groups)" to ease user comprehension

if not create_values.get('name'):

create_values['name'] = '%s (%s)' % (login, groups)

# automatically give a password equal to login

if not create_values.get('password'):

create_values['password'] = login + 'x' * (8 - len(login))

# generate email if not given as most test require an email

if 'email' not in create_values:

if single_email_re.match(login):

create_values['email'] = login

else:

create_values['email'] = '%s.%s@example.com' % (login[0], login[0])

# ensure company_id + allowed company constraint works if not given at create

if 'company_id' in create_values and 'company_ids' not in create_values:

create_values['company_ids'] = [(4, create_values['company_id'])]

def __init__(self, model, domain):

self._model = model

self._domain = domain

def __enter__(self):

self._before = self._model.search(self._domain)

self._after = None

return self

def __exit__(self, exc_type, exc_value, exc_traceback):

if exc_type is None:

self._after = self._model.search(self._domain) - self._before

@property

def records(self):

if self._after is None:

return self._model.search(self._domain) - self._before

""" Metaclass of test case classes to assign default 'test_tags':

'standard', 'at_install' and the name of the module.

"""

def __init__(cls, name, bases, attrs):

super(MetaCase, cls).__init__(name, bases, attrs)

# assign default test tags

if cls.__module__.startswith('odoo.addons.'):

if getattr(cls, 'test_tags', None) is None:

cls.test_tags = {'standard', 'at_install'}

cls.test_module = cls.__module__.split('.')[2]

cls.test_class = cls.__name__

"""Takes some xml and normalize it to make it comparable to other xml

in particular, blank text is removed, and the output is pretty-printed

:param str arch_string: the string representing an XML arch

:param str parser_method: an string representing which lxml.Parser class to use

when normalizing both archs. Takes either "xml" or "html"

:return: the normalized arch

:rtype str:

"""

Parser = None

if parser_method == 'xml':

Parser = etree.XMLParser

elif parser_method == 'html':

Parser = etree.HTMLParser

parser = Parser(remove_blank_text=True)

arch_string = etree.fromstring(arch_string, parser=parser)

""" Subclass of TestCase for Odoo-specific code. This class is abstract and

expects self.registry, self.cr and self.uid to be initialized by subclasses.

"""

longMessage = True # more verbose error message by default: https://www.odoo.com/r/Vmh

warm = True # False during warm-up phase (see :func:`warmup`)

_python_version = sys.version_info

def __init__(self, methodName='runTest'):

super().__init__(methodName)

self.addTypeEqualityFunc(etree._Element, self.assertTreesEqual)

self.addTypeEqualityFunc(html.HtmlElement, self.assertTreesEqual)

def run(self, result):

testMethod = getattr(self, self._testMethodName)

if getattr(testMethod, '_retry', True) and getattr(self, '_retry', True):

tests_run_count = int(os.environ.get('ODOO_TEST_FAILURE_RETRIES', 0)) + 1

else:

tests_run_count = 1

_logger.info('Auto retry disabled for %s', self)

failure = False

for retry in range(tests_run_count):

if retry:

_logger.runbot(f'Retrying a failed test: {self}')

if retry < tests_run_count-1:

with warnings.catch_warnings(), \

result.soft_fail(), \

lower_logging(25, logging.INFO) as quiet_log:

super().run(result)

failure = result.had_failure or quiet_log.had_error_log

else: # last try

super().run(result)

if not failure:

break

def cursor(self):

return self.registry.cursor()

@property

def uid(self):

""" Get the current uid. """

return self.env.uid

@uid.setter

def uid(self, user):

""" Set the uid by changing the test's environment. """

self.env = self.env(user=user)

def ref(self, xid):

""" Returns database ID for the provided :term:`external identifier`,

shortcut for ``_xmlid_lookup``

:param xid: fully-qualified :term:`external identifier`, in the form

:samp:`{module}.{identifier}`

:raise: ValueError if not found

:returns: registered id

"""

return self.browse_ref(xid).id

def browse_ref(self, xid):

""" Returns a record object for the provided

:term:`external identifier`

:param xid: fully-qualified :term:`external identifier`, in the form

:samp:`{module}.{identifier}`

:raise: ValueError if not found

:returns: :class:`~odoo.models.BaseModel`

"""

assert "." in xid, "this method requires a fully qualified parameter, in the following form: 'module.identifier'"

return self.env.ref(xid)

def patch(self, obj, key, val):

""" Do the patch ``setattr(obj, key, val)``, and prepare cleanup. """

patcher = patch.object(obj, key, val) # this is unittest.mock.patch

patcher.start()

self.addCleanup(patcher.stop)

@classmethod

def classPatch(cls, obj, key, val):

""" Do the patch ``setattr(obj, key, val)``, and prepare cleanup. """

patcher = patch.object(obj, key, val) # this is unittest.mock.patch

patcher.start()

cls.addClassCleanup(patcher.stop)

def startPatcher(self, patcher):

mock = patcher.start()

self.addCleanup(patcher.stop)

return mock

@classmethod

def startClassPatcher(cls, patcher):

mock = patcher.start()

cls.addClassCleanup(patcher.stop)

return mock

@contextmanager

def with_user(self, login):

""" Change user for a given test, like with self.with_user() ... """

old_uid = self.uid

try:

user = self.env['res.users'].sudo().search([('login', '=', login)])

assert user, "Login %s not found" % login

# switch user

self.uid = user.id

self.env = self.env(user=self.uid)

yield

finally:

# back

self.uid = old_uid

self.env = self.env(user=self.uid)

@contextmanager

def debug_mode(self):

""" Enable the effects of group 'base.group_no_one'; mainly useful with :class:`Form`. """

origin_user_has_groups = BaseModel.user_has_groups

def user_has_groups(self, groups):

group_set = set(groups.split(','))

if '!base.group_no_one' in group_set:

return False

elif 'base.group_no_one' in group_set:

group_set.remove('base.group_no_one')

return not group_set or origin_user_has_groups(self, ','.join(group_set))

return origin_user_has_groups(self, groups)

with patch('odoo.models.BaseModel.user_has_groups', user_has_groups):

yield

@contextmanager

def _assertRaises(self, exception, *, msg=None):

""" Context manager that clears the environment upon failure. """

with ExitStack() as init:

if hasattr(self, 'env'):

init.enter_context(self.env.cr.savepoint())

if issubclass(exception, AccessError):

# The savepoint() above calls flush(), which leaves the

# record cache with lots of data. This can prevent

# access errors to be detected. In order to avoid this

# issue, we clear the cache before proceeding.

self.env.cr.clear()

with ExitStack() as inner:

cm = inner.enter_context(super().assertRaises(exception, msg=msg))

# *moves* the cleanups from init to inner, this ensures the

# savepoint gets rolled back when `yield` raises `exception`,

# but still allows the initialisation to be protected *and* not

# interfered with by `assertRaises`.

inner.push(init.pop_all())

yield cm

def assertRaises(self, exception, func=None, *args, **kwargs):

if func:

with self._assertRaises(exception):

func(*args, **kwargs)

else:

return self._assertRaises(exception, **kwargs)

if sys.version_info < (3, 10):

# simplified backport of assertNoLogs()

@contextmanager

def assertNoLogs(self, logger: str, level: str):

# assertLogs ensures there is at least one log record when

# exiting the context manager. We insert one dummy record just

# so we pass that silly test while still capturing the logs.

with self.assertLogs(logger, level) as capture:

logging.getLogger(logger).log(getattr(logging, level), "Dummy log record")

yield

if len(capture.output) > 1:

raise self.failureException(f"Unexpected logs found: {capture.output[1:]}")

@contextmanager

def assertQueries(self, expected, flush=True):

""" Check the queries made by the current cursor. ``expected`` is a list

of strings representing the expected queries being made. Query strings

are matched against each other, ignoring case and whitespaces.

"""

Cursor_execute = Cursor.execute

actual_queries = []

def execute(self, query, params=None, log_exceptions=None):

actual_queries.append(query)

return Cursor_execute(self, query, params, log_exceptions)

def get_unaccent_wrapper(cr):

return lambda x: x

if flush:

self.env.flush_all()

self.env.cr.flush()

with patch('odoo.sql_db.Cursor.execute', execute):

with patch('odoo.osv.expression.get_unaccent_wrapper', get_unaccent_wrapper):

yield actual_queries

if flush:

self.env.flush_all()

self.env.cr.flush()

if not self.warm:

return

self.assertEqual(

len(actual_queries), len(expected),

"\n---- actual queries:\n%s\n---- expected queries:\n%s" % (

"\n".join(actual_queries), "\n".join(expected),

)

)

for actual_query, expect_query in zip(actual_queries, expected):

self.assertEqual(

"".join(actual_query.lower().split()),

"".join(expect_query.lower().split()),

"\n---- actual query:\n%s\n---- not like:\n%s" % (actual_query, expect_query),

)

@contextmanager

def assertQueryCount(self, default=0, flush=True, **counters):

""" Context manager that counts queries. It may be invoked either with

one value, or with a set of named arguments like ``login=value``::

with self.assertQueryCount(42):

...

with self.assertQueryCount(admin=3, demo=5):

...

The second form is convenient when used with :func:`users`.

"""

if self.warm:

# mock random in order to avoid random bus gc

with patch('random.random', lambda: 1):

login = self.env.user.login

expected = counters.get(login, default)

if flush:

self.env.flush_all()

self.env.cr.flush()

count0 = self.cr.sql_log_count

yield

if flush:

self.env.flush_all()

self.env.cr.flush()

count = self.cr.sql_log_count - count0

if count != expected:

# add some info on caller to allow semi-automatic update of query count

frame, filename, linenum, funcname, lines, index = inspect.stack()[2]

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

if "/odoo/addons/" in filename:

filename = filename.rsplit("/odoo/addons/", 1)[1]

if count > expected:

msg = "Query count more than expected for user %s: %d > %d in %s at %s:%s"

# add a subtest in order to continue the test_method in case of failures

with self.subTest():

self.fail(msg % (login, count, expected, funcname, filename, linenum))

else:

logger = logging.getLogger(type(self).__module__)

msg = "Query count less than expected for user %s: %d < %d in %s at %s:%s"

logger.info(msg, login, count, expected, funcname, filename, linenum)

else:

# flush before and after during warmup, in order to reproduce the

# same operations, otherwise the caches might not be ready!

if flush:

self.env.flush_all()

self.env.cr.flush()

yield

if flush:

self.env.flush_all()

self.env.cr.flush()

def assertRecordValues(self, records, expected_values):

''' Compare a recordset with a list of dictionaries representing the expected results.

This method performs a comparison element by element based on their index.

Then, the order of the expected values is extremely important.

Note that:

- Comparison between falsy values is supported: False match with None.

- Comparison between monetary field is also treated according the currency's rounding.

- Comparison between x2many field is done by ids. Then, empty expected ids must be [].

- Comparison between many2one field id done by id. Empty comparison can be done using any falsy value.

:param records: The records to compare.

:param expected_values: List of dicts expected to be exactly matched in records

'''

def _compare_candidate(record, candidate, field_names):

''' Compare all the values in `candidate` with a record.

:param record: record being compared

:param candidate: dict of values to compare

:return: A dictionary will encountered difference in values.

'''

diff = {}

for field_name in field_names:

record_value = record[field_name]

field = record._fields[field_name]

field_type = field.type

if field_type == 'monetary':

# Compare monetary field.

currency_field_name = record._fields[field_name].get_currency_field(record)

record_currency = record[currency_field_name]

if field_name not in candidate:

diff[field_name] = (record_value, None)

elif record_currency:

if record_currency.compare_amounts(candidate[field_name], record_value):

diff[field_name] = (record_value, record_currency.round(candidate[field_name]))

elif candidate[field_name] != record_value:

diff[field_name] = (record_value, candidate[field_name])

elif field_type == 'float' and field.get_digits(record.env):

prec = field.get_digits(record.env)[1]

if float_compare(candidate[field_name], record_value, precision_digits=prec) != 0:

diff[field_name] = (record_value, candidate[field_name])

elif field_type in ('one2many', 'many2many'):

# Compare x2many relational fields.

# Empty comparison must be an empty list to be True.

if field_name not in candidate:

diff[field_name] = (sorted(record_value.ids), None)

elif set(record_value.ids) != set(candidate[field_name]):

diff[field_name] = (sorted(record_value.ids), sorted(candidate[field_name]))

elif field_type == 'many2one':

# Compare many2one relational fields.

# Every falsy value is allowed to compare with an empty record.

if field_name not in candidate:

diff[field_name] = (record_value.id, None)

elif (record_value or candidate[field_name]) and record_value.id != candidate[field_name]:

diff[field_name] = (record_value.id, candidate[field_name])

else:

# Compare others fields if not both interpreted as falsy values.

if field_name not in candidate:

diff[field_name] = (record_value, None)

elif (candidate[field_name] or record_value) and record_value != candidate[field_name]:

diff[field_name] = (record_value, candidate[field_name])

return diff

# Compare records with candidates.

different_values = []

field_names = list(expected_values[0].keys())

for index, record in enumerate(records):

is_additional_record = index >= len(expected_values)

candidate = {} if is_additional_record else expected_values[index]

diff = _compare_candidate(record, candidate, field_names)

if diff:

different_values.append((index, 'additional_record' if is_additional_record else 'regular_diff', diff))

for index in range(len(records), len(expected_values)):

diff = {}

for field_name in field_names:

diff[field_name] = (None, expected_values[index][field_name])

different_values.append((index, 'missing_record', diff))

# Build error message.

if not different_values:

return

errors = ['The records and expected_values do not match.']

if len(records) != len(expected_values):

errors.append('Wrong number of records to compare: %d records versus %d expected values.' % (len(records), len(expected_values)))

for index, diff_type, diff in different_values:

if diff_type == 'regular_diff':

errors.append('\n==== Differences at index %s ====' % index)

record_diff = ['%s:%s' % (k, v[0]) for k, v in diff.items()]

candidate_diff = ['%s:%s' % (k, v[1]) for k, v in diff.items()]

errors.append('\n'.join(difflib.unified_diff(record_diff, candidate_diff)))

elif diff_type == 'additional_record':

errors += [

'\n==== Additional record ====',

pprint.pformat(dict((k, v[0]) for k, v in diff.items())),

]

elif diff_type == 'missing_record':

errors += [

'\n==== Missing record ====',

pprint.pformat(dict((k, v[1]) for k, v in diff.items())),

]

self.fail('\n'.join(errors))

# turns out this thing may not be quite as useful as we thought...

def assertItemsEqual(self, a, b, msg=None):

self.assertCountEqual(a, b, msg=None)

def assertTreesEqual(self, n1, n2, msg=None):

self.assertIsNotNone(n1, msg)

self.assertIsNotNone(n2, msg)

self.assertEqual(n1.tag, n2.tag, msg)

# Because lxml.attrib is an ordereddict for which order is important

# to equality, even though *we* don't care

self.assertEqual(dict(n1.attrib), dict(n2.attrib), msg)

self.assertEqual((n1.text or u'').strip(), (n2.text or u'').strip(), msg)

self.assertEqual((n1.tail or u'').strip(), (n2.tail or u'').strip(), msg)

for c1, c2 in izip_longest(n1, n2):

self.assertTreesEqual(c1, c2, msg)

def _assertXMLEqual(self, original, expected, parser="xml"):

"""Asserts that two xmls archs are equal

:param original: the xml arch to test

:type original: str

:param expected: the xml arch of reference

:type expected: str

:param parser: an string representing which lxml.Parser class to use

when normalizing both archs. Takes either "xml" or "html"

:type parser: str

"""

if original:

original = _normalize_arch_for_assert(original, parser)

if expected:

expected = _normalize_arch_for_assert(expected, parser)

self.assertEqual(original, expected)

def assertXMLEqual(self, original, expected):

return self._assertXMLEqual(original, expected)

def assertHTMLEqual(self, original, expected):

return self._assertXMLEqual(original, expected, 'html')

def profile(self, description='', **kwargs):

test_method = getattr(self, '_testMethodName', 'Unknown test method')

if not hasattr(self, 'profile_session'):

self.profile_session = profiler.make_session(test_method)

return profiler.Profiler(

description='%s uid:%s %s %s' % (test_method, self.env.user.id, 'warm' if self.warm else 'cold', description),

db=self.env.cr.dbname,

profile_session=self.profile_session,

**kwargs)

def patch_requests(self):

# requests.get -> requests.api.request -> Session().request

# TBD: enable by default & set side_effect=NotImplementedError to force an error

p = patch('requests.Session.request', Mock(spec_set=[]))

self.addCleanup(p.stop)

""" Test class in which all test methods are run in a single transaction,

but each test method is run in a sub-transaction managed by a savepoint.

The transaction's cursor is always closed without committing.

The data setup common to all methods should be done in the class method

`setUpClass`, so that it is done once for all test methods. This is useful

for test cases containing fast tests but with significant database setup

common to all cases (complex in-db test data).

After being run, each test method cleans up the record cache and the

registry cache. However, there is no cleanup of the registry models and

fields. If a test modifies the registry (custom models and/or fields), it

should prepare the necessary cleanup (`self.registry.reset_changes()`).

"""

registry: Registry = None

env: api.Environment = None

cr: Cursor = None

@classmethod

def _gc_filestore(cls):

# attachment can be created or unlink during the tests.

# they can addup during test and take some disc space.

# since cron are not running during tests, we need to gc manually

# We need to check the status of the file system outside of the test cursor

with odoo.registry(get_db_name()).cursor() as cr:

gc_env = api.Environment(cr, odoo.SUPERUSER_ID, {})

gc_env['ir.attachment']._gc_file_store_unsafe()

@classmethod

def setUpClass(cls):

super().setUpClass()

cls.addClassCleanup(cls._gc_filestore)

cls.registry = odoo.registry(get_db_name())

cls.addClassCleanup(cls.registry.reset_changes)

cls.addClassCleanup(cls.registry.clear_caches)

cls.cr = cls.registry.cursor()

cls.addClassCleanup(cls.cr.close)

cls.env = api.Environment(cls.cr, odoo.SUPERUSER_ID, {})

def setUp(self):

super().setUp()

# restore environments after the test to avoid invoking flush() with an

# invalid environment (inexistent user id) from another test

envs = self.env.all.envs

for env in list(envs):

self.addCleanup(env.clear)

# restore the set of known environments as it was at setUp

self.addCleanup(envs.update, list(envs))

self.addCleanup(envs.clear)

self.addCleanup(self.registry.clear_caches)

# This prevents precommit functions and data from piling up

# until cr.flush is called in 'assertRaises' clauses

# (these are not cleared in self.env.clear or envs.clear)

cr = self.env.cr

def _reset(cb, funcs, data):

cb._funcs = funcs

cb.data = data

for callback in [cr.precommit, cr.postcommit, cr.prerollback, cr.postrollback]:

self.addCleanup(_reset, callback, collections.deque(callback._funcs), dict(callback.data))

# flush everything in setUpClass before introducing a savepoint

self.env.flush_all()

self._savepoint_id = next(savepoint_seq)

self.cr.execute('SAVEPOINT test_%d' % self._savepoint_id)

self.addCleanup(self.cr.execute, 'ROLLBACK TO SAVEPOINT test_%d' % self._savepoint_id)

@classmethod

def __init_subclass__(cls):

super().__init_subclass__()

warnings.warn(

"Deprecated class SavepointCase has been merged into TransactionCase",

DeprecationWarning, stacklevel=2,

""" TestCase in which all test methods are run in the same transaction,

the transaction is started with the first test method and rolled back at

the end of the last.

"""

@classmethod

def __init_subclass__(cls):

super().__init_subclass__()

if issubclass(cls, TransactionCase):

_logger.warning("%s inherits from both TransactionCase and SingleTransactionCase")

@classmethod

def setUpClass(cls):

super().setUpClass()

cls.registry = odoo.registry(get_db_name())

cls.addClassCleanup(cls.registry.reset_changes)

cls.addClassCleanup(cls.registry.clear_caches)

cls.cr = cls.registry.cursor()

cls.addClassCleanup(cls.cr.close)

cls.env = api.Environment(cls.cr, odoo.SUPERUSER_ID, {})

def setUp(self):

super(SingleTransactionCase, self).setUp()

"""Maps a future's result through a callback.

Resolves to the application of ``map_fun`` to the result of ``future``.

.. warning:: this does *not* recursively resolve futures, if that's what

you need see :func:`fchain`

"""

fmap_future = Future()

@future.add_done_callback

def _(f):

try:

fmap_future.set_result(map_fun(f.result()))

except Exception as e:

fmap_future.set_exception(e)

"""Chains a future's result to a new future through a callback.

Corresponds to the ``bind`` monadic operation (aka flatmap aka then...

kinda).

"""

new_future = Future()

@future.add_done_callback

def _(f):

try:

n = next_callback(f.result())

@n.add_done_callback

def _(f):

try:

new_future.set_result(f.result())

except Exception as e:

new_future.set_exception(e)

except Exception as e:

new_future.set_exception(e)

assert re.fullmatch(r'\w*_', prefix)

assert re.fullmatch(r'[a-z]+', extension)

assert re.fullmatch(r'\w+', test_name)

now = datetime.now().strftime(date_format)

screenshots_dir = pathlib.Path(odoo.tools.config['screenshots']) / get_db_name() / 'screenshots'

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

fname = f'{prefix}{now}_{test_name}.{extension}'

full_path = screenshots_dir / fname

with full_path.open('wb') as f:

f.write(content)

""" Helper object to control a Chrome headless process. """

remote_debugging_port = 0 # 9222, change it in a non-git-tracked file

def __init__(self, test_class):

self._logger = test_class._logger

self.test_class = test_class

if websocket is None:

self._logger.warning("websocket-client module is not installed")

raise unittest.SkipTest("websocket-client module is not installed")

self.devtools_port = None

self.ws_url = '' # WebSocketUrl

self.ws = None # websocket

self.user_data_dir = tempfile.mkdtemp(suffix='_chrome_odoo')

self.chrome_pid = None

otc = odoo.tools.config

self.screenshots_dir = os.path.join(otc['screenshots'], get_db_name(), 'screenshots')

self.screencasts_dir = None

self.screencasts_frames_dir = None

if otc['screencasts']:

self.screencasts_dir = os.path.join(otc['screencasts'], get_db_name(), 'screencasts')

self.screencasts_frames_dir = os.path.join(self.screencasts_dir, 'frames')

os.makedirs(self.screencasts_frames_dir, exist_ok=True)

self.screencast_frames = []

os.makedirs(self.screenshots_dir, exist_ok=True)

self.window_size = test_class.browser_size.replace('x', ',')

self.touch_enabled = test_class.touch_enabled

self.sigxcpu_handler = None

self._chrome_start()

self._find_websocket()

self._logger.info('Websocket url found: %s', self.ws_url)

self._open_websocket()

self._request_id = itertools.count()

self._result = Future()

self.error_checker = None

self.had_failure = False

# maps request_id to Futures

self._responses = {}

# maps frame ids to callbacks

self._frames = {}

self._handlers = {

'Runtime.consoleAPICalled': self._handle_console,

'Runtime.exceptionThrown': self._handle_exception,

'Page.frameStoppedLoading': self._handle_frame_stopped_loading,

'Page.screencastFrame': self._handle_screencast_frame,

}

self._receiver = threading.Thread(

target=self._receive,

name="WebSocket events consumer",

args=(get_db_name(),)

)

self._receiver.start()

self._logger.info('Enable chrome headless console log notification')

self._websocket_send('Runtime.enable')

self._logger.info('Chrome headless enable page notifications')

self._websocket_send('Page.enable')

if os.name == 'posix':

self.sigxcpu_handler = signal.getsignal(signal.SIGXCPU)

signal.signal(signal.SIGXCPU, self.signal_handler)

def signal_handler(self, sig, frame):

if sig == signal.SIGXCPU:

_logger.info('CPU time limit reached, stopping Chrome and shutting down')

self.stop()

os._exit(0)

def stop(self):

if self.chrome_pid is not None:

self._logger.info("Closing chrome headless with pid %s", self.chrome_pid)

self._websocket_send('Browser.close')

self._logger.info("Closing websocket connection")

self.ws.close()

self._logger.info("Terminating chrome headless with pid %s", self.chrome_pid)

os.kill(self.chrome_pid, signal.SIGTERM)

if self.user_data_dir and os.path.isdir(self.user_data_dir) and self.user_data_dir != '/':

self._logger.info('Removing chrome user profile "%s"', self.user_data_dir)

shutil.rmtree(self.user_data_dir, ignore_errors=True)

# Restore previous signal handler

if self.sigxcpu_handler and os.name == 'posix':

signal.signal(signal.SIGXCPU, self.sigxcpu_handler)

@property

def executable(self):

system = platform.system()

if system == 'Linux':

for bin_ in ['google-chrome', 'chromium', 'chromium-browser', 'google-chrome-stable']:

try:

return find_in_path(bin_)

except IOError:

continue

elif system == 'Darwin':

bins = [

'/Applications/Google Chrome.app/Contents/MacOS/Google Chrome',

'/Applications/Chromium.app/Contents/MacOS/Chromium',

]

for bin_ in bins:

if os.path.exists(bin_):

return bin_

elif system == 'Windows':

bins = [

'%ProgramFiles%\\Google\\Chrome\\Application\\chrome.exe',

'%ProgramFiles(x86)%\\Google\\Chrome\\Application\\chrome.exe',

'%LocalAppData%\\Google\\Chrome\\Application\\chrome.exe',

]

for bin_ in bins:

bin_ = os.path.expandvars(bin_)

if os.path.exists(bin_):

return bin_

raise unittest.SkipTest("Chrome executable not found")

def _chrome_without_limit(self, cmd):

if os.name == 'posix' and platform.system() != 'Darwin':

# since the introduction of pointer compression in Chrome 80 (v8 v8.0),

# the memory reservation algorithm requires more than 8GiB of

# virtual mem for alignment this exceeds our default memory limits.

def preexec():

import resource

resource.setrlimit(resource.RLIMIT_AS, (resource.RLIM_INFINITY, resource.RLIM_INFINITY))

else:

preexec = None

# pylint: disable=subprocess-popen-preexec-fn

return subprocess.Popen(cmd, stderr=subprocess.DEVNULL, preexec_fn=preexec)

def _spawn_chrome(self, cmd):

proc = self._chrome_without_limit(cmd)

port_file = pathlib.Path(self.user_data_dir, 'DevToolsActivePort')

for _ in range(CHECK_BROWSER_ITERATIONS):

time.sleep(CHECK_BROWSER_SLEEP)

if port_file.is_file() and port_file.stat().st_size > 5:

with port_file.open('r', encoding='utf-8') as f:

self.devtools_port = int(f.readline())

return proc.pid

raise unittest.SkipTest(f'Failed to detect chrome devtools port after {BROWSER_WAIT :.1f}s.')

def _chrome_start(self):

if self.chrome_pid is not None:

return

switches = {

'--headless': '',

'--no-default-browser-check': '',

'--no-first-run': '',

'--disable-extensions': '',

'--disable-background-networking' : '',

'--disable-background-timer-throttling' : '',