xan/python-awips
1
0
Fork 0
mirror of https://github.com/Unidata/python-awips.git synced 2025-02-23 14:57:56 -05:00
Code Issues Projects Releases Packages Wiki Activity Actions

remove dafTests for unsupported types. cleanup attributes in PyData

Browse source
This commit is contained in:
Michael James 2018-10-15 14:10:08 -06:00
parent 0b9a06851f
commit c7ed7e6242
10 changed files with 6 additions and 480 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

5
awips/dataaccess/PyData.py
View file

@ -25,11 +25,10 @@ class PyData(IData):
self.__attributes = dataRecord.getAttributes()
def getAttribute(self, key):
value = self.__attributes[key]
return value
return self.__attributes[key]
def getAttributes(self):
return list(self.__attributes.keys())
return self.__attributes.keys()
def getDataTime(self):
return self.__time

23
awips/test/dafTests/testBufrMosAvn.py
View file

@ -1,23 +0,0 @@
from awips.test.dafTests import baseBufrMosTestCase
import unittest
#
# Test DAF support for bufrmosAVN data
#
# SOFTWARE HISTORY
#
# Date Ticket# Engineer Description
# ------------ ---------- ----------- --------------------------
# 01/19/16 4795 mapeters Initial Creation.
# 04/11/16 5548 tgurney Cleanup
# 04/18/16 5548 tgurney More cleanup
#
#
class BufrMosAvnTestCase(baseBufrMosTestCase.BufrMosTestCase):
"""Test DAF support for bufrmosAVN data"""
datatype = "bufrmosAVN"
# All tests inherited from superclass

23
awips/test/dafTests/testBufrMosEta.py
View file

@ -1,23 +0,0 @@
from awips.test.dafTests import baseBufrMosTestCase
import unittest
#
# Test DAF support for bufrmosETA data
#
# SOFTWARE HISTORY
#
# Date Ticket# Engineer Description
# ------------ ---------- ----------- --------------------------
# 01/19/16 4795 mapeters Initial Creation.
# 04/11/16 5548 tgurney Cleanup
# 04/18/16 5548 tgurney More cleanup
#
#
class BufrMosEtaTestCase(baseBufrMosTestCase.BufrMosTestCase):
"""Test DAF support for bufrmosETA data"""
datatype = "bufrmosETA"
# All tests inherited from superclass

29
awips/test/dafTests/testBufrMosHpc.py
View file

@ -1,29 +0,0 @@
from awips.dataaccess import DataAccessLayer as DAL
from awips.test.dafTests import baseBufrMosTestCase
from awips.test.dafTests import params
import unittest
#
# Test DAF support for bufrmosHPC data
#
# SOFTWARE HISTORY
#
# Date Ticket# Engineer Description
# ------------ ---------- ----------- --------------------------
# 01/19/16 4795 mapeters Initial Creation.
# 04/11/16 5548 tgurney Cleanup
# 04/18/16 5548 tgurney More cleanup
# 12/07/16 5981 tgurney Parameterize
# 12/20/16 5981 tgurney Inherit all tests
#
#
class BufrMosHpcTestCase(baseBufrMosTestCase.BufrMosTestCase):
"""Test DAF support for bufrmosHPC data"""
datatype = "bufrmosHPC"
data_params = "forecastHr", "maxTemp24Hour"
# All tests inherited from superclass

23
awips/test/dafTests/testBufrMosLamp.py
View file

@ -1,23 +0,0 @@
from awips.test.dafTests import baseBufrMosTestCase
import unittest
#
# Test DAF support for bufrmosLAMP data
#
# SOFTWARE HISTORY
#
# Date Ticket# Engineer Description
# ------------ ---------- ----------- --------------------------
# 01/19/16 4795 mapeters Initial Creation.
# 04/11/16 5548 tgurney Cleanup
# 04/18/16 5548 tgurney More cleanup
#
#
class BufrMosLampTestCase(baseBufrMosTestCase.BufrMosTestCase):
"""Test DAF support for bufrmosLAMP data"""
datatype = "bufrmosLAMP"
# All tests inherited from superclass

