mirror of
https://github.com/Unidata/python-awips.git
synced 2025-02-23 22:57:56 -05:00
211 lines
No EOL
8.7 KiB
Python
211 lines
No EOL
8.7 KiB
Python
##
|
|
##
|
|
|
|
from __future__ import print_function
|
|
from dynamicserialize.dstypes.com.raytheon.uf.common.dataquery.requests import RequestConstraint
|
|
from awips.dataaccess import DataAccessLayer as DAL
|
|
|
|
import baseDafTestCase
|
|
import params
|
|
import unittest
|
|
|
|
#
|
|
# Test DAF support for ffmp 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/18/16 5587 tgurney Add test for sane handling of
|
|
# zero records returned
|
|
# 06/20/16 5587 tgurney Add identifier values tests
|
|
# 07/01/16 5728 mapeters Add advanced query tests,
|
|
# include huc and accumHrs in
|
|
# id values tests, test that
|
|
# accumHrs id is never required
|
|
# 08/03/16 5728 mapeters Fixed minor bugs, replaced
|
|
# PRTM parameter since it isn't
|
|
# configured for ec-oma
|
|
# 11/08/16 5985 tgurney Do not check data times
|
|
# 12/07/16 5981 tgurney Parameterize
|
|
# 12/20/16 5981 tgurney Do not check data times
|
|
#
|
|
#
|
|
|
|
|
|
class FfmpTestCase(baseDafTestCase.DafTestCase):
|
|
"""Test DAF support for ffmp data"""
|
|
|
|
datatype = 'ffmp'
|
|
location = params.RADAR.lower()
|
|
|
|
@staticmethod
|
|
def addIdentifiers(req):
|
|
req.addIdentifier('wfo', params.SITE_ID)
|
|
req.addIdentifier('siteKey', 'hpe')
|
|
req.addIdentifier('dataKey', 'hpe')
|
|
req.addIdentifier('huc', 'ALL')
|
|
|
|
def testGetAvailableParameters(self):
|
|
req = DAL.newDataRequest(self.datatype)
|
|
self.runParametersTest(req)
|
|
|
|
def testGetAvailableLocations(self):
|
|
req = DAL.newDataRequest(self.datatype)
|
|
self.addIdentifiers(req)
|
|
self.runLocationsTest(req)
|
|
|
|
def testGetAvailableTimes(self):
|
|
req = DAL.newDataRequest(self.datatype)
|
|
self.addIdentifiers(req)
|
|
req.setParameters('DHRMOSAIC')
|
|
self.runTimesTest(req)
|
|
|
|
def testGetGeometryData(self):
|
|
req = DAL.newDataRequest(self.datatype)
|
|
self.addIdentifiers(req)
|
|
req.setParameters('DHRMOSAIC')
|
|
self.runGeometryDataTest(req, checkDataTimes=False)
|
|
|
|
def testGetGeometryDataEmptyResult(self):
|
|
req = DAL.newDataRequest(self.datatype)
|
|
self.addIdentifiers(req)
|
|
req.setParameters('blah blah blah') # force 0 records returned
|
|
result = self.runGeometryDataTest(req, checkDataTimes=False)
|
|
self.assertEqual(len(result), 0)
|
|
|
|
def testGetIdentifierValues(self):
|
|
req = DAL.newDataRequest(self.datatype)
|
|
optionalIds = set(DAL.getOptionalIdentifiers(req))
|
|
requiredIds = set(DAL.getRequiredIdentifiers(req))
|
|
ids = requiredIds | optionalIds
|
|
for id in ids:
|
|
req = DAL.newDataRequest(self.datatype)
|
|
if id == 'accumHrs':
|
|
req.setParameters('ARI6H2YR')
|
|
req.addIdentifier('wfo', params.SITE_ID)
|
|
req.addIdentifier('siteKey', self.location)
|
|
req.addIdentifier('huc', 'ALL')
|
|
idValues = DAL.getIdentifierValues(req, id)
|
|
self.assertTrue(hasattr(idValues, '__iter__'))
|
|
print(id + " values: " + str(idValues))
|
|
|
|
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('wfo', params.SITE_ID)
|
|
req.addIdentifier('huc', 'ALL')
|
|
req.setParameters('QPFSCAN')
|
|
return self.runGeometryDataTest(req, checkDataTimes=False)
|
|
|
|
def testGetDataWithEqualsString(self):
|
|
geometryData = self._runConstraintTest('siteKey', '=', self.location)
|
|
for record in geometryData:
|
|
self.assertEqual(record.getAttribute('siteKey'), self.location)
|
|
|
|
def testGetDataWithEqualsUnicode(self):
|
|
geometryData = self._runConstraintTest('siteKey', '=', unicode(self.location))
|
|
for record in geometryData:
|
|
self.assertEqual(record.getAttribute('siteKey'), self.location)
|
|
|
|
# No numeric tests since no numeric identifiers are available that support
|
|
# RequestConstraints.
|
|
|
|
def testGetDataWithEqualsNone(self):
|
|
geometryData = self._runConstraintTest('siteKey', '=', None)
|
|
for record in geometryData:
|
|
self.assertIsNone(record.getAttribute('siteKey'))
|
|
|
|
def testGetDataWithNotEquals(self):
|
|
geometryData = self._runConstraintTest('siteKey', '!=', self.location)
|
|
for record in geometryData:
|
|
self.assertNotEqual(record.getAttribute('siteKey'), self.location)
|
|
|
|
def testGetDataWithNotEqualsNone(self):
|
|
geometryData = self._runConstraintTest('siteKey', '!=', None)
|
|
for record in geometryData:
|
|
self.assertIsNotNone(record.getAttribute('siteKey'))
|
|
|
|
def testGetDataWithGreaterThan(self):
|
|
geometryData = self._runConstraintTest('siteKey', '>', self.location)
|
|
for record in geometryData:
|
|
self.assertGreater(record.getAttribute('siteKey'), self.location)
|
|
|
|
def testGetDataWithLessThan(self):
|
|
geometryData = self._runConstraintTest('siteKey', '<', self.location)
|
|
for record in geometryData:
|
|
self.assertLess(record.getAttribute('siteKey'), self.location)
|
|
|
|
def testGetDataWithGreaterThanEquals(self):
|
|
geometryData = self._runConstraintTest('siteKey', '>=', self.location)
|
|
for record in geometryData:
|
|
self.assertGreaterEqual(record.getAttribute('siteKey'), self.location)
|
|
|
|
def testGetDataWithLessThanEquals(self):
|
|
geometryData = self._runConstraintTest('siteKey', '<=', self.location)
|
|
for record in geometryData:
|
|
self.assertLessEqual(record.getAttribute('siteKey'), self.location)
|
|
|
|
def testGetDataWithInList(self):
|
|
collection = [self.location, 'kuex']
|
|
geometryData = self._runConstraintTest('siteKey', 'in', collection)
|
|
for record in geometryData:
|
|
self.assertIn(record.getAttribute('siteKey'), collection)
|
|
|
|
def testGetDataWithNotInList(self):
|
|
collection = [self.location, 'kuex']
|
|
geometryData = self._runConstraintTest('siteKey', 'not in', collection)
|
|
for record in geometryData:
|
|
self.assertNotIn(record.getAttribute('siteKey'), collection)
|
|
|
|
def testGetDataWithInvalidConstraintTypeThrowsException(self):
|
|
with self.assertRaises(ValueError):
|
|
self._runConstraintTest('siteKey', 'junk', self.location)
|
|
|
|
def testGetDataWithInvalidConstraintValueThrowsException(self):
|
|
with self.assertRaises(TypeError):
|
|
self._runConstraintTest('siteKey', '=', {})
|
|
|
|
def testGetDataWithEmptyInConstraintThrowsException(self):
|
|
with self.assertRaises(ValueError):
|
|
self._runConstraintTest('siteKey', 'in', [])
|
|
|
|
def testGetDataWithSiteKeyAndDataKeyConstraints(self):
|
|
siteKeys = [self.location, 'hpe']
|
|
dataKeys = ['kuex', 'kdmx']
|
|
|
|
req = DAL.newDataRequest(self.datatype)
|
|
req.addIdentifier('wfo', params.SITE_ID)
|
|
req.addIdentifier('huc', 'ALL')
|
|
|
|
siteKeysConstraint = RequestConstraint.new('in', siteKeys)
|
|
req.addIdentifier('siteKey', siteKeysConstraint)
|
|
dataKeysConstraint = RequestConstraint.new('in', dataKeys)
|
|
req.addIdentifier('dataKey', dataKeysConstraint)
|
|
|
|
req.setParameters('QPFSCAN')
|
|
geometryData = self.runGeometryDataTest(req, checkDataTimes=False)
|
|
for record in geometryData:
|
|
self.assertIn(record.getAttribute('siteKey'), siteKeys)
|
|
# dataKey attr. is comma-separated list of dataKeys that had data
|
|
for dataKey in record.getAttribute('dataKey').split(','):
|
|
self.assertIn(dataKey, dataKeys)
|
|
|
|
def testGetGuidanceDataWithoutAccumHrsIdentifierSet(self):
|
|
# Test that accumHrs identifier is not required for guidance data
|
|
req = DAL.newDataRequest(self.datatype)
|
|
req.addIdentifier('wfo', params.SITE_ID)
|
|
req.addIdentifier('siteKey', self.location)
|
|
req.addIdentifier('huc', 'ALL')
|
|
req.setParameters('FFG0124hr')
|
|
self.runGeometryDataTest(req, checkDataTimes=False) |