xan/awips2
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Omaha #4795 Add tests to python DAF regression test framework

Browse source 
Change-Id: Ic1096b2c6a5a0c5f081d4710ac18b512cdd5b348

Former-commit-id: e5999c9ed5e32317f57c1c0276c3896c17ede363
This commit is contained in:
Mark Peters 2016-01-29 13:42:13 -06:00
parent 9901cd4d60
commit abf916d5c8
34 changed files with 1225 additions and 318 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

13
pythonPackages/ufpy/test/dafTests/baseBufrMosTest.py → pythonPackages/ufpy/test/dafTests/baseBufrMosTestCase.py
View file

@ -20,10 +20,9 @@
from ufpy.dataaccess import DataAccessLayer as DAL
import dafTestsUtil
import unittest
import baseDafTestCase
class BaseBufrMosTest(unittest.TestCase):
class BufrMosTestCase(baseDafTestCase.DafTestCase):
"""
Base class for testing that bufrmos data can be retrieved through the DAF.
"""
@ -31,22 +30,22 @@ class BaseBufrMosTest(unittest.TestCase):
def testParameters(self):
req = DAL.newDataRequest(self.datatype)
dafTestsUtil.testParameters(req)
self.runParametersTest(req)
def testLocations(self):
req = DAL.newDataRequest(self.datatype)
dafTestsUtil.testLocations(req)
self.runLocationsTest(req)
def testTimes(self):
req = DAL.newDataRequest(self.datatype)
req.setLocationNames("KOMA")
dafTestsUtil.testTimes(req)
self.runTimesTest(req)
def testGeometryData(self):
req = DAL.newDataRequest(self.datatype)
req.setLocationNames("KOMA")
req.setParameters("temperature", "dewpoint")
dafTestsUtil.testGeometryData(req)
self.runGeometryDataTest(req)

174
pythonPackages/ufpy/test/dafTests/baseDafTestCase.py Normal file
View file

@ -0,0 +1,174 @@
##
# This software was developed and / or modified by Raytheon Company,
# pursuant to Contract DG133W-05-CQ-1067 with the US Government.
#
# U.S. EXPORT CONTROLLED TECHNICAL DATA
# This software product contains export-restricted data whose
# export/transfer/disclosure is restricted by U.S. law. Dissemination
# to non-U.S. persons whether in the United States or abroad requires
# an export license or other authorization.
#
# Contractor Name: Raytheon Company
# Contractor Address: 6825 Pine Street, Suite 340
# Mail Stop B8
# Omaha, NE 68106
# 402.291.0100
#
# See the AWIPS II Master Rights File ("Master Rights File.pdf") for
# further licensing information.
##
from __future__ import print_function
from ufpy.dataaccess import DataAccessLayer as DAL
import unittest
#
# Base TestCase for DAF tests. Note that this class doesn't provide any test
# methods that are automatically called when running this TestCase (methods
# named test*()). Instead, it provides default implementations of various tests
# that subclasses must explicity choose to call from their test*() methods.
#
# The tests primarily check that no unexpected exceptions are thrown when
# retrieving data for DAF requests, instead of explicity asserting against
# expected outcomes.
#
# SOFTWARE HISTORY
#
# Date Ticket# Engineer Description
# ------------ ---------- ----------- --------------------------
# 01/19/16 4795 mapeters Initial Creation.
#
#
#
class DafTestCase(unittest.TestCase):
sampleDataLimit = 5
"""
Maximum number of levels, locations, times, and geometry/grid data to
display
"""
numTimesToLimit = 3
"""
When limiting geometry/grid data requests with times, only retrieve data
for this many times
"""
def runParametersTest(self, req):
print("Testing getAvailableParameters()")
params = DAL.getAvailableParameters(req)
self.assertIsNotNone(params)
print(params)
print("getAvailableParameters() complete\n")
def runIdentifiersTest(self, datatype):
print("Testing get*Identifiers()")
required = DAL.getRequiredIdentifiers(datatype)
self.assertIsNotNone(required)
print("Required identifiers:", required)
optional = DAL.getOptionalIdentifiers(datatype)
self.assertIsNotNone(optional)
print("Optional identifiers:", optional)
print("get*Identifiers() complete\n")
def runLevelsTest(self, req):
print("Testing getAvailableLevels()")
levels = DAL.getAvailableLevels(req)
self.assertIsNotNone(levels)
print("Number of levels: " + str(len(levels)))
strLevels = map(str, levels[:self.sampleDataLimit])
print("Sample levels:\n" + str(strLevels))
print("getAvailableLevels() complete\n")
def runLocationsTest(self, req):
print("Testing getAvailableLocationNames()")
locs = DAL.getAvailableLocationNames(req)
self.assertIsNotNone(locs)
print("Number of location names: " + str(len(locs)))
print("Sample location names:\n" + str(locs[:self.sampleDataLimit]))
print("getAvailableLocationNames() complete\n")
def runTimesTest(self, req):
print("Testing getAvailableTimes()")
times = DAL.getAvailableTimes(req)
self.assertIsNotNone(times)
print("Number of times: " + str(len(times)))
strTimes = map(str, times[:self.sampleDataLimit])
print("Sample times:\n" + str(strTimes))
print("getAvailableTimes() complete\n")
def runGeometryDataTest(self, req, limitTimes=False):
"""
Test that we are able to successfully retrieve geometry data for the given
request.
Args:
limitTimes: whether or not to limit the request to the most recent times
"""
print("Testing getGeometryData()")
times = []
if limitTimes:
times = DAL.getAvailableTimes(req)
geomData = DAL.getGeometryData(req, times[:self.numTimesToLimit])
self.assertIsNotNone(geomData)
print("Number of geometry records: " + str(len(geomData)))
print("Sample geometry data:")
for record in geomData[:self.sampleDataLimit]:
print("geometry=" + str(record.getGeometry()), end="")
for p in req.getParameters():
print(" " + p + "=" + record.getString(p), end="")
print()
print("getGeometryData() complete\n")
def runGridDataTest(self, req, limitTimes=False, testSameShape=True):
"""
Test that we are able to successfully retrieve grid data for the given
request.
Args:
limitTimes: whether or not to limit the request to the most recent times
testSameShape: whether or not to verify that all the retrieved data have
the same shape (most data don't change shape)
"""
print("Testing getGridData()")
times = []
if limitTimes:
times = DAL.getAvailableTimes(req)
gridData = DAL.getGridData(req, times[:self.numTimesToLimit])
self.assertIsNotNone(gridData)
print("Number of grid records: " + str(len(gridData)))
if len(gridData) > 0:
print("Sample grid data shape:\n" + str(gridData[0].getRawData().shape) + "\n")
print("Sample grid data:\n" + str(gridData[0].getRawData()) + "\n")
print("Sample lat-lon data:\n" + str(gridData[0].getLatLonCoords()) + "\n")
if testSameShape:
allGridsPopulated = True
correctGridShape = gridData[0].getLatLonCoords()[0].shape
for record in gridData:
if record.getRawData() is None or record.getRawData().shape != correctGridShape:
allGridsPopulated = False
break
print("All grid data records have correct grid data: " + str(allGridsPopulated))
self.assertTrue(allGridsPopulated)
print("getGridData() complete\n")

66
pythonPackages/ufpy/test/dafTests/dafTestsArgsUtil.py Normal file
View file

@ -0,0 +1,66 @@
##
# This software was developed and / or modified by Raytheon Company,
# pursuant to Contract DG133W-05-CQ-1067 with the US Government.
#
# U.S. EXPORT CONTROLLED TECHNICAL DATA
# This software product contains export-restricted data whose
# export/transfer/disclosure is restricted by U.S. law. Dissemination
# to non-U.S. persons whether in the United States or abroad requires
# an export license or other authorization.
#
# Contractor Name: Raytheon Company
# Contractor Address: 6825 Pine Street, Suite 340
# Mail Stop B8
# Omaha, NE 68106
# 402.291.0100
#
# See the AWIPS II Master Rights File ("Master Rights File.pdf") for
# further licensing information.
##
from ufpy.dataaccess import DataAccessLayer as DAL
import argparse
import os
import sys
#
# Utility methods to be used by DAF tests for parsing arguments.
#
# SOFTWARE HISTORY
#
# Date Ticket# Engineer Description
# ------------ ---------- ----------- --------------------------
# 02/05/16 4795 mapeters Initial Creation.
#
#
#
def getParser():
"""
Return an ArgumentParser for parsing the standard arguments: the host to run
tests against and whether to display the data retrieved in the tests.
"""
parser = argparse.ArgumentParser(conflict_handler="resolve")
parser.add_argument("-h", action="store", dest="host", default="localhost",
help="EDEX server hostname",
metavar="hostname")
parser.add_argument("-v", action="store_true", dest="verbose", default=False,
help="Display data output")
return parser
def handleArgs(args):
"""
Handle the arguments specified in getParser().
"""
DAL.changeEDEXHost(args.host)
# Suppress stdout unless user-requested
if not args.verbose:
sys.stdout = open(os.devnull, "w")
def parseAndHandleArgs():
"""
Parses and handles the arguments specified in getParser(). Use this
method to handle arguments when no additional arguments need to be added.
"""
handleArgs(getParser().parse_args())