29
awips/test/dafTests/testBufrMosMrf.py
View file

@ -1,29 +0,0 @@
from awips.dataaccess import DataAccessLayer as DAL
from awips.test.dafTests import baseBufrMosTestCase
from awips.test.dafTests import params
import unittest
#
# Test DAF support for bufrmosMRF data
#
# SOFTWARE HISTORY
#
# Date Ticket# Engineer Description
# ------------ ---------- ----------- --------------------------
# 01/19/16 4795 mapeters Initial Creation.
# 04/11/16 5548 tgurney Cleanup
# 04/18/16 5548 tgurney More cleanup
# 12/07/16 5981 tgurney Parameterize
# 12/20/16 5981 tgurney Inherit all tests
#
#
class BufrMosMrfTestCase(baseBufrMosTestCase.BufrMosTestCase):
"""Test DAF support for bufrmosMRF data"""
datatype = "bufrmosMRF"
data_params = "forecastHr", "maxTempDay"
# All tests inherited from superclass

6
awips/test/dafTests/testCombinedTimeQuery.py
View file

@ -20,6 +20,8 @@ import os
class CombinedTimeQueryTestCase(unittest.TestCase):
modelName = "RAP13"
@classmethod
def setUp(cls):
host = os.environ.get('DAF_TEST_HOST')
@ -29,7 +31,7 @@ class CombinedTimeQueryTestCase(unittest.TestCase):
def testSuccessfulQuery(self):
req = DAL.newDataRequest('grid')
req.setLocationNames('RUC130')
req.setLocationNames(self.modelName)
req.setParameters('T','GH')
req.setLevels('300MB', '500MB','700MB')
times = CTQ.getAvailableTimes(req)
@ -40,7 +42,7 @@ class CombinedTimeQueryTestCase(unittest.TestCase):
Test that when a parameter is only available on one of the levels that no times are returned.
"""
req = DAL.newDataRequest('grid')
req.setLocationNames('RUC130')
req.setLocationNames(self.modelName)
req.setParameters('T','GH', 'LgSP1hr')
req.setLevels('300MB', '500MB','700MB','0.0SFC')
times = CTQ.getAvailableTimes(req)

248
awips/test/dafTests/testHydro.py
View file

@ -1,248 +0,0 @@
from __future__ import print_function
import datetime
from awips.dataaccess import DataAccessLayer as DAL
from awips.ThriftClient import ThriftRequestException
from dynamicserialize.dstypes.com.raytheon.uf.common.dataquery.requests import RequestConstraint
from dynamicserialize.dstypes.com.raytheon.uf.common.time import TimeRange
from awips.test.dafTests import baseDafTestCase
import unittest
#
# Test DAF support for hydro data
#
# SOFTWARE HISTORY
#
# Date Ticket# Engineer Description
# ------------ ---------- ----------- --------------------------
# 01/19/16 4795 mapeters Initial Creation.
# 04/11/16 5548 tgurney Cleanup
# 04/18/16 5548 tgurney More cleanup
# 04/21/16 5596 tgurney Add tests to verify #5596
# 04/26/16 5587 tgurney Add identifier values tests
# 06/09/16 5574 tgurney Add advanced query tests
# 06/13/16 5574 tgurney Fix checks for None
# 06/21/16 5548 tgurney Skip tests that cause errors
# 06/30/16 5725 tgurney Add test for NOT IN
# 10/06/16 5926 dgilling Add additional location tests.
#
#
class HydroTestCase(baseDafTestCase.DafTestCase):
"""Test DAF support for hydro data"""
datatype = 'hydro'
def testGetAvailableParameters(self):
req = DAL.newDataRequest(self.datatype)
req.addIdentifier('table', 'height')
self.runParametersTest(req)
def testGetAvailableParametersFullyQualifiedTable(self):
req = DAL.newDataRequest(self.datatype)
req.addIdentifier('table', 'public.height')
self.runParametersTest(req)
def testGetAvailableParamsNoTableThrowsInvalidIdentifiersException(self):
req = DAL.newDataRequest(self.datatype)
with self.assertRaises(ThriftRequestException) as cm:
self.runParametersTest(req)
self.assertIn('InvalidIdentifiersException', str(cm.exception))
def testGetAvailableLocations(self):
req = DAL.newDataRequest(self.datatype)
req.addIdentifier('table', 'height')
self.runLocationsTest(req)
def testGetAvailableLocationsWithConstraint(self):
req = DAL.newDataRequest(self.datatype)
req.addIdentifier('table', 'height')
req.addIdentifier('value', RequestConstraint.new('>', 5.0))
self.runLocationsTest(req)
def testGetAvailableLocationsWithInvalidTable(self):
req = DAL.newDataRequest(self.datatype)
req.addIdentifier('table', 'city')
with self.assertRaises(ThriftRequestException) as cm:
DAL.getAvailableLocationNames(req)
self.assertIn('IncompatibleRequestException', str(cm.exception))
def testGetAvailableTimes(self):
req = DAL.newDataRequest(self.datatype)
req.addIdentifier('table', 'height')
req.setParameters('lid', 'quality_code')
self.runTimesTest(req)
def testGetGeometryDataWithoutLocationSpecified(self):
req = DAL.newDataRequest(self.datatype)
req.addIdentifier('table', 'height')
req.setParameters('lid', 'quality_code')
self.runGeometryDataTest(req)
def testGetGeometryDataWithLocationSpecified(self):
req = DAL.newDataRequest(self.datatype)
req.addIdentifier('table', 'fcstheight')
locs = DAL.getAvailableLocationNames(req)
if locs:
req.setLocationNames(locs[0])
req.setParameters('probability', 'value')
data = self.runGeometryDataTest(req)
self.assertNotEqual(len(data), 0)
def testGetTableIdentifierValues(self):
self.runGetIdValuesTest(['table'])
def testGetColumnIdValuesWithTable(self):
req = DAL.newDataRequest(self.datatype)
req.addIdentifier('table', 'height')
idValues = DAL.getIdentifierValues(req, 'lid')
self.assertTrue(hasattr(idValues, '__iter__'))
@unittest.skip('avoid EDEX error')
def testGetColumnIdValuesWithNonexistentTableThrowsException(self):
req = DAL.newDataRequest(self.datatype)
req.addIdentifier('table', 'nonexistentjunk')
with self.assertRaises(ThriftRequestException):
idValues = DAL.getIdentifierValues(req, 'lid')
def testGetColumnIdValuesWithoutTableThrowsException(self):
req = DAL.newDataRequest(self.datatype)
with self.assertRaises(ThriftRequestException):
idValues = DAL.getIdentifierValues(req, 'lid')
@unittest.skip('avoid EDEX error')
def testGetNonexistentColumnIdValuesThrowsException(self):
req = DAL.newDataRequest(self.datatype)
req.addIdentifier('table', 'height')
with self.assertRaises(ThriftRequestException):
idValues = DAL.getIdentifierValues(req, 'nonexistentjunk')
def testGetInvalidIdentifierValuesThrowsException(self):
self.runInvalidIdValuesTest()
def testGetNonexistentIdentifierValuesThrowsException(self):
self.runNonexistentIdValuesTest()
def _runConstraintTest(self, key, operator, value):
req = DAL.newDataRequest(self.datatype)
constraint = RequestConstraint.new(operator, value)
req.addIdentifier(key, constraint)
req.addIdentifier('table', 'height')
req.addIdentifier('ts', 'RG')
req.setParameters('value', 'lid', 'quality_code')
return self.runGeometryDataTest(req)
def testGetDataWithEqualsString(self):
geometryData = self._runConstraintTest('value', '=', '3')
for record in geometryData:
self.assertEqual(record.getNumber('value'), 3)
def testGetDataWithEqualsUnicode(self):
geometryData = self._runConstraintTest('value', '=', u'3')
for record in geometryData:
self.assertEqual(record.getNumber('value'), 3)
def testGetDataWithEqualsInt(self):
geometryData = self._runConstraintTest('value', '=', 3)
for record in geometryData:
self.assertEqual(record.getNumber('value'), 3)
def testGetDataWithEqualsLong(self):
geometryData = self._runConstraintTest('value', '=', 3)
for record in geometryData:
self.assertEqual(record.getNumber('value'), 3)
def testGetDataWithEqualsFloat(self):
geometryData = self._runConstraintTest('value', '=', 3.0)
for record in geometryData:
self.assertEqual(round(record.getNumber('value'), 1), 3.0)
def testGetDataWithEqualsNone(self):
geometryData = self._runConstraintTest('value', '=', None)
self.assertEqual(len(geometryData), 0)
def testGetDataWithNotEquals(self):
geometryData = self._runConstraintTest('value', '!=', 3)
for record in geometryData:
self.assertNotEqual(record.getNumber('value'), '3')
def testGetDataWithNotEqualsNone(self):
geometryData = self._runConstraintTest('value', '!=', None)
self.assertNotEqual(len(geometryData), 0)
for record in geometryData:
self.assertNotEqual(record.getType('value'), 'NULL')
def testGetDataWithGreaterThan(self):
geometryData = self._runConstraintTest('value', '>', 3)
for record in geometryData:
self.assertGreater(record.getNumber('value'), 3)
def testGetDataWithLessThan(self):
geometryData = self._runConstraintTest('value', '<', 3)
for record in geometryData:
self.assertLess(record.getNumber('value'), 3)
def testGetDataWithGreaterThanEquals(self):
geometryData = self._runConstraintTest('value', '>=', 3)
for record in geometryData:
self.assertGreaterEqual(record.getNumber('value'), 3)
def testGetDataWithLessThanEquals(self):
geometryData = self._runConstraintTest('value', '<=', 3)
for record in geometryData:
self.assertLessEqual(record.getNumber('value'), 3)
def testGetDataWithInTuple(self):
collection = (3, 4)
geometryData = self._runConstraintTest('value', 'in', collection)
for record in geometryData:
self.assertIn(record.getNumber('value'), collection)
def testGetDataWithInList(self):
collection = [3, 4]
geometryData = self._runConstraintTest('value', 'in', collection)
for record in geometryData:
self.assertIn(record.getNumber('value'), collection)
def testGetDataWithInGenerator(self):
collection = (3, 4)
generator = (item for item in collection)
geometryData = self._runConstraintTest('value', 'in', generator)
for record in geometryData:
self.assertIn(record.getNumber('value'), collection)
def testGetDataWithNotInList(self):
collection = [3, 4]
geometryData = self._runConstraintTest('value', 'not in', collection)
for record in geometryData:
self.assertNotIn(record.getNumber('value'), collection)
def testGetDataWithTimeRange(self):
req = DAL.newDataRequest(self.datatype)
req.addIdentifier('table', 'height')
req.addIdentifier('ts', 'RG')
req.setParameters('value', 'lid', 'quality_code')
times = DAL.getAvailableTimes(req)
limitTimes = times[-self.numTimesToLimit:]
startTime = datetime.datetime.utcfromtimestamp(limitTimes[0].getRefTime().getTime()/1000)
endTime = datetime.datetime.utcnow()
tr = TimeRange(startTime, endTime)
self.runGeometryDataTestWithTimeRange(req, tr)
def testGetDataWithInvalidConstraintTypeThrowsException(self):
with self.assertRaises(ValueError):
self._runConstraintTest('value', 'junk', 3)
def testGetDataWithInvalidConstraintValueThrowsException(self):
with self.assertRaises(TypeError):
self._runConstraintTest('value', '=', {})
def testGetDataWithEmptyInConstraintThrowsException(self):
with self.assertRaises(ValueError):
self._runConstraintTest('value', 'in', [])
def testGetDataWithNestedInConstraintThrowsException(self):
collection = ('3', '4', ())
with self.assertRaises(TypeError):
self._runConstraintTest('value', 'in', collection)

71
awips/test/dafTests/testPirep.py
View file

@ -1,71 +0,0 @@
from __future__ import print_function
from awips.dataaccess import DataAccessLayer as DAL
from awips.test.dafTests import baseDafTestCase
from awips.test.dafTests import params
import unittest
#
# Test DAF support for pirep data
#
# SOFTWARE HISTORY
#
# Date Ticket# Engineer Description
# ------------ ---------- ----------- --------------------------
# 01/19/16 4795 mapeters Initial Creation.
# 04/11/16 5548 tgurney Cleanup
# 04/18/16 5548 tgurney More cleanup
# 12/07/16 5981 tgurney Parameterize
# 12/20/16 5981 tgurney Add envelope test
#
#
class PirepTestCase(baseDafTestCase.DafTestCase):
"""Test DAF support for pirep data"""
datatype = "pirep"
def testGetAvailableParameters(self):
req = DAL.newDataRequest(self.datatype)
self.runParametersTest(req)
def testGetAvailableLocations(self):
req = DAL.newDataRequest(self.datatype)
self.runLocationsTest(req)
def testGetAvailableTimes(self):
req = DAL.newDataRequest(self.datatype)
req.setLocationNames(params.AIRPORT)
self.runTimesTest(req)
def testGetGeometryData(self):
req = DAL.newDataRequest(self.datatype)
req.setLocationNames(params.AIRPORT)
req.setParameters("temperature", "windSpeed", "hazardType", "turbType")
print("Testing getGeometryData()")
geomData = DAL.getGeometryData(req)
self.assertIsNotNone(geomData)
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(" temperature=" + record.getString("temperature") + record.getUnit("temperature"), end="")
print(" windSpeed=" + record.getString("windSpeed") + record.getUnit("windSpeed"), end="")
else:
print(" hazardType=" + record.getString("hazardType"), end="")
print(" turbType=" + record.getString("turbType"), end="")
print(" geometry=", record.getGeometry())
print("getGeometryData() complete\n")
def testGetGeometryDataWithEnvelope(self):
req = DAL.newDataRequest(self.datatype)
req.setParameters("temperature", "windSpeed", "hazardType", "turbType")
req.setEnvelope(params.ENVELOPE)
print("Testing getGeometryData()")
data = DAL.getGeometryData(req)
for item in data:
self.assertTrue(params.ENVELOPE.contains(item.getGeometry()))

29
awips/test/dafTests/testPracticeWarning.py
View file

@ -1,29 +0,0 @@
from __future__ import print_function
from awips.dataaccess import DataAccessLayer as DAL
from awips.test.dafTests import baseDafTestCase
from awips.test.dafTests import testWarning
import unittest
#
# Test DAF support for practicewarning data
#
# SOFTWARE HISTORY
#
# Date Ticket# Engineer Description
# ------------ ---------- ----------- --------------------------
# 01/19/16 4795 mapeters Initial Creation.
# 04/11/16 5548 tgurney Cleanup
# 04/18/16 5548 tgurney More cleanup
# 06/10/16 5548 tgurney Inherit all tests from
# warning
#
class PracticeWarningTestCase(testWarning.WarningTestCase):
"""Test DAF support for practicewarning data"""
datatype = "practicewarning"
# All tests taken from testWarning