Moved hydra util package near editor python tests

This commit is contained in:
evanchia
2021-04-20 16:51:48 -07:00
parent 40f4b651ab
commit e79b1b4af1
18 changed files with 172 additions and 103 deletions
@@ -0,0 +1,49 @@
All or portions of this file Copyright (c) Amazon.com, Inc. or its affiliates or
its licensors.
For complete copyright and license terms please see the LICENSE at the root of this
distribution (the "License"). All use of this software is governed by the License,
or, if provided, by the license below or the license accompanying this file. Do not
remove or modify any license notices. This file is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
INTRODUCTION
------------
EditorPythonBindings is a Python project that contains a collection of editor testing tools
developed by the Lumberyard feature teams. The project contains tools for system level
editor tests.
REQUIREMENTS
------------
* Python 3.7.5 (64-bit)
It is recommended that you completely remove any other versions of Python
installed on your system.
INSTALL
-----------
It is recommended to set up these these tools with Lumberyard's CMake build commands.
Assuming CMake is already setup on your operating system, below are some sample build commands:
cd /path/to/od3e/
mkdir windows_vs2019
cd windows_vs2019
cmake .. -G "Visual Studio 16 2019" -A x64 -T host=x64 -DLY_3RDPARTY_PATH="%3RDPARTYPATH%" -DLY_PROJECTS=AutomatedTesting
NOTE:
Using the above command also adds EditorPythonTestTools to the PYTHONPATH OS environment variable.
Additionally, some CTest scripts will add the Python interpreter path to the PYTHON OS environment variable.
To manually install the project in development mode using your own installed Python interpreter:
cd /path/to/od3e/AutomatedTesting/Gem/PythonTests/EditorPythonTestTools
/path/to/your/python -m pip install -e .
UNINSTALLATION
--------------
The preferred way to uninstall the project is:
/path/to/your/python -m pip uninstall editor_python_test_tools
@@ -0,0 +1,10 @@
"""
All or portions of this file Copyright (c) Amazon.com, Inc. or its affiliates or
its licensors.
For complete copyright and license terms please see the LICENSE at the root of this
distribution (the "License"). All use of this software is governed by the License,
or, if provided, by the license below or the license accompanying this file. Do not
remove or modify any license notices. This file is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
"""
@@ -0,0 +1,110 @@
Metadata-Version: 1.0
Name: editor-python-test-tools
Version: 1.0.0
Summary: Lumberyard editor Python bindings test tools
Home-page: UNKNOWN
Author: UNKNOWN
Author-email: UNKNOWN
License: UNKNOWN
Description: All or portions of this file Copyright (c) Amazon.com, Inc. or its affiliates or
its licensors.
For complete copyright and license terms please see the LICENSE at the root of this
distribution (the "License"). All use of this software is governed by the License,
or, if provided, by the license below or the license accompanying this file. Do not
remove or modify any license notices. This file is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
INTRODUCTION
------------
EditorPythonBindings is a Python project that contains a collection of testing tools
developed by the Lumberyard Test Tech team. The project contains
the following tools:
* Workspace Manager:
A library to manipulate Lumberyard installations
* Launchers:
A library to test the game in a variety of platforms
REQUIREMENTS
------------
* Python 3.7.5 (64-bit)
It is recommended that you completely remove any other versions of Python
installed on your system.
INSTALL
-----------
It is recommended to set up these these tools with Lumberyard's CMake build commands.
Assuming CMake is already setup on your operating system, below are some sample build commands:
cd /path/to/od3e/
mkdir windows_vs2019
cd windows_vs2019
cmake .. -G "Visual Studio 16 2019" -A x64 -T host=x64 -DLY_3RDPARTY_PATH="%3RDPARTYPATH%" -DLY_PROJECTS=AutomatedTesting
NOTE:
Using the above command also adds LyTestTools to the PYTHONPATH OS environment variable.
Additionally, some CTest scripts will add the Python interpreter path to the PYTHON OS environment variable.
There is some LyTestTools functionality that will search for these, so feel free to populate them manually.
To manually install the project in development mode using your own installed Python interpreter:
cd /path/to/lumberyard/dev/Tools/LyTestTools/
/path/to/your/python -m pip install -e .
For console/mobile testing, update the following .ini file in your root user directory:
i.e. C:/Users/myusername/ly_test_tools/devices.ini (a.k.a. %USERPROFILE%/ly_test_tools/devices.ini)
You will need to add a section for the device, and a key holding the device identifier value (usually an IP or ID).
It should look similar to this for each device:
[android]
id = 988939353955305449
[gameconsole]
ip = 192.168.1.1
[gameconsole2]
ip = 192.168.1.2
PACKAGE STRUCTURE
-----------------
The project is organized into packages. Each package corresponds to a tool:
- LyTestTools.ly_test_tools._internal: contains logging setup, pytest fixture, and o3de workspace manager modules
- LyTestTools.ly_test_tools.builtin: builtin helpers and fixtures for quickly writing tests
- LyTestTools.ly_test_tools.console: modules used for consoles
- LyTestTools.ly_test_tools.environment: functions related to file/process management and cleanup
- LyTestTools.ly_test_tools.image: modules related to image capturing and processing
- LyTestTools.ly_test_tools.launchers: game launchers library
- LyTestTools.ly_test_tools.log: modules for interacting with generated or existing log files
- LyTestTools.ly_test_tools.o3de: modules used to interact with Open 3D Engine
- LyTestTools.ly_test_tools.mobile: modules used for android/ios
- LyTestTools.ly_test_tools.report: modules used for reporting
- LyTestTools.tests: LyTestTools integration, unit, and example usage tests
DIRECTORY STRUCTURE
-------------------
The directory structure corresponds to the package structure. For example, the
ly_test_tools.builtin package is located in the ly_test_tools/builtin/ directory.
ENTRY POINTS
------------
Deploying the project in development mode installs only entry points for pytest fixtures.
UNINSTALLATION
--------------
The preferred way to uninstall the project is:
/path/to/your/python -m pip uninstall ly_test_tools
Platform: UNKNOWN
@@ -0,0 +1,7 @@
README.txt
setup.py
editor_python_test_tools.egg-info/PKG-INFO
editor_python_test_tools.egg-info/SOURCES.txt
editor_python_test_tools.egg-info/dependency_links.txt
editor_python_test_tools.egg-info/requires.txt
editor_python_test_tools.egg-info/top_level.txt
@@ -0,0 +1,10 @@
"""
All or portions of this file Copyright (c) Amazon.com, Inc. or its affiliates or
its licensors.
For complete copyright and license terms please see the LICENSE at the root of this
distribution (the "License"). All use of this software is governed by the License,
or, if provided, by the license below or the license accompanying this file. Do not
remove or modify any license notices. This file is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
"""
@@ -0,0 +1,318 @@
"""
All or portions of this file Copyright (c) Amazon.com, Inc. or its affiliates or
its licensors.
For complete copyright and license terms please see the LICENSE at the root of this
distribution (the "License"). All use of this software is governed by the License,
or, if provided, by the license below or the license accompanying this file. Do not
remove or modify any license notices. This file is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
"""
# Built-in Imports
from __future__ import annotations
from typing import List, Tuple, Union
# Open 3D Engine Imports
import azlmbr
import azlmbr.bus as bus
import azlmbr.editor as editor
import azlmbr.math as math
import azlmbr.legacy.general as general
# Helper file Imports
from editor_python_test_tools.utils import Report
class EditorComponent:
"""
EditorComponent class used to set and get the component property value using path
EditorComponent object is returned from either of
EditorEntity.add_component() or Entity.add_components() or EditorEntity.get_component_objects()
which also assigns self.id and self.type_id to the EditorComponent object.
"""
# Methods
def get_component_name(self) -> str:
"""
Used to get name of component
:return: name of component
"""
type_names = editor.EditorComponentAPIBus(bus.Broadcast, "FindComponentTypeNames", [self.type_id])
assert len(type_names) != 0, "Component object does not have type id"
return type_names[0]
def get_property_tree(self):
"""
Used to get the property tree object of component that has following functions associated with it:
1. prop_tree.is_container(path)
2. prop_tree.get_container_count(path)
3. prop_tree.reset_container(path)
4. prop_tree.add_container_item(path, key, item)
5. prop_tree.remove_container_item(path, key)
6. prop_tree.update_container_item(path, key, value)
7. prop_tree.get_container_item(path, key)
:return: Property tree object of a component
"""
build_prop_tree_outcome = editor.EditorComponentAPIBus(
bus.Broadcast, "BuildComponentPropertyTreeEditor", self.id
)
assert (
build_prop_tree_outcome.IsSuccess()
), f"Failure: Could not build property tree of component: '{self.get_component_name()}'"
prop_tree = build_prop_tree_outcome.GetValue()
Report.info(prop_tree.build_paths_list())
return prop_tree
def get_component_property_value(self, component_property_path: str):
"""
Given a component property path, outputs the property's value
:param component_property_path: String of component property. (e.g. 'Settings|Visible')
:return: Value set in given component_property_path. Type is dependent on component property
"""
get_component_property_outcome = editor.EditorComponentAPIBus(
bus.Broadcast, "GetComponentProperty", self.id, component_property_path
)
assert (
get_component_property_outcome.IsSuccess()
), f"Failure: Could not get value from {self.get_component_name()} : {component_property_path}"
return get_component_property_outcome.GetValue()
def set_component_property_value(self, component_property_path: str, value: object):
"""
Used to set component property value
:param component_property_path: Path of property in the component to act on
:param value: new value for the variable being changed in the component
"""
outcome = editor.EditorComponentAPIBus(
bus.Broadcast, "SetComponentProperty", self.id, component_property_path, value
)
assert (
outcome.IsSuccess()
), f"Failure: Could not set value to '{self.get_component_name()}' : '{component_property_path}'"
@staticmethod
def get_type_ids(component_names: list) -> list:
"""
Used to get type ids of given components list
:param: component_names: List of components to get type ids
:return: List of type ids of given components.
"""
type_ids = editor.EditorComponentAPIBus(
bus.Broadcast, "FindComponentTypeIdsByEntityType", component_names, azlmbr.entity.EntityType().Game
)
return type_ids
class EditorEntity:
"""
Entity class is used to create and interact with Editor Entities.
Example: To create Editor Entity, Use the code:
test_entity = Entity.create_editor_entity("TestEntity")
# This creates a python object with 'test_entity' linked to entity name "TestEntity" in Editor.
# To add component, use:
test_entity.add_component(<COMPONENT_NAME>)
"""
def __init__(self, id: azlmbr.entity.EntityId):
self.id: azlmbr.entity.EntityId = id
# Creation functions
@classmethod
def find_editor_entity(cls, entity_name: str) -> EditorEntity:
"""
Given Entity name, outputs entity object
:param entity_name: Name of entity to find
:return: EditorEntity class object
"""
entity_id = general.find_editor_entity(entity_name)
assert entity_id.IsValid(), f"Failure: Couldn't find entity with name: '{entity_name}'"
entity = cls(entity_id)
return entity
@classmethod
def create_editor_entity(cls, name: str = None, parent_id=None) -> EditorEntity:
"""
Used to create entity at default position using 'CreateNewEntity' Bus
:param name: Name of the Entity to be created
:param parent_id: (optional) Used to create child entity under parent_id if specified
:return: EditorEntity class object
"""
if parent_id is None:
parent_id = azlmbr.entity.EntityId()
new_id = azlmbr.editor.ToolsApplicationRequestBus(bus.Broadcast, "CreateNewEntity", parent_id)
assert new_id.IsValid(), "Failure: Could not create Editor Entity"
entity = cls(new_id)
if name:
entity.set_name(name)
return entity
@classmethod
def create_editor_entity_at(
cls,
entity_position: Union[List, Tuple, math.Vector3],
name: str = None,
parent_id: azlmbr.entity.EntityId = None,
) -> EditorEntity:
"""
Used to create entity at position using 'CreateNewEntityAtPosition' Bus.
:param entity_position: World Position(X, Y, Z) of entity in viewport.
Example: [512.0, 512.0, 32.0]
:param name: Name of the Entity to be created
:parent_id: (optional) Used to create child entity under parent_id if specified
:Example: test_entity = EditorEntity.create_editor_entity_at([512.0, 512.0, 32.0], "TestEntity")
:return: EditorEntity class object
"""
def convert_to_azvector3(xyz) -> math.Vector3:
if isinstance(xyz, Tuple) or isinstance(xyz, List):
assert len(xyz) == 3, ValueError("vector must be a 3 element list/tuple or azlmbr.math.Vector3")
return math.Vector3(*xyz)
elif isinstance(xyz, type(math.Vector3())):
return xyz
else:
raise ValueError("vector must be a 3 element list/tuple or azlmbr.math.Vector3")
if parent_id is None:
parent_id = azlmbr.entity.EntityId()
new_id = azlmbr.editor.ToolsApplicationRequestBus(
bus.Broadcast, "CreateNewEntityAtPosition", convert_to_azvector3(entity_position), parent_id
)
assert new_id.IsValid(), "Failure: Could not create Editor Entity"
entity = cls(new_id)
if name:
entity.set_name(name)
return entity
# Methods
def set_name(self, entity_name: str):
"""
Given entity_name, sets name to Entity
:param: entity_name: Name of the entity to set
"""
editor.EditorEntityAPIBus(bus.Event, "SetName", self.id, entity_name)
def get_name(self) -> str:
"""
Used to get the name of entity
"""
return editor.EditorEntityInfoRequestBus(bus.Event, "GetName", self.id)
def set_parent_entity(self, parent_entity_id):
"""
Used to set this entity to be child of parent entity passed in
:param: parent_entity_id: Entity Id of parent to set
"""
assert (
parent_entity_id.IsValid()
), f"Failure: Could not set parent to entity: {self.get_name()}, Invalid parent id"
editor.EditorEntityAPIBus(bus.Event, "SetParent", self.id, parent_entity_id)
def get_parent_id(self) -> azlmbr.entity.EntityId:
"""
:return: Entity id of parent. Type: entity.EntityId()
"""
return editor.EditorEntityInfoRequestBus(bus.Event, "GetParent", self.id)
def add_component(self, component_name: str) -> EditorComponent:
"""
Used to add new component to Entity.
:param component_name: String of component name to add.
:return: Component object of newly added component.
"""
component = self.add_components([component_name])[0]
return component
def add_components(self, component_names: list) -> List[EditorComponent]:
"""
Used to add multiple components
:param: component_names: List of components to add to entity
:return: List of newly added components to the entity
"""
components = []
type_ids = EditorComponent.get_type_ids(component_names)
for type_id in type_ids:
new_comp = EditorComponent()
new_comp.type_id = type_id
add_component_outcome = editor.EditorComponentAPIBus(
bus.Broadcast, "AddComponentsOfType", self.id, [type_id]
)
assert (
add_component_outcome.IsSuccess()
), f"Failure: Could not add component: '{new_comp.get_component_name()}' to entity: '{self.get_name()}'"
new_comp.id = add_component_outcome.GetValue()[0]
components.append(new_comp)
return components
def get_components_of_type(self, component_names: list) -> List[EditorComponent]:
"""
Used to get components of type component_name that already exists on Entity
:param component_name: Name to component to check
:return: List of Entity Component objects of given component name
"""
component_list = []
type_ids = EditorComponent.get_type_ids(component_names)
for type_id in type_ids:
component = EditorComponent()
component.type_id = type_id
get_component_of_type_outcome = editor.EditorComponentAPIBus(
bus.Broadcast, "GetComponentOfType", self.id, type_id
)
assert (
get_component_of_type_outcome.IsSuccess()
), f"Failure: Entity: '{self.get_name()}' does not have component:'{component.get_component_name()}'"
component.id = get_component_of_type_outcome.GetValue()
component_list.append(component)
return component_list
def has_component(self, component_name: str) -> bool:
"""
Used to verify if the entity has the specified component
:param component_name: Name of component to check for
:return: True, if entity has specified component. Else, False
"""
type_ids = EditorComponent.get_type_ids([component_name])
return editor.EditorComponentAPIBus(bus.Broadcast, "HasComponentOfType", self.id, type_ids[0])
def get_start_status(self) -> int:
"""
This will return a value for an entity's starting status (active, inactive, or editor) in the form of azlmbr.globals.property.EditorEntityStartStatus_<start type>. For comparisons using this value, should compare against the azlmbr property and not the int value
"""
status = editor.EditorEntityInfoRequestBus(bus.Event, "GetStartStatus", self.id)
if status == azlmbr.globals.property.EditorEntityStartStatus_StartActive:
status_text = "active"
elif status == azlmbr.globals.property.EditorEntityStartStatus_StartInactive:
status_text = "inactive"
elif status == azlmbr.globals.property.EditorEntityStartStatus_EditorOnly:
status_text = "editor"
Report.info(f"The start status for {self.get_name} is {status_text}")
self.start_status = status
return status
def set_start_status(self, desired_start_status: str):
"""
Set an entity as active/inactive at beginning of runtime or it is editor-only,
given its entity id and the start status then return set success
:param desired_start_status: must be one of three choices: active, inactive, or editor
"""
if desired_start_status == "active":
status_to_set = azlmbr.globals.property.EditorEntityStartStatus_StartActive
elif desired_start_status == "inactive":
status_to_set = azlmbr.globals.property.EditorEntityStartStatus_StartInactive
elif desired_start_status == "editor":
status_to_set = azlmbr.globals.property.EditorEntityStartStatus_EditorOnly
else:
Report.info(
f"Invalid desired_start_status argument for {self.get_name} set_start_status command;\
Use editor, active, or inactive"
)
editor.EditorEntityAPIBus(bus.Event, "SetStartStatus", self.id, status_to_set)
set_status = self.get_start_status()
assert set_status == status_to_set, f"Failed to set start status of {desired_start_status} to {self.get_name}"
@@ -0,0 +1,324 @@
#
# All or portions of this file Copyright (c) Amazon.com, Inc. or its affiliates or
# its licensors.
#
# For complete copyright and license terms please see the LICENSE at the root of this
# distribution (the "License"). All use of this software is governed by the License,
# or, if provided, by the license below or the license accompanying this file. Do not
# remove or modify any license notices. This file is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
#
# NOTE: This code is used for tests in several feature areas. If changes are made to this file, please verify all
# dependent tests continue to run without issue.
#
import sys
import time
from typing import Sequence
# Open 3D Engine specific imports
import azlmbr.legacy.general as general
import azlmbr.legacy.settings as settings
from editor_python_test_tools.utils import Report
class EditorTestHelper:
def __init__(self, log_prefix: str, args: Sequence[str] = None) -> None:
self.log_prefix = log_prefix + ": "
self.test_success = True
# If the idle loop has already been enabled at test init time, the Editor is already running.
# If that's the case, we'll skip the "exit_no_prompt" at the end.
self.editor_already_running = general.is_idle_enabled()
self.args = {}
if args:
# Get the level name and heightmap name from command-line args
if len(sys.argv) == (len(args) + 1):
for arg_index in range(len(args)):
self.args[args[arg_index]] = sys.argv[arg_index + 1]
else:
self.test_success = False
self.log(f"Expected command-line args: {args}")
self.log(f"Check that cfg_args were passed into the test class")
# Test Setup
# Set helpers
# Set viewport size
# Turn off display mode, antialiasing
# set log prefix, log test started
def setup(self) -> None:
self.log("test started")
def after_level_load(self, bypass_viewport_resize: bool = False) -> bool:
success = True
# Enable the Editor to start running its idle loop.
# This is needed for Python scripts passed into the Editor startup. Since they're executed
# during the startup flow, they run before idle processing starts. Without this, the engine loop
# won't run during idle_wait, which will prevent our test level from working.
general.idle_enable(True)
# Give everything a second to initialize
general.idle_wait(1.0)
general.update_viewport()
general.idle_wait(0.5) # half a second is more than enough for updating the viewport.
self.original_settings = settings.get_misc_editor_settings()
self.helpers_visible = general.is_helpers_shown()
self.viewport_size = general.get_viewport_size()
self.viewport_layout = general.get_view_pane_layout()
# Turn off the helper gizmos if visible
if self.helpers_visible:
general.toggle_helpers()
general.idle_wait(1.0)
# Close the Error Report window so it doesn't interfere with testing hierarchies and focus
if general.is_pane_visible("Error Report"):
general.close_pane("Error Report")
if general.is_pane_visible("Error Log"):
general.close_pane("Error Log")
general.idle_wait(1.0)
if not bypass_viewport_resize:
# Set Editor viewport to a well-defined size
screen_width = 1600
screen_height = 900
general.set_viewport_expansion_policy("FixedSize")
general.set_viewport_size(screen_width, screen_height)
general.update_viewport()
general.idle_wait(1.0)
new_viewport_size = general.get_viewport_size()
new_viewport_width = int(new_viewport_size.x)
new_viewport_height = int(new_viewport_size.y)
if (new_viewport_width != screen_width) or (new_viewport_height != screen_height):
self.log(
f"set_viewport_size failed - expected ({screen_width},{screen_height}), got ({new_viewport_width},{new_viewport_height})"
)
self.test_success = False
success = False
# Turn off any display info like FPS, as that will mess up our image comparisons
# Turn off antialiasing as well
general.run_console("r_displayInfo=0")
general.run_console("r_antialiasingmode=0")
general.idle_wait(1.0)
return success
# Test Teardown
# Restore everything from above
# log test results, exit editor
def teardown(self) -> None:
# Restore the original Editor settings
settings.set_misc_editor_settings(self.original_settings)
# If the helper gizmos were on at the start, restore them
if self.helpers_visible:
general.toggle_helpers()
# Set the viewport back to whatever size it was at the start and restore the pane layout
general.set_viewport_size(int(self.viewport_size.x), int(self.viewport_size.y))
general.set_viewport_expansion_policy("AutoExpand")
general.set_view_pane_layout(self.viewport_layout)
general.update_viewport()
self.log("test finished")
if self.test_success:
self.log("result=SUCCESS")
general.set_result_to_success()
else:
self.log("result=FAILURE")
general.set_result_to_failure()
if not self.editor_already_running:
general.exit_no_prompt()
def run_test(self) -> None:
self.log("run")
def run(self) -> None:
self.setup()
# Only run the actual test if we didn't have setup issues
if self.test_success:
self.run_test()
self.teardown()
def get_arg(self, arg_name: str) -> str:
if arg_name in self.args:
return self.args[arg_name]
return ""
# general logger that adds prefix?
def log(self, log_line: str) -> None:
Report.info(self.log_prefix + log_line)
# isclose: Compares two floating-point values for "nearly-equal"
def isclose(self, a: float, b: float, rel_tol: float = 1e-9, abs_tol: float = 0.0) -> bool:
return abs(a - b) <= max(rel_tol * max(abs(a), abs(b)), abs_tol)
# Create a new empty level
def create_level(
self,
level_name: str,
heightmap_resolution: int = 1024,
heightmap_meters_per_pixel: int = 1,
terrain_texture_resolution: int = 4096,
use_terrain: bool = False,
bypass_viewport_resize: bool = False,
) -> bool:
self.log(f"Creating level {level_name}")
result = general.create_level_no_prompt(
level_name, heightmap_resolution, heightmap_meters_per_pixel, terrain_texture_resolution, use_terrain
)
# Result codes are ECreateLevelResult defined in CryEdit.h
if result == 1:
self.log(f"{level_name} level already exists")
elif result == 2:
self.log("Failed to create directory")
elif result == 3:
self.log("Directory length is too long")
elif result != 0:
self.log("Unknown error, failed to create level")
else:
self.log(f"{level_name} level created successfully")
# If the editor is already running, allow "level already exists" to count as success
if (result == 0) or (self.editor_already_running and (result == 1)):
# For successful level creation, call the post-load step.
if self.after_level_load(bypass_viewport_resize):
result = 0
else:
result = -1
return result == 0
def open_level(self, level_name: str, bypass_viewport_resize: bool = False) -> bool:
# Open the level non-interactively
if self.editor_already_running and (general.get_current_level_name() == level_name):
self.log(f"Level {level_name} already open")
result = True
else:
self.log(f"Opening level {level_name}")
result = general.open_level_no_prompt(level_name)
result = result and self.after_level_load(bypass_viewport_resize)
if result:
self.log(f"Successfully opened {level_name}")
else:
self.log(f"Unknown error, {level_name} level failed to open")
return result
# Take Screenshot
def take_viewport_screenshot(
self, posX: float, posY: float, posZ: float, rotX: float, rotY: float, rotZ: float
) -> None:
# Set our camera position / rotation and wait for the Editor to acknowledge it
general.set_current_view_position(posX, posY, posZ)
general.set_current_view_rotation(rotX, rotY, rotZ)
general.idle_wait(1.0)
# Request a screenshot and wait for the Editor to process it
general.run_console("r_GetScreenShot=2")
general.idle_wait(1.0)
def enter_game_mode(self, success_message: str) -> None:
"""
:param success_message: The str with the expected message for entering game mode.
:return: None
"""
Report.info("Entering game mode")
general.enter_game_mode()
general.idle_wait_frames(1)
self.critical_result(success_message, general.is_in_game_mode())
def exit_game_mode(self, success_message: str) -> None:
"""
:param success_message: The str with the expected message for exiting game mode.
:return: None
"""
Report.info("Exiting game mode")
general.exit_game_mode()
general.idle_wait_frames(1)
self.critical_result(success_message, not general.is_in_game_mode())
def critical_result(self, success_message: str, condition: bool, fast_fail_message: str = None) -> None:
"""
if condition is False we will fail fast
:param success_message: messages to print based on the condition
:param condition: success (True) or failure (False)
:param fast_fail_message: [optional] message to include on fast fail
"""
if not isinstance(condition, bool):
raise TypeError("condition argument must be a bool")
if not Report.result(success_message, condition):
self.test_success = False
self.fail_fast(fast_fail_message)
def fail_fast(self, message: str = None) -> None:
"""
A state has been reached where progressing in the test is not viable.
raises FailFast
:return: None
"""
Report.info("Failing fast. Raising an exception and shutting down the editor.")
if message:
Report.info(f"Fail fast message: {message}")
self.teardown()
raise RuntimeError
def wait_for_condition(self, function, timeout_in_seconds=1.0):
# type: (function, float) -> bool
"""
**** Will be replaced by a function of the same name exposed in the Engine*****
a function to run until it returns True or timeout is reached
the function can have no parameters and
waiting idle__wait_* is handled here not in the function
:param function: a function that returns a boolean indicating a desired condition is achieved
:param timeout_in_seconds: when reached, function execution is abandoned and False is returned
"""
with Timeout(timeout_in_seconds) as t:
while True:
try:
general.idle_wait_frames(1)
except Exception:
print("WARNING: Couldn't wait for frame")
if t.timed_out:
return False
ret = function()
if not isinstance(ret, bool):
raise TypeError("return value for wait_for_condition function must be a bool")
if ret:
return True
class Timeout:
# type: (float) -> None
"""
contextual timeout
:param seconds: float seconds to allow before timed_out is True
"""
def __init__(self, seconds):
self.seconds = seconds
def __enter__(self):
self.die_after = time.time() + self.seconds
return self
def __exit__(self, type, value, traceback):
pass
@property
def timed_out(self):
return time.time() > self.die_after
@@ -0,0 +1,425 @@
"""
All or portions of this file Copyright (c) Amazon.com, Inc. or its affiliates or
its licensors.
For complete copyright and license terms please see the LICENSE at the root of this
distribution (the "License"). All use of this software is governed by the License,
or, if provided, by the license below or the license accompanying this file. Do not
remove or modify any license notices. This file is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
"""
import azlmbr.bus as bus
import azlmbr.editor as editor
import azlmbr.entity as entity
import azlmbr.object
from typing import List
from math import isclose
import collections.abc
def find_entity_by_name(entity_name):
"""
Gets an entity ID from the entity with the given entity_name
:param entity_name: String of entity name to search for
:return entity ID
"""
search_filter = entity.SearchFilter()
search_filter.names = [entity_name]
matching_entity_list = entity.SearchBus(bus.Broadcast, 'SearchEntities', search_filter)
if matching_entity_list:
matching_entity = matching_entity_list[0]
if matching_entity.IsValid():
print(f'{entity_name} entity found with ID {matching_entity.ToString()}')
return matching_entity
else:
return matching_entity_list
def get_component_type_id(component_name):
"""
Gets the component_type_id from a given component name
:param component_name: String of component name to search for
:return component type ID
"""
type_ids_list = editor.EditorComponentAPIBus(bus.Broadcast, 'FindComponentTypeIdsByEntityType', [component_name],
entity.EntityType().Game)
component_type_id = type_ids_list[0]
return component_type_id
def add_level_component(component_name):
"""
Adds the specified component to the Level Inspector
:param component_name: String of component name to search for
:return Component object.
"""
level_component_list = [component_name]
level_component_type_ids_list = editor.EditorComponentAPIBus(bus.Broadcast, 'FindComponentTypeIdsByEntityType',
level_component_list, entity.EntityType().Level)
level_component_outcome = editor.EditorLevelComponentAPIBus(bus.Broadcast, 'AddComponentsOfType',
[level_component_type_ids_list[0]])
level_component = level_component_outcome.GetValue()[0]
return level_component
def add_component(componentName, entityId):
"""
Given a component name, finds component TypeId, adds to given entity, and verifies successful add/active state.
:param componentName: String of component name to add.
:param entityId: Entity to add component to.
:return: Component object.
"""
typeIdsList = editor.EditorComponentAPIBus(bus.Broadcast, 'FindComponentTypeIdsByEntityType', [componentName],
entity.EntityType().Game)
typeNamesList = editor.EditorComponentAPIBus(bus.Broadcast, 'FindComponentTypeNames', typeIdsList)
componentOutcome = editor.EditorComponentAPIBus(bus.Broadcast, 'AddComponentsOfType', entityId, typeIdsList)
isActive = editor.EditorComponentAPIBus(bus.Broadcast, 'IsComponentEnabled', componentOutcome.GetValue()[0])
hasComponent = editor.EditorComponentAPIBus(bus.Broadcast, 'HasComponentOfType', entityId, typeIdsList[0])
if componentOutcome.IsSuccess() and isActive:
print('{} component was added to entity'.format(typeNamesList[0]))
elif componentOutcome.IsSuccess() and not isActive:
print('{} component was added to entity, but the component is disabled'.format(typeNamesList[0]))
elif not componentOutcome.IsSuccess():
print('Failed to add {} component to entity'.format(typeNamesList[0]))
if hasComponent:
print('Entity has a {} component'.format(typeNamesList[0]))
return componentOutcome.GetValue()[0]
def add_component_of_type(componentTypeId, entityId):
typeIdsList = [componentTypeId]
componentOutcome = editor.EditorComponentAPIBus(
azlmbr.bus.Broadcast, 'AddComponentsOfType', entityId, typeIdsList)
return componentOutcome.GetValue()[0]
def remove_component(component_name, entity_id):
"""
Removes the specified component from the specified entity.
:param component_name: String of component name to remove.
:param entity_id: Entity to remove component from.
:return: EntityComponentIdPair if removal was successful, else None.
"""
type_ids_list = [get_component_type_id(component_name)]
outcome = editor.EditorComponentAPIBus(bus.Broadcast, 'GetComponentOfType', entity_id, type_ids_list[0])
if outcome.IsSuccess():
component_entity_pair = outcome.GetValue()
editor.EditorComponentAPIBus(bus.Broadcast, 'RemoveComponents', [component_entity_pair])
has_component = editor.EditorComponentAPIBus(bus.Broadcast, 'HasComponentOfType', entity_id, type_ids_list[0])
if has_component:
print(f"Failed to remove {component_name}")
return None
else:
print(f"{component_name} was successfully removed")
return component_entity_pair
else:
print(f"{component_name} not found on entity")
return None
def get_component_property_value(component, component_propertyPath):
"""
Given a component name and component property path, outputs the property's value.
:param component: Component object to act on.
:param componentPropertyPath: String of component property. (e.g. 'Settings|Visible')
:return: Value set in given componentPropertyPath
"""
componentPropertyObj = editor.EditorComponentAPIBus(bus.Broadcast, 'GetComponentProperty', component,
component_propertyPath)
if componentPropertyObj.IsSuccess():
componentProperty = componentPropertyObj.GetValue()
print(f'{component_propertyPath} set to {componentProperty}')
return componentProperty
else:
print(f'FAILURE: Could not get value from {component_propertyPath}')
return None
def get_property_tree(component):
"""
Given a configured component object, prints the property tree info from that component
:param component: Component object to act on.
"""
pteObj = editor.EditorComponentAPIBus(bus.Broadcast, 'BuildComponentPropertyTreeEditor', component)
pte = pteObj.GetValue()
print(pte.build_paths_list())
return pte
def compare_values(first_object: object, second_object: object, name: str) -> bool:
# Quick case - can we just directly compare the two objects successfully?
if (first_object == second_object):
result = True
# No, so get a lot more specific
elif isinstance(first_object, collections.abc.Container):
# If they aren't both containers, they're different
if not isinstance(second_object, collections.abc.Container):
result = False
# If they have different lengths, they're different
elif len(first_object) != len (second_object):
result = False
# If they're different strings, they're containers but they failed the == check so
# we know they're different
elif isinstance(first_object, str):
result = False
else:
# It's a collection of values, so iterate through them all...
collection_idx = 0
result = True
for val1, val2 in zip(first_object, second_object):
result = result and compare_values(val1, val2, f"{name} (index [{collection_idx}])")
collection_idx = collection_idx + 1
else:
# Do approximate comparisons for floats
if isinstance(first_object, float) and isclose(first_object, second_object, rel_tol=0.001):
result = True
# We currently don't have a generic way to compare PythonProxyObject contents, so return a
# false positive result for now.
elif isinstance(first_object, azlmbr.object.PythonProxyObject):
print(f"{name}: validation inconclusive, the two objects cannot be directly compared.")
result = True
else:
result = False
if not result:
print(f"compare_values failed: {first_object} ({type(first_object)}) vs {second_object} ({type(second_object)})")
print(f"{name}: {'SUCCESS' if result else 'FAILURE'}")
return result
class Entity:
"""
Entity class used to create entity objects
:param name: String for the name of the Entity
:param id: The ID of the entity
"""
def __init__(self, name: str, id: object = entity.EntityId()):
self.name: str = name
self.id: object = id
self.components: List[object] = None
self.parent_id = None
self.parent_name = None
def create_entity(self, entity_position, components, parent_id=entity.EntityId()):
self.id = editor.ToolsApplicationRequestBus(
bus.Broadcast, "CreateNewEntityAtPosition", entity_position, parent_id
)
if self.id.IsValid():
print(f"{self.name} Entity successfully created")
editor.EditorEntityAPIBus(bus.Event, 'SetName', self.id, self.name)
self.components = []
for component in components:
self.add_component(component)
def add_component(self, component):
new_component = add_component(component, self.id)
self.components.append(new_component)
def add_component_of_type(self, componentTypeId):
new_component = add_component_of_type(componentTypeId, self.id)
self.components.append(new_component)
def remove_component(self, component):
removed_component = remove_component(component, self.id)
if removed_component is not None:
self.components.remove(removed_component)
def get_parent_info(self):
"""
Sets the value for parent_id and parent_name on the entity (self)
Prints the string for papertrail
:return: None
"""
self.parent_id = editor.EditorEntityInfoRequestBus(bus.Event, "GetParent", self.id)
self.parent_name = editor.EditorEntityInfoRequestBus(bus.Event, "GetName", self.parent_id)
print(f"The parent entity of {self.name} is {self.parent_name}")
def set_test_parent_entity(self, parent_entity_obj):
editor.EditorEntityAPIBus(bus.Event, "SetParent", self.id, parent_entity_obj.id)
self.get_parent_info()
def get_set_test(self, component_index: int, path: str, value: object, expected_result: object = None) -> bool:
"""
Used to set and validate changes in component values
:param component_index: Index location in the self.components list
:param path: asset path in the component
:param value: new value for the variable being changed in the component
:param expected_result: (optional) check the result against a specific expected value
"""
if expected_result is None:
expected_result = value
# Test Get/Set (get old value, set new value, check that new value was set correctly)
print(f"Entity {self.name} Path {path} Component Index {component_index} ")
component = self.components[component_index]
old_value = get_component_property_value(component, path)
if old_value is not None:
print(f"SUCCESS: Retrieved property Value for {self.name}")
else:
print(f"FAILURE: Failed to find value in {self.name} {path}")
return False
if old_value == expected_result:
print((f"WARNING: get_set_test on {self.name} is setting the same value that already exists ({old_value})."
"The set results will be inconclusive."))
editor.EditorComponentAPIBus(bus.Broadcast, "SetComponentProperty", component, path, value)
new_value = get_component_property_value(self.components[component_index], path)
if new_value is not None:
print(f"SUCCESS: Retrieved new property Value for {self.name}")
else:
print(f"FAILURE: Failed to find new value in {self.name}")
return False
return compare_values(new_value, expected_result, f"{self.name} {path}")
def get_set_test(entity: object, component_index: int, path: str, value: object) -> bool:
"""
Used to set and validate changes in component values
:param component_index: Index location in the entity.components list
:param path: asset path in the component
:param value: new value for the variable being changed in the component
"""
return entity.get_set_test(component_index, path, value)
def get_set_property_test(ly_object: object, attribute_name: str, value: object, expected_result: object = None) -> bool:
"""
Used to set and validate BehaviorContext property changes in Open 3D Engine objects
:param ly_object: The Open 3D Engine object to test
:param attribute_name: property (attribute) name in the BehaviorContext
:param value: new value for the variable being changed in the component
:param expected_result: (optional) check the result against a specific expected value other than the one set
"""
if expected_result is None:
expected_result = value
# Test Get/Set (get old value, set new value, check that new value was set correctly)
print(f"Attempting to set {ly_object.typename}.{attribute_name} = {value} (expected result is {expected_result})")
if hasattr(ly_object, attribute_name):
print(f"SUCCESS: Located attribute {attribute_name} for {ly_object.typename}")
else:
print(f"FAILURE: Failed to find attribute {attribute_name} in {ly_object.typename}")
return False
old_value = getattr(ly_object, attribute_name)
if old_value is not None:
print(f"SUCCESS: Retrieved existing value {old_value} for {attribute_name} in {ly_object.typename}")
else:
print(f"FAILURE: Failed to retrieve value for {attribute_name} in {ly_object.typename}")
return False
if old_value == expected_result:
print((f"WARNING: get_set_test on {attribute_name} is setting the same value that already exists ({old_value})."
"The 'set' result for the test will be inconclusive."))
setattr(ly_object, attribute_name, expected_result)
new_value = getattr(ly_object, attribute_name)
if new_value is not None:
print(f"SUCCESS: Retrieved new value {new_value} for {attribute_name} in {ly_object.typename}")
else:
print(f"FAILURE: Failed to retrieve value for {attribute_name} in {ly_object.typename}")
return False
return compare_values(new_value, expected_result, f"{ly_object.typename}.{attribute_name}")
def has_components(entity_id: object, component_list: list) -> bool:
"""
Used to verify if a given entity has all the components of components_list. Returns True if all the
components are present, else False
:param entity_id: entity id of the entity
:param component_list: list of component names to be verified
"""
typeIdsList = editor.EditorComponentAPIBus(bus.Broadcast, 'FindComponentTypeIdsByEntityType', component_list,
entity.EntityType().Game)
for type_id in typeIdsList:
if not editor.EditorComponentAPIBus(bus.Broadcast, 'HasComponentOfType', entity_id, type_id):
return False
return True
class PathNotFoundError(Exception):
def __init__(self, path):
self.path = path
def __str__(self):
return f"Path \"{self.path}\" not found in Editor Settings"
def get_editor_settings_path_list():
"""
Get the list of Editor Settings paths
"""
paths = editor.EditorSettingsAPIBus(bus.Broadcast, 'BuildSettingsList')
return paths
def get_editor_settings_by_path(path):
"""
Get the value of Editor Settings based on the path.
:param path: path to the Editor Settings to get the value
"""
if path not in get_editor_settings_path_list():
raise PathNotFoundError(path)
outcome = editor.EditorSettingsAPIBus(bus.Broadcast, 'GetValue', path)
if outcome.isSuccess():
return outcome.GetValue()
raise RuntimeError(f"GetValue for path '{path}' failed")
def set_editor_settings_by_path(path, value, is_bool = False):
"""
Set the value of Editor Settings based on the path.
# NOTE: Some Editor Settings may need an Editor restart to apply.
# Ex: Enabling or disabling New Viewport Interaction Model
:param path: path to the Editor Settings to get the value
:param value: value to be set
:param is_bool: True for Boolean settings (enable/disable), False for other settings
"""
if path not in get_editor_settings_path_list():
raise PathNotFoundError(path)
if is_bool and not isinstance(value, bool):
def ParseBoolValue(value):
if(value == "0"):
return False
return True
value = ParseBoolValue(value)
outcome = editor.EditorSettingsAPIBus(bus.Broadcast, 'SetValue', path, value)
if not outcome.isSuccess():
raise RuntimeError(f"SetValue for path '{path}' failed")
print(f"Value for path '{path}' is set to {value}")
def get_component_type_id_map(component_name_list):
"""
Given a list of component names, returns a map of component name -> component type id
:param component_name_list: The Open 3D Engine object to test
:return: Dictionary of component name -> component type id pairs
"""
# Remove any duplicates so we don't have to query for the same TypeId
component_names = list(set(component_name_list))
type_ids_by_component = {}
type_ids = editor.EditorComponentAPIBus(bus.Broadcast, 'FindComponentTypeIdsByEntityType', component_names,
entity.EntityType().Game)
for i, typeId in enumerate(type_ids):
type_ids_by_component[component_names[i]] = typeId
return type_ids_by_component
@@ -0,0 +1,137 @@
"""
All or portions of this file Copyright (c) Amazon.com, Inc. or its affiliates or
its licensors.
For complete copyright and license terms please see the LICENSE at the root of this
distribution (the "License"). All use of this software is governed by the License,
or, if provided, by the license below or the license accompanying this file. Do not
remove or modify any license notices. This file is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
"""
import logging
import os
import tempfile
import ly_test_tools.log.log_monitor
import ly_test_tools.environment.process_utils as process_utils
import ly_test_tools.environment.waiter as waiter
from ly_remote_console.remote_console_commands import RemoteConsole as RemoteConsole
from ly_remote_console.remote_console_commands import send_command_and_expect_response as send_command_and_expect_response
logger = logging.getLogger(__name__)
def teardown_editor(editor):
"""
:param editor: Configured editor object
:return:
"""
process_utils.kill_processes_named('AssetProcessor.exe')
logger.debug('Ensuring Editor is stopped')
editor.ensure_stopped()
def launch_and_validate_results(request, test_directory, editor, editor_script, expected_lines, unexpected_lines=[],
halt_on_unexpected=False, run_python="--runpythontest", auto_test_mode=True, null_renderer=False, cfg_args=[],
timeout=300):
"""
Runs the Editor with the specified script, and monitors for expected log lines.
:param request: Special fixture providing information of the requesting test function.
:param test_directory: Path to test directory that editor_script lives in.
:param editor: Configured editor object to run test against.
:param editor_script: Name of script that will execute in the Editor.
:param expected_lines: Expected lines to search log for.
:param unexpected_lines: Unexpected lines to search log for. Defaults to none.
:param halt_on_unexpected: Halts test if unexpected lines are found. Defaults to False.
:param run_python: Defaults to "--runpythontest", other option is "--runpython".
:param auto_test_mode: Determines if Editor will launch in autotest_mode, suppressing modal dialogs. Defaults to True.
:param null_renderer: Specifies the test does not require the renderer. Defaults to True.
:param cfg_args: Additional arguments for CFG, such as LevelName.
:param timeout: Length of time for test to run. Default is 60.
"""
test_case = os.path.join(test_directory, editor_script)
request.addfinalizer(lambda: teardown_editor(editor))
logger.debug("Running automated test: {}".format(editor_script))
editor.args.extend(["--skipWelcomeScreenDialog", "--regset=/Amazon/Settings/EnableSourceControl=false",
"--regset=/Amazon/Preferences/EnablePrefabSystem=false", run_python, test_case,
"--runpythonargs", " ".join(cfg_args)])
if auto_test_mode:
editor.args.extend(["--autotest_mode"])
if null_renderer:
editor.args.extend(["-NullRenderer"])
with editor.start():
editorlog_file = os.path.join(editor.workspace.paths.project_log(), 'Editor.log')
# Initialize the log monitor and set time to wait for log creation
log_monitor = ly_test_tools.log.log_monitor.LogMonitor(launcher=editor, log_file_path=editorlog_file)
log_monitor.log_creation_max_wait_time = timeout
# Check for expected/unexpected lines
log_monitor.monitor_log_for_lines(expected_lines=expected_lines, unexpected_lines=unexpected_lines,
halt_on_unexpected=halt_on_unexpected, timeout=timeout)
def launch_and_validate_results_launcher(launcher, level, remote_console_instance, expected_lines, null_renderer=True,
port_listener_timeout=120, log_monitor_timeout=300, remote_console_port=4600):
"""
Runs the launcher with the specified level, and monitors Game.log for expected lines.
:param launcher: Configured launcher object to run test against.
:param level: The level to load in the launcher.
:param remote_console_instance: Configured Remote Console object.
:param expected_lines: Expected lines to search log for.
:oaram null_renderer: Specifies the test does not require the renderer. Defaults to True.
:param port_listener_timeout: Timeout for verifying successful connection to Remote Console.
:param log_monitor_timeout: Timeout for monitoring for lines in Game.log
:param remote_console_port: The port used to communicate with the Remote Console.
"""
def _check_for_listening_port(port):
"""
Checks to see if the connection to the designated port was established.
:param port: Port to listen to.
:return: True if port is listening.
"""
port_listening = False
for conn in psutil.net_connections():
if 'port={}'.format(port) in str(conn):
port_listening = True
return port_listening
if null_renderer:
launcher.args.extend(["-NullRenderer"])
# Start the Launcher
with launcher.start():
# Ensure Remote Console can be reached
waiter.wait_for(
lambda: _check_for_listening_port(remote_console_port),
port_listener_timeout,
exc=AssertionError("Port {} not listening.".format(remote_console_port)),
)
remote_console_instance.start(timeout=30)
# Load the specified level in the launcher
send_command_and_expect_response(remote_console_instance,
f"map {level}",
"LEVEL_LOAD_COMPLETE", timeout=30)
# Monitor the console for expected lines
for line in expected_lines:
assert remote_console_instance.expect_log_line(line, log_monitor_timeout), f"Expected line not found: {line}"
def remove_files(artifact_path, suffix):
"""
Removes files with the specified suffix from the specified path
:param artifact_path: Path to search for files
:param suffix: File extension to remove
"""
if not os.path.isdir(artifact_path):
return
for file_name in os.listdir(artifact_path):
if file_name.endswith(suffix):
os.remove(os.path.join(artifact_path, file_name))
@@ -0,0 +1,75 @@
"""
All or portions of this file Copyright (c) Amazon.com, Inc. or its affiliates or
its licensors.
For complete copyright and license terms please see the LICENSE at the root of this
distribution (the "License"). All use of this software is governed by the License,
or, if provided, by the license below or the license accompanying this file. Do not
remove or modify any license notices. This file is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
"""
import PySide2
import editor_python_test_tools.pyside_utils
def get_component_combobox_values(component_name, property_name, log_fn=None):
"""
Retrieves the Combo box values from a Component. Assumes the Entity has been selected and that the component is
visible in the Entity Inspector.
Works by inspecting the list of child widgets for the Entity from the Entity Inspector, looking for the right type
of widget (QFrame) with a label containing the name of the component. Then, the QFrame is inspected to find a
QComboBox with the property name.
:param component_name: Name of the component to inspect.
:param property_name: Name of the property to inspect.
:param log_fn: Function used to log messages, should take a string as argument like log_fn('message to log').
:return: A list containing the values from the Combo box.
"""
def _log_fn_wrapper(message):
log_fn(message) if log_fn else None
editor_window = pyside_utils.get_editor_main_window()
entity_inspector = editor_window.findChild(PySide2.QtWidgets.QDockWidget, 'Entity Inspector')
assert entity_inspector, 'Entity Inspector widget is not valid.'
entity_inspector.update()
component_list_widget = entity_inspector.findChild(PySide2.QtWidgets.QWidget, 'm_componentListContents')
component_list_children = component_list_widget.children()
assert component_list_children, 'Could not retrieve components for the entity.'
# Iterate over the widgets that are children of the component list. On each one, retrieve the first child QFrame
# that has a label with the "component_name" as text.
# If that label is found, the same QFrame is inspected for a child frame with the name matching the "property_name".
# This property frame should have the combo box as child.
for component_widget in component_list_children:
if type(component_widget) is PySide2.QtWidgets.QFrame:
component_label = pyside_utils.find_child_by_pattern(component_widget, {'text': component_name})
if component_label:
property_frame = component_widget.findChild(PySide2.QtWidgets.QFrame, property_name)
if not property_frame:
_log_fn_wrapper(f'QFrame not found as child of the component widget {component_widget}.')
continue
combobox = pyside_utils.find_child_by_pattern(property_frame, {'type': PySide2.QtWidgets.QComboBox})
if not combobox:
_log_fn_wrapper(f'QComboBox not found as child of the property frame {property_frame}.')
continue
item_count = combobox.count()
values = []
for index in range(item_count):
values.append(combobox.itemText(index))
if not values:
_log_fn_wrapper('The QComboBox does not have values to retrieve.')
return values
_log_fn_wrapper('Matching component and property not found in Component list, or the list is empty.')
return None
@@ -0,0 +1,938 @@
"""
All or portions of this file Copyright (c) Amazon.com, Inc. or its affiliates or
its licensors.
For complete copyright and license terms please see the LICENSE at the root of this
distribution (the "License"). All use of this software is governed by the License,
or, if provided, by the license below or the license accompanying this file. Do not
remove or modify any license notices. This file is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
"""
import azlmbr.qt
import azlmbr.qt_helpers
import asyncio
import re
from shiboken2 import wrapInstance, getCppPointer
from PySide2 import QtCore, QtWidgets, QtGui, QtTest
from PySide2.QtWidgets import QAction, QWidget
from PySide2.QtCore import Qt
from PySide2.QtTest import QTest
import azlmbr.legacy.general as general
import traceback
import threading
import types
qApp = QtWidgets.QApplication.instance()
class LmbrQtEventLoop(asyncio.AbstractEventLoop):
def __init__(self):
self.running = False
self.shutdown = threading.Event()
self.blocked_events = set()
self.finished_events = set()
self.queue = []
self._wait_future = None
self._event_loop_nesting = 0
def get_debug(self):
return False
def time(self):
return azlmbr.qt_helpers.time()
def wait_for_condition(self, condition, action, on_timeout=None, timeout=1.0):
timeout = self.time() + timeout if timeout is not None else None
def callback(time):
# Run our action and remove us from the queue if our condition is satisfied
if condition():
action()
return True
# Give up if timeout has elapsed
if time > timeout:
if on_timeout is not None:
on_timeout()
return True
return False
self.queue.append((callback))
def event_loop(self):
time = self.time()
def run_event(event):
if event in self.blocked_events or event in self.finished_events:
return False
self.blocked_events.add(event)
try:
if event(time):
self.finished_events.add(event)
except Exception:
traceback.print_exc()
self.finished_events.add(event)
finally:
self.blocked_events.remove(event)
self._event_loop_nesting += 1
try:
for event in self.queue:
run_event(event)
finally:
self._event_loop_nesting -= 1
# Clear out any finished events if the queue is safe to mutate
if self._event_loop_nesting == 0:
self.queue = [event for event in self.queue if event not in self.finished_events]
self.finished_events = set()
if not self.running or self._wait_future is not None and self._wait_future.done():
self.close()
def run_until_shutdown(self):
# Run our event loop callback (via azlmbr.qt_helpers) by pumping the Qt event loop
# azlmbr.qt_helpers will attempt to ensure our event loop is always run, even when a
# new event loop is started and run from the main event loop
self.running = True
self.shutdown.clear()
azlmbr.qt_helpers.set_loop_callback(self.event_loop)
while not self.shutdown.is_set():
qApp.processEvents(QtCore.QEventLoop.AllEvents, 0)
def run_forever(self):
self._wait_future = None
self.run_until_shutdown()
def run_until_complete(self, future):
# Wrap coroutines into Tasks (future-like analogs)
if isinstance(future, types.CoroutineType):
future = self.create_task(future)
self._wait_future = future
self.run_until_shutdown()
def _timer_handle_cancelled(self, handle):
pass
def is_running(self):
return self.running
def is_closed(self):
return not azlmbr.qt_helpers.loop_is_running()
def stop(self):
self.running = False
def close(self):
self.running = False
self.shutdown.set()
azlmbr.qt_helpers.clear_loop_callback()
def shutdown_asyncgens(self):
pass
def call_exception_handler(self, context):
try:
raise context.get('exception', None)
except:
traceback.print_exc()
def call_soon(self, callback, *args, **kw):
h = asyncio.Handle(callback, args, self)
def callback_wrapper(time):
if not h.cancelled():
h._run()
return True
self.queue.append(callback_wrapper)
return h
def call_later(self, delay, callback, *args, **kw):
if delay < 0:
raise Exception("Can't schedule in the past")
return self.call_at(self.time() + delay, callback, *args)
def call_at(self, when, callback, *args, **kw):
h = asyncio.TimerHandle(when, callback, args, self)
h._scheduled = True
def callback_wrapper(time):
if time > when:
if not h.cancelled():
h._run()
return True
return False
self.queue.append(callback_wrapper)
return h
def create_task(self, coro):
return asyncio.Task(coro, loop=self)
def create_future(self):
return asyncio.Future(loop=self)
class EventLoopTimeoutException(Exception):
pass
event_loop = LmbrQtEventLoop()
def wait_for_condition(condition, timeout=1.0):
"""
Asynchronously waits for `condition` to evaluate to True.
condition: A function with the signature def condition() -> bool
This condition will be evaluated until it evaluates to True or the timeout elapses
timeout: The time in seconds to wait - if 0, this will wait forever
Throws pyside_utils.EventLoopTimeoutException on timeout.
"""
future = event_loop.create_future()
def on_complete():
future.set_result(True)
def on_timeout():
future.set_exception(EventLoopTimeoutException())
event_loop.wait_for_condition(condition, on_complete, on_timeout=on_timeout, timeout=timeout)
return future
async def wait_for(expression, timeout=1.0):
"""
Asynchronously waits for "expression" to evaluate to a non-None value,
then returns that value.
expression: A function with the signature def expression() -> Generic[Any,None]
The result of expression will be returned as soon as it returns a non-None value.
timeout: The time in seconds to wait - if 0, this will wait forever
Throws pyside_utils.EventLoopTimeoutException on timeout.
"""
result = None
def condition():
nonlocal result
result = expression()
return result is not None
await wait_for_condition(condition, timeout)
return result
def run_soon(fn):
"""
Runs a function on the event loop to enable asynchronous execution.
fn: The function to run, should be a function that takes no arguments
Returns a future that will be popualted with the result of fn or the exception it threw.
"""
future = event_loop.create_future()
def coroutine():
try:
fn()
future.set_result(True)
except Exception as e:
future.set_exception(e)
event_loop.call_soon(coroutine)
return future
def run_async(awaitable):
"""
Synchronously runs a coroutine or a future on the event loop.
This can be used in lieu of "await" in non-async functions.
awaitable: The coroutine or future to await.
Returns the result of operation specified.
"""
if isinstance(awaitable, types.CoroutineType):
awaitable = event_loop.create_task(awaitable)
event_loop.run_until_complete(awaitable)
return awaitable.result()
def wrap_async(fn):
"""
This decorator enables an async function's execution from a synchronous one.
For example:
@pyside_utils.wrap_async
async def foo():
result = await long_operation()
return result
def non_async_fn():
x = foo() # this will return the correct result by executing the event loop
fn: The function to wrap
Returns the decorated function.
"""
def wrapper(*args, **kw):
result = fn(*args, **kw)
return run_async(result)
return wrapper
def get_editor_main_window():
"""
Fetches the main Editor instance of QMainWindow for use with PySide tests
:return Instance of QMainWindow for the Editor
"""
params = azlmbr.qt.QtForPythonRequestBus(azlmbr.bus.Broadcast, "GetQtBootstrapParameters")
editor_id = QtWidgets.QWidget.find(params.mainWindowId)
main_window = wrapInstance(int(getCppPointer(editor_id)[0]), QtWidgets.QMainWindow)
return main_window
def get_action_for_menu_path(editor_window: QtWidgets.QMainWindow, main_menu_item: str, *menu_item_path: str):
"""
main_menu_item: Main menu item among the MenuBar actions. Ex: "File"
menu_item_path: Path to any nested menu item. Ex: "Viewport", "Goto Coordinates"
returns: QAction object for the corresponding path.
"""
# Check if path is valid
menu_bar = editor_window.menuBar()
menu_bar_actions = [index.iconText() for index in menu_bar.actions()]
# Verify if the given Menu exists in the Menubar
if main_menu_item not in menu_bar_actions:
print(f"QAction not found for main menu item '{main_menu_item}'")
return None
curr_action = menu_bar.actions()[menu_bar_actions.index(main_menu_item)]
curr_menu = curr_action.menu()
for index, element in enumerate(menu_item_path):
curr_menu_actions = [index.iconText() for index in curr_menu.actions()]
if element not in curr_menu_actions:
print(f"QAction not found for menu item '{element}'")
return None
if index == len(menu_item_path) - 1:
return curr_menu.actions()[curr_menu_actions.index(element)]
curr_action = curr_menu.actions()[curr_menu_actions.index(element)]
curr_menu = curr_action.menu()
return None
def _pattern_to_dict(pattern, **kw):
"""
Helper function, turns a pattern match parameter into a normalized dictionary
"""
def is_string_or_regex(x):
return isinstance(x, str) or isinstance(x, re.Pattern)
# If it's None, just make an empty dict
if pattern is None:
pattern = {}
# If our pattern is a string or regex, turn it into a text match
elif is_string_or_regex(pattern):
pattern = dict(text=pattern)
# If our pattern is an (int, int) tuple, turn it into a row/column match
elif isinstance(pattern, tuple) and isinstance(pattern[0], int) and isinstance(pattern[1], int):
pattern = dict(row=pattern[0], column=pattern[1])
# If our pattern is a QObject type, turn it into a type match
elif isinstance(pattern, type(QtCore.QObject)):
pattern = dict(type=pattern)
# Otherwise assume it's a dict and make a copy
else:
pattern = dict(pattern)
# Merge with any kw arguments
for key, value in kw.items():
pattern[key] = value
return pattern
def _match_pattern(obj, pattern):
"""
Helper function, determines whether obj matches the pattern specified by pattern.
It is required that pattern is normalized into a dict before calling this.
"""
def compare(value1, value2):
# Do a regex search if it's a regex, otherwise do a normal compare
if isinstance(value2, re.Pattern):
return re.search(value2, value1)
return value1 == value2
item_roles = Qt.ItemDataRole.values.values()
for key, value in pattern.items():
if key == "type": # Class type
if not isinstance(obj, value):
return False
elif key == "text": # Default 'text' path, depends on type
text_values = []
def get_from_attrs(*args):
for attr in args:
try:
text_values.append(getattr(obj, attr)())
except Exception:
pass
# Use any of the following fields for default matching, if they're defined
get_from_attrs("text", "objectName", "windowTitle")
# Additionally, use the DisplayRole for QModelIndexes
if isinstance(obj, QtCore.QModelIndex):
text_values.append(obj.data(Qt.DisplayRole))
if not any(compare(text, value) for text in text_values):
return False
elif key in item_roles: # QAbstractItemModel display role
if not isinstance(obj, QtCore.QModelIndex):
raise RuntimeError(f"Attempted to match data role on unsupported object {obj}")
if not compare(obj.data(key), value):
return False
elif hasattr(obj, key):
# Look up our key on the object itself
objectValue = getattr(obj, key)
# Invoke it if it's a getter
if callable(objectValue):
objectValue = objectValue()
if not compare(objectValue, value):
return False
else:
return False
return True
def get_child_indexes(model, parent_index=QtCore.QModelIndex()):
indexes = [parent_index]
while len(indexes) > 0:
parent_index = indexes.pop(0)
for row in range(model.rowCount(parent_index)):
# FIXME
# PySide appears to have a bug where-in it thinks columnCount is private
# Bail gracefully for now, we can add a C++ wrapper to work around if needed
try:
column_count = model.columnCount(parent_index)
except Exception:
column_count = 1
for col in range(column_count):
cur_index = model.index(row, col, parent_index)
yield cur_index
def _get_children(obj):
"""
Helper function. Get the direct descendants from a given PySide object.
This includes all: QObject children, QActions owned by the object, and QModelIndexes if applicable
"""
if isinstance(obj, QtCore.QObject):
yield from obj.children()
if isinstance(obj, QtWidgets.QWidget):
yield from obj.actions()
if isinstance(obj, (QtWidgets.QAbstractItemView, QtCore.QModelIndex)):
model = obj.model()
if model is None:
return
# For a QAbstractItemView (e.g. QTreeView, QListView), the parent index
# will be an invalid QModelIndex(), which will use find all indexes on the root.
# For a QModelIndex, we use the actual QModelIndex as the parent_index so that
# it will find any child indexes under it
parent_index = QtCore.QModelIndex()
if isinstance(obj, QtCore.QModelIndex):
parent_index = obj
yield from get_child_indexes(model, parent_index)
def _get_parents_to_search(obj_entry_or_list):
"""
Helper function, turns obj_entry_or_list into a list of parents to search
If obj_entry_or_list is None, returns all visible top level widgets
If obj_entry_or_list is iterable, return it as a list
Otherwise, return a list containing obj_entry_or_list
"""
if obj_entry_or_list is None:
return [widget for widget in QtWidgets.QApplication.topLevelWidgets() if widget.isVisible()]
try:
return list(obj_entry_or_list)
except TypeError:
return [obj_entry_or_list]
def find_children_by_pattern(obj=None, pattern=None, recursive=True, **kw):
"""
Finds the children of an object that match a given pattern.
See find_child_by_pattern for more information on usage.
"""
pattern = _pattern_to_dict(pattern, **kw)
parents_to_search = _get_parents_to_search(obj)
while len(parents_to_search) > 0:
parent = parents_to_search.pop(0)
for child in _get_children(parent):
if _match_pattern(child, pattern):
yield child
if recursive:
parents_to_search.append(child)
def find_child_by_pattern(obj=None, pattern=None, recursive=True, **kw):
"""
Finds the child of an object that matches a given pattern.
A "child" in this context is not necessarily a QObject child.
QActions are also considered children, as are the QModelIndex children of QAbstractItemViews.
obj: The object to search - should be either a QObject or a QModelIndex, or a list of them
If None this will search all top level windows.
pattern: The pattern to match, the first child that matches all of the criteria specified will
be returned. This is a dictionary with any combination of the following:
- "text": generic text to match, will search object names for QObjects, display role text
for QModelIndexes, or action text() for QActions
- "type": a class type, e.g. QtWidgets.QMenu, a child will only match if it's of this type
- "row" / "column": integer row and column indices of a QModelIndex
- "type": type class (e.g. PySide.QtWidgets.QComboBox) that the object must inherit from
- A Qt.ItemDataRole: matches for QModelIndexes with data of a given value
- Any other fields will fall back on being looked up on the object itself by name, e.g.
{"windowTitle": "Foo"} would match a windowTitle named "Foo"
Any instances where a field is specified as text can also be specified as a regular expression:
find_child_by_pattern(obj, {text: re.compile("Foo_.*")}) would find a child with text starting
with "Foo_"
For convenience, these parameter types may also be specified as keyword arguments:
find_child_by_pattern(obj, text="foo", type=QtWidgets.QAction)
is equivalent to
find_child_by_pattern(obj, {"text": "foo", "type": QtWidgets.QAction})
If pattern is specified as a string, it will turn into a pattern matching "text":
find_child_by_pattern(obj, "foo")
is equivalent to
find_child_by_pattern(obj, {"text": "foo"})
If a pattern is specified as an (int, int) tuple, it will turn into a row/column match:
find_child_by_pattern(obj, (0, 2))
is equivalent to
find_child_by_pattern(obj, {"row": 0, "column": 2})
If a pattern is specified as a type, like PySide.QtWidgets.QLabel, it will turn into a type match:
find_child_by_pattern(obj, PySide.QtWidgets.QLabel)
is equivalent to
find_child_by_pattern(obj, {"type": PySide.QtWidgets.QLabel})
"""
# Return the first match result, if found
for match in find_children_by_pattern(obj, pattern=pattern, recursive=recursive, **kw):
return match
return None
def find_child_by_hierarchy(parent, *patterns):
"""
Searches for a hierarchy of children descending from parent.
parent: The Qt object (or list of Qt obejcts) to search within
If none, this will search all top level windows.
patterns: A list of patterns to match to find a hierarchy of descendants.
These patterns will be tested in order.
For example, to look for the QComboBox in a hierarchy like the following:
QWidget (window)
-QTabWidget
-QWidget named "m_exampleTab"
-QComboBox
One might invoke:
find_child_by_hierarchy(window, QtWidgets.QTabWidget, "m_exampleTab", QtWidgets.QComboBox)
Alternatively, "..." may be specified in place of a parent, where the hierarchy will match any
ancestors along the path, so the above might be shortened to:
find_child_by_hierarchy(window, ..., "m_exampleTab", QtWidgets.QComboBox)
"""
search_recursively = False
current_objects = _get_parents_to_search(parent)
for pattern in patterns:
# If it's an ellipsis, do the next search recursively as we're looking for any number of intermediate ancestors
if pattern is ...:
search_recursively = True
continue
candidates = []
for parent_candidate in current_objects:
candidates += find_children_by_pattern(parent_candidate, pattern=pattern, recursive=search_recursively)
if len(candidates) == 0:
return None
current_objects = candidates
search_recursively = False
return current_objects[0]
async def wait_for_child_by_hierarchy(parent, *patterns, timeout=1.0):
"""
Searches for a hierarchy of children descending from parent until timeout occurs.
Returns a future that will result in either the found child or an EventLoopTimeoutException.
See find_child_by_hierarchy for usage information.
"""
match = None
def condition():
nonlocal match
match = find_child_by_hierarchy(parent, *patterns)
return match is not None
await wait_for_condition(condition, timeout)
return match
async def wait_for_child_by_pattern(obj=None, pattern=None, recursive=True, timeout=1.0, **kw):
"""
Finds the child of an object that matches a given pattern.
Returns a future that will result in either the found child or an EventLoopTimeoutException.
See find_child_by_hierarchy for usage information.
"""
match = None
def condition():
nonlocal match
match = find_child_by_pattern(obj, pattern, recursive, **kw)
return match is not None
await wait_for_condition(condition, timeout)
return match
def find_child_by_property(obj, obj_type, property_name, property_value, reg_exp_search=False):
"""
Finds the child of an object which has the property name matching the property value
of type obj_type
obj: The property value is searched through obj children
obj_type: Type of object to be matched
property_name: Property of the child which should be verified for the required value.
property_value: Property value that needs to be matched
reg_exp_search: If True searches for the property_value based on re search. Defaults to False.
"""
for child in obj.children():
if reg_exp_search and re.search(property_value, getattr(child, property_name)()):
return child
if not reg_exp_search and isinstance(child, obj_type) and getattr(child, property_name)() == property_value:
return child
return None
def get_item_view_index(item_view, row, column=0, parent=QtCore.QModelIndex()):
"""
Retrieve the index for a specified row/column, with optional parent
This is necessary when needing to reference into nested hierarchies in a QTreeView
item_view: The QAbstractItemView instance
row: The requested row index
column: The requested column index (defaults to 0 in case of single column)
parent: Parent index (defaults to invalid)
"""
item_model = item_view.model()
model_index = item_model.index(row, column, parent)
return model_index
def get_item_view_index_rect(item_view, index):
"""
Gets the QRect for a given index in a QAbstractItemView (e.g. QTreeView, QTableView, QListView).
This is helpful because for sending mouse events to a QAbstractItemView, you have to send them to
the viewport() widget of the QAbstractItemView.
item_view: The QAbstractItemView instance
index: A QModelIndex for the item index
"""
return item_view.visualRect(index)
def item_view_index_mouse_click(item_view, index, button=QtCore.Qt.LeftButton, modifier=QtCore.Qt.NoModifier):
"""
Helper method version of QTest.mouseClick for injecting mouse clicks on a QAbstractItemView
item_view: The QAbstractItemView instance
index: A QModelIndex for the item index to be clicked
"""
item_index_rect = get_item_view_index_rect(item_view, index)
item_index_center = item_index_rect.center()
# For QAbstractItemView widgets, the events need to be forwarded to the actual viewport() widget
QTest.mouseClick(item_view.viewport(), button, modifier, item_index_center)
def item_view_mouse_click(item_view, row, column=0, button=QtCore.Qt.LeftButton, modifier=QtCore.Qt.NoModifier):
"""
Helper method version of 'item_view_index_mouse_click' using a row, column instead of a QModelIndex
item_view: The QAbstractItemView instance
row: The requested row index
column: The requested column index (defaults to 0 in case of single column)
"""
index = get_item_view_index(item_view, row, column)
item_view_index_mouse_click(item_view, index, button, modifier)
async def wait_for_action_in_menu(menu, pattern, timeout=1.0):
"""
Finds a QAction inside a menu, based on the specified pattern.
menu: The QMenu to search
pattern: The action text or pattern to match (see find_child_by_pattern)
If pattern specifies a QWidget, this will search for the associated QWidgetAction
"""
action = await wait_for_child_by_pattern(menu, pattern, timeout=timeout)
if action is None:
raise TimeoutError(f"Failed to find context menu action for {pattern}")
# If we've found a valid QAction, we're good to go
if hasattr(action, 'trigger'):
return action
# If pattern matches a widget and not a QAction, look for an associated QWidgetAction
widget_actions = find_children_by_pattern(menu, type=QtWidgets.QWidgetAction)
underlying_widget_action = None
for widget_action in widget_actions:
widgets_to_check = [widget_action.defaultWidget()] + widget_action.createdWidgets()
for check_widget in widgets_to_check:
if action in _get_children(check_widget):
underlying_widget_action = widget_action
break
if underlying_widget_action is not None:
action = underlying_widget_action
break
if not hasattr(action, 'trigger'):
raise RuntimeError(f"Failed to find action associated with widget {action}")
return action
def queue_hide_event(widget):
"""
Explicitly post a hide event for the next frame, this can be used to ensure modal dialogs exit correctly.
widget: The widget to hide
"""
qApp.postEvent(widget, QtGui.QHideEvent())
async def wait_for_destroyed(obj, timeout=1.0):
"""
Waits for a QObject (including a widget) to be fully destroyed
This can be used to wait for a modal dialog to shut down properly
obj: The object to wait on destruction
timeout: The time, in seconds to wait. 0 for an indefinite wait.
"""
was_destroyed = False
def on_destroyed():
nonlocal was_destroyed
was_destroyed = True
obj.destroyed.connect(on_destroyed)
return await wait_for_condition(lambda: was_destroyed, timeout=timeout)
async def close_modal(modal_widget, timeout=1.0):
"""
Closes a modal dialog and waits for it to be cleaned up.
This attempts to ensure the modal event loop gets properly exited.
modal_widget: The widget to close
timeout: The time, in seconds, to wait. 0 for an indefinite wait.
"""
queue_hide_event(modal_widget)
return await wait_for_destroyed(modal_widget, timeout=timeout)
def trigger_context_menu_entry(widget, pattern, pos=None, index=None):
"""
Trigger a context menu event on a widget and activate an entry
widget: The widget to trigger the event on
pattern: The action text or pattern to match (see find_child_by_pattern)
pos: Optional, the QPoint to set as the event origin
index: Optional, the QModelIndex to click in widget
widget must be a QAbstractItemView
"""
async def async_wrapper():
menu = await open_context_menu(widget, pos=pos, index=index)
action = await wait_for_action_in_menu(menu, pattern)
action.trigger()
queue_hide_event(menu)
result = async_wrapper()
# If we have an event loop, go ahead and just return the coroutine
# Otherwise, do a synchronous wait
if event_loop.is_running():
return result
else:
return run_async(result)
async def open_context_menu(widget, pos=None, index=None, timeout=1.0):
"""
Trigger a context menu event on a widget
widget: The widget to trigger the event on
pos: Optional, the QPoint to set as the event origin
index: Optional, the QModelIndex to click in widget
widget must be a QAbstractItemView
Returns the menu that was created.
"""
if index is not None:
if pos is not None:
raise RuntimeError("Error: 'index' and 'pos' are mutually exclusive")
pos = widget.visualRect(index).center()
parent = widget
widget = widget.viewport()
pos = widget.mapFrom(parent, pos)
if pos is None:
pos = widget.rect().center()
# Post both a mouse event and a context menu to let the widget handle whichever is appropriate
qApp.postEvent(widget, QtGui.QContextMenuEvent(QtGui.QContextMenuEvent.Mouse, pos))
QtTest.QTest.mouseClick(widget, Qt.RightButton, Qt.NoModifier, pos)
menu = None
# Wait for a menu popup
def menu_has_focus():
nonlocal menu
for fw in [QtWidgets.QApplication.activePopupWidget(), QtWidgets.QApplication.activeModalWidget(),
QtWidgets.QApplication.focusWidget(), QtWidgets.QApplication.activeWindow()]:
if fw and isinstance(fw, QtWidgets.QMenu) and fw.isVisible():
menu = fw
return True
return False
await wait_for_condition(menu_has_focus, timeout)
return menu
def move_mouse(widget, position):
"""
Helper method to move the mouse to a specified position on a widget
widget: The widget to trigger the event on
position: The QPoint (local to widget) to move the mouse to
"""
# For some reason, Qt wouldn't register the mouse movement correctly unless both of these ways are invoked.
# The QTest.mouseMove seems to update the global cursor position, but doesn't always result in the MouseMove event being
# triggered, which prevents drag/drop being able to be simulated.
# Similarly, if only the MouseMove event is sent by itself to the core application, the global cursor position wasn't
# updated properly, so drag/drop logic that depends on grabbing the globalPos didn't work.
QtTest.QTest.mouseMove(widget, position)
event = QtGui.QMouseEvent(QtCore.QEvent.MouseMove, position, widget.mapToGlobal(position), QtCore.Qt.LeftButton, QtCore.Qt.LeftButton, QtCore.Qt.NoModifier)
QtCore.QCoreApplication.sendEvent(widget, event)
def drag_and_drop(source, target, source_point = QtCore.QPoint(), target_point = QtCore.QPoint()):
"""
Simulate a drag/drop event from a source object to a specified target
This has special case handling if the source is a QDockWidget (for docking) vs normal drag/drop
source: The source object to initiate the drag from
This is either a QWidget, or a tuple of (QAbstractItemView, QModelIndex) for dragging an item view item
target: The target object to drop on after dragging
This is either a QWidget, or a tuple of (QAbstractItemView, QModelIndex) for dropping on an item view item
source_point: Optional, The QPoint to initiate the drag from. If none is specified, the center of the source will be used.
target_point: Optional, The QPoint to drop on. If none is specified, the center of the target will be used.
"""
# Flag if this drag/drop is for docking, which has some special cases
docking = False
# If the source is a tuple of (QAbstractItemView, QModelIndex), we need to use the
# viewport() as the source, and find the location of the index
if isinstance(source, tuple) and len(source) == 2:
source_item_view = source[0]
source_widget = source_item_view.viewport()
source_model_index = source[1]
source_rect = source_item_view.visualRect(source_model_index)
else:
# There are some special case actions if we are doing this drag for docking,
# so figure this out by checking if the source is a QDockWidget
if isinstance(source, QtWidgets.QDockWidget):
docking = True
source_widget = source
source_rect = source.rect()
# If the target is a tuple of (QAbstractItemView, QModelIndex), we need to use the
# viewport() as the target, and find the location of the index
if isinstance(target, tuple) and len(target) == 2:
target_item_view = target[0]
target_widget = target_item_view.viewport()
target_model_index = target[1]
target_rect = target_item_view.visualRect(target_model_index)
else:
# If we are doing a drag for docking, we actually want all the mouse events
# to still be directed through the source widget
if docking:
target_widget = source_widget
else:
target_widget = target
target_rect = target.rect()
# If no source_point is specified, we need to find the center point of
# the source widget
if source_point.isNull():
# If we are dragging for docking, initiate the drag from the center of the
# dock widget title bar
if docking:
title_bar_widget = source.titleBarWidget()
if title_bar_widget:
source_point = title_bar_widget.geometry().center()
else:
raise RuntimeError("No titleBarWidget found for QDockWidget")
# Otherwise, can just find the center of the rect
else:
source_point = source_rect.center()
# If no target_point was specified, we need to find the center point of the target widget
if target_point.isNull():
target_point = target_rect.center()
# If we are dragging for docking and we aren't dragging within the same source/target,
# the mouse movements need to be directed to the source_widget, so we need to use the
# difference in global positions of our source and target widgets to adjust the target_point
# to be relative to the source
if docking and source != target:
source_top_left = source.mapToGlobal(QtCore.QPoint(0, 0))
target_top_left = target.mapToGlobal(QtCore.QPoint(0, 0))
offset = target_top_left - source_top_left
target_point += offset
# Move the mouse to the source spot where we will start the drag
move_mouse(source_widget, source_point)
# Press the left-mouse button to begin the drag
QtTest.QTest.mousePress(source_widget, QtCore.Qt.LeftButton, QtCore.Qt.NoModifier, source_point)
# If we are dragging for docking, we first need to drag the mouse past the minimum distance to
# trigger the docking system properly
if docking:
drag_distance = QtWidgets.QApplication.startDragDistance() + 1
docking_trigger_point = source_point + QtCore.QPoint(drag_distance, drag_distance)
move_mouse(source_widget, docking_trigger_point)
# Drag the mouse to the target widget over the desired point
move_mouse(target_widget, target_point)
# Release the left-mouse button to complete the drop.
# If we are docking, we need to delay the actual mouse button release because the docking system has
# a delay before the drop zone becomes active after it has been hovered, which can be found here:
# FancyDockingDropZoneConstants::dockingTargetDelayMS = 110 ms
# So we need to delay greater than dockingTargetDelayMS after the final mouse move
# over the intended target.
delay = -1
if docking:
delay = 200
QtTest.QTest.mouseRelease(target_widget, QtCore.Qt.LeftButton, QtCore.Qt.NoModifier, target_point, delay)
# Some drag/drop events have extra processing on the following event tick, so let those processEvents
# first before we complete the drag/drop operation
QtWidgets.QApplication.processEvents()
def trigger_action_async(action):
"""
Convenience function. Triggers an action asynchronously.
This can be used if calling action.trigger might block (e.g. if it opens a modal dialog)
action: The action to trigger
"""
return run_soon(lambda: action.trigger())
def click_button_async(button):
"""
Convenience function. Clicks a button asynchronously.
This can be used if calling button.click might block (e.g. if it opens a modal dialog)
button: The button to click
"""
return run_soon(lambda: button.click())
async def wait_for_modal_widget(timeout=1.0):
"""
Waits for an active modal widget and returns it.
"""
return await wait_for(lambda: QtWidgets.QApplication.activeModalWidget(), timeout=timeout)
async def wait_for_popup_widget(timeout=1.0):
"""
Waits for an active popup widget and returns it.
"""
return await wait_for(lambda: QtWidgets.QApplication.activePopupWidget(), timeout=timeout)
@@ -0,0 +1,367 @@
"""
All or portions of this file Copyright (c) Amazon.com, Inc. or its affiliates or
its licensors.
For complete copyright and license terms please see the LICENSE at the root of this
distribution (the "License"). All use of this software is governed by the License,
or, if provided, by the license below or the license accompanying this file. Do not
remove or modify any license notices. This file is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
"""
import os
import time
import math
import azlmbr
import azlmbr.legacy.general as general
import azlmbr.debug
import traceback
class FailFast(Exception):
"""
Raise to stop proceeding through test steps.
"""
pass
class TestHelper:
@staticmethod
def init_idle():
general.idle_enable(True)
# JIRA: SPEC-2880
# general.idle_wait_frames(1)
@staticmethod
def open_level(directory, level):
# type: (str, ) -> None
"""
:param level: the name of the level folder in AutomatedTesting\\Physics\\
:return: None
"""
Report.info("Open level {}/{}".format(directory, level))
success = general.open_level_no_prompt(os.path.join(directory, level))
if not success:
open_level_name = general.get_current_level_name()
if open_level_name == level:
Report.info("{} was already opened".format(level))
else:
assert False, "Failed to open level: {} does not exist or is invalid".format(level)
# FIX-ME: Expose call for checking when has been finished loading and change this frame waiting
# Jira: LY-113761
general.idle_wait_frames(200)
@staticmethod
def enter_game_mode(msgtuple_success_fail):
# type: (tuple) -> None
"""
:param msgtuple_success_fail: The tuple with the expected/unexpected messages for entering game mode.
:return: None
"""
Report.info("Entering game mode")
general.enter_game_mode()
TestHelper.wait_for_condition(lambda : general.is_in_game_mode(), 1.0)
Report.critical_result(msgtuple_success_fail, general.is_in_game_mode())
@staticmethod
def exit_game_mode(msgtuple_success_fail):
# type: (tuple) -> None
"""
:param msgtuple_success_fail: The tuple with the expected/unexpected messages for exiting game mode.
:return: None
"""
Report.info("Exiting game mode")
general.exit_game_mode()
TestHelper.wait_for_condition(lambda : not general.is_in_game_mode(), 1.0)
Report.critical_result(msgtuple_success_fail, not general.is_in_game_mode())
@staticmethod
def close_editor():
general.exit_no_prompt()
@staticmethod
def fail_fast(message=None):
# type: (str) -> None
"""
A state has been reached where progressing in the test is not viable.
raises FailFast
:return: None
"""
Report.info("Failing fast. Raising an exception and shutting down the editor.")
if message:
Report.info("Fail fast message: {}".format(message))
TestHelper.close_editor()
raise FailFast()
@staticmethod
def wait_for_condition(function, timeout_in_seconds):
# type: (function, float) -> bool
"""
**** Will be replaced by a function of the same name exposed in the Engine*****
a function to run until it returns True or timeout is reached
the function can have no parameters and
waiting idle__wait_* is handled here not in the function
:param function: a function that returns a boolean indicating a desired condition is achieved
:param timeout_in_seconds: when reached, function execution is abandoned and False is returned
"""
with Timeout(timeout_in_seconds) as t:
while True:
try:
general.idle_wait_frames(1)
except:
Report.info("WARNING: Couldn't wait for frame")
if t.timed_out:
return False
ret = function()
if not isinstance(ret, bool):
raise TypeError("return value for wait_for_condition function must be a bool")
if ret:
return True
class Timeout:
# type: (float) -> None
"""
contextual timeout
:param seconds: float seconds to allow before timed_out is True
"""
def __init__(self, seconds):
self.seconds = seconds
def __enter__(self):
self.die_after = time.time() + self.seconds
return self
def __exit__(self, type, value, traceback):
pass
@property
def timed_out(self):
return time.time() > self.die_after
class Report:
_results = []
_exception = None
@staticmethod
def start_test(test_function):
try:
test_function()
except Exception as ex:
Report._exception = traceback.format_exc()
Report.report_results(test_function)
@staticmethod
def report_results(test_function):
success = True
report = f"Report for {test_function.__name__}:\n"
for result in Report._results:
passed, info = result
success = success and passed
if passed:
report += f"[SUCCESS] {info}\n"
else:
report += f"[FAILED ] {info}\n"
if Report._exception:
report += "EXCEPTION raised:\n %s\n" % Report._exception[:-1].replace("\n", "\n ")
success = False
report += "Test result: "
report += "SUCCESS" if success else "FAILURE"
print(report)
general.report_test_result(success, report)
@staticmethod
def info(msg):
print("Info: {}".format(msg))
@staticmethod
def success(msgtuple_success_fail):
outcome = "Success: {}".format(msgtuple_success_fail[0])
print(outcome)
Report._results.append((True, outcome))
@staticmethod
def failure(msgtuple_success_fail):
outcome = "Failure: {}".format(msgtuple_success_fail[1])
print(outcome)
Report._results.append((False, outcome))
@staticmethod
def result(msgtuple_success_fail, condition):
if not isinstance(condition, bool):
raise TypeError("condition argument must be a bool")
if condition:
Report.success(msgtuple_success_fail)
else:
Report.failure(msgtuple_success_fail)
return condition
@staticmethod
def critical_result(msgtuple_success_fail, condition, fast_fail_message=None):
# type: (tuple, bool, str) -> None
"""
if condition is False we will fail fast
:param msgtuple_success_fail: messages to print based on the condition
:param condition: success (True) or failure (False)
:param fast_fail_message: [optional] message to include on fast fail
"""
if not isinstance(condition, bool):
raise TypeError("condition argument must be a bool")
if not Report.result(msgtuple_success_fail, condition):
TestHelper.fail_fast(fast_fail_message)
# DEPRECATED: Use vector3_str()
@staticmethod
def info_vector3(vector3, label="", magnitude=None):
# type: (azlmbr.math.Vector3, str, float) -> None
"""
prints the vector to the Report.info log. If applied, label will print first,
followed by the vector's values (x, y, z,) to 2 decimal places. Lastly if the
magnitude is supplied, it will print on the third line.
:param vector3: a azlmbr.math.Vector3 object to print
prints in [x: , y: , z: ] format.
:param label: [optional] A string to print before printing the vector3's contents
:param magnitude: [optional] the vector's magnitude to print after the vector's contents
:return: None
"""
if label != "":
Report.info(label)
Report.info(" x: {:.2f}, y: {:.2f}, z: {:.2f}".format(vector3.x, vector3.y, vector3.z))
if magnitude is not None:
Report.info(" magnitude: {:.2f}".format(magnitude))
'''
Utility for scope tracing errors and warnings.
Usage:
...
with Tracer() as section_tracer:
# section were we are interested in capturing errors/warnings/asserts
...
Report.result(Tests.warnings_not_found_in_section, not section_tracer.has_warnings)
'''
class Tracer:
def __init__(self):
self.warnings = []
self.errors = []
self.asserts = []
self.has_warnings = False
self.has_errors = False
self.has_asserts = False
self.handler = None
class WarningInfo:
def __init__(self, args):
self.window = args[0]
self.filename = args[1]
self.line = args[2]
self.function = args[3]
self.message = args[4]
class ErrorInfo:
def __init__(self, args):
self.window = args[0]
self.filename = args[1]
self.line = args[2]
self.function = args[3]
self.message = args[4]
class AssertInfo:
def __init__(self, args):
self.filename = args[0]
self.line = args[1]
self.function = args[2]
self.message = args[3]
def _on_warning(self, args):
warningInfo = Tracer.WarningInfo(args)
self.warnings.append(warningInfo)
Report.info("Tracer caught Warning: %s" % warningInfo.message)
self.has_warnings = True
return False
def _on_error(self, args):
errorInfo = Tracer.ErrorInfo(args)
self.errors.append(errorInfo)
Report.info("Tracer caught Error: %s" % errorInfo.message)
self.has_errors = True
return False
def _on_assert(self, args):
assertInfo = Tracer.AssertInfo(args)
self.asserts.append(assertInfo)
Report.info("Tracer caught Assert: %s:%i[%s] \"%s\"" % (assertInfo.filename, assertInfo.line, assertInfo.function, assertInfo.message))
self.has_asserts = True
return False
def __enter__(self):
self.handler = azlmbr.debug.TraceMessageBusHandler()
self.handler.connect(None)
self.handler.add_callback("OnPreAssert", self._on_assert)
self.handler.add_callback("OnPreWarning", self._on_warning)
self.handler.add_callback("OnPreError", self._on_error)
return self
def __exit__(self, type, value, traceback):
self.handler.disconnect()
self.handler = None
return False
class AngleHelper:
@staticmethod
def is_angle_close(x_rad, y_rad, tolerance):
# type: (float, float , float) -> bool
"""
compare if 2 angles measured in radians are close
:param x_rad: angle in radians
:param y_rad: angle in radians
:param tolerance: the tolerance to define close
:return: bool
"""
sinx_sub_siny = math.sin(x_rad) - math.sin(y_rad)
cosx_sub_cosy = math.cos(x_rad) - math.cos(y_rad)
r = sinx_sub_siny * sinx_sub_siny + cosx_sub_cosy * cosx_sub_cosy
diff = math.acos((2.0 - r) / 2.0)
return abs(diff) <= tolerance
@staticmethod
def is_angle_close_deg(x_deg, y_deg, tolerance):
# type: (float, float , float) -> bool
"""
compare if 2 angles measured in degrees are close
:param x_deg: angle in degrees
:param y_deg: angle in degrees
:param tolerance: the tolerance to define close
:return: bool
"""
return AngleHelper.is_angle_close(math.radians(x_deg), math.radians(y_deg), tolerance)
def vector3_str(vector3):
return "(x: {:.2f}, y: {:.2f}, z: {:.2f})".format(vector3.x, vector3.y, vector3.z)
def aabb_str(aabb):
return "[Min: %s, Max: %s]" % (vector3_str(aabb.min), vector3_str(aabb.max))
@@ -0,0 +1,43 @@
"""
All or portions of this file Copyright (c) Amazon.com, Inc. or its affiliates or
its licensors.
For complete copyright and license terms please see the LICENSE at the root of this
distribution (the "License"). All use of this software is governed by the License,
or, if provided, by the license below or the license accompanying this file. Do not
remove or modify any license notices. This file is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
"""
import os
import platform
from setuptools import setup, find_packages
from setuptools.command.develop import develop
from setuptools.command.build_py import build_py
PACKAGE_ROOT = os.path.abspath(os.path.dirname(__file__))
PYTHON_64 = platform.architecture()[0] == '64bit'
if __name__ == '__main__':
if not PYTHON_64:
raise RuntimeError("32-bit Python is not a supported platform.")
with open(os.path.join(PACKAGE_ROOT, 'README.txt')) as f:
long_description = f.read()
setup(
name="editor_python_test_tools",
version="1.0.0",
description='Lumberyard editor Python bindings test tools',
long_description=long_description,
packages=find_packages(where='Tools', exclude=['tests']),
install_requires=[
"ly_test_tools"
],
tests_require=[
],
entry_points={
},
)