172
pythonPackages/ufpy/test/dafTests/dafTestsUtil.py
View file

@ -1,172 +0,0 @@
##
# This software was developed and / or modified by Raytheon Company,
# pursuant to Contract DG133W-05-CQ-1067 with the US Government.
#
# U.S. EXPORT CONTROLLED TECHNICAL DATA
# This software product contains export-restricted data whose
# export/transfer/disclosure is restricted by U.S. law. Dissemination
# to non-U.S. persons whether in the United States or abroad requires
# an export license or other authorization.
#
# Contractor Name: Raytheon Company
# Contractor Address: 6825 Pine Street, Suite 340
# Mail Stop B8
# Omaha, NE 68106
# 402.291.0100
#
# See the AWIPS II Master Rights File ("Master Rights File.pdf") for
# further licensing information.
##
from __future__ import print_function
from ufpy.dataaccess import DataAccessLayer as DAL
import argparse
import os
import sys
#
# Standard testing and utility methods for most DAF unit test requests.
# The tests primarily check that no unexpected exceptions are thrown when
# retrieving data for DAF requests, instead of explicity asserting against
# expected outcomes.
#
# SOFTWARE HISTORY
#
# Date Ticket# Engineer Description
# ------------ ---------- ----------- --------------------------
# 01/19/16 4795 mapeters Initial Creation.
#
#
#
# Maximum number of levels, locations, times, and geometry/grid data to display
sampleDataLimit = 5
def testParameters(req):
print("Testing getAvailableParameters()")
params = DAL.getAvailableParameters(req)
assert params is not None
print(params)
print("getAvailableParameters() complete\n")
def testIdentifiers(datatype):
print("Testing get*Identifiers()")
required = DAL.getRequiredIdentifiers(datatype)
assert required is not None
print("Required identifiers:", required)
optional = DAL.getOptionalIdentifiers(datatype)
assert optional is not None
print("Optional identifiers:", optional)
print("get*Identifiers() complete\n")
def testLevels(req):
print("Testing getAvailableLevels()")
levels = DAL.getAvailableLevels(req)
assert levels is not None
print("Number of levels: " + str(len(levels)))
strLevels = map(str, levels[:sampleDataLimit])
print("Sample levels:\n" + str(strLevels))
print("getAvailableLevels() complete\n")
def testLocations(req):
print("Testing getAvailableLocationNames()")
locs = DAL.getAvailableLocationNames(req)
assert locs is not None
print("Number of location names: " + str(len(locs)))
print("Sample location names:\n" + str(locs[:sampleDataLimit]))
print("getAvailableLocationNames() complete\n")
def testTimes(req):
print("Testing getAvailableTimes()")
times = DAL.getAvailableTimes(req)
assert times is not None
print("Number of times: " + str(len(times)))
strTimes = map(str, times[:sampleDataLimit])
print("Sample times:\n" + str(strTimes))
print("getAvailableTimes() complete\n")
def testGeometryData(req):
print("Testing getGeometryData()")
geomData = DAL.getGeometryData(req)
assert geomData is not None
print("Number of geometry records: " + str(len(geomData)))
print("Sample geometry data:")
for record in geomData[:sampleDataLimit]:
print("geometry=" + str(record.getGeometry()), end="")
for p in req.getParameters():
print(" " + p + "=" + record.getString(p), end="")
print()
print("getGeometryData() complete\n")
def testGridData(req, testSameShape=True):
"""
Test that we are able to successfully retrieve grid data for the given
request. Most data do not change shape, so an additional verification is
typically made to ensure that all the retrieved data have the same shape,
unless False is given for the testSameShape parameter.
"""
print("Testing getGridData()")
gridData = DAL.getGridData(req)
assert gridData is not None
print("Number of grid records: " + str(len(gridData)))
if len(gridData) > 0:
print("Sample grid data shape:\n" + str(gridData[0].getRawData().shape) + "\n")
print("Sample grid data:\n" + str(gridData[0].getRawData()) + "\n")
print("Sample lat-lon data:\n" + str(gridData[0].getLatLonCoords()) + "\n")
if testSameShape:
allGridsPopulated = True
correctGridShape = gridData[0].getLatLonCoords()[0].shape
for record in gridData:
if record.getRawData() is None or record.getRawData().shape != correctGridShape:
allGridsPopulated = False
break
print("All grid data records have correct grid data: " + str(allGridsPopulated))
assert allGridsPopulated
print("getGridData() complete\n")
def getParser():
"""
Return an ArgumentParser for parsing the standard arguments: the host to run
tests against and whether to display the data retrieved in the tests.
"""
parser = argparse.ArgumentParser(conflict_handler="resolve")
parser.add_argument("-h", action="store", dest="host", default="localhost",
help="EDEX server hostname",
metavar="hostname")
parser.add_argument("-v", action="store_true", dest="verbose", default=False,
help="Display data output")
return parser
def handleArgs(args):
"""
Handle the arguments specified in getParser().
"""
DAL.changeEDEXHost(args.host)
# Suppress stdout unless user-requested
if not args.verbose:
sys.stdout = open(os.devnull, "w")
def parseAndHandleArgs():
"""
Parses and handles the arguments specified in getParser(). Use this
method to handle arguments when no additional arguments need to be added.
"""
handleArgs(getParser().parse_args())

4
pythonPackages/ufpy/test/dafTests/runDafTests.py
View file

@ -18,7 +18,7 @@
# further licensing information.
##
import dafTestsUtil
import dafTestsArgsUtil
import os
import sys
import unittest
@ -41,5 +41,5 @@ def load_tests(loader, tests, pattern):
return loader.discover(start_dir=testDir)
if __name__ == "__main__":
dafTestsUtil.parseAndHandleArgs()
dafTestsArgsUtil.parseAndHandleArgs()
unittest.main(argv=sys.argv[:1])

71
pythonPackages/ufpy/test/dafTests/testAcars.py Normal file
View file

@ -0,0 +1,71 @@
##
# This software was developed and / or modified by Raytheon Company,
# pursuant to Contract DG133W-05-CQ-1067 with the US Government.
#
# U.S. EXPORT CONTROLLED TECHNICAL DATA
# This software product contains export-restricted data whose
# export/transfer/disclosure is restricted by U.S. law. Dissemination
# to non-U.S. persons whether in the United States or abroad requires
# an export license or other authorization.
#
# Contractor Name: Raytheon Company
# Contractor Address: 6825 Pine Street, Suite 340
# Mail Stop B8
# Omaha, NE 68106
# 402.291.0100
#
# See the AWIPS II Master Rights File ("Master Rights File.pdf") for
# further licensing information.
##
from __future__ import print_function
from ufpy.dataaccess import DataAccessLayer as DAL
import baseDafTestCase
import dafTestsArgsUtil
import sys
import unittest
class AcarsTestCase(baseDafTestCase.DafTestCase):
"""
Tests that acars data can be retrieved through the DAF, simply ensuring
that no unexpected exceptions are thrown while retrieving it and that the
returned data is not None.
"""
datatype = "acars"
@classmethod
def setUpClass(cls):
print("STARTING ACARS TESTS\n\n")
def testParameters(self):
req = DAL.newDataRequest(self.datatype)
self.runParametersTest(req)
def testLocations(self):
req = DAL.newDataRequest(self.datatype)
self.runLocationsTest(req)
def testTimes(self):
req = DAL.newDataRequest(self.datatype)
self.runTimesTest(req)
def testGeometryData(self):
req = DAL.newDataRequest(self.datatype)
req.setParameters("flightLevel", "tailNumber")
# Limit the times in the geometry data test to limit the amount of data
# returned.
self.runGeometryDataTest(req, limitTimes=True)
@classmethod
def tearDownClass(cls):
print("ACARS TESTS COMPLETE\n\n\n")
if __name__ == '__main__':
dafTestsArgsUtil.parseAndHandleArgs()
unittest.main(argv=sys.argv[:1])

15
pythonPackages/ufpy/test/dafTests/testAirep.py
View file

