Merge pull request #350

Roof Test and Refactor

pyobs/modules/roof/basedome.py

@@ -19,7 +19,6 @@ def __init__(self, **kwargs: Any):
         """Initialize a new base dome."""
         BaseRoof.__init__(self, **kwargs)
 
-        # register exception
         exc.register_exception(exc.MotionError, 3, timespan=600, callback=self._default_remote_error_callback)
 
     async def get_fits_header_before(
@@ -34,10 +33,8 @@ async def get_fits_header_before(
             Dictionary containing FITS headers.
         """
 
-        # get from parent
         hdr = await BaseRoof.get_fits_header_before(self, namespaces, **kwargs)
 
-        # add azimuth and return it
         _, az = await self.get_altaz()
         hdr["ROOF-AZ"] = (az, "Azimuth of roof slit, deg E of N")
         return hdr

pyobs/modules/roof/baseroof.py

@@ -19,7 +19,6 @@ def __init__(self, **kwargs: Any):
         """Initialize a new base roof."""
         Module.__init__(self, **kwargs)
 
-        # init mixins
         WeatherAwareMixin.__init__(self, **kwargs)
         MotionStatusMixin.__init__(self, **kwargs)
 
@@ -50,7 +49,7 @@ async def get_fits_header_before(
         }
 
     async def is_ready(self, **kwargs: Any) -> bool:
-        """Returns the device is "ready", whatever that means for the specific device.
+        """The roof is ready, if it is open.
 
         Returns:
             True, if roof is open.

pyobs/modules/roof/dummyroof.py

@@ -17,22 +17,23 @@ class DummyRoof(BaseRoof, IRoof):
 
     __module__ = "pyobs.modules.roof"
 
+    _ROOF_CLOSED_PERCENTAGE = 0
+    _ROOF_OPEN_PERCENTAGE = 100
+
     def __init__(self, **kwargs: Any):
         """Creates a new dummy root."""
         BaseRoof.__init__(self, **kwargs)
 
         # dummy state
-        self.open_percentage = 0
+        self._open_percentage: int = self._ROOF_CLOSED_PERCENTAGE
 
-        # allow to abort motion
         self._lock_motion = asyncio.Lock()
         self._abort_motion = asyncio.Event()
 
     async def open(self) -> None:
         """Open module."""
         await BaseRoof.open(self)
 
-        # register event
         await self.comm.register_event(RoofOpenedEvent)
         await self.comm.register_event(RoofClosingEvent)
 
@@ -44,34 +45,19 @@ async def init(self, **kwargs: Any) -> None:
             AcquireLockFailed: If current motion could not be aborted.
         """
 
-        # already open?
-        if self.open_percentage != 100:
-            # acquire lock
-            with LockWithAbort(self._lock_motion, self._abort_motion):
-                # change status
-                await self._change_motion_status(MotionStatus.INITIALIZING)
-
-                # open roof
-                while self.open_percentage < 100:
-                    # open more
-                    self.open_percentage += 1
+        if self._is_open():
+            return
 
-                    # abort?
-                    if self._abort_motion.is_set():
-                        await self._change_motion_status(MotionStatus.IDLE)
-                        return
+        async with LockWithAbort(self._lock_motion, self._abort_motion):
+            await self._change_motion_status(MotionStatus.INITIALIZING)
 
-                    # wait a little
-                    await asyncio.sleep(0.1)
+            await self._move_roof(self._ROOF_OPEN_PERCENTAGE)
 
-                # open fully
-                self.open_percentage = 100
-
-                # change status
-                await self._change_motion_status(MotionStatus.IDLE)
+            await self._change_motion_status(MotionStatus.IDLE)
+            self.comm.send_event(RoofOpenedEvent())
 
-                # send event
-                self.comm.send_event(RoofOpenedEvent())
+    def _is_open(self):
+        return self._open_percentage == self._ROOF_OPEN_PERCENTAGE
 
     @timeout(15)
     async def park(self, **kwargs: Any) -> None:
@@ -81,35 +67,34 @@ async def park(self, **kwargs: Any) -> None:
             AcquireLockFailed: If current motion could not be aborted.
         """
 
-        # already closed?
-        if self.open_percentage != 0:
-            # acquire lock
-            with LockWithAbort(self._lock_motion, self._abort_motion):
-                # change status
-                await self._change_motion_status(MotionStatus.PARKING)
+        if self._is_closed():
+            return
+
+        async with LockWithAbort(self._lock_motion, self._abort_motion):
+            await self._change_motion_status(MotionStatus.PARKING)
+            self.comm.send_event(RoofClosingEvent())
 
-                # send event
-                self.comm.send_event(RoofClosingEvent())
+            await self._move_roof(self._ROOF_CLOSED_PERCENTAGE)
 
-                # close roof
-                while self.open_percentage > 0:
-                    # close more
-                    self.open_percentage -= 1
+            await self._change_motion_status(MotionStatus.PARKED)
 
-                    # abort?
-                    if self._abort_motion.is_set():
-                        await self._change_motion_status(MotionStatus.IDLE)
-                        return
+    def _is_closed(self):
+        return self._open_percentage == self._ROOF_CLOSED_PERCENTAGE
 
-                    # wait a little
-                    await asyncio.sleep(0.1)
+    async def _move_roof(self, target_pos: int) -> None:
+        step = 1 if target_pos > self._open_percentage else -1
+
+        while self._open_percentage != target_pos:
+            if self._abort_motion.is_set():
+                await self._change_motion_status(MotionStatus.IDLE)
+                return
 
-                # change status
-                await self._change_motion_status(MotionStatus.PARKED)
+            self._open_percentage += step
+            await asyncio.sleep(0.1)
 
     def get_percent_open(self) -> float:
         """Get the percentage the roof is open."""
-        return self.open_percentage
+        return self._open_percentage
 
     async def stop_motion(self, device: Optional[str] = None, **kwargs: Any) -> None:
         """Stop the motion.
@@ -121,13 +106,9 @@ async def stop_motion(self, device: Optional[str] = None, **kwargs: Any) -> None
             AcquireLockFailed: If current motion could not be aborted.
         """
 
-        # change status
         await self._change_motion_status(MotionStatus.ABORTING)
 
-        # abort
-        # acquire lock
-        with LockWithAbort(self._lock_motion, self._abort_motion):
-            # change status
+        async with LockWithAbort(self._lock_motion, self._abort_motion):
             await self._change_motion_status(MotionStatus.IDLE)
 
 

tests/modules/roof/test_basedome.py

@@ -0,0 +1,35 @@
+from typing import Any, Tuple, Optional
+
+import pytest
+
+from pyobs.modules.roof import BaseDome
+
+
+class TestBaseDome(BaseDome):
+
+    async def init(self, **kwargs: Any) -> None:
+        pass
+
+    async def park(self, **kwargs: Any) -> None:
+        pass
+
+    async def stop_motion(self, device: Optional[str] = None, **kwargs: Any) -> None:
+        pass
+
+    async def move_altaz(self, alt: float, az: float, **kwargs: Any) -> None:
+        pass
+
+    async def get_altaz(self, **kwargs: Any) -> Tuple[float, float]:
+        return 60.0, 0.0
+
+
+@pytest.mark.asyncio
+async def test_get_fits_header_before(mocker):
+    dome = TestBaseDome()
+
+    mocker.patch("pyobs.modules.roof.BaseRoof.get_fits_header_before", return_value={"ROOF-OPN": (True, "")})
+
+    header = await dome.get_fits_header_before()
+
+    assert "ROOF-OPN" in header
+    assert header["ROOF-AZ"] == (0.0, "Azimuth of roof slit, deg E of N")

tests/modules/roof/test_baseroof.py

@@ -0,0 +1,69 @@
+from typing import Optional, Any
+from unittest.mock import AsyncMock
+
+import pytest
+
+import pyobs
+from pyobs.modules.roof import BaseRoof
+from pyobs.utils.enums import MotionStatus
+
+
+class TestBaseRoof(BaseRoof):
+    async def init(self, **kwargs: Any) -> None:
+        pass
+
+    async def park(self, **kwargs: Any) -> None:
+        pass
+
+    async def stop_motion(self, device: Optional[str] = None, **kwargs: Any) -> None:
+        pass
+
+
+@pytest.mark.asyncio
+async def test_open(mocker):
+    mocker.patch("pyobs.mixins.WeatherAwareMixin.open")
+    mocker.patch("pyobs.mixins.MotionStatusMixin.open")
+    mocker.patch("pyobs.modules.Module.open")
+
+    telescope = TestBaseRoof()
+    await telescope.open()
+
+    pyobs.mixins.WeatherAwareMixin.open.assert_called_once_with(telescope)
+    pyobs.mixins.MotionStatusMixin.open.assert_called_once_with(telescope)
+    pyobs.modules.Module.open.assert_called_once_with(telescope)
+
+
+@pytest.mark.asyncio
+async def test_get_fits_header_before_open():
+    telescope = TestBaseRoof()
+
+    telescope.get_motion_status = AsyncMock(return_value=MotionStatus.POSITIONED)
+    header = await telescope.get_fits_header_before()
+
+    assert header["ROOF-OPN"] == (True, "True for open, false for closed roof")
+
+
+@pytest.mark.asyncio
+async def test_get_fits_header_before_closed():
+    telescope = TestBaseRoof()
+
+    telescope.get_motion_status = AsyncMock(return_value=MotionStatus.PARKED)
+    header = await telescope.get_fits_header_before()
+
+    assert header["ROOF-OPN"] == (False, "True for open, false for closed roof")
+
+
+@pytest.mark.asyncio
+async def test_ready():
+    telescope = TestBaseRoof()
+
+    telescope.get_motion_status = AsyncMock(return_value=MotionStatus.TRACKING)
+    assert await telescope.is_ready() is True
+
+
+@pytest.mark.asyncio
+async def test_not_ready():
+    telescope = TestBaseRoof()
+
+    telescope.get_motion_status = AsyncMock(return_value=MotionStatus.PARKING)
+    assert await telescope.is_ready() is False

tests/modules/roof/test_dummyroof.py

@@ -0,0 +1,105 @@
+from unittest.mock import AsyncMock, Mock
+
+import pytest
+
+import pyobs
+from pyobs.events import RoofOpenedEvent, RoofClosingEvent
+from pyobs.modules.roof import DummyRoof
+from pyobs.utils.enums import MotionStatus
+
+
+@pytest.mark.asyncio
+async def test_open(mocker) -> None:
+    mocker.patch("pyobs.modules.roof.BaseRoof.open")
+    roof = DummyRoof()
+    roof.comm.register_event = AsyncMock()
+
+    await roof.open()
+
+    pyobs.modules.roof.BaseRoof.open.assert_called_once()
+
+    assert roof.comm.register_event.call_args_list[0][0][0] == RoofOpenedEvent
+    assert roof.comm.register_event.call_args_list[1][0][0] == RoofClosingEvent
+
+
+@pytest.mark.asyncio
+async def test_init(mocker) -> None:
+    mocker.patch("asyncio.sleep")
+
+    roof = DummyRoof()
+    roof._change_motion_status = AsyncMock()
+    roof.comm.send_event = Mock()
+
+    await roof.init()
+
+    roof._change_motion_status.assert_awaited_with(MotionStatus.IDLE)
+    roof.comm.send_event(RoofOpenedEvent())
+
+
+@pytest.mark.asyncio
+async def test_park(mocker) -> None:
+    mocker.patch("asyncio.sleep")
+
+    roof = DummyRoof()
+    roof._open_percentage = 100
+
+    roof._change_motion_status = AsyncMock()
+    roof.comm.send_event = Mock()
+
+    await roof.park()
+
+    roof._change_motion_status.assert_awaited_with(MotionStatus.PARKED)
+
+
+@pytest.mark.asyncio
+async def test_move_roof_open(mocker) -> None:
+    mocker.patch("asyncio.sleep")
+
+    roof = DummyRoof()
+
+    await roof._move_roof(roof._ROOF_OPEN_PERCENTAGE)
+
+    assert roof._open_percentage == 100
+
+
+@pytest.mark.asyncio
+async def test_move_roof_closed(mocker) -> None:
+    mocker.patch("asyncio.sleep")
+
+    roof = DummyRoof()
+
+    await roof._move_roof(roof._ROOF_CLOSED_PERCENTAGE)
+
+    assert roof._open_percentage == 0
+
+
+@pytest.mark.asyncio
+async def test_move_roof_abort(mocker) -> None:
+    mocker.patch("asyncio.sleep")
+
+    roof = DummyRoof()
+
+    roof._abort_motion.set()
+    await roof._move_roof(roof._ROOF_OPEN_PERCENTAGE)
+
+    assert roof._open_percentage == 0
+
+
+@pytest.mark.asyncio
+async def test_move_roof_open(mocker) -> None:
+    mocker.patch("asyncio.sleep")
+
+    roof = DummyRoof()
+
+    await roof._move_roof(roof._ROOF_OPEN_PERCENTAGE)
+
+    assert roof._open_percentage == 100
+
+
+@pytest.mark.asyncio
+async def test_stop_motion() -> None:
+    roof = DummyRoof()
+    roof._change_motion_status = AsyncMock()
+    await roof.stop_motion()
+
+    roof._change_motion_status.assert_awaited_with(MotionStatus.IDLE)