# This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this
# file, You can obtain one at http://mozilla.org/MPL/2.0/.
import os
import sys
import mozinfo
import yaml
from marionette_driver.errors
import JavascriptException, ScriptTimeoutException
from mozproxy
import get_playback
AWSY_PATH = os.path.dirname(os.path.dirname(os.path.realpath(__file__)))
if AWSY_PATH
not in sys.path:
sys.path.append(AWSY_PATH)
from awsy
import process_perf_data, webservers
from awsy.awsy_test_case
import AwsyTestCase
class TestMemoryUsage(AwsyTestCase):
"""Provides a test that collects memory usage at various checkpoints:
-
"Start" - Just after startup
-
"StartSettled" - After an additional wait time
-
"TabsOpen" - After opening all provided URLs
-
"TabsOpenSettled" - After an additional wait time
-
"TabsOpenForceGC" - After forcibly invoking garbage collection
-
"TabsClosed" - After closing all tabs
-
"TabsClosedSettled" - After an additional wait time
-
"TabsClosedForceGC" - After forcibly invoking garbage collection
"""
def urls(self):
return self._urls
def perf_suites(self):
return process_perf_data.PERF_SUITES
def perf_checkpoints(self):
return process_perf_data.CHECKPOINTS
def perf_extra_opts(self):
return self._extra_opts
def setupTp5(self):
urls =
None
default_tp5n_manifest = os.path.join(
self._webroot_dir,
"page_load_test",
"tp5n",
"tp5n.manifest"
)
tp5n_manifest = self.testvars.get(
"pageManifest", default_tp5n_manifest)
with open(tp5n_manifest)
as fp:
urls = fp.readlines()
# pylint --py3k: W1636
urls = list(map(
lambda x: x.replace(
"localhost",
"localhost:{}"), urls))
# We haven't set self._urls yet, so this value might be zero if
# 'entities' wasn't specified.
to_load = self.pages_to_load()
if not to_load:
to_load = len(urls)
self._webservers = webservers.WebServers(
"localhost", 8001, self._webroot_dir, to_load
)
self._webservers.start()
for url, server
in zip(urls, self._webservers.servers):
self._urls.append(url.strip().format(server.port))
def setupTp6(self):
# tp5n stores its manifest in the zip file that gets extracted, tp6
# doesn't so we just keep one in our project dir for now.
default_tp6_pages_manifest = os.path.join(AWSY_PATH,
"conf",
"tp6-pages.yml")
tp6_pages_manifest = self.testvars.get(
"pageManifest", default_tp6_pages_manifest
)
urls = []
with open(tp6_pages_manifest)
as f:
d = yaml.safe_load(f)
for r
in d:
url = r[
"url"]
if isinstance(url, list):
urls.extend(url)
else:
urls.append(url)
self._urls = urls
# Indicate that we're using tp6 in the perf data.
self._extra_opts = [
"tp6"]
if self.marionette.get_pref(
"fission.autostart"):
self._extra_opts.append(
"fission")
# Now we setup the mitm proxy with our tp6 pageset.
tp6_pageset_manifest = os.path.join(AWSY_PATH,
"tp6-pageset.manifest")
config = {
"playback_tool":
"mitmproxy",
"playback_version":
"5.1.1",
"playback_files": [tp6_pageset_manifest],
"platform": mozinfo.os,
"obj_path": self._webroot_dir,
"binary": self._binary,
"run_local": self._run_local,
"app":
"firefox",
"host":
"127.0.0.1",
"ignore_mitmdump_exit_failure":
True,
}
self._playback = get_playback(config)
self._playback.start()
# We need to reload after the mitmproxy cert is installed
self.marionette.restart(in_app=
False, clean=
False)
# Setup WebDriver capabilities that we need
self.marionette.delete_session()
caps = {
"unhandledPromptBehavior":
"dismiss",
# Ignore page navigation warnings
}
self.marionette.start_session(caps)
self.marionette.set_context(
"chrome")
def setUp(self):
AwsyTestCase.setUp(self)
self.logger.info(
"setting up")
self._webroot_dir = self.testvars[
"webRootDir"]
self._urls = []
self._extra_opts =
None
if self.testvars.get(
"tp5",
False):
self.setupTp5()
else:
self.setupTp6()
self.logger.info(
"areweslimyet run by %d pages, %d iterations,"
" %d perTabPause, %d settleWaitTime"
% (
self._pages_to_load,
self._iterations,
self._perTabPause,
self._settleWaitTime,
)
)
self.logger.info(
"done setting up!")
def tearDown(self):
self.logger.info(
"tearing down!")
self.logger.info(
"tearing down webservers!")
if self.testvars.get(
"tp5",
False):
self._webservers.stop()
else:
self._playback.stop()
AwsyTestCase.tearDown(self)
self.logger.info(
"done tearing down!")
def clear_preloaded_browser(self):
"""
Clears out the preloaded browser.
"""
self.logger.info(
"closing preloaded browser")
script =
"""
if (window.NewTabPagePreloading) {
return NewTabPagePreloading.removePreloadedBrowser(window);
}
return "NewTabPagePreloading.removePreloadedBrowser not available";
"""
try:
result = self.marionette.execute_script(script, script_timeout=180000)
except JavascriptException
as e:
self.logger.error(
"removePreloadedBrowser() JavaScript error: %s" % e)
except ScriptTimeoutException:
self.logger.error(
"removePreloadedBrowser() timed out")
except Exception:
self.logger.error(
"removePreloadedBrowser() Unexpected error: %s" % sys.exc_info()[0]
)
else:
if result:
self.logger.info(result)
def test_open_tabs(self):
"""Marionette test entry that returns an array of checkpoint arrays.
This will generate a set of checkpoints
for each iteration requested.
Upon successful completion the results will be stored
in
|self.testvars[
"results"]|
and accessible to the test runner via the
|testvars| object it passed
in.
"""
# setup the results array
results = [[]
for _
in range(self.iterations())]
def create_checkpoint(name, iteration, minimize=
False):
checkpoint = self.do_memory_report(name, iteration, minimize)
self.assertIsNotNone(checkpoint,
"Checkpoint was recorded")
results[iteration].append(checkpoint)
# The first iteration gets Start and StartSettled entries before
# opening tabs
create_checkpoint(
"Start", 0)
self.settle()
create_checkpoint(
"StartSettled", 0)
for itr
in range(self.iterations()):
self.open_pages()
create_checkpoint(
"TabsOpen", itr)
self.settle()
create_checkpoint(
"TabsOpenSettled", itr)
create_checkpoint(
"TabsOpenForceGC", itr, minimize=
True)
# Close all tabs
self.reset_state()
with self.marionette.using_context(
"content"):
self.logger.info(
"navigating to about:blank")
self.marionette.navigate(
"about:blank")
self.logger.info(
"navigated to about:blank")
self.signal_user_active()
# Create checkpoint that may contain retained processes that will
# be reused.
create_checkpoint(
"TabsClosedExtraProcesses", itr)
# Clear out the retained processes and measure again.
self.clear_preloaded_browser()
create_checkpoint(
"TabsClosed", itr)
self.settle()
create_checkpoint(
"TabsClosedSettled", itr)
create_checkpoint(
"TabsClosedForceGC", itr, minimize=
True)
# TODO(ER): Temporary hack until bug 1121139 lands
self.logger.info(
"setting results")
self.testvars[
"results"] = results
