diff --git a/tangostationcontrol/tangostationcontrol/common/measures.py b/tangostationcontrol/tangostationcontrol/common/measures.py index efba34a91d31dd3f727ce38a2ff3bab8b2d21410..e6c5767cd619218e3633985664d3a7a4955b7df4 100644 --- a/tangostationcontrol/tangostationcontrol/common/measures.py +++ b/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,7 +73,8 @@ def use_measures_directory(newdir): # switch to new directory current_symlink = pathlib.Path(IERS_ROOTDIR, "current") - current_symlink.unlink() + if current_symlink.exists(): + current_symlink.unlink() current_symlink.symlink_to(newdir) def restart_python(): @@ -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)) diff --git a/tangostationcontrol/tangostationcontrol/test/common/test_measures.py b/tangostationcontrol/tangostationcontrol/test/common/test_measures.py new file mode 100644 index 0000000000000000000000000000000000000000..d70e012573c3bf4d4c3fdd61c1740762f393d0a0 --- /dev/null +++ b/tangostationcontrol/tangostationcontrol/test/common/test_measures.py @@ -0,0 +1,73 @@ +# -*- 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())