L2SS-497: Added tests for measures

tangostationcontrol/tangostationcontrol/common/measures.py

@@ -39,7 +39,7 @@ import sys
 IERS_ROOTDIR = "/opt/IERS"
 
 # Where to download files to
-TEMPDIR = "/tmp"
+DOWNLOAD_DIR = "/tmp"
 
 # Where new measures can be downloaded
 MEASURES_URL = "ftp://ftp.astron.nl/outgoing/Measures/WSRT_Measures.ztar"
@@ -73,6 +73,7 @@ def use_measures_directory(newdir):
 
     # switch to new directory
     current_symlink = pathlib.Path(IERS_ROOTDIR, "current")
+    if current_symlink.exists():
         current_symlink.unlink()
     current_symlink.symlink_to(newdir)
 
@@ -100,7 +101,7 @@ def download_measures() -> str:
     iers_dir.mkdir()
 
     try:
-        measures_filename = pathlib.Path(TEMPDIR, "WSRT_Measures.ztar")
+        measures_filename = pathlib.Path(DOWNLOAD_DIR, "WSRT_Measures.ztar")
 
         # download measures
         urllib.request.urlretrieve(MEASURES_URL, str(measures_filename))

tangostationcontrol/tangostationcontrol/test/common/test_measures.py

+# -*- coding: utf-8 -*-
+#
+# This file is part of the LOFAR 2.0 Station Software
+#
+#
+#
+# Distributed under the terms of the APACHE license.
+# See LICENSE.txt for more info.
+
+import urllib.request
+import os.path
+import datetime
+from unittest import mock
+import shutil
+import tempfile
+import time
+
+from tangostationcontrol.common import measures
+
+from tangostationcontrol.test import base
+
+# where our WSRT_Measures.ztar surrogate is located
+fake_measures = os.path.dirname(__file__) + "/fake_measures.ztar"
+
+class TestMeasures(base.TestCase):
+    @mock.patch.object(urllib.request, 'urlretrieve')
+    def test_download_and_use(self, m_urlretrieve):
+        """ Test downloading and using new measures tables. """
+
+        with tempfile.TemporaryDirectory() as tmpdirname, \
+             mock.patch('tangostationcontrol.common.measures.IERS_ROOTDIR', tmpdirname) as rootdir, \
+             mock.patch('tangostationcontrol.common.measures.DOWNLOAD_DIR', tmpdirname) as downloaddir:
+                
+            # emulate the download
+            m_urlretrieve.side_effect = lambda *args, **kw: shutil.copyfile(fake_measures, tmpdirname + "/WSRT_Measures.ztar")
+
+            # 'download' and process our fake measures
+            newdir = measures.download_measures()
+
+            # active them
+            measures.use_measures_directory(newdir)
+
+            # check if they're activated
+            self.assertIn(newdir, measures.get_available_measures_directories())
+            self.assertEqual(newdir, measures.get_measures_directory())
+
+    @mock.patch.object(urllib.request, 'urlretrieve')
+    def test_switch_tables(self, m_urlretrieve):
+        """ Test switching between available sets of measures tables. """
+
+        with tempfile.TemporaryDirectory() as tmpdirname, \
+             mock.patch('tangostationcontrol.common.measures.IERS_ROOTDIR', tmpdirname) as rootdir, \
+             mock.patch('tangostationcontrol.common.measures.DOWNLOAD_DIR', tmpdirname) as downloaddir:
+                
+            # emulate the download
+            m_urlretrieve.side_effect = lambda *args, **kw: shutil.copyfile(fake_measures, tmpdirname + "/WSRT_Measures.ztar")
+
+            # 'download' two measures with different timestamps
+            newdir1 = measures.download_measures()
+            time.sleep(1)
+            newdir2 = measures.download_measures()
+
+            # check if both are available
+            self.assertIn(newdir1, measures.get_available_measures_directories())
+            self.assertIn(newdir2, measures.get_available_measures_directories())
+
+            # switch between the two
+            measures.use_measures_directory(newdir1)
+            self.assertEqual(newdir1, measures.get_measures_directory())
+            measures.use_measures_directory(newdir2)
+            self.assertEqual(newdir2, measures.get_measures_directory())
+            measures.use_measures_directory(newdir1)
+            self.assertEqual(newdir1, measures.get_measures_directory())