Commits

David Mehringer authored b2ad643370d
check point
No tags

casatasks/tests/tasks/test_task_antposalma.py

Modified
15 15 # License for more details.
16 16 #
17 17 # [Add the link to the JIRA ticket here once it exists]
18 18 #
19 19 # Based on the requirements listed in plone found here:
20 20 # https://casadocs.readthedocs.io/en/stable/api/tt/casatasks.manipulation.phaseshift.html
21 21 #
22 22 #
23 23 ##########################################################################
24 24 # import glob
25 -import http.server
26 -import os
27 -from pathlib import Path
28 25 """
29 26 import numpy as np
30 27 import re
31 28 import shutil
32 29 import sys
33 -import threading
34 30 """
31 +import casatestutils
32 +import http.server
33 +import os
34 +import threading
35 35 import unittest
36 +from pathlib import Path
37 +from urllib.parse import urlparse, parse_qs
36 38 """
37 39 from urllib import request
38 40 from urllib.error import URLError
39 -from urllib.parse import urlparse, parse_qs
40 41
41 -from casatasks import casalog
42 42
43 43 from casatools import componentlist, measures
44 44 """
45 45
46 -from casatasks import antposalma
46 +from casatasks import antposalma, casalog
47 47
48 -# import casatestutils
49 48
50 -"""
51 49 # NOTE be certain to specify the top-level casatestutils directory
52 50 # in your PYTHONPATH so you load the casatestutils directory which
53 51 # is a subdir of that
54 52
55 53
56 54 class MockHTTPRequestHandler(http.server.BaseHTTPRequestHandler):
57 - """"HTTPServer mock request handler""""
55 + """HTTPServer mock request handler"""
58 56
59 57 def do_GET(self):
60 58 casalog.post('server path ' + self.path, 'WARN')
61 59 parms = parse_qs(urlparse(self.path).query)
62 60 casalog.post(f'server parms {parms}', 'INFO')
63 61 good_sources = ("3C48", "3C286", "3C138", "3C147")
64 62 if 'source' in parms and parms['source'][0].upper() not in good_sources:
65 63 explain = f'source must be one of {good_sources}'
66 64 self.send_error(400, message='Invalid input', explain=explain)
67 65 self.end_headers()
68 66 return
69 - """"Handle GET requests""""
67 + """Handle GET requests"""
70 68 self.send_response(200)
71 69 self.send_header("Content-Type", "application/json")
72 70 self.end_headers()
73 71 myfile = os.sep.join([
74 - casatestutils.__path__[0], 'calmod_helpers', 'query1.json'
72 + casatestutils.__path__[0], 'antposalma_helpers', 'query1.json'
75 73 ])
76 74 with open(myfile, 'r') as f:
77 75 file_contents = f.read()
78 76 self.wfile.write(str.encode(file_contents))
79 77
78 +
80 79 def log_request(self, code=None, size=None):
81 - """"Don't log anything""""
80 + """Don't log anything"""
81 + pass
82 82
83 -"""
84 83
85 84 class antposalma_test(unittest.TestCase):
86 - """
87 85
88 - hostname = 'http://127.0.0.1:8080'
89 - """
86 + hostname = "http://127.0.0.1:8080"
87 +
88 + outfile = "antposalma.json"
90 89
91 90 def setUp(self):
92 - pass
91 + if os.path.exists(self.outfile):
92 + os.remove(self.outfile)
93 +
93 94
94 95 def tearDown(self):
95 - pass
96 + if os.path.exists(self.outfile):
97 + os.remove(self.outfile)
98 +
96 99
97 100 def exception_verification(self, cm, expected_msg):
98 101 exc = cm.exception
99 102 pos = str(exc).find(expected_msg)
100 103 self.assertNotEqual(
101 104 pos, -1, msg=f'Unexpected exception was thrown: {exc}'
102 105 )
103 106
104 107
105 108 def _query_server(self, method):
210 213 with self.assertRaises(RuntimeError) as cm:
211 214 antposalma(
212 215 outfile="myants.json", hosts=["http://www.bogus.edu"]
213 216 )
214 217 self.exception_verification(
215 218 cm,
216 219 "All URLs failed to return an antenna position list."
217 220 )
218 221
219 222
220 - """
221 223 def test_json_file_writing(self):
222 - """"Test successful writing of json file of antenna positions""""
224 + """Test successful writing of json file of antenna positions"""
223 225 hosts = [self.hostname]
224 - self.query_server(
225 - lambda: calmod(
226 - self.clname, '3C48', band='Q',obsdate=50000, hosts=hosts
226 + self._query_server(
227 + lambda: antposalma(
228 + self.outfile, asdm="uid://A002/X10ac6bc/X896d", hosts=hosts
227 229 )
228 230 )
229 - self.cl.open(self.clname)
230 - self.assertEqual(self.cl.length(), 385, 'Incorrect number of components')
231 - ws = self.cl.getkeyword('web_service')
232 - self.assertEqual(ws['band'], 'Q', 'Incorrect band in web_service metadata')
233 - self.assertEqual(ws['source'], '3C48', 'Incorrect source in web_service metadata')
234 - """
235 - """
236 - def test_bad_source_name(self):
237 - hosts = [self.hostname]
238 - with self.assertRaises(RuntimeError) as cm:
239 - self.query_server(
240 - lambda: calmod(
241 - 'my.cl', 'mysource', band='L', hosts=[self.hostname],
242 - obsdate=50000
243 - )
244 - )
245 - self.exception_verification(cm, 'All URLs failed to return a component list')
246 - found = False
247 - pattern = "source must be one of \('3C48', '3C286', '3C138', '3C147'\)"
248 - with open(casalog.logfile()) as logfile:
249 - for line in logfile:
250 - if re.search(pattern, line):
251 - found = True
252 - break
253 - self.assertTrue(found)
254 -
231 + self.assertTrue(os.path.exists(self.outfile))
255 232
256 - def test_obsdate_as_string(self):
257 - hosts = [self.hostname]
258 - self.query_server(
259 - lambda: calmod(
260 - self.clname, '3C48', band='Q',obsdate='2002-04-20', hosts=hosts
261 - )
262 - )
263 - self.cl.open(self.clname)
264 - self.assertEqual(self.cl.length(), 385, 'Incorrect number of components')
265 - ws = self.cl.getkeyword('web_service')
266 - self.assertEqual(ws['band'], 'Q', 'Incorrect band in web_service metadata')
267 - self.assertEqual(ws['source'], '3C48', 'Incorrect source in web_service metadata')
268 -
269 -
270 - def test_direction(self):
271 - """"Test direction input""""
272 - hosts = [self.hostname]
273 - direction = 'J2000 01:37:41.1 33.09.32'
274 - self.query_server(
275 - lambda: calmod(
276 - self.clname, direction=direction, band='Q',obsdate=50000, hosts=hosts
277 - )
278 - )
279 - self.cl.open(self.clname)
280 - self.assertEqual(self.cl.length(), 385, 'Incorrect number of components')
281 - ws = self.cl.getkeyword('web_service')
282 - self.assertEqual(ws['band'], 'Q', 'Incorrect band in web_service metadata')
283 - """
284 233
285 234 if __name__ == '__main__':
286 235 unittest.main()

Everything looks good. We'll let you know here if there's anything you should know about.

Add shortcut