"""Tools for writing Cockpit test cases."""
import argparse
import asyncio
import base64
import contextlib
import errno
import fnmatch
import glob
import io
import json
import logging
import os
import re
import shutil
import socket
import subprocess
import sys
import tempfile
import threading
import time
import traceback
import unittest
from collections.abc import Collection, Container, Coroutine, Iterator, Mapping, Sequence
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Callable, ClassVar, Literal, TypedDict, TypeVar
import webdriver_bidi
from lcov import write_lcov
from lib.constants import OSTREE_IMAGES
from machine import testvm
from PIL import Image
from pixelmatch.contrib.PIL import pixelmatch
_T = TypeVar('_T')
_FT = TypeVar("_FT", bound=Callable[..., Any])
JsonObject = dict[str, Any]
BASE_DIR = os.path.realpath(f'{__file__}/../../..')
TEST_DIR = f'{BASE_DIR}/test'
BOTS_DIR = f'{BASE_DIR}/bots'
os.environ["PATH"] = f"{os.environ.get('PATH')}:{BOTS_DIR}:{TEST_DIR}"
UNEXPECTED_MESSAGE = "FAIL: Test completed, but found unexpected "
PIXEL_TEST_MESSAGE = "Some pixel tests have failed"
__all__ = (
'PIXEL_TEST_MESSAGE',
'TEST_DIR',
'UNEXPECTED_MESSAGE',
'Browser',
'Error',
'MachineCase',
'arg_parser',
'destructive',
'no_retry_when_changed',
'nondestructive',
'onlyImage',
'opts',
'sit',
'skipBrowser',
'skipImage',
'skipOstree',
'test_main',
'timeout',
'todo',
'wait',
)
opts = argparse.Namespace()
opts.sit = False
opts.trace = False
opts.attachments = None
opts.revision = None
opts.address = None
opts.fetch = True
opts.coverage = False
WEBDRIVER_KEYS = {
"Backspace": "\uE003",
"Tab": "\uE004",
"Return": "\uE006",
"Enter": "\uE007",
"Shift": "\uE008",
"Control": "\uE009",
"Alt": "\uE00A",
"Escape": "\uE00C",
"Space": "\uE00D",
"PageUp": "\uE00E",
"PageDown": "\uE00F",
"End": "\uE010",
"Home": "\uE011",
"ArrowLeft": "\uE012",
"ArrowUp": "\uE013",
"ArrowRight": "\uE014",
"ArrowDown": "\uE015",
"Insert": "\uE016",
"Delete": "\uE017",
"Meta": "\uE03D",
"F2": "\uE032",
}
class BrowserLayout(TypedDict):
name: str
theme: Literal["light"] | Literal["dark"]
shell_size: tuple[int, int]
content_size: tuple[int, int]
default_layouts: Sequence[BrowserLayout] = (
{
"name": "desktop",
"theme": "light",
"shell_size": (1920, 1200),
"content_size": (1680, 1130)
},
{
"name": "medium",
"theme": "light",
"shell_size": (1280, 768),
"content_size": (1040, 698)
},
{
"name": "mobile",
"theme": "light",
"shell_size": (414, 1920),
"content_size": (414, 1856)
},
{
"name": "dark",
"theme": "dark",
"shell_size": (1920, 1200),
"content_size": (1680, 1130)
},
{
"name": "rtl",
"theme": "light",
"shell_size": (1920, 1200),
"content_size": (1680, 1130)
},
)
def attach(filename: str, move: bool = False) -> None:
"""Put a file into the attachments directory.
:param filename: file to put in attachments directory
:param move: set this to true to move dynamically generated files which
are not touched by destructive tests. (default False)
"""
if not opts.attachments:
return
dest = os.path.join(opts.attachments, os.path.basename(filename))
if os.path.exists(filename) and not os.path.exists(dest):
if move:
shutil.move(filename, dest)
else:
shutil.copy(filename, dest)
def unique_filename(base: str, ext: str) -> str:
for i in range(20):
if i == 0:
f = f"{base}.{ext}"
else:
f = f"{base}-{i}.{ext}"
if not os.path.exists(f):
return f
return f"{base}.{ext}"
class Browser:
driver: webdriver_bidi.WebdriverBidi
browser: str
layouts: Sequence[BrowserLayout]
current_layout: BrowserLayout | None
port: str | int
def __init__(
self,
address: str,
label: str,
machine: 'MachineCase',
pixels_label: str | None = None,
coverage_label: str | None = None,
port: int | str | None = None
) -> None:
if ":" in address:
self.address, _, self.port = address.rpartition(":")
else:
self.address = address
self.port = 9090
if port is not None:
self.port = port
self.default_user = "admin"
self.label = label
self.pixels_label = pixels_label
self.used_pixel_references = set[str]()
self.coverage_label = coverage_label
self.machine = machine
self.chromium_fake_mouse = False
headless = os.environ.get("TEST_SHOW_BROWSER", '0') == '0'
self.browser = os.environ.get("TEST_BROWSER", "chromium")
if self.browser == "chromium":
self.driver = webdriver_bidi.ChromiumBidi(headless=headless)
elif self.browser == "firefox":
self.driver = webdriver_bidi.FirefoxBidi(headless=headless)
else:
raise ValueError(f"unknown browser {self.browser}")
self.loop = asyncio.new_event_loop()
self.bidi_thread = threading.Thread(target=self.asyncio_loop_thread, args=(self.loop,))
self.bidi_thread.start()
self.run_async(self.driver.start_session())
if opts.trace:
logging.basicConfig(level=logging.INFO)
webdriver_bidi.log_command.setLevel(logging.INFO if opts.trace else logging.WARNING)
test_functions = (Path(__file__).parent / "test-functions.js").read_text()
test_functions = "if (window.ph_select) return; " + test_functions
self.bidi("script.addPreloadScript", quiet=True, functionDeclaration=f"() => {{ {test_functions} }}")
try:
sizzle_js = (Path(__file__).parent.parent.parent / "node_modules/sizzle/dist/sizzle.js").read_text()
sizzle_js = sizzle_js.replace('function assert( fn ) {',
'function assert( fn ) { if (true) return true; else ')
sizzle_js = sizzle_js.replace('context = context || document;', 'context = context || window.document;')
self.bidi("script.addPreloadScript", quiet=True, functionDeclaration=f"() => {{ {sizzle_js} }}")
except FileNotFoundError:
pass
if coverage_label:
self.cdp_command("Profiler.enable")
self.cdp_command("Profiler.startPreciseCoverage", callCount=False, detailed=True)
self.password = "foobar"
self.timeout_factor = float(os.getenv("TEST_TIMEOUT_FACTOR", "1"))
self.timeout = 15
self.failed_pixel_tests = 0
self.allow_oops = False
try:
with open(f'{TEST_DIR}/browser-layouts.json') as fp:
self.layouts = json.load(fp)
except FileNotFoundError:
self.layouts = default_layouts
self.current_layout = None
def _is_running(self) -> bool:
"""True initially, false after calling .kill()"""
return self.driver.bidi_session is not None
def have_test_api(self) -> bool:
"""Check if the browser is running and has a Cockpit page
I.e. are our test-functions.js available? This is only true after
opening cockpit, not for the initial blank page (before login_and_go)
or other URLs like Grafana.
"""
if not self._is_running():
return False
return self.eval_js("!!window.ph_find")
def run_async(self, coro: Coroutine[Any, Any, Any]) -> JsonObject:
"""Run coro in main loop in our BiDi thread
Wait for the result and return it.
"""
return asyncio.run_coroutine_threadsafe(coro, self.loop).result()
@staticmethod
def asyncio_loop_thread(loop: asyncio.AbstractEventLoop) -> None:
asyncio.set_event_loop(loop)
loop.run_forever()
def kill(self) -> None:
if not self._is_running():
return
self.run_async(self.driver.close())
self.loop.call_soon_threadsafe(self.loop.stop)
self.bidi_thread.join()
def bidi(self, method: str, **params: Any) -> webdriver_bidi.JsonObject:
"""Send a Webdriver BiDi command and return the JSON response"""
try:
return self.run_async(self.driver.bidi(method, **params))
except webdriver_bidi.WebdriverError as e:
raise Error(str(e)) from None
def cdp_command(self, method: str, **params: Any) -> webdriver_bidi.JsonObject:
"""Send a Chrome DevTools Protocol command and return the JSON response"""
if self.browser == "chromium":
assert isinstance(self.driver, webdriver_bidi.ChromiumBidi)
reply = self.run_async(self.driver.cdp(method, **params))
if 'error' in reply:
raise Error(str(reply['error'])) from None
return reply['result']
else:
raise webdriver_bidi.WebdriverError("CDP is only supported in Chromium")
def open(self, href: str, cookie: Mapping[str, str] | None = None, tls: bool = False) -> None:
"""Load a page into the browser.
:param href: the path of the Cockpit page to load, such as "/users". Either PAGE or URL needs to be given.
:param cookie: a dictionary object representing a cookie.
:param tls: load the page using https (default False)
Raises:
Error: When a timeout occurs waiting for the page to load.
"""
if href.startswith("/"):
schema = "https" if tls else "http"
href = f"{schema}://{self.address}:{self.port}{href}"
if not self.current_layout and os.environ.get("TEST_SHOW_BROWSER") in [None, "pixels"]:
self.current_layout = self.layouts[0]
size = self.current_layout["shell_size"]
self._set_window_size(size[0], size[1])
if cookie:
c = {**cookie, "value": {"type": "string", "value": cookie["value"]}}
self.bidi("storage.setCookie", cookie=c)
self.switch_to_top()
self.bidi("browsingContext.navigate", context=self.driver.context, url="about:blank", wait="complete")
self.bidi("browsingContext.navigate", context=self.driver.context, url=href, wait="complete", timeout=30)
def set_user_agent(self, ua: str) -> None:
"""Set the user agent of the browser
:param ua: user agent string
:type ua: str
"""
if self.browser == "chromium":
self.cdp_command("Emulation.setUserAgentOverride", userAgent=ua)
else:
raise NotImplementedError
def reload(self, ignore_cache: bool = False) -> None:
"""Reload the current page
:param ignore_cache: if true browser cache is ignored (default False)
:type ignore_cache: bool
"""
self.switch_to_top()
self.wait_js_cond("ph_select('iframe.container-frame').every(e => e.getAttribute('data-loaded'))")
if self.browser == "firefox":
if ignore_cache:
webdriver_bidi.log_command.warning(
"Browser.reload(): ignore_cache==True not yet supported with Firefox, ignoring")
self.bidi("browsingContext.reload", context=self.driver.context, wait="complete")
else:
self.bidi("browsingContext.reload", context=self.driver.context, ignoreCache=ignore_cache,
wait="complete")
self.machine.allow_restart_journal_messages()
def switch_to_frame(self, name: str | None) -> None:
"""Switch to frame in browser tab
Each page has a main frame and can have multiple subframes, usually
iframes.
:param name: frame name
"""
if name is None:
self.switch_to_top()
else:
self.run_async(self.driver.switch_to_frame(name))
def switch_to_top(self) -> None:
"""Switch to the main frame
Switch to the main frame from for example an iframe.
"""
self.driver.switch_to_top()
def allow_download(self) -> None:
"""Allow browser downloads"""
if self.browser == "chromium":
self.cdp_command("Browser.setDownloadBehavior", behavior="allow",
downloadPath=str(self.driver.download_dir))
def upload_files(self, selector: str, files: Sequence[str]) -> None:
"""Upload a local file to the browser
The selector should select the <input type="file"/> element.
Files is a list of absolute paths to files which should be uploaded.
"""
element = self.eval_js(f"ph_find({jsquote(selector)})")
self.bidi("input.setFiles", context=self.driver.context, element=element, files=files)
def eval_js(self, code: str, no_trace: bool = False) -> Any:
"""Execute JS code that returns something
:param code: a string containing JavaScript code
:param no_trace: do not print information about unknown return values (default False)
"""
return self.bidi("script.evaluate", expression=code, quiet=no_trace,
awaitPromise=True, target={"context": self.driver.context})["result"]
def call_js_func(self, func: str, *args: object) -> Any:
"""Call a JavaScript function
:param func: JavaScript function to call
:param args: arguments for the JavaScript function
"""
return self.eval_js("{}({})".format(func, ','.join(map(jsquote, args))))
def set_mock(self, mock: Mapping[str, str], base: str = "") -> None:
"""Replace some DOM elements with mock text
The 'mock' parameter is a dictionary from CSS selectors to the
text that the elements matching the selector should be
replaced with.
XXX - There is no way to easily undo the effects of this
function. There is no coordination with React. This
will improve as necessary.
:param mock: the mock data, see above
:param base: if given, all selectors are relative to this one
"""
self.call_js_func('ph_set_texts', {base + " " + k: v for k, v in mock.items()})
def cookie(self, name: str) -> Mapping[str, object] | None:
"""Retrieve a browser cookie by name
:param name: the name of the cookie
:type name: str
"""
cookies = self.bidi("storage.getCookies", filter={"name": name})["cookies"]
if len(cookies) > 0:
c = cookies[0]
assert c["value"]["type"] == "string"
c["value"] = c["value"]["value"]
return c
return None
def go(self, path_hash: str) -> None:
"""Change URL to a different path or hash
This must be called with either a path (starting with `/`) or hash (starting with `#`).
"""
if path_hash.startswith("#"):
self.call_js_func('cockpit.location.go', path_hash[1:])
elif path_hash.startswith("/"):
self.call_js_func('ph_jump', path_hash)
else:
raise ValueError(f"go({path_hash}): must be called with a path or hash value")
def mouse(
self,
selector: str,
event: str,
x: int | None = None,
y: int | None = None,
btn: int = 0,
*,
ctrlKey: bool = False,
shiftKey: bool = False,
altKey: bool = False,
metaKey: bool = False,
scrollVisible: bool = True,
synthetic: bool = False,
) -> None:
"""Do a mouse event in the browser.
:param selector: the element to interact with
:param type: click, dblclick, mousemove; you can also use "mouseenter" (alias for mousemove) or "mouseleave"
(but this is just a heuristic by moving the mouse 500x500 pixels away; prefer moving to an explicit
target)
:param x, y: coordinates; when not given, default to center of element
:param btn: mouse button to click https://developer.mozilla.org/en-US/docs/Web/API/MouseEvent/buttons
:param crtlKey: press the ctrl key
:param shiftKey: press the shift key
:param altKey: press the alt key
:param metaKey: press the meta key
:param scrollVisible: set to False in rare cases where scrolling an element into view triggers side effects
"""
self.wait_visible(selector)
if x is not None or y is not None:
synthetic = True
if self.browser == "chromium" and self.chromium_fake_mouse:
synthetic = True
if synthetic:
self.call_js_func('ph_mouse', selector, event, x or 0, y or 0, btn, ctrlKey, shiftKey, altKey, metaKey)
return
ev_id = f"pointer-{self.driver.last_id}"
def key(type_: str, name: str) -> JsonObject:
return {"type": "key", "id": ev_id + type_, "actions": [{"type": type_, "value": WEBDRIVER_KEYS[name]}]}
for _retry in range(3):
element = self.call_js_func('ph_find_scroll_into_view' if scrollVisible else 'ph_find', selector)
actions: list[JsonObject] = [
{"type": "pointerMove", "x": 0, "y": 0, "origin": {"type": "element", "element": element}}
]
down = {"type": "pointerDown", "button": btn}
up = {"type": "pointerUp", "button": btn}
if event == "click":
actions.extend([down, up])
elif event == "dblclick":
actions.extend([down, up, down, up])
elif event in ["mousemove", "mouseenter"]:
pass
elif event == "mouseleave":
actions = [{"type": "pointerMove", "x": 500, "y": 500, "origin": "pointer"}]
else:
raise NotImplementedError(f"unknown event {event}")
keys_pre: list[JsonObject] = []
keys_post: list[JsonObject] = []
if altKey:
keys_pre.append(key("keyDown", "Alt"))
keys_post.append(key("keyUp", "Alt"))
if ctrlKey:
keys_pre.append(key("keyDown", "Control"))
keys_post.append(key("keyUp", "Control"))
if shiftKey:
keys_pre.append(key("keyDown", "Shift"))
keys_post.append(key("keyUp", "Shift"))
if metaKey:
keys_pre.append(key("keyDown", "Meta"))
keys_post.append(key("keyUp", "Meta"))
actions = [{
"id": ev_id,
"type": "pointer",
"parameters": {"pointerType": "mouse"},
"actions": actions,
}]
try:
self.bidi("input.performActions", context=self.driver.context, actions=keys_pre + actions + keys_post)
break
except Error as e:
if "no such element" in str(e):
webdriver_bidi.log_command.warning(
"Browser.mouse(%r, %r): retrying after %s", selector, event, e.msg)
scrollVisible = False
else:
raise
def click(self, selector: str) -> None:
"""Click on a ui element
:param selector: the selector to click on
"""
self.mouse(selector + ":not([disabled]):not([aria-disabled=true])", "click")
def val(self, selector: str) -> Any:
"""Get the value attribute of a selector.
:param selector: the selector to get the value of
"""
self.wait_visible(selector)
return self.call_js_func('ph_val', selector)
def set_val(self, selector: str, val: object) -> None:
"""Set the value attribute of a non disabled DOM element.
This also emits a change DOM change event.
:param selector: the selector to set the value of
:param val: the value to set
"""
self.wait_visible(selector + ':not([disabled]):not([aria-disabled=true])')
self.call_js_func('ph_set_val', selector, val)
def text(self, selector: str) -> str:
"""Get an element's textContent value.
:param selector: the selector to get the value of
"""
self.wait_visible(selector)
return self.call_js_func('ph_text', selector) or ''
def attr(self, selector: str, attr: str) -> Any:
"""Get the value of a given attribute of an element.
:param selector: the selector to get the attribute of
:param attr: the DOM element attribute
"""
self._wait_present(selector)
return self.call_js_func('ph_attr', selector, attr)
def set_attr(self, selector: str, attr: str, val: object) -> None:
"""Set an attribute value of an element.
:param selector: the selector
:param attr: the element attribute
:param val: the value of the attribute
"""
self._wait_present(selector + ':not([disabled]):not([aria-disabled=true])')
self.call_js_func('ph_set_attr', selector, attr, val)
def get_checked(self, selector: str) -> bool:
"""Get checked state of a given selector.
:param selector: the selector
:return: the checked state
"""
self.wait_visible(selector + ':not([disabled]):not([aria-disabled=true])')
return self.call_js_func('ph_get_checked', selector)
def set_checked(self, selector: str, val: bool) -> None:
"""Set checked state of a given selector.
:param selector: the selector
:param val: boolean value to enable or disable checkbox
"""
checked = self.get_checked(selector)
if checked != val:
self.click(selector)
def focus(self, selector: str) -> None:
"""Set focus on selected element.
:param selector: the selector
"""
self.wait_visible(selector + ':not([disabled]):not([aria-disabled=true])')
self.call_js_func('ph_focus', selector)
def blur(self, selector: str) -> None:
"""Remove keyboard focus from selected element.
:param selector: the selector
"""
self.wait_visible(selector + ':not([disabled]):not([aria-disabled=true])')
self.call_js_func('ph_blur', selector)
def input_text(self, text: str) -> None:
actions: list[JsonObject] = []
for c in text:
if c == '\n':
c = WEBDRIVER_KEYS["Enter"]
actions.append({"type": "keyDown", "value": c})
actions.append({"type": "keyUp", "value": c})
self.bidi("input.performActions", context=self.driver.context, actions=[
{"type": "key", "id": "key-0", "actions": actions}])
def key(self, name: str, repeat: int = 1, modifiers: list[str] | None = None) -> None:
"""Press and release a named keyboard key.
Use this function to input special characters or modifiers.
:param name: ASCII value or key name like "Enter", "Delete", or "ArrowLeft" (entry in WEBDRIVER_KEYS)
:param repeat: number of times to repeat this key (default 1)
:param modifiers: "Shift", "Control", "Alt"
"""
actions: list[JsonObject] = []
actions_pre: list[JsonObject] = []
actions_post: list[JsonObject] = []
keycode = WEBDRIVER_KEYS.get(name, name)
for m in (modifiers or []):
actions_pre.append({"type": "keyDown", "value": WEBDRIVER_KEYS[m]})
actions_post.append({"type": "keyUp", "value": WEBDRIVER_KEYS[m]})
for _ in range(repeat):
actions.append({"type": "keyDown", "value": keycode})
actions.append({"type": "keyUp", "value": keycode})
self.bidi("input.performActions", context=self.driver.context, actions=[
{"type": "key", "id": "key-0", "actions": actions_pre + actions + actions_post}])
def paste(self, val: str) -> None:
"""Simulate pasting a string into the currently focused element.
:param val: The text to paste. It will be left in the clipboard.
"""
self.call_js_func("((val) => navigator.clipboard.writeText(val))", val)
self.key("v", modifiers=["Control"])
def select_from_dropdown(self, selector: str, value: object) -> None:
"""For an actual <select> HTML component"""
self.wait_visible(selector + ':not([disabled]):not([aria-disabled=true])')
text_selector = f"{selector} option[value='{value}']"
self._wait_present(text_selector)
self.set_val(selector, value)
self.wait_val(selector, value)
def select_PF(self, selector: str, value: str, menu_class: str = ".pf-v6-c-menu") -> None:
"""For a PatternFly Select-like component
For things like <Select> or <TimePicker>. Unfortunately none of them render as an actual <select>, but a
<button> or <div> with a detached menu div (which can even be in the parent).
For similar components like the deprecated <Select> you can specify a different menu class.
"""
self.click(selector)
self.click(f"{menu_class} button:contains('{value}')")
self.wait_not_present(menu_class)
def click_button(self, compid: str, prefix: str = "", component: str = "PF6/Button") -> None:
"""Click on a button identified by its OUIA component id."""
self.click(f'{prefix} [data-ouia-component-type="{component}"][data-ouia-component-id="{compid}"]')
def set_input_text(
self, selector: str, val: str, append: bool = False, value_check: bool = True, blur: bool = True, paste: bool = False
) -> None:
self.focus(selector)
if not append:
self.key("a", modifiers=["Control"])
if val == "":
self.key("Backspace")
elif paste:
self.paste(val)
else:
self.input_text(val)
if blur:
self.blur(selector)
if value_check:
self.wait_val(selector, val)
def set_file_autocomplete_val(self, group_identifier: str, location: str) -> None:
self.set_input_text(f"{group_identifier} .pf-v6-c-menu-toggle input", location)
self.wait_text(".pf-v6-c-menu ul li:nth-child(1) button", location)
self.click(".pf-v6-c-menu ul li:nth-child(1) button")
self.wait_not_present(".pf-v6-c-menu")
self.wait_val(f"{group_identifier} .pf-v6-c-menu-toggle input", location)
def count(self, selector: str) -> int:
"""Returns the total count of matches DOM elements.
:param selector: the selector
"""
return self.call_js_func('ph_count', selector)
def wait_count(self, selector: str, total: int) -> None:
"""Wait until the total count matches the selected DOM elements.
:param selector: the selector
:param total: the total amount of elements to wait for
"""
return self.wait_js_func("ph_count_check", selector, total)
@contextlib.contextmanager
def wait_timeout(self, timeout: int) -> Iterator[None]:
old_timeout = self.timeout
self.timeout = timeout
try:
yield
finally:
self.timeout = old_timeout
def wait(self, predicate: Callable[[], _T | None]) -> _T:
for _ in range(int(self.timeout * self.timeout_factor * 5)):
val = predicate()
if val:
return val
time.sleep(0.2)
raise Error('timed out waiting for predicate to become true')
def wait_js_cond(self, cond: str, error_description: str = "null") -> None:
timeout = self.timeout * self.timeout_factor
start = time.time()
last_error = None
for _retry in range(5):
try:
self.bidi("script.evaluate",
expression=f"ph_wait_cond(() => {cond}, {timeout * 1000}, {error_description})",
awaitPromise=True, timeout=timeout + 5, target={"context": self.driver.context})
duration = time.time() - start
percent = int(duration / timeout * 100)
if percent >= 50:
print(f"WARNING: Waiting for {cond} took {duration:.1f} seconds, "
f"which is {percent}% of the timeout.")
return
except Error as e:
last_error = e
if any(pattern in str(e) for pattern in [
"is not a function",
"Execution context was destroyed",
"Cannot find context",
"Inspected target navigated or closed",
"execution contexts cleared",
"MessageHandlerFrame' destroyed",
"DiscardedBrowsingContextError",
"ph_wait_cond is not defined",
]):
if time.time() - start < timeout:
webdriver_bidi.log_command.info("wait_js_cond: Ignoring/retrying %r", e)
time.sleep(1)
continue
break
assert last_error
raise Error(f"timeout\nwait_js_cond({cond}): {last_error.msg}") from None
def wait_js_func(self, func: str, *args: object) -> None:
self.wait_js_cond("{}({})".format(func, ','.join(map(jsquote, args))))
def is_present(self, selector: str) -> bool:
return self.call_js_func('ph_is_present', selector)
def _wait_present(self, selector: str) -> None:
self.wait_js_func('ph_is_present', selector)
def wait_not_present(self, selector: str) -> None:
self.wait_js_func('!ph_is_present', selector)
def is_visible(self, selector: str) -> bool:
return self.call_js_func('ph_is_visible', selector)
def wait_visible(self, selector: str) -> None:
self._wait_present(selector)
self.wait_js_func('ph_is_visible', selector)
def wait_val(self, selector: str, val: object) -> None:
self.wait_visible(selector)
self.wait_js_func('ph_has_val', selector, val)
def wait_not_val(self, selector: str, val: str) -> None:
self.wait_visible(selector)
self.wait_js_func('!ph_has_val', selector, val)
def wait_attr(self, selector: str, attr: str, val: str | None) -> None:
self._wait_present(selector)
self.wait_js_func('ph_has_attr', selector, attr, val)
def wait_attr_contains(self, selector: str, attr: str, val: str) -> None:
self._wait_present(selector)
self.wait_js_func('ph_attr_contains', selector, attr, val)
def wait_attr_not_contains(self, selector: str, attr: str, val: object) -> None:
self._wait_present(selector)
self.wait_js_func('!ph_attr_contains', selector, attr, val)
def wait_not_attr(self, selector: str, attr: str, val: object) -> None:
self._wait_present(selector)
self.wait_js_func('!ph_has_attr', selector, attr, val)
def wait_not_visible(self, selector: str) -> None:
self.wait_js_func('!ph_is_visible', selector)
def wait_in_text(self, selector: str, text: str) -> None:
self.wait_visible(selector)
self.wait_js_cond(f"ph_in_text({jsquote(selector)},{jsquote(text)})",
error_description="() => 'actual text: ' + ph_text(%s)" % jsquote(selector))
def wait_not_in_text(self, selector: str, text: str) -> None:
self.wait_visible(selector)
self.wait_js_func('!ph_in_text', selector, text)
def wait_collected_text(self, selector: str, text: str) -> None:
self.wait_js_func('ph_collected_text_is', selector, text)
def wait_text(self, selector: str, text: str) -> None:
self.wait_visible(selector)
self.wait_js_cond(f"ph_text_is({jsquote(selector)},{jsquote(text)})",
error_description="() => 'actual text: ' + ph_text(%s)" % jsquote(selector))
def wait_text_not(self, selector: str, text: str) -> None:
self.wait_visible(selector)
self.wait_js_func('!ph_text_is', selector, text)
def wait_text_matches(self, selector: str, pattern: str) -> None:
self.wait_visible(selector)
self.wait_js_func('ph_text_matches', selector, pattern)
def wait_popup(self, elem_id: str) -> None:
"""Wait for a popup to open.
:param id: the 'id' attribute of the popup.
"""
self.wait_visible('#' + elem_id)
def wait_language(self, lang: str) -> None:
parts = lang.split("-")
code_1 = parts[0]
code_2 = parts[0]
if len(parts) > 1:
code_2 += "_" + parts[1].upper()
self.wait_js_cond(f"cockpit.language == '{code_1}' || cockpit.language == '{code_2}'")
def dialog_cancel(self, sel: str, button: str = "button[data-dismiss='modal']") -> None:
self.click(sel + " " + button)
self.wait_not_visible(sel)
def enter_page(self, path: str, host: str | None = None, reconnect: bool = True) -> None:
"""Wait for a page to become current.
:param path: The identifier the page. This is a string starting with "/"
:type path: str
:param host: The host to connect too
:type host: str
:param reconnect: Try to reconnect
:type reconnect: bool
"""
assert path.startswith("/")
if host:
frame = host + path
else:
frame = "localhost" + path
frame = "cockpit1:" + frame
self.switch_to_top()
def wait_no_curtain() -> None:
self.wait_js_cond('!ph_is_present(".curtains-ct") || !ph_is_visible(".curtains-ct")')
while True:
try:
self._wait_present("iframe.container-frame[name='%s'][data-loaded]" % frame)
wait_no_curtain()
self.wait_visible("iframe.container-frame[name='%s']" % frame)
break
except Error as ex:
if reconnect and ex.msg.startswith('timeout'):
reconnect = False
if self.is_present("#machine-reconnect"):
self.click("#machine-reconnect")
wait_no_curtain()
continue
raise
self.switch_to_frame(frame)
self.wait_visible("body")
def leave_page(self) -> None:
self.switch_to_top()
def try_login(
self,
user: str | None = None,
password: str | None = None,
*,
superuser: bool | None = True,
legacy_authorized: bool | None = None
) -> None:
"""Fills in the login dialog and clicks the button.
This differs from login_and_go() by not expecting any particular result.
:param user: the username to login with
:type user: str
:param password: the password of the user
:type password: str
:param superuser: determines whether the new session will try to get Administrative Access (default true)
:type superuser: bool
:param legacy_authorized: old versions of the login dialog that still
have the "[ ] Reuse my password for magic things" checkbox. Such a
dialog is encountered when testing against old bastion hosts, for
example.
"""
if user is None:
user = self.default_user
if password is None:
password = self.password
self.wait_visible("#login")
self.set_val('#login-user-input', user)
self.set_val('#login-password-input', password)
if legacy_authorized is not None:
self.set_checked('#authorized-input', legacy_authorized)
if superuser is not None:
self.eval_js('window.localStorage.setItem("superuser:{}", "{}");'.format(user, "any" if superuser else "none"))
self.click('#login-button')
def login_and_go(
self,
path: str | None = None,
*,
user: str | None = None,
host: str | None = None,
superuser: bool | None = True,
urlroot: str | None = None,
tls: bool = False,
password: str | None = None,
legacy_authorized: bool | None = None
) -> None:
"""Fills in the login dialog, clicks the button and navigates to the given path
:param user: the username to login with
:type user: str
:param password: the password of the user
:type password: str
:param superuser: determines whether the new session will try to get Administrative Access (default true)
:type superuser: bool
:param legacy_authorized: old versions of the login dialog that still
have the "[ ] Reuse my password for magic things" checkbox. Such a
dialog is encountered when testing against old bastion hosts, for
example.
"""
href = path
if not href:
href = "/"
if urlroot:
href = urlroot + href
if host:
href = "/@" + host + href
self.open(href, tls=tls)
self.try_login(user=user, password=password, superuser=superuser, legacy_authorized=legacy_authorized)
self.wait_visible('#content')
if path:
self.enter_page(path.split("#")[0], host=host)
def logout(self) -> None:
self.assert_no_oops()
self.switch_to_top()
self.wait_visible("#toggle-menu")
if self.is_present("button#machine-reconnect") and self.is_visible("button#machine-reconnect"):
self.click("button#machine-reconnect")
else:
self.open_session_menu()
try:
self.mouse('#logout', "click", synthetic=True)
except RuntimeError as e:
if "Execution context was destroyed" not in str(e):
raise
self.wait_visible('#login')
self.machine.allow_restart_journal_messages()
def relogin(
self,
path: str | None = None,
user: str | None = None,
*,
password: str | None = None,
superuser: bool | None = None,
wait_remote_session_machine: testvm.Machine | None = None
) -> None:
self.logout()
if wait_remote_session_machine:
wait_remote_session_machine.execute("while pgrep -af '[c]ockpit.beiboot'; do sleep 1; done")
self.try_login(user=user, password=password, superuser=superuser)
self.wait_visible('#content')
if path:
if path.startswith("/@"):
host = path[2:].split("/")[0]
else:
host = None
self.enter_page(path.split("#")[0], host=host)
def open_session_menu(self) -> None:
self.wait_visible("#toggle-menu")
if (self.attr("#toggle-menu", "aria-expanded") != "true"):
self.click("#toggle-menu")
self.wait_visible("button.display-language-menu")
def layout_is_mobile(self) -> bool:
if not self.current_layout:
return False
return self.current_layout["shell_size"][0] < 420
def open_superuser_dialog(self) -> None:
if self.layout_is_mobile():
self.open_session_menu()
self.click("#super-user-indicator-mobile button")
else:
self.click("#super-user-indicator button")
def check_superuser_indicator(self, expected: str) -> None:
if self.layout_is_mobile():
self.open_session_menu()
self.wait_text("#super-user-indicator-mobile", expected)
self.click("#toggle-menu")
else:
self.wait_text("#super-user-indicator", expected)
def become_superuser(
self,
user: str | None = None,
password: str | None = None,
passwordless: bool | None = False
) -> None:
with self.driver.restore_context():
self.switch_to_top()
self.open_superuser_dialog()
if user is None:
user = "root" if "suse" in self.machine.image else "admin"
if passwordless:
self.wait_in_text("div[role=dialog]", "Administrative access")
self.wait_in_text("div[role=dialog] .pf-v6-c-modal-box__body", "You now have administrative access.")
self.click("div[role=dialog] .pf-v6-c-modal-box__footer button")
else:
self.wait_in_text("div[role=dialog]", "Switch to administrative access")
self.wait_in_text("div[role=dialog]", "Password")
self.set_input_text("div[role=dialog] input", password or "foobar")
self.click("div[role=dialog] button.pf-m-primary")
self.wait_not_present("div[role=dialog]")
self.check_superuser_indicator("Administrative access")
def drop_superuser(self) -> None:
with self.driver.restore_context():
self.switch_to_top()
self.open_superuser_dialog()
self.wait_in_text("div[role=dialog]", "Switch to limited access")
self.click("div[role=dialog] button.pf-m-primary")
self.wait_not_present("div[role=dialog]")
self.check_superuser_indicator("Limited access")
def click_system_menu(self, path: str, enter: bool = True) -> None:
"""Click on a "System" menu entry with given URL path
Enters the given target frame afterwards, unless enter=False is given
(useful for remote hosts).
"""
self.switch_to_top()
self.click(f"#host-apps a[href='{path}']")
if enter:
self.enter_page(path.split('#')[0].rstrip('/'))
def get_pf_progress_value(self, progress_bar_sel: str) -> int:
"""Get numeric value of a PatternFly <ProgressBar> component"""
sel = progress_bar_sel + " .pf-v6-c-progress__indicator"
self.wait_visible(sel)
self.wait_attr_contains(sel, "style", "width:")
style = self.attr(sel, "style")
m = re.search(r"width: (\d+)%;", style)
assert m is not None
return int(m.group(1))
def get_raw_pf_progress_value(self, progress_bar_sel: str) -> int:
"""
Get raw numeric value of a PatternFly <ProgressBar> component
Value from aria-valuenow which is set by PatternFly
"""
sel = progress_bar_sel + " .pf-v6-c-progress__bar"
self.wait_visible(f"{sel}[aria-valuenow]")
component_value = self.attr(sel, "aria-valuenow")
assert component_value is not None
return int(component_value)
def start_machine_troubleshoot(
self,
new: bool = False,
known_host: bool = False,
password: str | None = None,
expect_closed_dialog: bool = True,
expect_warning: bool = True,
expect_curtain: bool = True
) -> None:
if expect_curtain:
self.click('#machine-troubleshoot')
if not new and expect_warning:
self.wait_visible('#hosts_connect_server_dialog')
self.click("#hosts_connect_server_dialog button.pf-m-warning")
self.wait_visible('#hosts_setup_server_dialog')
if new:
self.wait_text("#hosts_setup_server_dialog button.pf-m-primary", "Add")
self.click("#hosts_setup_server_dialog button.pf-m-primary")
if expect_warning:
self.wait_visible('#hosts_connect_server_dialog')
self.click("#hosts_connect_server_dialog button.pf-m-warning")
if not known_host:
self.wait_in_text('#hosts_setup_server_dialog', "You are connecting to")
self.wait_in_text('#hosts_setup_server_dialog', "for the first time.")
self.wait_text("#hosts_setup_server_dialog button.pf-m-primary", "Trust and add host")
self.click("#hosts_setup_server_dialog button.pf-m-primary")
if password:
self.wait_in_text('#hosts_setup_server_dialog', "Unable to log in")
self.set_input_text('#login-custom-password', password)
self.wait_text("#hosts_setup_server_dialog button.pf-m-primary", "Log in")
self.click("#hosts_setup_server_dialog button.pf-m-primary")
if expect_closed_dialog:
self.wait_not_present('#hosts_setup_server_dialog')
def add_machine(self, address: str, known_host: bool = False, password: str | None = "foobar",
expect_warning: bool = True) -> None:
self.switch_to_top()
self.go(f"/@{address}")
self.start_machine_troubleshoot(new=True,
known_host=known_host,
password=password,
expect_warning=expect_warning)
self.enter_page("/system", host=address)
def grant_permissions(self, *args: str) -> None:
"""Grant permissions to the browser"""
for perm in args:
self.bidi("permissions.setPermission", descriptor={"name": perm}, state="granted",
origin=f"http://{self.address}:{self.port}")
def snapshot(self, title: str, label: str | None = None) -> None:
"""Take a snapshot of the current screen and save it as a PNG and HTML.
Arguments:
title: Used for the filename.
"""
if self._is_running():
filename = unique_filename(f"{label or self.label}-{title}", "png")
try:
ret = self.bidi("browsingContext.captureScreenshot", quiet=True,
context=self.driver.top_context, origin="document")
with open(filename, 'wb') as f:
f.write(base64.standard_b64decode(ret["data"]))
attach(filename, move=True)
print("Wrote screenshot to " + filename)
except Error as e:
print("Screenshot not available:", e)
filename = unique_filename(f"{label or self.label}-{title}", "html")
try:
html = self.eval_js("document.documentElement.outerHTML", no_trace=True)
with open(filename, 'wb') as f:
f.write(html.encode())
attach(filename, move=True)
print("Wrote HTML dump to " + filename)
except Error as e:
print("HTML dump not available:", e)
def _set_window_size(self, width: int, height: int) -> None:
self.bidi("browsingContext.setViewport", context=self.driver.top_context,
viewport={"width": width, "height": height})
def _set_emulated_media_theme(self, name: str) -> None:
if self.browser == "chromium":
self.cdp_command("Emulation.setEmulatedMedia", features=[{'name': 'prefers-color-scheme', 'value': name}])
def _set_direction(self, direction: str) -> None:
with self.driver.restore_context():
if self.is_present("#shell-page"):
self.switch_to_top()
self.set_attr("#shell-page", "dir", direction)
self.set_attr("html", "dir", direction)
def set_layout(self, name: str) -> None:
layout = next(lo for lo in self.layouts if lo["name"] == name)
if layout != self.current_layout:
if layout["name"] == "rtl":
self._set_direction("rtl")
elif layout["name"] != "rtl" and self.current_layout and self.current_layout["name"] == "rtl":
self._set_direction("ltr")
self.current_layout = layout
size = layout["shell_size"]
self._set_window_size(size[0], size[1])
self._adjust_window_for_fixed_content_size()
self._set_emulated_media_theme(layout["theme"])
def _adjust_window_for_fixed_content_size(self) -> None:
assert self.current_layout is not None
if self.eval_js("window.name").startswith("cockpit1:"):
shell_size = self.current_layout["shell_size"]
want_size = self.current_layout["content_size"]
have_size = self.eval_js("[ document.body.offsetWidth, document.body.offsetHeight ]")
delta = (want_size[0] - have_size[0], want_size[1] - have_size[1])
if delta[0] != 0 or delta[1] != 0:
self._set_window_size(shell_size[0] + delta[0], shell_size[1] + delta[1])
def assert_pixels_in_current_layout(
self,
selector: str,
key: str,
*,
ignore: Collection[str] = (),
mock: Mapping[str, str] | None = None,
sit_after_mock: bool = False,
scroll_into_view: str | None = None,
wait_animations: bool = True,
wait_delay: float = 0.5,
chrome_hack_double_shots: bool = False,
abs_tolerance: float = 20
) -> None:
"""Compare the given element with its reference in the current layout"""
if not (Image and self.pixels_label):
return
assert self.current_layout is not None
self._adjust_window_for_fixed_content_size()
self.call_js_func('ph_scrollIntoViewIfNeeded', scroll_into_view or selector)
self.call_js_func('ph_blur_active')
if wait_animations:
time.sleep(wait_delay)
self.wait_js_cond('ph_count_animations(%s) == 0' % jsquote(selector))
if mock is not None:
self.set_mock(mock, base=selector)
if sit_after_mock:
sit()
rect = self.call_js_func('ph_element_clip', selector)
def relative_clips(sels: Collection[str]) -> Collection[tuple[int, int, int, int]]:
return [(
r['x'] - rect['x'],
r['y'] - rect['y'],
r['x'] - rect['x'] + r['width'],
r['y'] - rect['y'] + r['height'])
for r in self.call_js_func('ph_selector_clips', sels)]
reference_dir = os.path.join(TEST_DIR, 'reference')
if not os.path.exists(os.path.join(reference_dir, '.git')):
raise SystemError("Pixel test references are missing, please run: test/common/pixel-tests pull")
ignore_rects = relative_clips([f"{selector} {item}" for item in ignore])
base = self.pixels_label + "-" + key
if self.current_layout != self.layouts[0]:
base += "-" + self.current_layout["name"]
filename = base + "-pixels.png"
ref_filename = os.path.join(reference_dir, filename)
self.used_pixel_references.add(ref_filename)
rect["type"] = "box"
ret = self.bidi("browsingContext.captureScreenshot", quiet=True,
context=self.driver.top_context,
clip=rect)
if chrome_hack_double_shots:
ret1 = ret
ret = self.bidi("browsingContext.captureScreenshot", quiet=True,
context=self.driver.top_context,
clip=rect)
if ret1["data"] != ret["data"]:
print("WARNING: Inconsistent screenshots for", base)
png_now = base64.standard_b64decode(ret["data"])
png_ref = os.path.exists(ref_filename) and open(ref_filename, "rb").read()
if not png_ref:
with open(filename, 'wb') as f:
f.write(png_now)
attach(filename, move=True)
print("New pixel test reference " + filename)
self.failed_pixel_tests += 1
else:
img_now = Image.open(io.BytesIO(png_now))
img_ref = Image.open(io.BytesIO(png_ref))
img_delta = Image.new("RGBA",
(max(img_now.size[0], img_ref.size[0]), max(img_now.size[1], img_ref.size[1])),
(255, 0, 0, 255))
def img_eq(ref: Image.Image, now: Image.Image, delta: Image.Image) -> str | None:
if ref.size != now.size:
return f"size mismatch: reference {ref.size[0]}x{ref.size[1]}, now {now.size[0]}x{now.size[1]}"
diff_count = pixelmatch(
img1=ref, img2=now, output=delta, aa_color=(222, 172, 0), masked_areas=ignore_rects, alpha=0.6
)
if diff_count > abs_tolerance:
return f"differs by {diff_count} pixels, tolerance {abs_tolerance}"
return None
reason = img_eq(img_ref, img_now, img_delta)
if reason:
img_now.save(filename)
attach(filename, move=True)
ref_filename_for_attach = base + "-reference.png"
img_ref.save(ref_filename_for_attach)
attach(ref_filename_for_attach, move=True)
delta_filename = base + "-delta.png"
img_delta.save(delta_filename)
attach(delta_filename, move=True)
print(f"Pixel test {base} failed: {reason}")
self.failed_pixel_tests += 1
def assert_pixels(
self,
selector: str,
key: str,
*,
ignore: Collection[str] = (),
mock: Mapping[str, str] | None = None,
sit_after_mock: bool = False,
skip_layouts: Container[str] = (),
scroll_into_view: str | None = None,
wait_animations: bool = True,
wait_after_layout_change: bool = False,
wait_delay: float = 0.5,
layout_change_hook: Callable[[], None] | None = None,
chrome_hack_double_shots: bool = False,
abs_tolerance: float = 20
) -> None:
"""Compare the given element with its reference in all layouts"""
if ignore is None:
ignore = []
if not (Image and self.pixels_label):
return
if not self.is_present("#login-details") and not self.is_present("#system-terminal-page"):
classes = self.attr("body", "class")
self.set_attr("body", "class", f"{classes} pixel-test")
self.bidi("input.performActions", context=self.driver.context, actions=[{
"id": "move-away",
"type": "pointer",
"parameters": {"pointerType": "mouse"},
"actions": [{"type": "pointerMove", "x": 2000, "y": 0, "origin": "viewport"}]
}])
if self.current_layout:
previous_layout = self.current_layout["name"]
for layout in self.layouts:
if layout["name"] not in skip_layouts:
self.set_layout(layout["name"])
if wait_after_layout_change:
time.sleep(wait_delay)
if layout_change_hook:
layout_change_hook()
self.assert_pixels_in_current_layout(selector, key, ignore=ignore,
mock=mock, sit_after_mock=sit_after_mock,
scroll_into_view=scroll_into_view,
wait_animations=wait_animations,
wait_delay=wait_delay,
chrome_hack_double_shots=chrome_hack_double_shots,
abs_tolerance=abs_tolerance)
self.set_layout(previous_layout)
def assert_no_unused_pixel_test_references(self) -> None:
"""Check whether all reference images in test/reference have been used."""
if not (Image and self.pixels_label):
return
pixel_references = set(glob.glob(os.path.join(TEST_DIR, "reference", self.pixels_label + "-*-pixels.png")))
unused = pixel_references - self.used_pixel_references
for u in unused:
print("Unused reference image " + os.path.basename(u))
self.failed_pixel_tests += 1
def get_js_log(self) -> Sequence[str]:
"""Return the current javascript log"""
if self._is_running():
return [str(log) for log in self.driver.logs]
return []
def copy_js_log(self, title: str, label: str | None = None) -> None:
"""Copy the current javascript log"""
logs = self.get_js_log()
if logs:
filename = unique_filename(f"{label or self.label}-{title}", "js.log")
with open(filename, 'wb') as f:
f.write('\n'.join(logs).encode())
attach(filename, move=True)
print("Wrote JS log to " + filename)
def write_coverage_data(self) -> None:
if self.coverage_label and self._is_running():
coverage = self.cdp_command("Profiler.takePreciseCoverage")
write_lcov(coverage['result'], self.coverage_label)
def assert_no_oops(self) -> None:
if self.allow_oops:
return
if self.have_test_api():
self.switch_to_top()
if self.eval_js("!!document.getElementById('navbar-oops')"):
assert not self.is_visible("#navbar-oops"), "Cockpit shows an Oops"
class MachineCase(unittest.TestCase):
image = testvm.DEFAULT_IMAGE
libexecdir: str | None = None
sshd_socket: str | None = None
runner = None
machine: testvm.Machine
machines: Mapping[str, testvm.Machine]
machine_class: type[testvm.VirtMachine] | None = None
browser: Browser
network = None
journal_start: str | None = None
provision: ClassVar[Mapping[str, Mapping[str, Any]] | None] = None
global_machine = None
@classmethod
def get_global_machine(cls) -> testvm.Machine:
if cls.global_machine:
return cls.global_machine
cls.global_machine = cls().new_machine(restrict=True, cleanup=False)
if opts.trace:
print(f"Starting global machine {cls.global_machine.label}")
cls.global_machine.start()
return cls.global_machine
@classmethod
def kill_global_machine(cls) -> None:
if cls.global_machine:
cls.global_machine.kill()
cls.global_machine = None
def label(self) -> str:
return self.__class__.__name__ + '-' + self._testMethodName
def new_machine(
self,
image: str | None = None,
forward: Mapping[str, int] | None = None,
restrict: bool = True,
cleanup: bool = True,
inherit_machine_class: bool = True,
**kwargs: Any
) -> testvm.Machine:
machine_class = (inherit_machine_class and self.machine_class) or testvm.VirtMachine
if opts.address:
if forward:
raise unittest.SkipTest("Cannot run this test when specific machine address is specified")
machine = testvm.Machine(address=opts.address, image=image or self.image,
verbose=opts.trace, browser=opts.browser)
if cleanup:
self.addCleanup(machine.disconnect)
else:
if image is None:
image = os.path.join(TEST_DIR, "images", self.image)
if not os.path.exists(image):
raise FileNotFoundError("Can't run tests without a prepared image; use test/image-prepare")
if not self.network:
network = testvm.VirtNetwork(image=image)
if cleanup:
self.addCleanup(network.kill)
self.network = network
networking = self.network.host(restrict=restrict, forward=forward or {})
machine = machine_class(verbose=opts.trace, networking=networking, image=image, **kwargs)
image_file = machine.image_file
if opts.fetch and not os.path.exists(image_file):
machine.pull(image_file)
if cleanup:
self.addCleanup(machine.kill)
return machine
def new_browser(self, machine: testvm.Machine | None = None, coverage: bool = False) -> Browser:
if machine is None:
machine = self.machine
label = self.label() + "-" + machine.label
pixels_label = None
if os.environ.get("TEST_BROWSER", "chromium") == "chromium" and not self.is_devel_build():
try:
with open(f'{TEST_DIR}/reference-image') as fp:
reference_image = fp.read().strip()
except FileNotFoundError:
pass
else:
if machine.image == reference_image:
pixels_label = self.label()
browser = Browser(machine.web_address,
label=label, pixels_label=pixels_label, coverage_label=self.label() if coverage else None,
port=machine.web_port, machine=self)
self.addCleanup(browser.kill)
return browser
def getError(self) -> str | None:
errors = []
assert hasattr(self, '_outcome')
if hasattr(self._outcome, 'errors'):
assert hasattr(self, '_feedErrorsToResult')
result = self.defaultTestResult()
errors = result.errors
self._feedErrorsToResult(result, self._outcome.errors)
elif hasattr(self._outcome, 'result') and getattr(self._outcome.result, '_excinfo', None) is not None:
if isinstance(self._outcome.result._excinfo, str):
return self._outcome.result._excinfo
else:
assert isinstance(self._outcome.result._excinfo, list)
return str(self._outcome.result._excinfo[0])
else:
errors = getattr(self._outcome.result, 'errors', []) + getattr(self._outcome.result, 'failures', [])
try:
return errors[0][1]
except IndexError:
return None
def is_nondestructive(self) -> bool:
test_method = getattr(self.__class__, self._testMethodName)
return get_decorator(test_method, self.__class__, "nondestructive")
def is_devel_build(self) -> bool:
return os.environ.get('NODE_ENV') == 'development'
def disable_preload(self, *packages: str, machine: testvm.Machine | None = None) -> None:
if machine is None:
machine = self.machine
for pkg in packages:
machine.write(f"/etc/cockpit/{pkg}.override.json", '{ "preload": [ ] }')
def enable_preload(self, package: str, *pages: str) -> None:
pages_str = ', '.join(f'"{page}"' for page in pages)
self.machine.write(f"/etc/cockpit/{package}.override.json", f'{{ "preload": [ {pages_str} ] }}')
def system_before(self, version: int) -> bool:
try:
v = self.machine.execute("""rpm -q --qf '%{V}' cockpit-system ||
dpkg-query -W -f '${source:Upstream-Version}' cockpit-system ||
(pacman -Q cockpit | cut -f2 -d' ' | cut -f1 -d-)
""").split(".")
except subprocess.CalledProcessError:
return False
return int(v[0]) < version
def setUp(self) -> None:
self.allowed_messages = self.default_allowed_messages
self.allowed_console_errors = self.default_allowed_console_errors
self.allow_core_dumps = False
if os.getenv("MACHINE"):
opts.address = os.getenv("MACHINE")
if self.is_nondestructive():
pass
elif os.getenv("DESTRUCTIVE") and not self.is_nondestructive():
print("Run destructive test, be careful, may lead to unpredictable state of machine")
else:
raise unittest.SkipTest("Skip destructive test by default")
if os.getenv("BROWSER"):
opts.browser = os.getenv("BROWSER")
if os.getenv("TRACE"):
opts.trace = True
if os.getenv("SIT"):
opts.sit = True
if opts.address and self.provision is not None:
raise unittest.SkipTest("Cannot provision multiple machines if a specific machine address is specified")
self.machines = {}
provision = self.provision or {'machine1': {}}
self.tmpdir = tempfile.mkdtemp()
self.vm_tmpdir = "/var/lib/cockpittest"
if self.is_nondestructive() and not opts.address:
if self.provision:
raise unittest.SkipTest("Cannot provision machines if test is marked as nondestructive")
self.machine = self.machines['machine1'] = MachineCase.get_global_machine()
else:
MachineCase.kill_global_machine()
first_machine = True
for key in sorted(provision.keys()):
options = dict(provision[key])
options.pop('address', None)
options.pop('address6', None)
options.pop('dns', None)
options.pop('dhcp', None)
machine = self.new_machine(**options)
self.machines[key] = machine
if first_machine:
first_machine = False
self.machine = machine
if opts.trace:
print(f"Starting {key} {machine.label}")
machine.start()
self.danger_btn_class = '.pf-m-danger'
self.primary_btn_class = '.pf-m-primary'
self.default_btn_class = '.pf-m-secondary'
for key in self.machines.keys():
machine = self.machines[key]
machine.wait_boot()
address = provision[key].get("address", None)
address6 = provision[key].get("address6", None)
if address is not None or address6 is not None:
machine.set_address(address, address6)
dns = provision[key].get("dns")
if address or dns:
machine.set_dns(dns)
dhcp = provision[key].get("dhcp", False)
if dhcp:
machine.dhcp_server()
m = self.machine
self.journal_start = m.journal_cursor()
self.browser = self.new_browser(coverage=opts.coverage)
m.write("/etc/cockpit/cockpit.conf", "[Log]\nFatal = criticals\n")
if self.is_nondestructive():
self.nonDestructiveSetup()
if self.is_devel_build():
self.disable_preload("packagekit", "systemd")
image = m.image
if image.startswith(('debian', 'ubuntu')) or image == 'arch':
self.libexecdir = '/usr/lib/cockpit'
else:
self.libexecdir = '/usr/libexec'
if image.startswith(('debian', 'ubuntu')):
self.sshd_service = 'ssh.service'
self.sshd_socket = 'ssh.socket'
else:
self.sshd_service = 'sshd.service'
if image == 'arch':
self.sshd_socket = None
else:
self.sshd_socket = 'sshd.socket'
self.restart_sshd = f'systemctl try-restart {self.sshd_service}'
self.multihost_enabled = image.startswith(("rhel-9", "centos-9"))
self.has_sudo_rs = image.startswith("ubuntu")
def nonDestructiveSetup(self) -> None:
"""generic setUp/tearDown for @nondestructive tests"""
m = self.machine
name = f"{self.__class__.__name__}.{self._testMethodName}"
m.execute("logger -p user.info 'COCKPITTEST: start %s'" % name)
self.addCleanup(m.execute, "logger -p user.info 'COCKPITTEST: end %s'" % name)
self.addCleanup(m.execute, "find /var/lib/systemd/coredump -type f -delete")
self.addCleanup(m.execute, f"""
if [ -d {self.vm_tmpdir} ]; then
findmnt --list --noheadings --output TARGET | grep ^{self.vm_tmpdir} | xargs -r umount
rm -r {self.vm_tmpdir}
fi""")
self.restore_file("/etc/passwd")
self.restore_file("/etc/group")
self.restore_file("/etc/shadow")
self.restore_file("/etc/gshadow")
self.restore_file("/etc/subuid")
self.restore_file("/etc/subgid")
self.restore_file("/var/log/wtmp")
def get_home_dirs() -> Sequence[str]:
return m.execute("if [ -d /home ]; then ls /home; fi").strip().split()
initial_home_dirs = get_home_dirs()
def cleanup_home_dirs() -> None:
for d in get_home_dirs():
if d not in initial_home_dirs:
m.execute("rm -r /home/" + d)
self.addCleanup(cleanup_home_dirs)
if m.image == "arch":
self.addCleanup(m.execute, "rm -rf /run/faillock")
self.restore_dir("/etc/cockpit")
self.restore_file("/etc/fstab")
self.restore_file("/etc/crypttab")
if not m.ws_container:
self.addCleanup(m.execute, "systemctl stop --quiet cockpit")
m.execute("systemctl try-restart sssd || true")
self.addCleanup(self.machine.execute,
"set -e; [ -e /sys/module/scsi_debug ] || exit 0; "
"for dev in $(ls /sys/bus/pseudo/drivers/scsi_debug/adapter*/host*/target*/*:*/block); do "
" for s in /sys/block/*/slaves/${dev}*; do [ -e $s ] || break; "
" d=/dev/$(dirname $(dirname ${s#/sys/block/})); "
" while fuser --mount $d --kill; do sleep 0.1; done; "
" umount $d || true; dmsetup remove --force $d || true; "
" done; "
" while fuser --mount /dev/$dev --kill; do sleep 0.1; done; "
" umount /dev/$dev || true; "
" swapon --show=NAME --noheadings | grep $dev | xargs -r swapoff; "
"done; until rmmod scsi_debug; do sleep 0.2; done", stdout=None)
def _terminate_sessions(self) -> None:
m = self.machine
m.execute("""for u in $(loginctl --no-legend list-users | awk '{ if ($2 != "root") print $1 }'); do
loginctl terminate-user $u 2>/dev/null || true
loginctl kill-user $u 2>/dev/null || true
pkill -9 -u $u || true
while pgrep -u $u; do sleep 0.2; done
while mountpoint -q /run/user/$u && ! umount /run/user/$u; do sleep 0.2; done
rm -rf /run/user/$u
done""")
sessions = m.execute("loginctl --no-legend list-sessions | awk '/web console/ { print $1 }'").strip().split()
for s in sessions:
m.execute(f"loginctl kill-session {s} || true; loginctl terminate-session {s} || true")
m.execute("systemctl stop systemd-logind")
sessions = m.execute("loginctl --no-legend list-sessions | awk '/web console/ { print $1 }'").strip().split()
for s in sessions:
try:
m.execute(f"while loginctl show-session {s}; do sleep 0.2; done", timeout=30)
except RuntimeError:
m.execute(f"loginctl session-status {s}; systemd-cgls", stdout=None)
raise
m.execute("cd /run/systemd/users/; "
"for f in $(ls); do [ $f -le 500 ] || systemctl stop user@$f; done")
m.execute("systemctl stop systemd-logind; cd /run/systemd/users/; "
"for f in $(ls); do [ $f -le 500 ] || rm $f; done")
def tearDown(self) -> None:
error = self.getError()
if error:
print(error, file=sys.stderr)
try:
self.snapshot("FAIL")
self.copy_js_log("FAIL")
self.copy_journal("FAIL")
self.copy_cores("FAIL")
except (OSError, RuntimeError):
sys.stderr.write("Failed to generate debug artifact:\n")
traceback.print_exc(file=sys.stderr)
if opts.sit:
sit(self.machines)
if self.browser:
self.browser.write_coverage_data()
if self.machine.ssh_reachable:
self.check_journal_messages()
if not error:
self.check_browser_errors()
self.check_pixel_tests()
if self.is_nondestructive():
self._terminate_sessions()
shutil.rmtree(self.tmpdir, ignore_errors=True)
def enable_multihost(self, machine: testvm.Machine) -> None:
if isBeibootLogin():
raise NotImplementedError("multi-host config change not currently implemented for beiboot scenario")
if not self.multihost_enabled:
machine.write("/etc/cockpit/cockpit.conf",
'[WebService]\nAllowMultiHost=yes\n')
machine.restart_cockpit()
def login_and_go(
self,
path: str | None = None,
*,
user: str | None = None,
password: str | None = None,
host: str | None = None,
superuser: bool = True,
urlroot: str | None = None,
tls: bool = False,
enable_root_login: bool = False
) -> None:
if enable_root_login:
self.enable_root_login()
self.machine.start_cockpit(tls=tls)
with self.browser.wait_timeout(30):
self.browser.login_and_go(path, user=user, password=password, host=host, superuser=superuser,
urlroot=urlroot, tls=tls)
default_allowed_messages = [
'.*Reauthorizing unix-user:.*',
'.*user .* was reauthorized.*',
"Error executing command as another user: Not authorized",
"This incident has been reported.",
"-- Reboot --",
"Lost the name com.redhat.Cockpit on the session message bus",
"GLib-GIO:ERROR:gdbusobjectmanagerserver\\.c:.*:g_dbus_object_manager_server_emit_interfaces_.*: assertion failed \\(error == NULL\\): The connection is closed \\(g-io-error-quark, 18\\)",
"Error sending message: The connection is closed",
"cockpit-session: pam: Creating directory .*",
"cockpit-session: pam: Changing password for .*",
"cockpit-session: pam: Last failed login:.*",
"cockpit-session: pam: Last login: .*",
"cockpit-session: pam: There .* failed login attempts? since the last successful login.",
".*/var/log/lastlog: No such file or directory",
'10.*: dropping message while waiting for child to exit',
"xargs: basename: terminated by signal 13",
"(audit: )?type=1403 audit.*",
"(audit: )?type=1404 audit.*",
"(audit: )?type=1405 audit.*",
"(audit: )?type=1400.*apparmor=\"STATUS\".*",
"(audit: )?type=1400.*apparmor=\"ALLOWED\".*",
'Successfully loaded SELinux database in.*',
'calling: info',
'Sent message type=method_call sender=.*',
'Got message type=method_return sender=.*',
"Journal file.*truncated, ignoring file.",
"Failed to send coredump datagram:.*",
"Failed to get (COMM|EXE).*: No such process",
"sudo: unable to resolve host.*",
"We trust you have received the usual lecture from the local System",
"Administrator. It usually boils down to these three things:",
r"#1\) Respect the privacy of others.",
r"#2\) Think before you type.",
r"#3\) With great power comes great responsibility.",
"For security reasons, the password you type will not be visible",
"(direct|pcp-archive): no such metric: .*: Unknown metric name",
"(direct|pcp-archive): instance name lookup failed:.*",
"(direct|pcp-archive): couldn't create pcp archive context for.*",
"cockpit.channels.pcp-ERROR: no such metric: .*",
".*org.freedesktop.timedate1: couldn't get all properties.*Error:org.freedesktop.DBus.Error.NoReply.*",
'.*apparmor="DENIED" operation="capable" class="cap" profile="systemd-detect-virt".*',
'.*apparmor="DENIED" operation="sendmsg" class="file" profile="systemd-detect-virt".*',
]
default_allowed_messages += os.environ.get("TEST_ALLOW_JOURNAL_MESSAGES", "").split(",")
default_allowed_console_errors = [
"Warning: .* setState.*on an unmounted component",
"Warning: Can't perform a React state update on an unmounted component",
"Warning: Cannot update a component.*while rendering a different component",
"Warning: A component is changing an uncontrolled input to be controlled",
"Warning: A component is changing a controlled input to be uncontrolled",
"Warning: Can't call.*on a component that is not yet mounted. This is a no-op",
"Warning: Cannot update during an existing state transition",
r"Warning: You are calling ReactDOMClient.createRoot\(\) on a container that has already been passed to createRoot",
"aria-label",
"error: Could not determine kpatch packages:.*PackageKit crashed",
]
env_allow = os.environ.get("TEST_ALLOW_BROWSER_ERRORS")
if env_allow:
default_allowed_console_errors += env_allow.split(",")
def allow_journal_messages(self, *patterns: str) -> None:
"""Don't fail if the journal contains a entry completely matching the given regexp"""
for p in patterns:
self.allowed_messages.append(p)
def allow_hostkey_messages(self) -> None:
self.allow_journal_messages('.*: .* host key for server is not known: .*',
'.*: refusing to connect to unknown host: .*',
'.*: .* host key for server has changed to: .*',
'.*: host key for this server changed key type: .*',
'.*: failed to retrieve resource: hostkey-unknown')
def allow_restart_journal_messages(self) -> None:
self.allow_journal_messages(".*Connection reset by peer.*",
"connection unexpectedly closed by peer",
".*gnutls_handshake failed.*",
"cockpit-tls: .* No such file or directory",
".*Broken pipe.*",
"g_dbus_connection_real_closed: Remote peer vanished with error: Underlying GIOStream returned 0 bytes on an async read \\(g-io-error-quark, 0\\). Exiting.",
"cockpit-session: .*timed out.*",
"ignoring failure from session process:.*",
"peer did not close io when expected",
"request timed out, closing",
"PolicyKit daemon disconnected from the bus.",
".*couldn't create polkit session subject: No session for pid.*",
"We are no longer a registered authentication agent.",
".*: failed to retrieve resource: terminated",
".*: external channel failed:.*",
".*: truncated data in external channel",
'audit:.*denied.*comm="systemd-user-se".*nologin.*',
".*No session for cookie",
'localhost: dropping message while waiting for child to exit',
'.*: GDBus.Error:org.freedesktop.PolicyKit1.Error.Failed: .*',
'.*g_dbus_connection_call_finish_internal.*G_IS_DBUS_CONNECTION.*',
'.*Message recipient disconnected from message bus without replying.*',
'.*Unable to shutdown socket: Transport endpoint is not connected.*',
'which: no python in .*'
)
self.allow_browser_errors("Tracer failed:.*internal-error")
def check_journal_messages(self, machine: testvm.Machine | None = None) -> None:
"""Check for unexpected journal entries."""
machine = machine or self.machine
cursor = self.journal_start if machine == self.machine else None
matches = [
"SYSLOG_IDENTIFIER=cockpit-ws",
"SYSLOG_IDENTIFIER=cockpit-tls",
"SYSLOG_IDENTIFIER=cockpit-bridge",
"SYSLOG_IDENTIFIER=cockpit/ssh",
"_COMM=cockpit-ws",
"GLIB_DOMAIN=cockpit-ws",
"GLIB_DOMAIN=cockpit-bridge",
]
if not self.allow_core_dumps:
matches += ["SYSLOG_IDENTIFIER=systemd-coredump"]
self.allowed_messages.append("Resource limits disable core dumping for process.*")
self.allowed_messages.append("Failed to connect to coredump service: No such file or directory")
self.allowed_messages.append("Failed to connect to coredump service: Connection refused")
messages = machine.journal_messages(matches, 6, cursor=cursor)
if "TEST_AUDIT_NO_SELINUX" not in os.environ:
messages += machine.audit_messages("14", cursor=cursor)
self.allowed_messages += self.machine.allowed_messages()
all_found = True
first = None
for m in messages:
m = m.strip()
if not m:
continue
found = False
if m.startswith("Failed to generate stack trace"):
self.allowed_messages.append("Process .* of user .* dumped core.*")
continue
for p in self.allowed_messages:
match = re.match(p, m)
if match and match.group(0) == m:
found = True
break
if not found:
all_found = False
if not first:
first = m
print(m)
if not all_found:
assert first is not None
self.copy_js_log("FAIL")
self.copy_journal("FAIL")
self.copy_cores("FAIL")
if not self.getError():
raise Error(UNEXPECTED_MESSAGE + "journal messages:\n" + first)
def allow_browser_errors(self, *patterns: str) -> None:
"""Don't fail if the test caused a console error contains the given regexp"""
for p in patterns:
self.allowed_console_errors.append(p)
def check_browser_errors(self) -> None:
if not self.browser:
return
for log in self.browser.get_js_log():
if not log.startswith("error: "):
continue
for p in self.allowed_console_errors:
if re.search(p, log):
break
else:
raise Error(UNEXPECTED_MESSAGE + "browser errors:\n" + log)
self.browser.assert_no_oops()
def check_pixel_tests(self) -> None:
if self.browser:
self.browser.assert_no_unused_pixel_test_references()
if self.browser.failed_pixel_tests > 0:
raise Error(PIXEL_TEST_MESSAGE)
def snapshot(self, title: str, label: str | None = None) -> None:
"""Take a snapshot of the current screen and save it as a PNG.
Arguments:
title: Used for the filename.
"""
if self.browser is not None:
try:
self.browser.snapshot(title, label)
except Error:
sys.stderr.write("Unexpected exception in snapshot():\n")
sys.stderr.write(traceback.format_exc())
def copy_js_log(self, title: str, label: str | None = None) -> None:
if self.browser is not None:
try:
self.browser.copy_js_log(title, label)
except RuntimeError:
sys.stderr.write("Unexpected exception in copy_js_log():\n")
sys.stderr.write(traceback.format_exc())
def copy_journal(self, title: str, label: str | None = None) -> None:
for _, m in self.machines.items():
if m.ssh_reachable:
log = unique_filename(f"{label or self.label()}-{m.label}-{title}", "log.gz")
with open(log, "w") as fp:
m.execute("journalctl|gzip", stdout=fp)
print("Journal extracted to %s" % (log))
attach(log, move=True)
def copy_cores(self, title: str, label: str | None = None) -> None:
if self.allow_core_dumps:
return
for _, m in self.machines.items():
if m.ssh_reachable:
directory = f"{label or self.label()}-{m.label}-{title}.core"
dest = os.path.abspath(directory)
if os.path.exists(dest):
shutil.rmtree(dest)
m.download_dir("/var/lib/systemd/coredump", dest)
try:
os.rmdir(dest)
except OSError as ex:
if ex.errno == errno.ENOTEMPTY:
print("Core dumps downloaded to %s" % (dest))
def settle_cpu(self) -> None:
"""Wait until CPU usage in the VM settles down
Wait until the process with the highest CPU usage drops below 20%
usage. Wait for up to a minute, then return. There is no error if the
CPU stays busy, as usually a test then should just try to run anyway.
"""
for _ in range(20):
busy_proc = self.machine.execute("ps --no-headers -eo pcpu,pid,args | sort -k 1 -n -r | head -n1")
if float(busy_proc.split()[0]) < 20.0:
break
time.sleep(3)
def sed_file(self, expr: str, path: str, apply_change_action: str | None = None) -> None:
"""sed a file on primary machine
This is safe for @nondestructive tests, the file will be restored during cleanup.
The optional apply_change_action will be run both after sedding and after restoring the file.
"""
m = self.machine
m.execute(f"sed -i.cockpittest '{expr}' {path}")
if apply_change_action:
m.execute(apply_change_action)
if self.is_nondestructive():
if apply_change_action:
self.addCleanup(m.execute, apply_change_action)
self.addCleanup(m.execute, f"mv {path}.cockpittest {path}")
def file_exists(self, path: str) -> bool:
"""Check if file exists on test machine"""
return self.machine.execute(f"if test -e {path}; then echo yes; fi").strip() != ""
def restore_dir(
self,
path: str,
post_restore_action: str | None = None,
reboot_safe: bool = False,
restart_unit: str | None = None
) -> None:
"""Backup/restore a directory for a nondestructive test
This takes care to not ever touch the original content on disk, but uses transient overlays.
As this uses a bind mount, it does not work for files that get changed atomically (with mv);
use restore_file() for these.
`restart_unit` will be stopped before restoring path, and restarted afterwards if it was running.
The optional post_restore_action will run after restoring the original content.
If the directory needs to survive reboot, `reboot_safe=True` needs to be specified; then this
will just backup/restore the directory instead of bind-mounting, which is less robust.
"""
if not self.is_nondestructive():
return
exe = self.machine.execute
if not self.file_exists(path):
self.addCleanup(exe, f"rm -rf '{path}'")
return
backup = os.path.join(self.vm_tmpdir, path.replace('/', '_'))
exe(f"mkdir -p {self.vm_tmpdir}; cp -a {path}/ {backup}/")
if not reboot_safe:
exe(f"mount -o bind {backup} {path}")
if restart_unit:
restart_stamp = f"/run/cockpit_restart_{restart_unit}"
self.addCleanup(
exe,
f"if [ -e {restart_stamp} ]; then systemctl start {restart_unit}; rm {restart_stamp}; fi"
)
if post_restore_action:
self.addCleanup(exe, post_restore_action)
if reboot_safe:
self.addCleanup(exe, f"rm -rf {path}; mv {backup} {path}")
else:
if path.startswith("/home"):
cmd = f"umount -lf {path}"
else:
cmd = f"umount {path} || {{ fuser -uvk {path} {path}/* >&2 || true; sleep 1; umount {path}; }}"
self.addCleanup(exe, cmd)
if restart_unit:
self.addCleanup(exe, f"if systemctl --quiet is-active {restart_unit}; then touch {restart_stamp}; fi; "
f"systemctl stop {restart_unit}")
def restore_file(self, path: str, post_restore_action: str | None = None) -> None:
"""Backup/restore a file for a nondestructive test
This is less robust than restore_dir(), but works for files that need to get changed atomically.
If path does not currently exist, it will be removed again on cleanup.
"""
if not self.is_nondestructive():
return
if post_restore_action:
self.addCleanup(self.machine.execute, post_restore_action)
if self.file_exists(path):
backup = os.path.join(self.vm_tmpdir, path.replace('/', '_'))
self.machine.execute(f"mkdir -p {self.vm_tmpdir}; cp -a {path} {backup}")
self.addCleanup(self.machine.execute, f"mv {backup} {path}")
else:
self.addCleanup(self.machine.execute, f"rm -f {path}")
def write_file(
self,
path: str,
content: str,
append: bool = False,
owner: str | None = None,
perm: str | None = None,
post_restore_action: str | None = None
) -> None:
"""Write a file on primary machine
This is safe for @nondestructive tests, the file will be removed during cleanup.
If @append is True, append to existing file instead of replacing it.
@owner is the desired file owner as chown shell string (e.g. "admin:nogroup")
@perm is the desired file permission as chmod shell string (e.g. "0600")
"""
m = self.machine
self.restore_file(path, post_restore_action=post_restore_action)
m.write(path, content, append=append, owner=owner, perm=perm)
def enable_root_login(self) -> None:
"""Enable root login
By default root login is disabled in cockpit, removing the root entry of
/etc/cockpit/disallowed-users allows root to login.
"""
disallowed_conf = '/etc/cockpit/disallowed-users'
if not self.machine.ws_container and self.file_exists(disallowed_conf):
self.sed_file('/root/d', disallowed_conf)
def get_sudo_user(self) -> str:
return "root" if "suse" in self.image else "admin"
def reboot(self, timeout_sec: int | None = None) -> None:
self.allow_restart_journal_messages()
if timeout_sec is None:
self.machine.reboot()
else:
self.machine.reboot(timeout_sec=timeout_sec)
def wait_reboot(self, timeout_sec: int | None = None) -> None:
self.allow_restart_journal_messages()
if timeout_sec is None:
self.machine.wait_reboot()
else:
self.machine.wait_reboot(timeout_sec=timeout_sec)
def setup_provisioned_hosts(self, disable_preload: bool = False) -> None:
"""Setup provisioned hosts for testing
This sets the hostname of all machines to the name given in the
provision dictionary and optionally disabled preload.
"""
for name, m in self.machines.items():
m.execute(f"hostnamectl set-hostname {name}")
if disable_preload:
self.disable_preload("packagekit", "playground", "systemd", machine=m)
@staticmethod
def authorize_pubkey(machine: testvm.Machine, account: str, pubkey: str) -> None:
machine.execute(f"a={account} d=/home/$a/.ssh; mkdir -p $d; chown $a:$a $d; chmod 700 $d")
machine.write(f"/home/{account}/.ssh/authorized_keys", pubkey)
machine.execute(f"a={account}; chown $a:$a /home/$a/.ssh/authorized_keys")
@staticmethod
def get_pubkey(machine: testvm.Machine, account: str) -> str:
return machine.execute(f"cat /home/{account}/.ssh/id_rsa.pub")
def setup_ssh_auth(self) -> None:
self.machine.execute("d=/home/admin/.ssh; mkdir -p $d; chown admin:admin $d; chmod 700 $d")
self.machine.execute("test -f /home/admin/.ssh/id_rsa || ssh-keygen -f /home/admin/.ssh/id_rsa -t rsa -N ''")
self.machine.execute("chown admin:admin /home/admin/.ssh/id_rsa*")
pubkey = self.get_pubkey(self.machine, "admin")
for m in self.machines:
self.authorize_pubkey(self.machines[m], "admin", pubkey)
def enterAnacondaMode(self, config: JsonObject, page_identifier: str) -> None:
b = self.browser
b.call_js_func("window.sessionStorage.setItem", "cockpit_anaconda", json.dumps(config))
b.reload()
b.enter_page(page_identifier)
def jsquote(js: object) -> str:
return json.dumps(js)
def get_decorator(method: object, class_: object, name: str, default: Any = None) -> Any:
"""Get decorator value of a test method or its class
Return None if the decorator was not set.
"""
attr = "_testlib__" + name
return getattr(method, attr, getattr(class_, attr, default))
def skipBrowser(reason: str, *browsers: str) -> Callable[[_FT], _FT]:
"""Decorator for skipping a test on given browser(s)
Skips a test for provided *reason* on *browsers*.
"""
browser = os.environ.get("TEST_BROWSER", "chromium")
if browser in browsers:
return unittest.skip(f"{browser}: {reason}")
return lambda testEntity: testEntity
def skipImage(reason: str, *images: str) -> Callable[[_FT], _FT]:
"""Decorator for skipping a test for given image(s)
Skip a test for a provided *reason* for given *images*. These
support Unix shell style patterns via fnmatch.fnmatch.
Example: @skipImage("no btrfs support on RHEL", "rhel-*")
"""
if any(fnmatch.fnmatch(testvm.DEFAULT_IMAGE, img) for img in images):
return unittest.skip(f"{testvm.DEFAULT_IMAGE}: {reason}")
return lambda testEntity: testEntity
def onlyImage(reason: str, *images: str) -> Callable[[_FT], _FT]:
"""Decorator to only run a test on given image(s)
Only run this test on provided *images* for *reason*. These
support Unix shell style patterns via fnmatch.fnmatch.
"""
if not any(fnmatch.fnmatch(testvm.DEFAULT_IMAGE, arg) for arg in images):
return unittest.skip(f"{testvm.DEFAULT_IMAGE}: {reason}")
return lambda testEntity: testEntity
def skipOstree(reason: str) -> Callable[[_FT], _FT]:
"""Decorator for skipping a test on OSTree images
Skip test for *reason* on OSTree images defined in OSTREE_IMAGES in bots/lib/constants.py.
"""
if testvm.DEFAULT_IMAGE in OSTREE_IMAGES:
return unittest.skip(f"{testvm.DEFAULT_IMAGE}: {reason}")
return lambda testEntity: testEntity
def isBeibootLogin() -> bool:
return "ws-container" in os.getenv("TEST_SCENARIO", "")
def skipWsContainer(reason: str) -> Callable[[_FT], _FT]:
"""Decorator for skipping a test with cockpit/ws"""
if testvm.DEFAULT_IMAGE in OSTREE_IMAGES or isBeibootLogin():
return unittest.skip(f"{testvm.DEFAULT_IMAGE}: {reason}")
return lambda testEntity: testEntity
def skipBeiboot(reason: str) -> Callable[[_FT], _FT]:
"""Decorator for skipping a test with cockpit/ws in beiboot mode"""
if isBeibootLogin():
return unittest.skip(f"{testvm.DEFAULT_IMAGE}: {reason}")
return lambda testEntity: testEntity
def skipDaily(reason: str) -> Callable[[_FT], _FT]:
"""Decorator for skipping a test with fedora-*/nightly"""
if "daily" in os.getenv("TEST_SCENARIO", ""):
return unittest.skip(f"{testvm.DEFAULT_IMAGE}: {reason}")
return lambda testEntity: testEntity
def nondestructive(testEntity: _T) -> _T:
"""Tests decorated as nondestructive will all run against the same VM
Can be used on test classes and individual test methods.
"""
setattr(testEntity, '_testlib__nondestructive', True)
return testEntity
def destructive(testEntity: _T) -> _T:
"""Tests decorated as destructive will get their own VM
Can be used on test classes and individual test methods.
"""
setattr(testEntity, '_testlib__nondestructive', False)
return testEntity
def no_retry_when_changed(testEntity: _T) -> _T:
"""Tests decorated with no_retry_when_changed will only run once if they've been changed
Tests that have been changed are expected to succeed 3 times, if the test
takes a long time, this prevents timeouts. Can be used on test classes and
individual methods.
"""
setattr(testEntity, '_testlib__no_retry_when_changed', True)
return testEntity
def todo(reason: str = '') -> Callable[[_T], _T]:
"""Tests decorated with @todo are expected to fail.
An optional reason can be given, and will appear in the TAP output if run
via run-tests.
"""
def wrapper(testEntity: _T) -> _T:
setattr(testEntity, '_testlib__todo', reason)
return testEntity
return wrapper
def timeout(seconds: int) -> Callable[[_T], _T]:
"""Change default test timeout of 600s, for long running tests
Can be applied to an individual test method or the entire class. This only
applies to test/common/run-tests, not to calling check-* directly.
"""
def wrapper(testEntity: _T) -> _T:
setattr(testEntity, '_testlib__timeout', seconds)
return testEntity
return wrapper
class TapRunner:
def __init__(self, verbosity: int = 1):
self.verbosity = verbosity
def runOne(self, test: unittest.TestCase) -> unittest.TestResult:
result = unittest.TestResult()
print('# ----------------------------------------------------------------------')
print('#', test)
try:
unittest.TestSuite([test]).run(result)
except KeyboardInterrupt:
result.addError(test, sys.exc_info())
return result
except Exception:
result.addError(test, sys.exc_info())
sys.stderr.write(f"Unexpected exception while running {test}\n")
sys.stderr.write(traceback.format_exc())
return result
else:
result.printErrors()
if result.skipped:
print(f"# Result {test} skipped: {result.skipped[0][1]}")
elif result.wasSuccessful():
print(f"# Result {test} succeeded")
else:
for failure in result.failures:
print(failure[1])
for error in result.errors:
print(error[1])
print(f"# Result {test} failed")
return result
def run(self, testable: unittest.TestSuite) -> int:
tests: list[unittest.TestCase] = []
def collapse(test: unittest.TestCase | unittest.TestSuite, tests: list[unittest.TestCase]) -> None:
if isinstance(test, unittest.TestCase):
tests.append(test)
else:
for t in test:
collapse(t, tests)
collapse(testable, tests)
test_count = len(tests)
start = time.time()
failures = 0
skips = []
while tests:
test = tests.pop(0)
result = self.runOne(test)
if not result.wasSuccessful():
failures += 1
skips += result.skipped
duration = int(time.time() - start)
hostname = socket.gethostname().split(".")[0]
details = f"[{duration}s on {hostname}]"
MachineCase.kill_global_machine()
if len(skips) == test_count:
skipstr = ", ".join([f"{s[0]!s} {s[1]}" for s in skips])
sys.stdout.write(f"# SKIP {skipstr}\n")
return 77
if failures:
plural = "S" if failures > 1 else ""
sys.stdout.write(f"# {failures} TEST{plural} FAILED {details}\n")
return 1
else:
plural = "S" if test_count > 1 else ""
sys.stdout.write(f"# {test_count} TEST{plural} PASSED {details}\n")
return 0
def print_tests(tests: unittest.TestSuite | Collection[unittest.TestSuite | unittest.TestCase]) -> None:
assert hasattr(unittest.loader, '_FailedTest')
for test in tests:
if isinstance(test, unittest.TestSuite):
print_tests(test)
elif isinstance(test, unittest.loader._FailedTest):
name = test.id().replace("unittest.loader._FailedTest.", "")
print(f"Error: '{name}' does not match a test", file=sys.stderr)
else:
print(test.id().replace("__main__.", ""))
@dataclass
class TestlibArgs:
coverage: bool
enable_network: bool
fetch: bool
list: bool
sit: bool
trace: bool
verbosity: int
tests: Sequence[str]
def arg_parser(enable_sit: bool = True) -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(description='Run Cockpit test(s)')
parser.add_argument('-v', '--verbose', dest="verbosity", action='store_const',
const=2, help='Verbose output')
parser.add_argument('-t', "--trace", dest='trace', action='store_true',
help='Trace machine boot and commands')
parser.add_argument('-q', '--quiet', dest='verbosity', action='store_const',
const=0, help='Quiet output')
if enable_sit:
parser.add_argument('-s', "--sit", dest='sit', action='store_true',
help="Sit and wait after test failure")
else:
parser.set_defaults(sit=False)
parser.add_argument('--nonet', dest="fetch", action="store_false",
help="Don't go online to download images or data")
parser.add_argument('--enable-network', dest='enable_network', action='store_true',
help="Enable network access for tests")
parser.add_argument('--coverage', action='store_true',
help="Collect code coverage data")
parser.add_argument("-l", "--list", action="store_true", help="Print the list of tests that would be executed")
parser.add_argument('tests', nargs='*', default=os.getenv("TEST_NAMES", '').split())
parser.set_defaults(verbosity=1, fetch=True)
return parser
def test_main(
options: argparse.Namespace | None = None,
suite: unittest.TestSuite | None = None,
attachments: str | None = None,
) -> int:
"""
Run all test cases, as indicated by arguments.
If no arguments are given on the command line, all test cases are
executed. Otherwise only the given test cases are run.
"""
global opts
buf_arg = 0
os.environ['PYTHONUNBUFFERED'] = '1'
buf_arg = 1
sys.stdout.flush()
sys.stdout = os.fdopen(sys.stdout.fileno(), 'w', buf_arg)
parser = arg_parser()
parser.add_argument('--machine', metavar="hostname[:port]", dest="address",
default=None, help="Run this test against an already running machine")
parser.add_argument('--browser', metavar="hostname[:port]", dest="browser",
default=None, help="When using --machine, use this cockpit web address")
if options is None:
options = parser.parse_args()
standalone = True
else:
standalone = False
if options.sit:
options.verbosity = 2
for (key, value) in vars(options).items():
setattr(opts, key, value)
opts.address = getattr(opts, "address", None)
opts.browser = getattr(opts, "browser", None)
opts.attachments = os.environ.get("TEST_ATTACHMENTS", attachments)
if opts.attachments:
os.makedirs(opts.attachments, exist_ok=True)
import __main__
if len(opts.tests) > 0:
if suite:
parser.error("tests may not be specified when running a predefined test suite")
suite = unittest.TestLoader().loadTestsFromNames(opts.tests, module=__main__)
elif not suite:
suite = unittest.TestLoader().loadTestsFromModule(__main__)
if options.list:
print_tests(suite)
return 0
attach(os.path.join(TEST_DIR, "common/pixeldiff.html"))
attach(os.path.join(TEST_DIR, "common/link-patterns.json"))
runner = TapRunner(verbosity=opts.verbosity)
ret = runner.run(suite)
if not standalone:
return ret
sys.exit(ret)
class Error(Exception):
def __init__(self, msg: str) -> None:
self.msg = msg
def __str__(self) -> str:
return self.msg
def wait(func: Callable[[], _T | None], msg: str | None = None, delay: int = 1, tries: int = 60) -> _T:
"""Wait for FUNC to return something truthy, and return that.
FUNC is called repeatedly until it returns a true value or until a
timeout occurs. In the latter case, a exception is raised that
describes the situation. The exception is either the last one
thrown by FUNC, or includes MSG, or a default message.
:param func: The function to call
:param msg: A error message to use when the timeout occurs. Defaults
to a generic message.
:param delay: How long to wait between calls to FUNC, in seconds. (default 1)
:param tries: How often to call FUNC. (defaults 60)
:raises Error: When a timeout occurs.
"""
t = 0
while t < tries:
try:
val = func()
if val:
return val
except Exception:
if t == tries - 1:
raise
t = t + 1
time.sleep(delay)
raise Error(msg or "Condition did not become true.")
def sit(machines: Mapping[str, testvm.Machine] = {}) -> None:
"""
Wait until the user confirms to continue.
The current test case is suspended so that the user can inspect
the browser.
"""
for machine in machines.values():
sys.stderr.write(machine.diagnose())
print("Press RET to continue...")
sys.stdin.readline()
