1#!/usr/bin/env python
2#
3# Copyright 2020 - The Android Open Source Project
4#
5# Licensed under the Apache License, Version 2.0 (the "License");
6# you may not use this file except in compliance with the License.
7# You may obtain a copy of the License at
8#
9#     http://www.apache.org/licenses/LICENSE-2.0
10#
11# Unless required by applicable law or agreed to in writing, software
12# distributed under the License is distributed on an "AS IS" BASIS,
13# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14# See the License for the specific language governing permissions and
15# limitations under the License.
16"""Tests for LocalInstanceLock."""
17
18import fcntl
19import os
20import shutil
21import tempfile
22import unittest
23
24from unittest import mock
25
26from acloud import errors
27from acloud.internal.lib import local_instance_lock
28
29
30class LocalInstanceLockTest(unittest.TestCase):
31    """Test LocalInstanceLock methods."""
32
33    def setUp(self):
34        self._temp_dir = tempfile.mkdtemp()
35        self._lock_path = os.path.join(self._temp_dir, "temp.lock")
36        self._lock = local_instance_lock.LocalInstanceLock(
37            self._lock_path)
38
39    def tearDown(self):
40        shutil.rmtree(self._temp_dir, ignore_errors=True)
41
42    def testLock(self):
43        """Test the method calls that don't raise errors."""
44        self.assertTrue(self._lock.LockIfNotInUse())
45        self.assertTrue(os.path.isfile(self._lock_path))
46        self._lock.Unlock()
47
48        self.assertTrue(self._lock.LockIfNotInUse(timeout_secs=0))
49        self._lock.SetInUse(True)
50        self._lock.Unlock()
51
52        self.assertFalse(self._lock.LockIfNotInUse())
53
54        self.assertTrue(self._lock.Lock())
55        self._lock.SetInUse(False)
56        self._lock.Unlock()
57
58        self.assertTrue(self._lock.Lock(timeout_secs=0))
59        self._lock.Unlock()
60
61    def testOperationsWithoutLock(self):
62        """Test raising errors when the file is not locked."""
63        self.assertRaises(RuntimeError, self._lock.Unlock)
64        self.assertRaises(RuntimeError, self._lock.SetInUse, True)
65        self.assertRaises(RuntimeError, self._lock.SetInUse, False)
66
67    def testNonBlockingLock(self):
68        """Test failing to lock in non-blocking mode."""
69        lock = local_instance_lock.LocalInstanceLock(self._lock_path)
70        self.assertTrue(lock.Lock(timeout_secs=0))
71        try:
72            self.assertFalse(self._lock.Lock(timeout_secs=0))
73            self.assertFalse(self._lock.LockIfNotInUse(timeout_secs=0))
74        finally:
75            lock.Unlock()
76
77    @mock.patch("acloud.internal.lib.local_instance_lock."
78                "utils.TimeoutException")
79    def testLockWithTimeout(self, mock_timeout_exception):
80        """Test failing to lock due to timeout."""
81        mock_wrapped_flock = mock.Mock(side_effect=errors.FunctionTimeoutError)
82        mock_wrapper = mock.Mock(return_value=mock_wrapped_flock)
83        mock_timeout_exception.return_value = mock_wrapper
84
85        self.assertFalse(self._lock.Lock(timeout_secs=1))
86
87        mock_wrapper.assert_called_once_with(fcntl.flock)
88        mock_wrapped_flock.assert_called_once_with(mock.ANY, fcntl.LOCK_EX)
89        mock_wrapper.reset_mock()
90        mock_wrapped_flock.reset_mock()
91
92        self.assertFalse(self._lock.LockIfNotInUse(timeout_secs=1))
93
94        mock_wrapper.assert_called_once_with(fcntl.flock)
95        mock_wrapped_flock.assert_called_once_with(mock.ANY, fcntl.LOCK_EX)
96
97
98if __name__ == "__main__":
99    unittest.main()
100