Fixed physics tests on linux (#5579)

* Fixed physics tests on linux

* Fixed unit tests because of new implementation

* Fixed bug in lock, fixed unit tests

Co-authored-by: aljanru <aljanru@uc5564ff5a5ee55.ant.amazon.com>
monroegm-disable-blank-issue-2
AMZN-AlexOteiza 4 years ago committed by GitHub
parent 8bbb99d741
commit cb93bafc4f
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23

@ -218,6 +218,7 @@ class FileManagement:
src_file_path = os.path.join(src_path, src_file) src_file_path = os.path.join(src_path, src_file)
if os.path.exists(target_file_path): if os.path.exists(target_file_path):
fs.unlock_file(target_file_path) fs.unlock_file(target_file_path)
os.makedirs(target_path, exist_ok=True)
shutil.copyfile(src_file_path, target_file_path) shutil.copyfile(src_file_path, target_file_path)
@staticmethod @staticmethod

@ -219,7 +219,8 @@ def unlock_file(file_name):
:return: True if unlock succeeded, else False :return: True if unlock succeeded, else False
""" """
if not os.access(file_name, os.W_OK): if not os.access(file_name, os.W_OK):
os.chmod(file_name, stat.S_IWRITE) file_stat = os.stat(file_name)
os.chmod(file_name, file_stat.st_mode | stat.S_IWRITE)
logger.warning(f'Clearing write lock for file {file_name}.') logger.warning(f'Clearing write lock for file {file_name}.')
return True return True
else: else:
@ -235,7 +236,8 @@ def lock_file(file_name):
:return: True if lock succeeded, else False :return: True if lock succeeded, else False
""" """
if os.access(file_name, os.W_OK): if os.access(file_name, os.W_OK):
os.chmod(file_name, stat.S_IREAD) file_stat = os.stat(file_name)
os.chmod(file_name, file_stat.st_mode & (~stat.S_IWRITE))
logger.warning(f'Write locking file {file_name}') logger.warning(f'Write locking file {file_name}')
return True return True
else: else:

@ -8,6 +8,7 @@ SPDX-License-Identifier: Apache-2.0 OR MIT
import errno import errno
import logging import logging
import os import os
import stat
import psutil import psutil
import subprocess import subprocess
import sys import sys
@ -454,24 +455,33 @@ class TestChangePermissions(unittest.TestCase):
self.assertEqual(file_system.change_permissions('.', 0o777), False) self.assertEqual(file_system.change_permissions('.', 0o777), False)
class MockStatResult():
def __init__(self, st_mode):
self.st_mode = st_mode
class TestUnlockFile(unittest.TestCase): class TestUnlockFile(unittest.TestCase):
def setUp(self): def setUp(self):
self.file_name = 'file' self.file_name = 'file'
@mock.patch('os.stat')
@mock.patch('os.chmod') @mock.patch('os.chmod')
@mock.patch('os.access') @mock.patch('os.access')
def test_UnlockFile_WriteLocked_UnlockFile(self, mock_access, mock_chmod): def test_UnlockFile_WriteLocked_UnlockFile(self, mock_access, mock_chmod, mock_stat):
mock_access.return_value = False mock_access.return_value = False
os.stat.return_value = MockStatResult(stat.S_IREAD)
success = file_system.unlock_file(self.file_name) success = file_system.unlock_file(self.file_name)
mock_chmod.assert_called_once_with(self.file_name, stat.S_IREAD | stat.S_IWRITE)
self.assertTrue(success) self.assertTrue(success)
@mock.patch('os.stat')
@mock.patch('os.chmod') @mock.patch('os.chmod')
@mock.patch('os.access') @mock.patch('os.access')
def test_UnlockFile_AlreadyUnlocked_LogAlreadyUnlocked(self, mock_access, mock_chmod): def test_UnlockFile_AlreadyUnlocked_LogAlreadyUnlocked(self, mock_access, mock_chmod, mock_stat):
mock_access.return_value = True mock_access.return_value = True
os.stat.return_value = MockStatResult(stat.S_IREAD | stat.S_IWRITE)
success = file_system.unlock_file(self.file_name) success = file_system.unlock_file(self.file_name)
@ -483,19 +493,24 @@ class TestLockFile(unittest.TestCase):
def setUp(self): def setUp(self):
self.file_name = 'file' self.file_name = 'file'
@mock.patch('os.stat')
@mock.patch('os.chmod') @mock.patch('os.chmod')
@mock.patch('os.access') @mock.patch('os.access')
def test_UnlockFile_UnlockedFile_FileLockedSuccessReturnsTrue(self, mock_access, mock_chmod): def test_LockFile_UnlockedFile_FileLockedSuccessReturnsTrue(self, mock_access, mock_chmod, mock_stat):
mock_access.return_value = True mock_access.return_value = True
os.stat.return_value = MockStatResult(stat.S_IREAD | stat.S_IWRITE)
success = file_system.lock_file(self.file_name) success = file_system.lock_file(self.file_name)
mock_chmod.assert_called_once_with(self.file_name, stat.S_IREAD)
self.assertTrue(success) self.assertTrue(success)
@mock.patch('os.stat')
@mock.patch('os.chmod') @mock.patch('os.chmod')
@mock.patch('os.access') @mock.patch('os.access')
def test_UnlockFile_AlreadyLocked_FileLockedFailedReturnsFalse(self, mock_access, mock_chmod): def test_LockFile_AlreadyLocked_FileLockedFailedReturnsFalse(self, mock_access, mock_chmod, mock_stat):
mock_access.return_value = False mock_access.return_value = False
os.stat.return_value = MockStatResult(stat.S_IREAD)
success = file_system.lock_file(self.file_name) success = file_system.lock_file(self.file_name)

Loading…
Cancel
Save