Commits
Sandra M Castro authored 92130a29383
3 3 | import os |
4 4 | import shutil |
5 5 | import unittest |
6 6 | import math |
7 7 | import numpy as np |
8 8 | import numbers |
9 9 | |
10 10 | try: |
11 11 | from casatools import ctsys, table, msmetadata |
12 12 | from casatasks import exportuvfits, importuvfits |
13 + | |
13 14 | _tb = table() |
14 15 | ctsys_resolve = ctsys.resolve |
15 16 | _msmd = msmetadata() |
16 17 | is_CASA6 = True |
17 18 | except ImportError: |
18 19 | from tasks import * |
19 20 | from taskinit import * |
20 21 | import casac |
21 22 | from __main__ import * |
23 + | |
22 24 | _tb = tbtool() |
23 25 | _msmd = msmdtool() |
24 26 | is_CASA6 = False |
25 27 | from casa_stack_manip import stack_frame_find |
28 + | |
26 29 | casa_stack_rethrow = stack_frame_find().get('__rethrow_casa_exceptions', False) |
27 30 | |
28 31 | data_root = os.environ.get('CASAPATH').split()[0] + '/casatestdata/' |
32 + | |
33 + | |
29 34 | def ctsys_resolve(apath): |
30 35 | return os.path.join(data_root, apath) |
31 36 | |
32 37 | datapath = 'unittest/exportuvfits/' |
33 38 | |
39 + | |
34 40 | class exportuvfits_test(unittest.TestCase): |
35 41 | |
36 42 | def setUp(self): |
37 43 | self.datapath = ctsys_resolve(datapath) |
38 - | self.msname = '' |
39 - | self.reimportms = '' |
40 - | self.fitsname = '' |
41 - | |
44 + | |
42 45 | def tearDown(self): |
43 - | if os.path.exists(self.msname): |
44 - | shutil.rmtree(self.msname) |
45 - | if os.path.exists(self.reimportms): |
46 - | shutil.rmtree(self.reimportms) |
47 - | if os.path.exists(self.fitsname): |
48 - | os.remove(self.fitsname) |
46 + | self.assertTrue(len(_tb.showcache()) == 0) |
47 + | # make sure directory is clean as per verification test requirement |
48 + | cwd = os.getcwd() |
49 + | for filename in os.listdir(cwd): |
50 + | file_path = os.path.join(cwd, filename) |
51 + | try: |
52 + | if os.path.isfile(file_path) or os.path.islink(file_path): |
53 + | os.unlink(file_path) |
54 + | elif os.path.isdir(file_path): |
55 + | # CASA 5 tests need this directory |
56 + | if filename != 'xml': |
57 + | shutil.rmtree(file_path) |
58 + | except Exception as e: |
59 + | print('Failed to delete %s. Reason: %s' % (file_path, e)) |
49 60 | |
50 61 | def test_export_overwrite(self): |
51 62 | """CAS-5492: test the overwrite parameter when exporting MSes to uvfits""" |
52 - | self.msname = "uvfits_test.ms" |
53 - | shutil.copytree(os.path.join(self.datapath, self.msname), self.msname) |
54 - | self.fitsname = "CAS-5492.uvfits" |
55 - | res = exportuvfits(vis=self.msname, fitsfile=self.fitsname) |
63 + | msname = "uvfits_test.ms" |
64 + | shutil.copytree(os.path.join(self.datapath, msname), msname) |
65 + | fitsname = "CAS-5492.uvfits" |
66 + | res = exportuvfits(vis=msname, fitsfile=fitsname) |
56 67 | if is_CASA6: |
57 68 | # Not sure why all of a sudden CASA6 is returning None for tasks |
58 69 | self.assertTrue(res == None, "Failed exportuvfits") |
59 70 | else: |
60 71 | self.assertTrue(res, "Failed exportuvfits") |
61 72 | # fail because overwrite=False. |
62 73 | # CASA 6 throws an exception, CASA 5 returns False |
63 74 | if is_CASA6 or casa_stack_rethrow: |
64 75 | self.assertRaises( |
65 - | Exception, exportuvfits, vis=self.msname, fitsfile=self.fitsname, overwrite=False, |
76 + | Exception, exportuvfits, vis=msname, fitsfile=fitsname, overwrite=False, |
66 77 | msg="exportuvfits succeeded but should have failed because " |
67 - | + "overwrite=False" |
78 + | + "overwrite=False" |
68 79 | ) |
69 80 | else: |
70 81 | self.assertFalse( |
71 - | exportuvfits(vis=self.msname, fitsfile=self.fitsname, overwrite=False), |
82 + | exportuvfits(vis=msname, fitsfile=fitsname, overwrite=False), |
72 83 | "exportuvfits succeeded but should have failed because " |
73 84 | + "overwrite=False" |
74 85 | ) |
75 86 | # succeed because overwrite=True |
76 - | res = exportuvfits(vis=self.msname, fitsfile=self.fitsname, overwrite=True) |
87 + | res = exportuvfits(vis=msname, fitsfile=fitsname, overwrite=True) |
77 88 | if is_CASA6: |
78 89 | self.assertTrue( |
79 90 | res == None, |
80 91 | "exportuvfits failed but should have succeeded because " |
81 92 | + "overwrite=True" |
82 93 | ) |
83 94 | else: |
84 95 | self.assertTrue( |
85 96 | res, |
86 97 | "exportuvfits failed but should have succeeded because " |
87 98 | + "overwrite=True" |
88 99 | ) |
89 - | |
90 100 | |
91 101 | def test_no_rest_freqs(self): |
92 102 | """CAS-11514: test exporting an MS with no rest frequencies in the SOURCE table""" |
93 - | self.msname = "rest_freq_test.ms" |
94 - | shutil.copytree(os.path.join(self.datapath, self.msname), self.msname) |
95 - | self.fitsname = "no_rest_freqs.uvfits" |
96 - | res = exportuvfits(vis=self.msname, fitsfile=self.fitsname) |
103 + | msname = "rest_freq_test.ms" |
104 + | shutil.copytree(os.path.join(self.datapath, msname), msname) |
105 + | fitsname = "no_rest_freqs.uvfits" |
106 + | res = exportuvfits(vis=msname, fitsfile=fitsname) |
97 107 | if is_CASA6: |
98 108 | self.assertTrue(res == None, "Failed exportuvfits with no rest freqs") |
99 109 | else: |
100 110 | self.assertTrue(res, "Failed exportuvfits with no rest freqs") |
101 111 | # import and check the rest freqs |
102 112 | # importuvfits doesn't return anything, so we cannot test the |
103 113 | # return value for success |
104 - | self.reimportms = "imported_no_restfreqs.ms" |
105 - | importuvfits(fitsfile=self.fitsname, vis=self.reimportms) |
106 - | _msmd.open(self.reimportms) |
114 + | msname = "imported_no_restfreqs.ms" |
115 + | importuvfits(fitsfile=fitsname, vis=msname) |
116 + | _msmd.open(msname) |
107 117 | restfreqs = _msmd.restfreqs() |
108 118 | _msmd.done() |
109 119 | expec = { |
110 120 | '0': { |
111 121 | 'type': 'frequency', 'm0': {'value': 0.0, 'unit': 'Hz'}, |
112 122 | 'refer': 'LSRK' |
113 123 | } |
114 124 | } |
115 125 | self.assertEqual( |
116 126 | restfreqs, expec, "Got wrong restfreqs from re-imported dataset" |
117 127 | ) |
118 128 | |
119 129 | def test_no_source_table(self): |
120 130 | """CAS-11514: test exporting an MS with no rest frequencies in the SOURCE table""" |
121 - | self.msname = "no_source_table.ms" |
122 - | shutil.copytree(os.path.join(self.datapath, self.msname), self.msname) |
123 - | self.fitsname = "no_source_table.uvfits" |
124 - | res = exportuvfits(vis=self.msname, fitsfile=self.fitsname) |
131 + | msname = "no_source_table.ms" |
132 + | shutil.copytree(os.path.join(self.datapath, msname), msname) |
133 + | fitsname = "no_source_table.uvfits" |
134 + | res = exportuvfits(vis=msname, fitsfile=fitsname) |
125 135 | if is_CASA6: |
126 136 | self.assertTrue(res == None, "Failed exportuvfits with no SOURCE table") |
127 137 | else: |
128 138 | self.assertTrue(res, "Failed exportuvfits with no SOURCE table") |
129 - | |
139 + | |
140 + | |
130 141 | def suite(): |
131 - | return [exportuvfits_test] |
142 + | return [exportuvfits_test] |
143 + | |
132 144 | |
133 145 | if __name__ == '__main__': |
134 146 | unittest.main() |