@ -21,11 +21,12 @@
from __future__ import print_function
from ufpy.dataaccess import DataAccessLayer as DAL
import dafTestsUtil
import baseDafTestCase
import dafTestsArgsUtil
import sys
import unittest
class TestAirep(unittest.TestCase):
class AirepTestCase(baseDafTestCase.DafTestCase):
"""
Tests that airep data can be retrieved through the DAF, simply ensuring
that no unexpected exceptions are thrown while retrieving it and that the
@ -41,28 +42,28 @@ class TestAirep(unittest.TestCase):
def testParameters(self):
req = DAL.newDataRequest(self.datatype)
dafTestsUtil.testParameters(req)
self.runParametersTest(req)
def testLocations(self):
req = DAL.newDataRequest(self.datatype)
dafTestsUtil.testLocations(req)
self.runLocationsTest(req)
def testTimes(self):
req = DAL.newDataRequest(self.datatype)
dafTestsUtil.testTimes(req)
self.runTimesTest(req)
def testGeometryData(self):
req = DAL.newDataRequest(self.datatype)
req.setParameters("flightLevel", "reportType")
dafTestsUtil.testGeometryData(req)
self.runGeometryDataTest(req)
@classmethod
def tearDownClass(cls):
print("AIREP TESTS COMPLETE\n\n\n")
if __name__ == '__main__':
dafTestsUtil.parseAndHandleArgs()
dafTestsArgsUtil.parseAndHandleArgs()
unittest.main(argv=sys.argv[:1])

79
pythonPackages/ufpy/test/dafTests/testBinLightning.py Normal file
View file

@ -0,0 +1,79 @@
##
# This software was developed and / or modified by Raytheon Company,
# pursuant to Contract DG133W-05-CQ-1067 with the US Government.
#
# U.S. EXPORT CONTROLLED TECHNICAL DATA
# This software product contains export-restricted data whose
# export/transfer/disclosure is restricted by U.S. law. Dissemination
# to non-U.S. persons whether in the United States or abroad requires
# an export license or other authorization.
#
# Contractor Name: Raytheon Company
# Contractor Address: 6825 Pine Street, Suite 340
# Mail Stop B8
# Omaha, NE 68106
# 402.291.0100
#
# See the AWIPS II Master Rights File ("Master Rights File.pdf") for
# further licensing information.
##
from __future__ import print_function
from ufpy.dataaccess import DataAccessLayer as DAL
import baseDafTestCase
import dafTestsArgsUtil
import sys
import unittest
class BinLightningTestCase(baseDafTestCase.DafTestCase):
"""
Tests that binlightning data can be retrieved through the DAF, simply
ensuring that no unexpected exceptions are thrown while retrieving it and
that the returned data is not None.
"""
datatype = "binlightning"
params = "intensity"
@classmethod
def setUpClass(cls):
print("STARTING BINLIGHTNING TESTS\n\n")
def testParameters(self):
req = DAL.newDataRequest(self.datatype)
self.runParametersTest(req)
def testTimes(self):
req = DAL.newDataRequest(self.datatype)
req.addIdentifier("source", "NLDN")
self.runTimesTest(req)
def testGeometryData(self):
req = DAL.newDataRequest(self.datatype)
req.addIdentifier("source", "NLDN")
req.setParameters(*self.params.split(','))
# Limit the times in the geometry data test to limit the amount of data
# returned.
self.runGeometryDataTest(req, limitTimes=True)
@classmethod
def tearDownClass(cls):
print("BINLIGHTNING TESTS COMPLETE\n\n\n")
def getArgs():
parser = dafTestsArgsUtil.getParser()
parser.add_argument("-p", action="store", dest="params", default=BinLightningTestCase.params,
help="Lightning parameters comma separated",
metavar="parameters")
return parser.parse_args()
if __name__ == '__main__':
args = getArgs()
dafTestsArgsUtil.handleArgs(args)
BinLightningTestCase.params = args.params
unittest.main(argv=sys.argv[:1])

9
pythonPackages/ufpy/test/dafTests/testBufrMosAvn.py
View file

@ -19,14 +19,13 @@
##
from __future__ import print_function
from ufpy.dataaccess import DataAccessLayer as DAL
import baseBufrMosTest
import dafTestsUtil
import baseBufrMosTestCase
import dafTestsArgsUtil
import sys
import unittest
class TestBufrMosAvn(baseBufrMosTest.BaseBufrMosTest):
class BufrMosAvnTestCase(baseBufrMosTestCase.BufrMosTestCase):
"""
Tests that bufrmosAVN data can be retrieved through the DAF, simply ensuring
that no unexpected exceptions are thrown while retrieving it and that the
@ -46,5 +45,5 @@ class TestBufrMosAvn(baseBufrMosTest.BaseBufrMosTest):
print("BUFRMOSAVN TESTS COMPLETE\n\n\n")
if __name__ == '__main__':
dafTestsUtil.parseAndHandleArgs()
dafTestsArgsUtil.parseAndHandleArgs()
unittest.main(argv=sys.argv[:1])

9
pythonPackages/ufpy/test/dafTests/testBufrMosEta.py
View file

@ -19,14 +19,13 @@
##
from __future__ import print_function
from ufpy.dataaccess import DataAccessLayer as DAL
import baseBufrMosTest
import dafTestsUtil
import baseBufrMosTestCase
import dafTestsArgsUtil
import sys
import unittest
class TestBufrMosEta(baseBufrMosTest.BaseBufrMosTest):
class BufrMosEtaTestCase(baseBufrMosTestCase.BufrMosTestCase):
"""
Tests that bufrmosETA data can be retrieved through the DAF, simply ensuring
that no unexpected exceptions are thrown while retrieving it and that the
@ -46,5 +45,5 @@ class TestBufrMosEta(baseBufrMosTest.BaseBufrMosTest):
print("BUFRMOSETA TESTS COMPLETE\n\n\n")
if __name__ == '__main__':
dafTestsUtil.parseAndHandleArgs()
dafTestsArgsUtil.parseAndHandleArgs()
unittest.main(argv=sys.argv[:1])

9
pythonPackages/ufpy/test/dafTests/testBufrMosGfs.py
View file

@ -19,14 +19,13 @@
##
from __future__ import print_function
from ufpy.dataaccess import DataAccessLayer as DAL
import baseBufrMosTest
import dafTestsUtil
import baseBufrMosTestCase
import dafTestsArgsUtil
import sys
import unittest
class TestBufrMosGfs(baseBufrMosTest.BaseBufrMosTest):
class BufrMosGfsTestCase(baseBufrMosTestCase.BufrMosTestCase):
"""
Tests that bufrmosGFS data can be retrieved through the DAF, simply ensuring
that no unexpected exceptions are thrown while retrieving it and that the
@ -46,5 +45,5 @@ class TestBufrMosGfs(baseBufrMosTest.BaseBufrMosTest):
print("BUFRMOSGFS TESTS COMPLETE\n\n\n")
if __name__ == '__main__':
dafTestsUtil.parseAndHandleArgs()
dafTestsArgsUtil.parseAndHandleArgs()
unittest.main(argv=sys.argv[:1])

10
pythonPackages/ufpy/test/dafTests/testBufrMosHpc.py
View file

@ -21,12 +21,12 @@
from __future__ import print_function
from ufpy.dataaccess import DataAccessLayer as DAL
import baseBufrMosTest
import dafTestsUtil
import baseBufrMosTestCase
import dafTestsArgsUtil
import sys
import unittest
class TestBufrMosHpc(baseBufrMosTest.BaseBufrMosTest):
class BufrMosHpcTestCase(baseBufrMosTestCase.BufrMosTestCase):
"""
Tests that bufrmosHPC data can be retrieved through the DAF, simply ensuring
that no unexpected exceptions are thrown while retrieving it and that the
@ -46,12 +46,12 @@ class TestBufrMosHpc(baseBufrMosTest.BaseBufrMosTest):
req.setLocationNames("KOMA")
req.setParameters("forecastHr", "maxTemp24Hour")
dafTestsUtil.testGeometryData(req)
self.runGeometryDataTest(req)
@classmethod
def tearDownClass(cls):
print("BUFRMOSHPC TESTS COMPLETE\n\n\n")
if __name__ == '__main__':
dafTestsUtil.parseAndHandleArgs()
dafTestsArgsUtil.parseAndHandleArgs()
unittest.main(argv=sys.argv[:1])

9
pythonPackages/ufpy/test/dafTests/testBufrMosLamp.py
View file

@ -19,14 +19,13 @@
##
from __future__ import print_function
from ufpy.dataaccess import DataAccessLayer as DAL
import baseBufrMosTest
import dafTestsUtil
import baseBufrMosTestCase
import dafTestsArgsUtil
import sys
import unittest
class TestBufrMosLamp(baseBufrMosTest.BaseBufrMosTest):
class BufrMosLampTestCase(baseBufrMosTestCase.BufrMosTestCase):
"""
Tests that bufrmosLAMP data can be retrieved through the DAF, simply
ensuring that no unexpected exceptions are thrown while retrieving it and
@ -46,5 +45,5 @@ class TestBufrMosLamp(baseBufrMosTest.BaseBufrMosTest):
print("BUFRMOSLAMP TESTS COMPLETE\n\n\n")
if __name__ == '__main__':
dafTestsUtil.parseAndHandleArgs()
dafTestsArgsUtil.parseAndHandleArgs()
unittest.main(argv=sys.argv[:1])

10
pythonPackages/ufpy/test/dafTests/testBufrMosMrf.py
View file

