file_utils.py 6.1 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167
  1. """
  2. Copyright (c) Contributors to the Open 3D Engine Project.
  3. For complete copyright and license terms please see the LICENSE at the root of this distribution.
  4. SPDX-License-Identifier: Apache-2.0 OR MIT
  5. """
  6. import os
  7. import shutil
  8. import logging
  9. import stat
  10. import ly_test_tools.environment.file_system as file_system
  11. import ly_test_tools.environment.waiter as waiter
  12. logger = logging.getLogger(__name__)
  13. def clear_out_file(file_path):
  14. """
  15. Clears out the specified config file to be empty.
  16. :param file_path: The full path to the file.
  17. """
  18. if os.path.exists(file_path):
  19. file_system.unlock_file(file_path)
  20. with open(file_path, 'w') as file_to_write:
  21. file_to_write.write('')
  22. else:
  23. logger.debug(f'{file_path} not found while attempting to clear out file.')
  24. def add_commands_to_config_file(config_file_dir, config_file_name, command_list):
  25. """
  26. From the command list, appends each command to the specified config file.
  27. :param config_file_dir: The directory the config file is contained in.
  28. :param config_file_name: The config file name.
  29. :param command_list: The commands to add to the file.
  30. :return:
  31. """
  32. config_file_path = os.path.join(config_file_dir, config_file_name)
  33. os.chmod(config_file_path, stat.S_IWRITE | stat.S_IREAD | stat.S_IEXEC)
  34. with open(config_file_path, 'w') as launch_config_file:
  35. for command in command_list:
  36. launch_config_file.write("{}\n".format(command))
  37. def gather_error_logs(workspace):
  38. """
  39. Grabs all error logs (if there are any) and puts them into the specified logs path.
  40. :param workspace: The AbstractWorkspaceManager object that contains all of the paths
  41. """
  42. error_log_path = os.path.join(workspace.paths.project_log(), 'error.log')
  43. error_dump_path = os.path.join(workspace.paths.project_log(), 'error.dmp')
  44. if os.path.exists(error_dump_path):
  45. workspace.artifact_manager.save_artifact(error_dump_path)
  46. if os.path.exists(error_log_path):
  47. workspace.artifact_manager.save_artifact(error_log_path)
  48. def delete_screenshot_folder(workspace):
  49. """
  50. Deletes screenshot folder from platform path
  51. :param workspace: The AbstractWorkspaceManager object that contains all of the paths
  52. """
  53. shutil.rmtree(workspace.paths.project_screenshots(), ignore_errors=True)
  54. def move_file(src_dir, dest_dir, file_name, timeout=120):
  55. """
  56. Attempts to move a file from the source directory to the destination directory. Raises an IOError if
  57. the file is in use.
  58. :param src_dir: Directory of the file to be moved.
  59. :param dest_dir: Directory where the file will be moved to.
  60. :param file_name: Name of the file to be moved.
  61. :param timeout: Number of seconds to wait for the file to be released.
  62. """
  63. file_path = os.path.join(src_dir, file_name)
  64. if os.path.exists(file_path):
  65. waiter.wait_for(lambda: move_file_check(src_dir, dest_dir, file_name), timeout=timeout,
  66. exc=IOError('Cannot move file {} while in use'.format(file_path)))
  67. def move_file_check(src_dir, dest_dir, file_name):
  68. """
  69. Moves file and checks if the file has been moved from the source to the destination directory.
  70. :param src_dir: Source directory of the file to be moved
  71. :param dest_dir: Destination directory where the file should move to
  72. :param file_name: The name of the file to be moved
  73. :return:
  74. """
  75. try:
  76. shutil.move(os.path.join(src_dir, file_name), os.path.join(dest_dir, file_name))
  77. except OSError as e:
  78. logger.info(e)
  79. return False
  80. return True
  81. def rename_file(file_path, dest_path, timeout=10):
  82. # type: (str, str, int) -> None
  83. """
  84. Renames a file by moving it. Waits for file to become available and raises and exception if timeout occurs.
  85. :param file_path: absolute path to the source file
  86. :param dest_path: absolute path to the new file
  87. :param timeout: timeout to wait for function to complete
  88. :return: None
  89. """
  90. def _rename_file_check():
  91. try:
  92. shutil.move(file_path, dest_path)
  93. except OSError as e:
  94. logger.debug(f'Attempted to rename file: {file_path} but an error occurred, retrying.'
  95. f'\nError: {e}',
  96. stackinfo=True)
  97. return False
  98. return True
  99. if os.path.exists(file_path):
  100. waiter.wait_for(lambda: _rename_file_check(), timeout=timeout,
  101. exc=OSError('Cannot rename file {} while in use'.format(file_path)))
  102. def delete_level(workspace, level_dir, timeout=120):
  103. """
  104. Attempts to delete an entire level folder from the project.
  105. :param workspace: The workspace instance to delete the level from.
  106. :param level_dir: The level folder to delete
  107. """
  108. if not level_dir:
  109. logger.warning("level_dir is empty, nothing to delete.")
  110. return
  111. full_level_dir = os.path.join(workspace.paths.project(), 'Levels', level_dir)
  112. if not os.path.isdir(full_level_dir):
  113. if os.path.exists(full_level_dir):
  114. logger.error("level '{}' isn't a directory, it won't be deleted.".format(full_level_dir))
  115. else:
  116. logger.info("level '{}' doesn't exist, nothing to delete.".format(full_level_dir))
  117. return
  118. waiter.wait_for(lambda: delete_check(full_level_dir),
  119. timeout=timeout,
  120. exc=IOError('Cannot delete directory {} while in use'.format(full_level_dir)))
  121. def delete_check(src_dir):
  122. """
  123. Deletes directory and verifies that it's been deleted.
  124. :param src_dir: The directory to delete
  125. """
  126. try:
  127. def handle_delete_error(action, path, exception_info):
  128. # The deletion could fail if the file is read-only, so set the permissions to writeable and try again
  129. os.chmod(path, stat.S_IWRITE)
  130. # Try the passed-in action (delete) again
  131. action(path)
  132. shutil.rmtree(src_dir, onerror=handle_delete_error)
  133. except OSError as e:
  134. logger.debug("Delete for '{}' failed: {}".format(src_dir, e))
  135. return False
  136. return not os.path.exists(src_dir)