Source
xxxxxxxxxx
self.assertIsInstance(flagsdict1, dict, 'mode=list should return a dictionary of the flag versions')
#########################################################################
# test_task_flagmanager.py
# Copyright (C) 2018
# Associated Universities, Inc. Washington DC, USA.
#
# This script is free software; you can redistribute it and/or modify it
# under the terms of the GNU Library General Public License as published by
# the Free Software Foundation; either version 2 of the License, or (at your
# option) any later version.
#
# This library 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 Library General Public
# License for more details.
#
#
# Based on the requirements listed in casadocs found here:
# https://casadocs.readthedocs.io/en/stable/api/tt/casatasks.flagging.flagmanager.html
#
##########################################################################
import shutil
import unittest
import os
import sys
from casatasks import flagdata, flagmanager
from casatools import ctsys, agentflagger
from casatasks.private.parallel.parallel_task_helper import ParallelTaskHelper
# Path for data
datapath = ctsys.resolve('unittest/flagmanager/')
# Pick up alternative data directory to run tests on MMSs
testmms = False
if 'TEST_DATADIR' in os.environ:
DATADIR = str(os.environ.get('TEST_DATADIR'))+'/flagdata/'
if os.path.isdir(DATADIR):
testmms = True
datapath = DATADIR
print('flagmanager tests will use data from '+datapath)
# jagonzal (CAS-4287): Add a cluster-less mode to by-pass parallel processing for MMSs as requested
if 'BYPASS_PARALLEL_PROCESSING' in os.environ:
ParallelTaskHelper.bypassParallelProcessing(1)
# Local copy of the agentflagger tool
aflocal = agentflagger()
# Base class which defines setUp functions for importing different data sets
class test_base(unittest.TestCase):
def setUp_flagdatatest(self):
'''VLA data set, scan=2500~2600 spw=0 1 chan, RR,LL'''
self.vis = "flagdatatest.ms"
if os.path.exists(self.vis):
print("The MS is already around, just unflag")
else:
print("Moving data...")
os.system('cp -RH ' + os.path.join(datapath,self.vis) + ' '+ self.vis)
os.system('rm -rf ' + self.vis + '.flagversions')