@ -21,12 +21,12 @@
from __future__ import print_function
from ufpy.dataaccess import DataAccessLayer as DAL
import baseBufrMosTest
import dafTestsUtil
import baseBufrMosTestCase
import dafTestsArgsUtil
import sys
import unittest
class TestBufrMosMrf(baseBufrMosTest.BaseBufrMosTest):
class BufrMosMrfTestCase(baseBufrMosTestCase.BufrMosTestCase):
"""
Tests that bufrmosMRF data can be retrieved through the DAF, simply ensuring
that no unexpected exceptions are thrown while retrieving it and that the
@ -46,12 +46,12 @@ class TestBufrMosMrf(baseBufrMosTest.BaseBufrMosTest):
req.setLocationNames("KOMA")
req.setParameters("forecastHr", "maxTempDay")
dafTestsUtil.testGeometryData(req)
self.runGeometryDataTest(req)
@classmethod
def tearDownClass(cls):
print("BUFRMOSMRF TESTS COMPLETE\n\n\n")
if __name__ == '__main__':
dafTestsUtil.parseAndHandleArgs()
dafTestsArgsUtil.parseAndHandleArgs()
unittest.main(argv=sys.argv[:1])

26
pythonPackages/ufpy/test/dafTests/testBufrUa.py
View file

@ -21,11 +21,12 @@
from __future__ import print_function
from ufpy.dataaccess import DataAccessLayer as DAL
import dafTestsUtil
import baseDafTestCase
import dafTestsArgsUtil
import sys
import unittest
class TestBufrUa(unittest.TestCase):
class BufrUaTestCase(baseDafTestCase.DafTestCase):
"""
Tests that bufrua data can be retrieved through the DAF, simply ensuring
that no unexpected exceptions are thrown while retrieving it and that the
@ -35,6 +36,9 @@ class TestBufrUa(unittest.TestCase):
datatype = "bufrua"
location = "72558"
"""
stationid corresponding to KOAX
"""
@classmethod
def setUpClass(cls):
@ -43,20 +47,20 @@ class TestBufrUa(unittest.TestCase):
def testParameters(self):
req = DAL.newDataRequest(self.datatype)
dafTestsUtil.testParameters(req)
self.runParametersTest(req)
def testLocations(self):
req = DAL.newDataRequest(self.datatype)
req.addIdentifier("reportType", "2020")
dafTestsUtil.testLocations(req)
self.runLocationsTest(req)
def testTimes(self):
req = DAL.newDataRequest(self.datatype)
req.setLocationNames(self.location)
req.addIdentifier("reportType", "2020")
dafTestsUtil.testTimes(req)
self.runTimesTest(req)
def testGeometryData(self):
req = DAL.newDataRequest(self.datatype)
@ -67,10 +71,10 @@ class TestBufrUa(unittest.TestCase):
print("Testing getGeometryData()")
geomData = DAL.getGeometryData(req)
assert geomData is not None
self.assertIsNotNone(geomData)
print("Number of geometry records: " + str(len(geomData)))
print("Sample geometry data:")
for record in geomData[:dafTestsUtil.sampleDataLimit]:
for record in geomData[:self.sampleDataLimit]:
print("level=", record.getLevel(), end="")
# One dimensional parameters are reported on the 0.0UNKNOWN level.
# 2D parameters are reported on MB levels from pressure.
@ -89,14 +93,14 @@ class TestBufrUa(unittest.TestCase):
print("BUFRUA TESTS COMPLETE\n\n\n")
def getArgs():
parser = dafTestsUtil.getParser()
parser.add_argument("-l", action="store", dest="loc", default=TestBufrUa.location,
parser = dafTestsArgsUtil.getParser()
parser.add_argument("-l", action="store", dest="loc", default=BufrUaTestCase.location,
help="bufrua location id",
metavar="location")
return parser.parse_args()
if __name__ == '__main__':
args = getArgs()
dafTestsUtil.handleArgs(args)
TestBufrUa.location = args.loc
dafTestsArgsUtil.handleArgs(args)
BufrUaTestCase.location = args.loc
unittest.main(argv=sys.argv[:1])

77
pythonPackages/ufpy/test/dafTests/testClimate.py Normal file
View file

@ -0,0 +1,77 @@
##
# This software was developed and / or modified by Raytheon Company,
# pursuant to Contract DG133W-05-CQ-1067 with the US Government.
#
# U.S. EXPORT CONTROLLED TECHNICAL DATA
# This software product contains export-restricted data whose
# export/transfer/disclosure is restricted by U.S. law. Dissemination
# to non-U.S. persons whether in the United States or abroad requires
# an export license or other authorization.
#
# Contractor Name: Raytheon Company
# Contractor Address: 6825 Pine Street, Suite 340
# Mail Stop B8
# Omaha, NE 68106
# 402.291.0100
#
# See the AWIPS II Master Rights File ("Master Rights File.pdf") for
# further licensing information.
##
from __future__ import print_function
from ufpy.dataaccess import DataAccessLayer as DAL
import baseDafTestCase
import dafTestsArgsUtil
import sys
import unittest
class ClimateTestCase(baseDafTestCase.DafTestCase):
"""
Tests that climate data can be retrieved through the DAF, simply ensuring
that no unexpected exceptions are thrown while retrieving it and that the
returned data is not None.
"""
datatype = "climate"
table = "public.cli_asos_monthly"
@classmethod
def setUpClass(cls):
print("STARTING CLIMATE TESTS\n\n")
def testParameters(self):
req = DAL.newDataRequest(self.datatype)
req.addIdentifier("table", self.table)
self.runParametersTest(req)
def testLocations(self):
req = DAL.newDataRequest(self.datatype)
req.addIdentifier("table", self.table)
self.runLocationsTest(req)
def testTimes(self):
req = DAL.newDataRequest(self.datatype)
req.addIdentifier("table", self.table)
req.setParameters("maxtemp_mon", "min_sea_press")
self.runTimesTest(req)
def testGeometryData(self):
req = DAL.newDataRequest(self.datatype)
req.addIdentifier("table", self.table)
req.setLocationNames("KFNB")
req.setParameters("maxtemp_mon", "min_sea_press")
self.runGeometryDataTest(req)
@classmethod
def tearDownClass(cls):
print("CLIMATE TESTS COMPLETE\n\n\n")
if __name__ == '__main__':
dafTestsArgsUtil.parseAndHandleArgs()
unittest.main(argv=sys.argv[:1])

24
pythonPackages/ufpy/test/dafTests/testCommonObsSpatial.py
View file

@ -22,11 +22,12 @@ from __future__ import print_function
from shapely.geometry import box
from ufpy.dataaccess import DataAccessLayer as DAL
import dafTestsUtil
import baseDafTestCase
import dafTestsArgsUtil
import sys
import unittest
class TestCommonObsSpatial(unittest.TestCase):
class CommonObsSpatialTestCase(baseDafTestCase.DafTestCase):
"""
Tests that common_obs_spatial data can be retrieved through the DAF, simply
ensuring that no unexpected exceptions are thrown while retrieving it and
@ -36,6 +37,9 @@ class TestCommonObsSpatial(unittest.TestCase):
datatype = "common_obs_spatial"
bbox = ["-96", "41", "-97", "42"]
"""
Default request area (box around KOAX)
"""
envelope = None
@ -57,38 +61,38 @@ class TestCommonObsSpatial(unittest.TestCase):
def testParameters(self):
req = DAL.newDataRequest(self.datatype)
dafTestsUtil.testParameters(req)
self.runParametersTest(req)
def testLocations(self):
req = DAL.newDataRequest(self.datatype)
req.addIdentifier("country", ["US", "CN"])
dafTestsUtil.testLocations(req)
self.runLocationsTest(req)
def testIdentifiers(self):
dafTestsUtil.testIdentifiers(self.datatype)
self.runIdentifiersTest(self.datatype)
def testGeometryData(self):
req = DAL.newDataRequest(self.datatype)
req.setEnvelope(self.getReqEnvelope())
req.setParameters("name", "stationid")
dafTestsUtil.testGeometryData(req)
self.runGeometryDataTest(req)
@classmethod
def tearDownClass(cls):
print("COMMON_OBS_SPATIAL TESTS COMPLETE\n\n\n")
def getArgs():
parser = dafTestsUtil.getParser()
parser.add_argument("-p", action="store", dest="bbox", nargs=4, default=TestCommonObsSpatial.bbox,
parser = dafTestsArgsUtil.getParser()
parser.add_argument("-p", action="store", dest="bbox", nargs=4, default=CommonObsSpatialTestCase.bbox,
help="Request area",
metavar="point")
return parser.parse_args()
if __name__ == '__main__':
args = getArgs()
dafTestsUtil.handleArgs(args)
TestCommonObsSpatial.bbox = args.bbox
dafTestsArgsUtil.handleArgs(args)
CommonObsSpatialTestCase.bbox = args.bbox
unittest.main(argv=sys.argv[:1])

79
pythonPackages/ufpy/test/dafTests/testFfmp.py Normal file
View file

@ -0,0 +1,79 @@
##
# This software was developed and / or modified by Raytheon Company,
# pursuant to Contract DG133W-05-CQ-1067 with the US Government.
#
# U.S. EXPORT CONTROLLED TECHNICAL DATA
# This software product contains export-restricted data whose
# export/transfer/disclosure is restricted by U.S. law. Dissemination
# to non-U.S. persons whether in the United States or abroad requires
# an export license or other authorization.
#
# Contractor Name: Raytheon Company
# Contractor Address: 6825 Pine Street, Suite 340
# Mail Stop B8
# Omaha, NE 68106
# 402.291.0100
#
# See the AWIPS II Master Rights File ("Master Rights File.pdf") for
# further licensing information.
##
from __future__ import print_function
from ufpy.dataaccess import DataAccessLayer as DAL
import baseDafTestCase
import dafTestsArgsUtil
import sys
import unittest
class FfmpTestCase(baseDafTestCase.DafTestCase):
"""
Tests that ffmp data can be retrieved through the DAF, simply ensuring
that no unexpected exceptions are thrown while retrieving it and that the
returned data is not None.
"""
datatype = "ffmp"
@staticmethod
def addIdentifiers(req):
req.addIdentifier("wfo", "OAX")
req.addIdentifier("siteKey", "hpe")
req.addIdentifier("dataKey", "hpe")
req.addIdentifier("huc", "ALL")
@classmethod
def setUpClass(cls):
print("STARTING FFMP TESTS\n\n")
def testParameters(self):
req = DAL.newDataRequest(self.datatype)
self.runParametersTest(req)
def testLocations(self):
req = DAL.newDataRequest(self.datatype)
self.addIdentifiers(req)
self.runLocationsTest(req)
def testTimes(self):
req = DAL.newDataRequest(self.datatype)
self.addIdentifiers(req)
self.runTimesTest(req)
def testGeometryData(self):
req = DAL.newDataRequest(self.datatype)
self.addIdentifiers(req)
req.setParameters("DHR")
self.runGeometryDataTest(req)
@classmethod
def tearDownClass(cls):
print("FFMP TESTS COMPLETE\n\n\n")
if __name__ == '__main__':
dafTestsArgsUtil.parseAndHandleArgs()
unittest.main(argv=sys.argv[:1])

15
pythonPackages/ufpy/test/dafTests/testGfe.py
View file

@ -21,11 +21,12 @@
from __future__ import print_function
from ufpy.dataaccess import DataAccessLayer as DAL
import dafTestsUtil
import baseDafTestCase
import dafTestsArgsUtil
import sys
import unittest
class TestGfe(unittest.TestCase):
class GfeTestCase(baseDafTestCase.DafTestCase):
"""
Tests that gfe data can be retrieved through the DAF, primarily ensuring
that no unexpected exceptions are thrown while retrieving it and that the
@ -42,20 +43,20 @@ class TestGfe(unittest.TestCase):
def testParameters(self):
req = DAL.newDataRequest(self.datatype)
dafTestsUtil.testParameters(req)
self.runParametersTest(req)
def testLocations(self):
req = DAL.newDataRequest(self.datatype)
req.addIdentifier("modelName", "Fcst")
dafTestsUtil.testLocations(req)
self.runLocationsTest(req)
def testTimes(self):
req = DAL.newDataRequest(self.datatype)
req.addIdentifier("modelName", "Fcst")
req.addIdentifier("siteId", "OAX")
dafTestsUtil.testTimes(req)
self.runTimesTest(req)
def testGridData(self):
req = DAL.newDataRequest(self.datatype)
@ -63,12 +64,12 @@ class TestGfe(unittest.TestCase):
req.addIdentifier("siteId", "OAX")
req.setParameters("T")
dafTestsUtil.testGridData(req)
self.runGridDataTest(req)
@classmethod
def tearDownClass(cls):
print("GFE TESTS COMPLETE\n\n\n")
if __name__ == '__main__':
dafTestsUtil.parseAndHandleArgs()
dafTestsArgsUtil.parseAndHandleArgs()
unittest.main(argv=sys.argv[:1])

23
pythonPackages/ufpy/test/dafTests/testGrid.py
View file

@ -21,11 +21,12 @@
from __future__ import print_function
from ufpy.dataaccess import DataAccessLayer as DAL
import dafTestsUtil
import baseDafTestCase
import dafTestsArgsUtil
import sys
import unittest
class TestGrid(unittest.TestCase):
class GridTestCase(baseDafTestCase.DafTestCase):
"""
Tests that grid data can be retrieved through the DAF, primarily ensuring
that no unexpected exceptions are thrown while retrieving it and that the
@ -45,26 +46,26 @@ class TestGrid(unittest.TestCase):
req = DAL.newDataRequest(self.datatype)
req.addIdentifier("info.datasetId", self.model)
dafTestsUtil.testParameters(req)
self.runParametersTest(req)
def testLocations(self):
req = DAL.newDataRequest(self.datatype)
req.addIdentifier("info.datasetId", self.model)
dafTestsUtil.testLocations(req)
self.runLocationsTest(req)
def testLevels(self):
req = DAL.newDataRequest(self.datatype)
req.addIdentifier("info.datasetId", self.model)
dafTestsUtil.testLevels(req)
self.runLevelsTest(req)
def testTimes(self):
req = DAL.newDataRequest(self.datatype)
req.addIdentifier("info.datasetId", self.model)
req.setLevels("2FHAG")
dafTestsUtil.testTimes(req)
self.runTimesTest(req)
def testGridData(self):
req = DAL.newDataRequest(self.datatype)
@ -72,21 +73,21 @@ class TestGrid(unittest.TestCase):
req.setLevels("2FHAG")
req.setParameters("T")
dafTestsUtil.testGridData(req)
self.runGridDataTest(req)
@classmethod
def tearDownClass(cls):
print("GRID TESTS COMPLETE\n\n\n")
def getArgs():
parser = dafTestsUtil.getParser()
parser.add_argument("-m", action="store", dest="model", default=TestGrid.model,
parser = dafTestsArgsUtil.getParser()
parser.add_argument("-m", action="store", dest="model", default=GridTestCase.model,
help="model to retrieve data for",
metavar="modelName")
return parser.parse_args()
if __name__ == '__main__':
args = getArgs()
dafTestsUtil.handleArgs(args)
TestGrid.model = args.model
dafTestsArgsUtil.handleArgs(args)
GridTestCase.model = args.model
unittest.main(argv=sys.argv[:1])

78
pythonPackages/ufpy/test/dafTests/testHydro.py Normal file
View file

@ -0,0 +1,78 @@
##
# This software was developed and / or modified by Raytheon Company,
# pursuant to Contract DG133W-05-CQ-1067 with the US Government.
#
# U.S. EXPORT CONTROLLED TECHNICAL DATA
# This software product contains export-restricted data whose
# export/transfer/disclosure is restricted by U.S. law. Dissemination
# to non-U.S. persons whether in the United States or abroad requires
# an export license or other authorization.
#
# Contractor Name: Raytheon Company
# Contractor Address: 6825 Pine Street, Suite 340
# Mail Stop B8
# Omaha, NE 68106
# 402.291.0100
#
# See the AWIPS II Master Rights File ("Master Rights File.pdf") for
# further licensing information.
##
from __future__ import print_function
from ufpy.dataaccess import DataAccessLayer as DAL
import baseDafTestCase
import dafTestsArgsUtil
import sys
import unittest
class HydroTestCase(baseDafTestCase.DafTestCase):
"""
Tests that hydro data can be retrieved through the DAF, simply ensuring
that no unexpected exceptions are thrown while retrieving it and that the
returned data is not None.
"""
datatype = "hydro"
table = "public.height"
@classmethod
def setUpClass(cls):
print("STARTING HYDRO TESTS\n\n")
def testParameters(self):
req = DAL.newDataRequest(self.datatype)
req.addIdentifier("table", self.table)
self.runParametersTest(req)
def testLocations(self):
req = DAL.newDataRequest(self.datatype)
req.addIdentifier("table", self.table)
self.runLocationsTest(req)
def testTimes(self):
req = DAL.newDataRequest(self.datatype)
req.addIdentifier("table", self.table)
req.setParameters("lid", "quality_code")
self.runTimesTest(req)
def testGeometryData(self):
req = DAL.newDataRequest(self.datatype)
req.addIdentifier("table", self.table)
req.setParameters("lid", "quality_code")
# Limit the times in the geometry data test to limit the amount of data
# returned.
self.runGeometryDataTest(req, limitTimes=True)
@classmethod
def tearDownClass(cls):
print("HYDRO TESTS COMPLETE\n\n\n")
if __name__ == '__main__':
dafTestsArgsUtil.parseAndHandleArgs()
unittest.main(argv=sys.argv[:1])

15
pythonPackages/ufpy/test/dafTests/testLdadMesonet.py
View file

@ -22,11 +22,12 @@ from __future__ import print_function
from shapely.geometry import Polygon
from ufpy.dataaccess import DataAccessLayer as DAL
import dafTestsUtil
import baseDafTestCase
import dafTestsArgsUtil
import sys
import unittest
class TestLdadMesonet(unittest.TestCase):
class LdadMesonetTestCase(baseDafTestCase.DafTestCase):
"""
Tests that ldadmesonet data can be retrieved through the DAF, simply
ensuring that no unexpected exceptions are thrown while retrieving it and
@ -54,31 +55,31 @@ class TestLdadMesonet(unittest.TestCase):
def testParameters(self):
req = DAL.newDataRequest(self.datatype)
dafTestsUtil.testParameters(req)
self.runParametersTest(req)
def testLocations(self):
req = DAL.newDataRequest(self.datatype)
req.setEnvelope(self.getReqEnvelope())
dafTestsUtil.testLocations(req)
self.runLocationsTest(req)
def testTimes(self):
req = DAL.newDataRequest(self.datatype)
req.setEnvelope(self.getReqEnvelope())
dafTestsUtil.testTimes(req)
self.runTimesTest(req)
def testGeometryData(self):
req = DAL.newDataRequest(self.datatype)
req.setParameters("highLevelCloud", "pressure")
req.setEnvelope(self.getReqEnvelope())
dafTestsUtil.testGeometryData(req)
self.runGeometryDataTest(req)
@classmethod
def tearDownClass(cls):
print("LDADMESONET TESTS COMPLETE\n\n\n")
if __name__ == '__main__':
dafTestsUtil.parseAndHandleArgs()
dafTestsArgsUtil.parseAndHandleArgs()
unittest.main(argv=sys.argv[:1])

19
pythonPackages/ufpy/test/dafTests/testMaps.py
View file

@ -21,11 +21,12 @@
from __future__ import print_function
from ufpy.dataaccess import DataAccessLayer as DAL
import dafTestsUtil
import baseDafTestCase
import dafTestsArgsUtil
import sys
import unittest
class TestMaps(unittest.TestCase):
class MapsTestCase(baseDafTestCase.DafTestCase):
"""
Tests that maps data can be retrieved through the DAF, simply ensuring
that no unexpected exceptions are thrown while retrieving it and that the
@ -45,7 +46,7 @@ class TestMaps(unittest.TestCase):
req.addIdentifier("table", "mapdata.county")
req.addIdentifier("geomField", "the_geom")
dafTestsUtil.testParameters(req)
self.runParametersTest(req)
def testLocations(self):
req = DAL.newDataRequest(self.datatype)
@ -53,7 +54,7 @@ class TestMaps(unittest.TestCase):
req.addIdentifier("geomField", "the_geom")
req.addIdentifier("locationField", "cwa")
dafTestsUtil.testLocations(req)
self.runLocationsTest(req)
def testGeometryData(self):
req = DAL.newDataRequest(self.datatype)
@ -65,21 +66,21 @@ class TestMaps(unittest.TestCase):
req.addIdentifier("cwa", self.site)
req.setParameters("countyname", "state", "fips")
dafTestsUtil.testGeometryData(req)
self.runGeometryDataTest(req)
@classmethod
def tearDownClass(cls):
print("MAPS TESTS COMPLETE\n\n\n")
def getArgs():
parser = dafTestsUtil.getParser()
parser.add_argument("-s", action="store", dest="site", default=TestMaps.site,
parser = dafTestsArgsUtil.getParser()
parser.add_argument("-s", action="store", dest="site", default=MapsTestCase.site,
help="site to retrieve data for",
metavar="siteID")
return parser.parse_args()
if __name__ == '__main__':
args = getArgs()
dafTestsUtil.handleArgs(args)
TestMaps.site = args.site
dafTestsArgsUtil.handleArgs(args)
MapsTestCase.site = args.site
unittest.main(argv=sys.argv[:1])

94
pythonPackages/ufpy/test/dafTests/testModelSounding.py Normal file
View file

@ -0,0 +1,94 @@
##
# This software was developed and / or modified by Raytheon Company,
# pursuant to Contract DG133W-05-CQ-1067 with the US Government.
#
# U.S. EXPORT CONTROLLED TECHNICAL DATA
# This software product contains export-restricted data whose
# export/transfer/disclosure is restricted by U.S. law. Dissemination
# to non-U.S. persons whether in the United States or abroad requires
# an export license or other authorization.
#
# Contractor Name: Raytheon Company
# Contractor Address: 6825 Pine Street, Suite 340
# Mail Stop B8
# Omaha, NE 68106
# 402.291.0100
#
# See the AWIPS II Master Rights File ("Master Rights File.pdf") for
# further licensing information.
##
from __future__ import print_function
from ufpy.dataaccess import DataAccessLayer as DAL
import baseDafTestCase
import dafTestsArgsUtil
import sys
import unittest
class ModelSoundingTestCase(baseDafTestCase.DafTestCase):
"""
Tests that modelsounding data can be retrieved through the DAF, simply
ensuring that no unexpected exceptions are thrown while retrieving it and
that the returned data is not None.
"""
datatype = "modelsounding"
@classmethod
def setUpClass(cls):
print("STARTING MODELSOUNDING TESTS\n\n")
def testParameters(self):
req = DAL.newDataRequest(self.datatype)
self.runParametersTest(req)
def testLocations(self):
req = DAL.newDataRequest(self.datatype)
req.addIdentifier("reportType", "ETA")
self.runLocationsTest(req)
def testTimes(self):
req = DAL.newDataRequest(self.datatype)
req.addIdentifier("reportType", "ETA")
req.setLocationNames("KOMA")
self.runTimesTest(req)
def testGeometryData(self):
req = DAL.newDataRequest(self.datatype)
req.addIdentifier("reportType", "ETA")
req.setLocationNames("KOMA")
req.setParameters("temperature", "pressure", "specHum", "sfcPress", "temp2", "q2")
print("Testing getGeometryData()")
geomData = DAL.getGeometryData(req)
print("Number of geometry records: " + str(len(geomData)))
print("Sample geometry data:")
for record in geomData[:self.sampleDataLimit]:
print("level=" + record.getLevel(), end="")
# One dimensional parameters are reported on the 0.0UNKNOWN level.
# 2D parameters are reported on MB levels from pressure.
if record.getLevel() == "0.0UNKNOWN":
print(" sfcPress=" + record.getString("sfcPress") + record.getUnit("sfcPress"), end="")
print(" temp2=" + record.getString("temp2") + record.getUnit("temp2"), end="")
print(" q2=" + record.getString("q2") + record.getUnit("q2"), end="")
else:
print(" pressure=" + record.getString("pressure") + record.getUnit("pressure"), end="")
print(" temperature=" + record.getString("temperature") + record.getUnit("temperature"), end="")
print(" specHum=" + record.getString("specHum") + record.getUnit("specHum"), end="")
print(" geometry=" + str(record.getGeometry()))
print("getGeometryData() complete\n\n")
@classmethod
def tearDownClass(cls):
print("MODELSOUNDING TESTS COMPLETE\n\n\n")
if __name__ == '__main__':
dafTestsArgsUtil.parseAndHandleArgs()
unittest.main(argv=sys.argv[:1])

15
pythonPackages/ufpy/test/dafTests/testObs.py
View file

@ -21,11 +21,12 @@
from __future__ import print_function
from ufpy.dataaccess import DataAccessLayer as DAL
import dafTestsUtil
import baseDafTestCase
import dafTestsArgsUtil
import sys
import unittest
class TestObs(unittest.TestCase):
class ObsTestCase(baseDafTestCase.DafTestCase):
"""
Tests that obs data can be retrieved through the DAF, simply ensuring
that no unexpected exceptions are thrown while retrieving it and that the
@ -41,30 +42,30 @@ class TestObs(unittest.TestCase):
def testParameters(self):
req = DAL.newDataRequest(self.datatype)
dafTestsUtil.testParameters(req)
self.runParametersTest(req)
def testLocations(self):
req = DAL.newDataRequest(self.datatype)
dafTestsUtil.testLocations(req)
self.runLocationsTest(req)
def testTimes(self):
req = DAL.newDataRequest(self.datatype)
req.setLocationNames("KOMA")
dafTestsUtil.testTimes(req)
self.runTimesTest(req)
def testGeometryData(self):
req = DAL.newDataRequest(self.datatype)
req.setLocationNames("KOMA")
req.setParameters("temperature", "seaLevelPress", "dewpoint")
dafTestsUtil.testGeometryData(req)
self.runGeometryDataTest(req)
@classmethod
def tearDownClass(cls):
print("OBS TESTS COMPLETE\n\n\n")
if __name__ == '__main__':
dafTestsUtil.parseAndHandleArgs()
dafTestsArgsUtil.parseAndHandleArgs()
unittest.main(argv=sys.argv[:1])

23
pythonPackages/ufpy/test/dafTests/testPirep.py
View file

@ -21,11 +21,12 @@
from __future__ import print_function
from ufpy.dataaccess import DataAccessLayer as DAL
import dafTestsUtil
import baseDafTestCase
import dafTestsArgsUtil
import sys
import unittest
class TestPirep(unittest.TestCase):
class PirepTestCase(baseDafTestCase.DafTestCase):
"""
Tests that pirep data can be retrieved through the DAF, simply ensuring
that no unexpected exceptions are thrown while retrieving it and that the
@ -43,18 +44,18 @@ class TestPirep(unittest.TestCase):
def testParameters(self):
req = DAL.newDataRequest(self.datatype)
dafTestsUtil.testParameters(req)
self.runParametersTest(req)
def testLocations(self):
req = DAL.newDataRequest(self.datatype)
dafTestsUtil.testLocations(req)
self.runLocationsTest(req)
def testTimes(self):
req = DAL.newDataRequest(self.datatype)
req.setLocationNames(self.station)
dafTestsUtil.testTimes(req)
self.runTimesTest(req)
def testGeometryData(self):
req = DAL.newDataRequest(self.datatype)
@ -64,10 +65,10 @@ class TestPirep(unittest.TestCase):
print("Testing getGeometryData()")
geomData = DAL.getGeometryData(req)
assert geomData is not None
self.assertIsNotNone(geomData)
print("Number of geometry records: " + str(len(geomData)))
print("Sample geometry data:")
for record in geomData[:dafTestsUtil.sampleDataLimit]:
for record in geomData[:self.sampleDataLimit]:
print("level=", record.getLevel(), end="")
# One dimensional parameters are reported on the 0.0UNKNOWN level.
# 2D parameters are reported on MB levels from pressure.
@ -86,14 +87,14 @@ class TestPirep(unittest.TestCase):
print("PIREP TESTS COMPLETE\n\n\n")
def getArgs():
parser = dafTestsUtil.getParser()
parser.add_argument("-s", action="store", dest="station", default=TestPirep.station,
parser = dafTestsArgsUtil.getParser()
parser.add_argument("-s", action="store", dest="station", default=PirepTestCase.station,
help="stationId",
metavar="station")
return parser.parse_args()
if __name__ == '__main__':
args = getArgs()
dafTestsUtil.handleArgs(args)
TestPirep.station = args.station
dafTestsArgsUtil.handleArgs(args)
PirepTestCase.station = args.station
unittest.main(argv=sys.argv[:1])

70
pythonPackages/ufpy/test/dafTests/testPracticeWarning.py Normal file
View file

@ -0,0 +1,70 @@
##
# This software was developed and / or modified by Raytheon Company,
# pursuant to Contract DG133W-05-CQ-1067 with the US Government.
#
# U.S. EXPORT CONTROLLED TECHNICAL DATA
# This software product contains export-restricted data whose
# export/transfer/disclosure is restricted by U.S. law. Dissemination
# to non-U.S. persons whether in the United States or abroad requires
# an export license or other authorization.
#
# Contractor Name: Raytheon Company
# Contractor Address: 6825 Pine Street, Suite 340
# Mail Stop B8
# Omaha, NE 68106
# 402.291.0100
#
# See the AWIPS II Master Rights File ("Master Rights File.pdf") for
# further licensing information.
##
from __future__ import print_function
from ufpy.dataaccess import DataAccessLayer as DAL
import baseDafTestCase
import dafTestsArgsUtil
import sys
import unittest
class PracticeWarningTestCase(baseDafTestCase.DafTestCase):
"""
Tests that practicewarning data can be retrieved through the DAF, simply
ensuring that no unexpected exceptions are thrown while retrieving it and
that the returned data is not None.
"""
datatype = "practicewarning"
@classmethod
def setUpClass(cls):
print("STARTING PRACTICE WARNING TESTS\n\n")
def testParameters(self):
req = DAL.newDataRequest(self.datatype)
self.runParametersTest(req)
def testLocations(self):
req = DAL.newDataRequest(self.datatype)
self.runLocationsTest(req)
def testTimes(self):
req = DAL.newDataRequest(self.datatype)
req.setParameters("countyheader", "productclass")
self.runTimesTest(req)
def testGeometryData(self):
req = DAL.newDataRequest(self.datatype)
req.setParameters("countyheader", "productclass")
self.runGeometryDataTest(req)
@classmethod
def tearDownClass(cls):
print("PRACTICE WARNING TESTS COMPLETE\n\n\n")
if __name__ == '__main__':
dafTestsArgsUtil.parseAndHandleArgs()
unittest.main(argv=sys.argv[:1])

17
pythonPackages/ufpy/test/dafTests/testProfiler.py
View file

@ -21,11 +21,12 @@
from __future__ import print_function
from ufpy.dataaccess import DataAccessLayer as DAL
import dafTestsUtil
import baseDafTestCase
import dafTestsArgsUtil
import sys
import unittest
class TestProfiler(unittest.TestCase):
class ProfilerTestCase(baseDafTestCase.DafTestCase):
"""
Tests that profiler data can be retrieved through the DAF, simply ensuring
that no unexpected exceptions are thrown while retrieving it and that the
@ -41,17 +42,17 @@ class TestProfiler(unittest.TestCase):
def testParameters(self):
req = DAL.newDataRequest(self.datatype)
dafTestsUtil.testParameters(req)
self.runParametersTest(req)
def testLocations(self):
req = DAL.newDataRequest(self.datatype)
dafTestsUtil.testLocations(req)
self.runLocationsTest(req)
def testTimes(self):
req = DAL.newDataRequest(self.datatype)
dafTestsUtil.testTimes(req)
self.runTimesTest(req)
def testGeometryData(self):
req = DAL.newDataRequest(self.datatype)
@ -60,10 +61,10 @@ class TestProfiler(unittest.TestCase):
print("Testing getGeometryData()")
geomData = DAL.getGeometryData(req)
assert geomData is not None
self.assertIsNotNone(geomData)
print("Number of geometry records: " + str(len(geomData)))
print("Sample geometry data:")
for record in geomData[:dafTestsUtil.sampleDataLimit]:
for record in geomData[:self.sampleDataLimit]:
print("level:", record.getLevel(), end="")
# One dimensional parameters are reported on the 0.0UNKNOWN level.
# 2D parameters are reported on MB levels from pressure.
@ -82,5 +83,5 @@ class TestProfiler(unittest.TestCase):
print("PROFILER TESTS COMPLETE\n\n\n")
if __name__ == '__main__':
dafTestsUtil.parseAndHandleArgs()
dafTestsArgsUtil.parseAndHandleArgs()
unittest.main(argv=sys.argv[:1])

98
pythonPackages/ufpy/test/dafTests/testRadar.py Normal file
View file

@ -0,0 +1,98 @@
##
# This software was developed and / or modified by Raytheon Company,
# pursuant to Contract DG133W-05-CQ-1067 with the US Government.
#
# U.S. EXPORT CONTROLLED TECHNICAL DATA
# This software product contains export-restricted data whose
# export/transfer/disclosure is restricted by U.S. law. Dissemination
# to non-U.S. persons whether in the United States or abroad requires
# an export license or other authorization.
#
# Contractor Name: Raytheon Company
# Contractor Address: 6825 Pine Street, Suite 340
# Mail Stop B8
# Omaha, NE 68106
# 402.291.0100
#
# See the AWIPS II Master Rights File ("Master Rights File.pdf") for
# further licensing information.
##
from __future__ import print_function
from shapely.geometry import box
from ufpy.dataaccess import DataAccessLayer as DAL
import baseDafTestCase
import dafTestsArgsUtil
import sys
import unittest
class RadarTestCase(baseDafTestCase.DafTestCase):
"""
Tests that radar data can be retrieved through the DAF, simply ensuring
that no unexpected exceptions are thrown while retrieving it and that the
returned data is not None.
"""
datatype = "radar"
bbox = ["-96", "41", "-97", "42"]
"""
Request area (box around KOAX)
"""
envelope = None
@classmethod
def getReqEnvelope(cls):
if not cls.envelope:
x1, y1, x2, y2 = map(float, cls.bbox)
minX = min(x1, x2)
maxX = max(x1, x2)
minY = min(y1, y2)
maxY = max(y1, y2)
cls.envelope = box(minX, minY, maxX, maxY)
return cls.envelope
@classmethod
def setUpClass(cls):
print("STARTING RADAR TESTS\n\n")
def testParameters(self):
req = DAL.newDataRequest(self.datatype)
self.runParametersTest(req)
def testLocations(self):
req = DAL.newDataRequest(self.datatype)
self.runLocationsTest(req)
def testLevels(self):
req = DAL.newDataRequest(self.datatype)
self.runLevelsTest(req)
def testTimes(self):
req = DAL.newDataRequest(self.datatype)
req.setEnvelope(self.getReqEnvelope())
self.runTimesTest(req)
def testGridData(self):
req = DAL.newDataRequest(self.datatype)
req.setEnvelope(self.getReqEnvelope())
req.setLocationNames("koax")
req.setParameters("94")
# Limit the times in the grid data test to limit the amount of data
# returned, don't test shapes since they may differ.
self.runGridDataTest(req, limitTimes=True, testSameShape=False)
@classmethod
def tearDownClass(cls):
print("RADAR TESTS COMPLETE\n\n\n")
if __name__ == '__main__':
dafTestsArgsUtil.parseAndHandleArgs()
unittest.main(argv=sys.argv[:1])

104
pythonPackages/ufpy/test/dafTests/testRadarSpatial.py Normal file
View file

@ -0,0 +1,104 @@
##
# This software was developed and / or modified by Raytheon Company,
# pursuant to Contract DG133W-05-CQ-1067 with the US Government.
#
# U.S. EXPORT CONTROLLED TECHNICAL DATA
# This software product contains export-restricted data whose
# export/transfer/disclosure is restricted by U.S. law. Dissemination
# to non-U.S. persons whether in the United States or abroad requires
# an export license or other authorization.
#
# Contractor Name: Raytheon Company
# Contractor Address: 6825 Pine Street, Suite 340
# Mail Stop B8
# Omaha, NE 68106
# 402.291.0100
#
# See the AWIPS II Master Rights File ("Master Rights File.pdf") for
# further licensing information.
##
from __future__ import print_function
from shapely.geometry import box
from ufpy.dataaccess import DataAccessLayer as DAL
import baseDafTestCase
import dafTestsArgsUtil
import sys
import unittest
class RadarSpatialTestCase(baseDafTestCase.DafTestCase):
"""
Tests that radar_spatial data can be retrieved through the DAF, simply
ensuring that no unexpected exceptions are thrown while retrieving it and
that the returned data is not None.
"""
datatype = "radar_spatial"
bbox = ["-96", "41", "-97", "42"]
"""
Default request area (box around KOAX)
"""
locations = ["TORD", "TMDW"]
envelope = None
@classmethod
def getReqEnvelope(cls):
if not cls.envelope:
x1, y1, x2, y2 = map(float, cls.bbox)
minX = min(x1, x2)
maxX = max(x1, x2)
minY = min(y1, y2)
maxY = max(y1, y2)
cls.envelope = box(minX, minY, maxX, maxY)
return cls.envelope
@classmethod
def setUpClass(cls):
print("STARTING RADAR_SPATIAL TESTS\n\n")
def testLocations(self):
req = DAL.newDataRequest(self.datatype)
req.setEnvelope(self.getReqEnvelope())
self.runLocationsTest(req)
def testParameters(self):
req = DAL.newDataRequest(self.datatype)
self.runParametersTest(req)
def testIdentifiers(self):
self.runIdentifiersTest(self.datatype)
def testGeometryData(self):
req = DAL.newDataRequest(self.datatype)
req.setLocationNames(*self.locations)
req.setParameters("wfo_id", "name", "elevmeter")
self.runGeometryDataTest(req)
@classmethod
def tearDownClass(cls):
print("RADAR_SPATIAL TESTS COMPLETE\n\n\n")
def getArgs():
parser = dafTestsArgsUtil.getParser()
parser.add_argument("-p", action="store", dest="bbox", nargs=4, default=RadarSpatialTestCase.bbox,
help="Request area for location names query",
metavar="point")
parser.add_argument("-r", dest="locations", nargs="+", default=RadarSpatialTestCase.locations,
help="IDs of radar stations to retrieve geometry data for.",
metavar="locations")
return parser.parse_args()
if __name__ == '__main__':
args = getArgs()
dafTestsArgsUtil.handleArgs(args)
RadarSpatialTestCase.bbox = args.bbox
RadarSpatialTestCase.locations = args.locations
unittest.main(argv=sys.argv[:1])

75
pythonPackages/ufpy/test/dafTests/testSatellite.py Normal file
View file

@ -0,0 +1,75 @@
#!/usr/bin/env python
##
# This software was developed and / or modified by Raytheon Company,
# pursuant to Contract DG133W-05-CQ-1067 with the US Government.
#
# U.S. EXPORT CONTROLLED TECHNICAL DATA
# This software product contains export-restricted data whose
# export/transfer/disclosure is restricted by U.S. law. Dissemination
# to non-U.S. persons whether in the United States or abroad requires
# an export license or other authorization.
#
# Contractor Name: Raytheon Company
# Contractor Address: 6825 Pine Street, Suite 340
# Mail Stop B8
# Omaha, NE 68106
# 402.291.0100
#
# See the AWIPS II Master Rights File ("Master Rights File.pdf") for
# further licensing information.
##
from __future__ import print_function
from ufpy.dataaccess import DataAccessLayer as DAL
import baseDafTestCase
import dafTestsArgsUtil
import sys
import unittest
class SatelliteTestCase(baseDafTestCase.DafTestCase):
"""
Tests that satellite data can be retrieved through the DAF, primarily
ensuring that no unexpected exceptions are thrown while retrieving it and
that the returned data is not None. The only data validation that is
performed is to check that all retrieved grid data have the same shape.
"""
datatype = "satellite"
@classmethod
def setUpClass(cls):
print("STARTING SATELLITE TESTS\n\n")
def testParameters(self):
req = DAL.newDataRequest(self.datatype)
self.runParametersTest(req)
def testLocations(self):
req = DAL.newDataRequest(self.datatype)
self.runLocationsTest(req)
def testTimes(self):
req = DAL.newDataRequest(self.datatype)
req.setLocationNames("West CONUS")
self.runTimesTest(req)
def testGridData(self):
req = DAL.newDataRequest(self.datatype)
req.setParameters("Imager 11 micron IR")
req.setLocationNames("West CONUS")
# Limit the times in the grid data test to limit the amount of data
# returned.
self.runGridDataTest(req, limitTimes=True)
@classmethod
def tearDownClass(cls):
print("SATELLITE TESTS COMPLETE\n\n\n")
if __name__ == '__main__':
dafTestsArgsUtil.parseAndHandleArgs()
unittest.main(argv=sys.argv[:1])

15
pythonPackages/ufpy/test/dafTests/testSfcObs.py
View file

@ -21,11 +21,12 @@
from __future__ import print_function
from ufpy.dataaccess import DataAccessLayer as DAL
import dafTestsUtil
import baseDafTestCase
import dafTestsArgsUtil
import sys
import unittest
class TestSfcObs(unittest.TestCase):
class SfcObsTestCase(baseDafTestCase.DafTestCase):
"""
Tests that sfcobs data can be retrieved through the DAF, simply ensuring
that no unexpected exceptions are thrown while retrieving it and that the
@ -41,30 +42,30 @@ class TestSfcObs(unittest.TestCase):
def testParameters(self):
req = DAL.newDataRequest(self.datatype)
dafTestsUtil.testParameters(req)
self.runParametersTest(req)
def testLocations(self):
req = DAL.newDataRequest(self.datatype)
dafTestsUtil.testLocations(req)
self.runLocationsTest(req)
def testTimes(self):
req = DAL.newDataRequest(self.datatype)
req.setLocationNames("14547")
dafTestsUtil.testTimes(req)
self.runTimesTest(req)
def testGeometryData(self):
req = DAL.newDataRequest(self.datatype)
req.setLocationNames("14547")
req.setParameters("temperature", "seaLevelPress", "dewpoint")
dafTestsUtil.testGeometryData(req)
self.runGeometryDataTest(req)
@classmethod
def tearDownClass(cls):
print("SFCOBS TESTS COMPLETE\n\n\n")
if __name__ == '__main__':
dafTestsUtil.parseAndHandleArgs()
dafTestsArgsUtil.parseAndHandleArgs()
unittest.main(argv=sys.argv[:1])

11
pythonPackages/ufpy/test/dafTests/testTopo.py
View file

@ -22,12 +22,13 @@ from __future__ import print_function
from ufpy.dataaccess import DataAccessLayer as DAL
from ufpy.ThriftClient import ThriftRequestException
import dafTestsUtil
import baseDafTestCase
import dafTestsArgsUtil
import shapely.geometry
import sys
import unittest
class TestTopo(unittest.TestCase):
class TopoTestCase(baseDafTestCase.DafTestCase):
"""
Tests that topo data can be retrieved through the DAF, simply ensuring
that no unexpected exceptions are thrown while retrieving it and that the
@ -51,7 +52,7 @@ class TestTopo(unittest.TestCase):
poly = shapely.geometry.LinearRing(((-70, 40), (-71, 40), (-71, 42), (-70, 42)))
req.setEnvelope(poly)
gridData = DAL.getGridData(req)
assert gridData is not None
self.assertIsNotNone(gridData)
print("Number of grid records: " + str(len(gridData)))
print("Sample grid data shape:\n" + str(gridData[0].getRawData().shape) + "\n")
print("Sample grid data:\n" + str(gridData[0].getRawData()) + "\n")
@ -60,7 +61,7 @@ class TestTopo(unittest.TestCase):
print("\n" + topoFile)
req.addIdentifier("topoFile", topoFile)
gridData = DAL.getGridData(req)
assert gridData is not None
self.assertIsNotNone(gridData)
print("Number of grid records: " + str(len(gridData)))
print("Sample grid data shape:\n" + str(gridData[0].getRawData().shape) + "\n")
print("Sample grid data:\n" + str(gridData[0].getRawData()) + "\n")
@ -87,5 +88,5 @@ class TestTopo(unittest.TestCase):
print("TOPO TESTS COMPLETE\n\n\n")
if __name__ == '__main__':
dafTestsUtil.parseAndHandleArgs()
dafTestsArgsUtil.parseAndHandleArgs()
unittest.main(argv=sys.argv[:1])

15
pythonPackages/ufpy/test/dafTests/testWarning.py
View file

@ -21,11 +21,12 @@
from __future__ import print_function
from ufpy.dataaccess import DataAccessLayer as DAL
import dafTestsUtil
import baseDafTestCase
import dafTestsArgsUtil
import sys
import unittest
class TestWarning(unittest.TestCase):
class WarningTestCase(baseDafTestCase.DafTestCase):
"""
Tests that warning data can be retrieved through the DAF, simply ensuring
that no unexpected exceptions are thrown while retrieving it and that the
@ -41,29 +42,29 @@ class TestWarning(unittest.TestCase):
def testParameters(self):
req = DAL.newDataRequest(self.datatype)
dafTestsUtil.testParameters(req)
self.runParametersTest(req)
def testLocations(self):
req = DAL.newDataRequest(self.datatype)
dafTestsUtil.testLocations(req)
self.runLocationsTest(req)
def testTimes(self):
req = DAL.newDataRequest(self.datatype)
req.setParameters("etn", "wmoid")
dafTestsUtil.testTimes(req)
self.runTimesTest(req)
def testGeometryData(self):
req = DAL.newDataRequest(self.datatype)
req.setParameters("etn", "wmoid")
dafTestsUtil.testGeometryData(req)
self.runGeometryDataTest(req)
@classmethod
def tearDownClass(cls):
print("WARNING TESTS COMPLETE\n\n\n")
if __name__ == '__main__':
dafTestsUtil.parseAndHandleArgs()
dafTestsArgsUtil.parseAndHandleArgs()
unittest.main(argv=sys.argv[:1])