diff options
Diffstat (limited to 'testenv/tests/test_aioregion.py')
| -rw-r--r-- | testenv/tests/test_aioregion.py | 113 |
1 files changed, 113 insertions, 0 deletions
diff --git a/testenv/tests/test_aioregion.py b/testenv/tests/test_aioregion.py new file mode 100644 index 00000000..601f5280 --- /dev/null +++ b/testenv/tests/test_aioregion.py @@ -0,0 +1,113 @@ +# ========================================================================== # +# # +# KVMD - The main Pi-KVM daemon. # +# # +# Copyright (C) 2018 Maxim Devaev <[email protected]> # +# # +# This program is free software: you can redistribute it and/or modify # +# it under the terms of the GNU General Public License as published by # +# the Free Software Foundation, either version 3 of the License, or # +# (at your option) any later version. # +# # +# This program is distributed in the hope that it will be useful, # +# but WITHOUT ANY WARRANTY; without even the implied warranty of # +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # +# GNU General Public License for more details. # +# # +# You should have received a copy of the GNU General Public License # +# along with this program. If not, see <https://www.gnu.org/licenses/>. # +# # +# ========================================================================== # + + +import asyncio + +import pytest + +from kvmd.aioregion import RegionIsBusyError +from kvmd.aioregion import AioExclusiveRegion + + +# ===== +async def test_ok__access_one() -> None: + region = AioExclusiveRegion(RegionIsBusyError) + + async def func() -> None: + assert not region.is_busy() + with region: + assert region.is_busy() + assert not region.is_busy() + + await func() + + assert not region.is_busy() + region.exit() + assert not region.is_busy() + + +async def test_fail__access_one() -> None: + region = AioExclusiveRegion(RegionIsBusyError) + + async def func() -> None: + assert not region.is_busy() + with region: + assert region.is_busy() + region.enter() + assert not region.is_busy() + + with pytest.raises(RegionIsBusyError): + await func() + + assert not region.is_busy() + region.exit() + assert not region.is_busy() + + +# ===== +async def test_ok__access_two() -> None: + region = AioExclusiveRegion(RegionIsBusyError) + + async def func1() -> None: + with region: + await asyncio.sleep(1) + print("done func1()") + + async def func2() -> None: + await asyncio.sleep(2) + print("waiking up func2()") + with region: + await asyncio.sleep(1) + print("done func2()") + + await asyncio.gather(func1(), func2()) + + assert not region.is_busy() + region.exit() + assert not region.is_busy() + + +async def test_fail__access_two() -> None: + region = AioExclusiveRegion(RegionIsBusyError) + + async def func1() -> None: + with region: + await asyncio.sleep(2) + print("done func1()") + + async def func2() -> None: + await asyncio.sleep(1) + with region: + await asyncio.sleep(1) + print("done func2()") + + results = await asyncio.gather(func1(), func2(), return_exceptions=True) + assert results[0] is None + assert type(results[1]) == RegionIsBusyError # pylint: disable=unidiomatic-typecheck + + assert not region.is_busy() + region.exit() + assert not region.is_busy() |