Add starting point for Hydra/EPB Atom tests in AutomatedTesting project

This commit is contained in:
jromnoa
2021-04-20 18:56:49 -07:00
parent 707f5a07b2
commit 23dec3c10e
7 changed files with 674 additions and 0 deletions
@@ -69,4 +69,6 @@ set(GEM_DEPENDENCIES
Gem::AtomFont
Gem::AtomToolsFramework.Editor
Gem::Blast.Editor
Gem::DccScriptingInterface.Editor
Gem::QtForPython.Editor
)
@@ -16,6 +16,7 @@
ly_get_list_relative_pal_filename(pal_dir ${CMAKE_CURRENT_LIST_DIR}/Platform/${PAL_PLATFORM_NAME})
add_subdirectory(assetpipeline)
add_subdirectory(atom_renderer)
## Physics ##
# DISABLED - see LYN-2536
@@ -0,0 +1,46 @@
#
# 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.
#
#
# 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.
#
################################################################################
# Atom Renderer Automated Tests
# Runs EditorPythonBindings scripts inside the Editor to verify test results.
################################################################################
add_subdirectory(atom_python_scripts)
add_subdirectory(atom_utils)
add_subdirectory(epb_utils)
add_subdirectory(epb_scripts)
if(PAL_TRAIT_BUILD_HOST_TOOLS AND PAL_TRAIT_BUILD_TESTS_SUPPORTED AND AutomatedTesting IN_LIST LY_PROJECTS)
ly_add_pytest(
NAME AtomRenderer::HydraEPBTestsMain
TEST_REQUIRES gpu
TEST_SUITE main
PATH ${CMAKE_CURRENT_LIST_DIR}/atom_python_scripts/test_Atom_MainSuite.py
TEST_SERIAL
TIMEOUT 1200
RUNTIME_DEPENDENCIES
AssetProcessor
AtomTest.Assets
Editor
)
endif()
@@ -0,0 +1,91 @@
"""
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
from pathlib import PurePath
import pytest
# Bail on the test if ly_test_tools doesn't exist.
pytest.importorskip("ly_test_tools")
import ly_test_tools.environment.file_system as file_system
from atom_renderer.atom_utils import hydra_test_utils as hydra
logger = logging.getLogger(__name__)
EDITOR_TIMEOUT = 60
TEST_DIRECTORY = os.path.dirname(__file__)
# Go to the project root directory
PROJECT_DIRECTORY = PurePath(TEST_DIRECTORY)
if len(PROJECT_DIRECTORY.parents) > 5:
for _ in range(5):
PROJECT_DIRECTORY = PROJECT_DIRECTORY.parent
@pytest.mark.parametrize("project", ["AutomatedTesting"])
@pytest.mark.parametrize("launcher_platform", ['windows_editor'])
@pytest.mark.parametrize("level", ["tmp_level"])
class TestAllLevelsOpenClose(object):
@pytest.fixture(autouse=True)
def setup_teardown(self, request, workspace, project, level):
# Cleanup our temp level
file_system.delete(
[os.path.join(workspace.paths.engine_root(), project, "Levels", "AtomLevels", level)], True, True)
def teardown():
# Cleanup our temp level
file_system.delete(
[os.path.join(workspace.paths.engine_root(), project, "Levels", "AtomLevels", level)], True, True)
request.addfinalizer(teardown)
@pytest.mark.test_case_id(
"C34428159",
"C34428160",
"C34428161",
"C34428162",
"C34428163",
"C34428165",
"C34428166",
"C34428167",
"C34428158",
"C34428172",
"C34428173",
"C34428174",
"C34428175",
)
def test_AllLevelsOpenClose(self, request, editor, level, workspace, project, launcher_platform):
cfg_args = [level]
test_levels = os.path.join(str(PROJECT_DIRECTORY), "Levels", "AtomLevels")
expected_lines = []
for level in test_levels:
expected_lines.append(f"Successfully opened {level}")
unexpected_lines = [
"failed to open",
"Traceback (most recent call last):",
]
hydra.launch_and_validate_results(
request,
TEST_DIRECTORY,
editor,
"AllLevelsOpenClose_test_case.py",
timeout=EDITOR_TIMEOUT,
expected_lines=expected_lines,
unexpected_lines=unexpected_lines,
halt_on_unexpected=True,
cfg_args=cfg_args,
)
@@ -0,0 +1,266 @@
"""
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 time
import azlmbr.legacy.general as general
import azlmbr.atom
class FailFast(BaseException):
"""
Raise to stop proceeding through test steps.
"""
pass
class TestHelper:
@staticmethod
def init_idle():
general.idle_enable(True)
general.idle_wait_frames(1)
@staticmethod
def open_level(level):
# type: (str, ) -> None
"""
:param level: the name of the level folder in MestTest\\
:return: None
"""
result = general.open_level(level) # TO-DO: Check if success opening level
if result:
Report.info("Open level {}".format(level))
else:
Report.failure("Assert: failed to open level {}".format(level))
general.idle_wait_frames(1)
@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()
general.idle_wait_frames(1)
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
"""
general.exit_game_mode()
general.idle_wait_frames(1)
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=2.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:
general.idle_wait(1.0)
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
@staticmethod
def find_entities(entity_name):
search_filter = azlmbr.entity.SearchFilter()
search_filter.names = [entity_name]
searched_entities = azlmbr.entity.SearchBus(azlmbr.bus.Broadcast, 'SearchEntities', search_filter)
return searched_entities
@staticmethod
def attach_component_to_entity(entityId, componentName):
# type: (azlmbr.entity.EntityId, str) -> azlmbr.entity.EntityComponentIdPair
"""
Adds the component if not added already.
If successful, returns the EntityComponentIdPair, otherwise returns None.
"""
typeIdsList = azlmbr.editor.EditorComponentAPIBus(azlmbr.bus.Broadcast, 'FindComponentTypeIdsByEntityType',
[componentName], 0)
general.log("Components found = {}".format(len(typeIdsList)))
if len(typeIdsList) < 1:
general.log(f"ERROR: A component class with name {componentName} doesn't exist")
return None
elif len(typeIdsList) > 1:
general.log(f"ERROR: Found more than one component classes with same name: {componentName}")
return None
# Before adding the component let's check if it is already attached to the entity.
componentOutcome = azlmbr.editor.EditorComponentAPIBus(azlmbr.bus.Broadcast, 'GetComponentOfType', entityId,
typeIdsList[0])
if componentOutcome.IsSuccess():
return componentOutcome.GetValue() # In this case the value is not a list.
componentOutcome = azlmbr.editor.EditorComponentAPIBus(azlmbr.bus.Broadcast, 'AddComponentsOfType', entityId,
typeIdsList)
if componentOutcome.IsSuccess():
general.log(f"{componentName} Component added to entity.")
return componentOutcome.GetValue()[0]
general.log(f"ERROR: Failed to add component [{componentName}] to entity")
return None
@staticmethod
def get_component_property(component, propertyPath):
return azlmbr.editor.EditorComponentAPIBus(
azlmbr.bus.Broadcast,
'GetComponentProperty',
component,
propertyPath)
@staticmethod
def set_component_property(component, propertyPath, value):
azlmbr.editor.EditorComponentAPIBus(
azlmbr.bus.Broadcast,
'SetComponentProperty',
component,
propertyPath,
value)
@staticmethod
def get_property_list(Component):
property_list = azlmbr.editor.EditorComponentAPIBus(azlmbr.bus.Broadcast, 'BuildComponentPropertyList',
Component)
return property_list
@staticmethod
def compare_property_list(Component, PropertyList):
property_list = TestHelper.get_property_list(Component)
if set(property_list) == set(PropertyList):
general.log("Property list of component is correct.")
@staticmethod
def isclose(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)
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
# NOTE: implementation of reports will be changed to use a better mechanism rather than print
class Report:
@staticmethod
def info(msg):
print("Info: {}".format(msg))
@staticmethod
def success(msgtuple_success_fail):
print("Success: {}".format(msgtuple_success_fail[0]))
@staticmethod
def failure(msgtuple_success_fail):
print("Failure: {}".format(msgtuple_success_fail[1]))
@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)
@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))
@@ -0,0 +1,166 @@
"""
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 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 (
send_command_and_expect_response as send_command_and_expect_response,
)
from automatedtesting_shared.network_utils import check_for_listening_port
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,
log_file_name="Editor.log",
cfg_args=[],
timeout=60,
):
"""
Creates a temporary config file for Hydra execution, 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 log_file_name: Name of the log file created by the editor. Defaults to 'Editor.log'
: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))
if editor_script != "":
editor.args.extend(
[
"--skipWelcomeScreenDialog",
"--autotest_mode",
"--runpython",
test_case,
"--runpythonargs",
]
)
editor.args.extend([" ".join(cfg_args)])
with editor.start():
editorlog_file = os.path.join(editor.workspace.paths.project_log(), log_file_name)
# Log monitor requires the file to exist.
logger.debug("Waiting until log file <{}> exists...".format(editorlog_file))
waiter.wait_for(
lambda: os.path.exists(editorlog_file),
timeout=60,
exc=("Log file '{}' was never created by another process.".format(editorlog_file)),
interval=1,
)
logger.debug("Done! log file <{}> exists.".format(editorlog_file))
log_monitor = ly_test_tools.log.log_monitor.LogMonitor(launcher=editor, log_file_path=editorlog_file)
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,
unexpected_lines=[],
halt_on_unexpected=False,
port_listener_timeout=120,
log_monitor_timeout=60,
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.
: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 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.
"""
with launcher.start():
gamelog_file = os.path.join(launcher.workspace.paths.project_log(), "Game.log")
# 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)
# Log monitor requires the file to exist
logger.debug("Waiting until log file <{}> exists...".format(gamelog_file))
waiter.wait_for(
lambda: os.path.exists(gamelog_file),
timeout=60,
exc=("Log file '{}' was never created by another process.".format(gamelog_file)),
interval=1,
)
logger.debug("Done! log file <{}> exists.".format(gamelog_file))
log_monitor = ly_test_tools.log.log_monitor.LogMonitor(launcher=launcher, log_file_path=gamelog_file)
# Workaround for LY-110925 - Wait for log file to be opened before checking for expected lines. This is done in
# monitor_log_for_lines as well, but has a low timeout with no way to currently override
logger.debug("Waiting for log file '{}' to be opened by another process.".format(gamelog_file))
# 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=log_monitor_timeout,
)
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,102 @@
"""
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.
This hydra/EPB script opens and closes every possible Atom level.
"""
import os
import sys
import azlmbr.legacy.general as general
import azlmbr.legacy.settings as settings
import azlmbr.paths
sys.path.append(os.path.join(azlmbr.paths.devroot, 'AutomatedTesting', 'Gem', 'PythonTests'))
from atom_renderer.atom_utils.automated_test_utils import TestHelper as helper
LEVELS = os.listdir(os.path.join(azlmbr.paths.devroot, "AutomatedTesting", "Levels", "AtomLevels"))
class TestAllLevelsOpenClose(object):
"""Reserved for the test name."""
pass
def run():
"""
1. Open & close all valid test levels in the Editor.
2. Every time a level is opened, verify it loads correctly and the Editor remains stable.
"""
def after_level_load():
"""Function to call after creating/opening a level to ensure it loads."""
# Give everything a second to initialize.
general.idle_enable(True)
general.update_viewport()
general.idle_wait(0.5) # half a second is more than enough for updating the viewport.
# Close out problematic windows, FPS meters, and anti-aliasing.
if general.is_helpers_shown(): # Turn off the helper gizmos if visible
general.toggle_helpers()
if general.is_pane_visible("Error Report"): # Close Error Report windows that block focus.
general.close_pane("Error Report")
if general.is_pane_visible("Error Log"): # Close Error Log windows that block focus.
general.close_pane("Error Log")
general.run_console("r_displayInfo=0")
general.run_console("r_antialiasingmode=0")
return True
# Create a new level.
new_level_name = "tmp_level" # Specified in AllLevelsOpenClose_test.py
heightmap_resolution = 512
heightmap_meters_per_pixel = 1
terrain_texture_resolution = 412
use_terrain = False
# Return codes are ECreateLevelResult defined in CryEdit.h
return_code = general.create_level_no_prompt(
new_level_name, heightmap_resolution, heightmap_meters_per_pixel, terrain_texture_resolution, use_terrain)
if return_code == 1:
general.log(f"{new_level_name} level already exists")
elif return_code == 2:
general.log("Failed to create directory")
elif return_code == 3:
general.log("Directory length is too long")
elif return_code != 0:
general.log("Unknown error, failed to create level")
else:
general.log(f"{new_level_name} level created successfully")
after_level_load()
# Open all valid test levels.
failed_to_open = []
LEVELS.append(new_level_name) # Update LEVELS constant for created level.
for level in LEVELS:
if general.is_idle_enabled() and (general.get_current_level_name() == level):
general.log(f"Level {level} already open.")
else:
general.log(f"Opening level {level}")
general.open_level_no_prompt(level)
helper.wait_for_condition(function=lambda: general.get_current_level_name() == level,
timeout_in_seconds=2.0)
result = (general.get_current_level_name() == level) and after_level_load()
if result:
general.log(f"Successfully opened {level}")
else:
general.log(f"{level} failed to open")
failed_to_open.append(level)
if failed_to_open:
general.log(f"The following levels failed to open: {failed_to_open}")
if __name__ == "__main__":
run()