Not a member of Pastebin yet?
Sign Up,
it unlocks many cool features!
- import unittest
- import sys
- import os
- import numpy as np
- from rasdapy import core
- # Run this test first to open a connection to rasserver and keep it
- class ConfigManager(unittest.TestCase):
- db = None
- script_dir = None
- def setUp(self):
- print "Open connection to rasserver..."
- if ConfigManager.db is None:
- self.hostname = "localhost"
- self.username = "rasadmin"
- self.password = "rasadmin"
- self.con = core.Connection(hostname=self.hostname,
- username=self.username, password=self.password)
- ConfigManager.db = self.con.database("RASBASE")
- ConfigManager.script_dir = os.path.dirname(os.path.realpath(__file__))
- def test_00(self):
- # NOTE: without this dummy method, ConfigManager will not run first by unittest and cannot open connection to rasserver for other test classes
- print "Done"
- @staticmethod
- def execute_read(rasql_query):
- print rasql_query
- txn = ConfigManager.db.transaction(rw=False)
- query = txn.query(rasql_query)
- res = query.eval()
- txn.abort()
- return res
- @staticmethod
- def execute_write(rasql_query):
- print rasql_query
- txn = ConfigManager.db.transaction(rw=True)
- query = txn.query(rasql_query)
- data = query.eval()
- txn.commit()
- return data
- # Test CRUD on a fiction collection GreySet (1 band)
- class Test01(unittest.TestCase):
- def setUp(self):
- self.db = ConfigManager.db
- self.collName = "test_rasdapy_grey"
- def test_01_create_collection(self):
- rasql_query = "create collection " + self.collName + " GreySet"
- res = ConfigManager.execute_write(rasql_query)
- self.assertEqual(res, None)
- print "Done"
- def test_02_insert_in_collection(self):
- rasql_query = "insert into " + self.collName + " values marray x in [0:2, 0:2] values 1c"
- res = ConfigManager.execute_write(rasql_query)
- self.assertEqual(res, None)
- print "Done"
- def test_03_read_scalar_from_collection(self):
- rasql_query = "select avg_cells(" + self.collName + ") from " + self.collName
- data = ConfigManager.execute_read(rasql_query)
- self.assertIsInstance(data, list)
- self.assertEqual(data, [1.0])
- print "Done"
- def test_04_read_raw_data_from_collection(self):
- rasql_query = 'select ' + self.collName + ' from ' + self.collName
- data = ConfigManager.execute_read(rasql_query)
- # convert rasdapy.core.Array to numpy.ndarray
- ndarray = data.to_array()
- avg_ndarray = ndarray.mean()
- self.assertEqual(avg_ndarray, 1.0)
- print "Done"
- def test_10_drop_collection(self):
- rasql_query = "drop collection " + self.collName
- res = ConfigManager.execute_write(rasql_query)
- self.assertEqual(res, None)
- print "Done"
- # Test CRUD with a real import file GreySet (1 band)
- class Test02(unittest.TestCase):
- def setUp(self):
- self.db = ConfigManager.db
- self.collName = "test_rasdapy_mr"
- def test_01_create_collection(self):
- rasql_query = "create collection " + self.collName + " GreySet"
- res = ConfigManager.execute_write(rasql_query)
- self.assertEqual(res, None)
- print "Done"
- def test_02_insert_initial_values(self):
- rasql_query = "INSERT INTO " + self.collName + " VALUES <[0:0,0:0] 0c> TILING ALIGNED [0:1023, 0:1023] TILE SIZE 4194304"
- res = ConfigManager.execute_write(rasql_query)
- self.assertEqual(res, None)
- print "Done"
- def test_03_update_collection_from_file(self):
- # TODO - ticket:1671 fixed then, run this query via raspy
- rasql_query = "UPDATE " + self.collName + " SET " + self.collName + "[0:255,0:210] ASSIGN shift(decode(<[0:0] 1c>, \"GDAL\", \"{\\\"filePaths\\\":[\\\"/home/rasdaman/TEST/rasdapy/tests/mr_1.png\\\"]}\"), [0,0])"
- print rasql_query
- os.system("rasql -q '" + rasql_query + "' --user rasadmin --passwd rasadmin >> /dev/null")
- print "Done"
- def test_04_get_the_sdom_by_api(self):
- # NOTE: select sdom(c) returns a string which cannot be converted to rasdapy.core.Array object, then use the API to get the sdom
- rasql_query = "select c from " + self.collName + " as c"
- data = ConfigManager.execute_read(rasql_query)
- # sdom of test_mr is [0:250, 0:210]
- self.assertEqual(data.metadata.spatial_domain.interval_params, [('0', '255'), ('0', '210')])
- # grey set
- self.assertEqual(data.metadata.band_types, {'base_type': 'marray', 'type': 'char'})
- print "Done"
- def test_05_average_duplicate_collections(self):
- rasql_query = "select avg_cells(c + d + c + c + c + d + d + d) from " + self.collName + " as c, " + self.collName + " as d"
- data = ConfigManager.execute_read(rasql_query)
- self.assertEqual(data, [68.75666469194313])
- print "Done"
- def test_06_mean_np_array(self):
- rasql_query = 'select encode(c, "csv") from ' + self.collName + ' as c '
- data = ConfigManager.execute_read(rasql_query)
- # convert rasdapy.core.Array to numpy ndarray
- nparray = data.to_array()
- avg_of_nparray = nparray.mean()
- self.assertEqual(avg_of_nparray, 48.530484016224634)
- print "Done"
- def test_10_drop_collection(self):
- rasql_query = "drop collection " + self.collName
- res = ConfigManager.execute_write(rasql_query)
- self.assertEqual(res, None)
- print "Done"
- # Test CRUD on a fiction collection RGBSet (3 bands)
- class Test03(unittest.TestCase):
- def setUp(self):
- self.db = ConfigManager.db
- self.collName = "test_rasdapy_rgb"
- """ def test_01_create_collection(self):
- rasql_query = "create collection " + self.collName + " RGBSet"
- res = ConfigManager.execute_write(rasql_query)
- self.assertEqual(res, None)
- print "Done"
- def test_02_insert_in_collection(self):
- rasql_query = "insert into " + self.collName + " values marray x in [0:2, 0:2] values {1c, 1c, 1c}"
- res = ConfigManager.execute_write(rasql_query)
- self.assertEqual(res, None)
- print "Done"
- """
- def test_03_read_scalar_1band_from_collection(self):
- rasql_query = "select avg_cells(" + self.collName + ".0) from " + self.collName
- data = ConfigManager.execute_read(rasql_query)
- self.assertIsInstance(data, list)
- self.assertEqual(data, [1.0])
- print "Done"
- def test_04_read_scalar_3bands_from_collection(self):
- rasql_query = "select avg_cells(" + self.collName + ") from " + self.collName
- data = ConfigManager.execute_read(rasql_query)
- self.assertIsInstance(data, list)
- # TODO - ticket:1672 - avg_cells() are wrong for band2, 3, only compare band1 here
- self.assertEqual(data[0][0], 1.0)
- print "Done"
- def test_05_read_scalar_from_plus_collections(self):
- rasql_query = "select c + d + e + f from " + self.collName + " as c, " + self.collName + " as d, " + self.collName + " as e, " + self.collName +" as f"
- data = ConfigManager.execute_read(rasql_query)
- ndarray = data.to_array()
- avg_ndarray = ndarray.mean()
- self.assertEqual(avg_ndarray, 4.0)
- print "Done"
- def test_06_read_with_star_intervals(self):
- rasql_query = "select c[*:*, *:*] from " + self.collName + " as c"
- data = ConfigManager.execute_read(rasql_query)
- ndarray = data.to_array()
- avg_ndarray = ndarray.mean()
- self.assertEqual(avg_ndarray, 1.0)
- print "Done"
- def test_07_read_specific_with_star_interval(self):
- rasql_query = 'select c[0:2, 0:*] from ' + self.collName + ' as c'
- data = ConfigManager.execute_read(rasql_query)
- ndarray = data.to_array()
- avg_ndarray = ndarray.mean()
- self.assertEqual(avg_ndarray, 1.0)
- print "Done"
- def test_08_array_constants(self):
- # TODO - ticket:1674 - error parsing from array constants
- rasql_query = 'select < [0:2,0:4] 0, 1, 2, 3, 4; 1, 2, 3, 4, 5; 2, 3, 4, 5, 6 > ' + self.collName + ' as c'
- #data = ConfigManager.execute_read(rasql_query)
- #ndarray = data.to_array()
- #print ndarray
- self.assertEqual(None, None)
- # print "Done"
- def test_09_atomic_constant_char(self):
- rasql_query = 'select 24c from ' + self.collName + ' as c'
- data = ConfigManager.execute_read(rasql_query)
- self.assertEqual(data, [24])
- print "Done"
- def test_10_atomic_scientific_number(self):
- rasql_query = 'select 4e-5D from ' + self.collName + ' as c'
- data = ConfigManager.execute_read(rasql_query)
- print data
- def test_11_encode_png(self):
- #rasql_query = 'select encode(c, "png") from test_mr as c'
- rasql_query = 'SELECT encode(c[160:580,120:620], "tiff" , "{\\\"geoReference\\\":{\\\"crs\\\":\\\"EPSG:4326\\\",\\\"bbox\\\":{\\\"xmin\\\":119.975,\\\"ymin\\\":-40.025,\\\"xmax\\\":141.025,\\\"ymax\\\":-14.975}},\\\"formatParameters\\\":{\\\"INTERLEAVE\\\":\\\"BAND\\\"},\\\"metadata\\\":{\\\"new_metadata\\\":\\\"This is a new added metadata\\\"},\\\"nodata\\\":[0]}") FROM test_mean_summer_airtemp AS c'
- data = ConfigManager.execute_read(rasql_query)
- print data.to_array().shape
- data.to_array().astype('uint8').tofile(ConfigManager.script_dir + "/test.tiff")
- def test_12_no_encode(self):
- rasql_query = 'select c from ' + self.collName + ' as c'
- data = ConfigManager.execute_read(rasql_query)
- print data.to_array()
- """ def test_30_drop_collection(self):
- rasql_query = "drop collection " + self.collName
- res = ConfigManager.execute_write(rasql_query)
- self.assertEqual(res, None)
- print "Done" """
- if __name__ == '__main__':
- test_suite = unittest.defaultTestLoader.discover('.', 'test_crud.py')
- test_runner = unittest.TextTestRunner(resultclass=unittest.TextTestResult)
- result = test_runner.run(test_suite)
- os._exit(not result.wasSuccessful())
Add Comment
Please, Sign In to add comment