Guest User

Untitled

a guest
Jan 12th, 2018
48
0
Never
Not a member of Pastebin yet? Sign Up, it unlocks many cool features!
Python 10.33 KB | None | 0 0
  1. import unittest
  2. import sys
  3. import os
  4. import numpy as np
  5. from rasdapy import core
  6.  
  7. # Run this test first to open a connection to rasserver and keep it
  8. class ConfigManager(unittest.TestCase):
  9.    
  10.     db = None
  11.     script_dir = None
  12.  
  13.     def setUp(self):
  14.         print "Open connection to rasserver..."
  15.         if ConfigManager.db is None:
  16.             self.hostname = "localhost"
  17.             self.username = "rasadmin"
  18.             self.password = "rasadmin"
  19.             self.con = core.Connection(hostname=self.hostname,
  20.                                 username=self.username, password=self.password)
  21.             ConfigManager.db = self.con.database("RASBASE")
  22.             ConfigManager.script_dir = os.path.dirname(os.path.realpath(__file__))
  23.  
  24.     def test_00(self):
  25.         # NOTE: without this dummy method, ConfigManager will not run first by unittest and cannot open connection to rasserver for other test classes
  26.         print "Done"
  27.  
  28.     @staticmethod
  29.     def execute_read(rasql_query):
  30.         print rasql_query
  31.         txn = ConfigManager.db.transaction(rw=False)
  32.         query = txn.query(rasql_query)
  33.         res = query.eval()
  34.         txn.abort()
  35.  
  36.         return res
  37.    
  38.     @staticmethod
  39.     def execute_write(rasql_query):
  40.         print rasql_query
  41.         txn = ConfigManager.db.transaction(rw=True)
  42.         query = txn.query(rasql_query)
  43.         data = query.eval()
  44.         txn.commit()
  45.  
  46.         return data
  47.  
  48. # Test CRUD on a fiction collection GreySet (1 band)
  49. class Test01(unittest.TestCase):
  50.  
  51.     def setUp(self):
  52.         self.db = ConfigManager.db
  53.         self.collName = "test_rasdapy_grey"
  54.    
  55.     def test_01_create_collection(self):
  56.         rasql_query = "create collection " + self.collName + " GreySet"
  57.         res = ConfigManager.execute_write(rasql_query)
  58.         self.assertEqual(res, None)
  59.         print "Done"
  60.  
  61.     def test_02_insert_in_collection(self):
  62.         rasql_query = "insert into " + self.collName + " values marray x in [0:2, 0:2] values 1c"
  63.         res = ConfigManager.execute_write(rasql_query)
  64.         self.assertEqual(res, None)
  65.         print "Done"
  66.  
  67.     def test_03_read_scalar_from_collection(self):
  68.         rasql_query = "select avg_cells(" + self.collName + ") from " + self.collName
  69.         data = ConfigManager.execute_read(rasql_query)
  70.         self.assertIsInstance(data, list)
  71.         self.assertEqual(data, [1.0])
  72.         print "Done"
  73.  
  74.     def test_04_read_raw_data_from_collection(self):
  75.         rasql_query = 'select ' + self.collName + ' from ' + self.collName
  76.         data = ConfigManager.execute_read(rasql_query)
  77.         # convert rasdapy.core.Array to numpy.ndarray
  78.         ndarray = data.to_array()
  79.         avg_ndarray = ndarray.mean()
  80.         self.assertEqual(avg_ndarray, 1.0)
  81.         print "Done"
  82.    
  83.     def test_10_drop_collection(self):
  84.         rasql_query = "drop collection " + self.collName
  85.         res = ConfigManager.execute_write(rasql_query)
  86.         self.assertEqual(res, None)
  87.         print "Done"        
  88.  
  89.  
  90. # Test CRUD with a real import file GreySet (1 band)
  91. class Test02(unittest.TestCase):
  92.  
  93.     def setUp(self):
  94.         self.db = ConfigManager.db
  95.         self.collName = "test_rasdapy_mr"
  96.    
  97.     def test_01_create_collection(self):
  98.         rasql_query = "create collection " + self.collName + " GreySet"
  99.         res = ConfigManager.execute_write(rasql_query)
  100.         self.assertEqual(res, None)
  101.         print "Done"
  102.  
  103.     def test_02_insert_initial_values(self):
  104.         rasql_query = "INSERT INTO " + self.collName + " VALUES <[0:0,0:0] 0c> TILING ALIGNED [0:1023, 0:1023] TILE SIZE 4194304"
  105.         res = ConfigManager.execute_write(rasql_query)
  106.         self.assertEqual(res, None)
  107.         print "Done"
  108.    
  109.     def test_03_update_collection_from_file(self):
  110.         # TODO - ticket:1671 fixed then, run this query via raspy
  111.         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])"
  112.         print rasql_query
  113.         os.system("rasql -q '" + rasql_query + "' --user rasadmin --passwd rasadmin >> /dev/null")
  114.         print "Done"
  115.  
  116.     def test_04_get_the_sdom_by_api(self):
  117.         # NOTE: select sdom(c) returns a string which cannot be converted to rasdapy.core.Array object, then use the API to get the sdom
  118.         rasql_query = "select c from " + self.collName + " as c"
  119.         data = ConfigManager.execute_read(rasql_query)
  120.         # sdom of test_mr is [0:250, 0:210]
  121.         self.assertEqual(data.metadata.spatial_domain.interval_params, [('0', '255'), ('0', '210')])
  122.         # grey set
  123.         self.assertEqual(data.metadata.band_types, {'base_type': 'marray', 'type': 'char'})        
  124.         print "Done"
  125.  
  126.     def test_05_average_duplicate_collections(self):
  127.         rasql_query = "select avg_cells(c + d + c + c + c + d + d + d) from " + self.collName + " as c, " + self.collName + " as d"
  128.         data = ConfigManager.execute_read(rasql_query)      
  129.         self.assertEqual(data, [68.75666469194313])                        
  130.         print "Done"
  131.  
  132.     def test_06_mean_np_array(self):
  133.         rasql_query = 'select encode(c, "csv") from ' + self.collName + ' as c '
  134.         data = ConfigManager.execute_read(rasql_query)
  135.         # convert rasdapy.core.Array to numpy ndarray
  136.         nparray = data.to_array()
  137.         avg_of_nparray = nparray.mean()
  138.         self.assertEqual(avg_of_nparray, 48.530484016224634)
  139.         print "Done"
  140.  
  141.     def test_10_drop_collection(self):
  142.         rasql_query = "drop collection " + self.collName
  143.         res = ConfigManager.execute_write(rasql_query)
  144.         self.assertEqual(res, None)
  145.         print "Done"
  146.  
  147. # Test CRUD on a fiction collection RGBSet (3 bands)
  148. class Test03(unittest.TestCase):
  149.  
  150.     def setUp(self):
  151.         self.db = ConfigManager.db
  152.         self.collName = "test_rasdapy_rgb"
  153.    
  154.     """ def test_01_create_collection(self):
  155.        rasql_query = "create collection " + self.collName + " RGBSet"
  156.        res = ConfigManager.execute_write(rasql_query)
  157.        self.assertEqual(res, None)
  158.        print "Done"
  159.  
  160.    def test_02_insert_in_collection(self):
  161.        rasql_query = "insert into " + self.collName + " values marray x in [0:2, 0:2] values {1c, 1c, 1c}"
  162.        res = ConfigManager.execute_write(rasql_query)
  163.        self.assertEqual(res, None)
  164.        print "Done"
  165.    """
  166.  
  167.     def test_03_read_scalar_1band_from_collection(self):
  168.         rasql_query = "select avg_cells(" + self.collName + ".0) from " + self.collName
  169.         data = ConfigManager.execute_read(rasql_query)        
  170.         self.assertIsInstance(data, list)
  171.         self.assertEqual(data, [1.0])
  172.         print "Done"
  173.  
  174.     def test_04_read_scalar_3bands_from_collection(self):
  175.         rasql_query = "select avg_cells(" + self.collName + ") from " + self.collName
  176.         data = ConfigManager.execute_read(rasql_query)
  177.         self.assertIsInstance(data, list)
  178.         # TODO - ticket:1672 - avg_cells() are wrong for band2, 3, only compare band1 here
  179.         self.assertEqual(data[0][0], 1.0)
  180.         print "Done"
  181.  
  182.     def test_05_read_scalar_from_plus_collections(self):
  183.         rasql_query = "select c + d + e + f from " + self.collName + " as c, " + self.collName + " as d, " + self.collName + " as e, " + self.collName +" as f"
  184.         data = ConfigManager.execute_read(rasql_query)
  185.         ndarray = data.to_array()
  186.         avg_ndarray = ndarray.mean()
  187.         self.assertEqual(avg_ndarray, 4.0)
  188.         print "Done"
  189.  
  190.     def test_06_read_with_star_intervals(self):
  191.         rasql_query = "select c[*:*, *:*] from " + self.collName + " as c"
  192.         data = ConfigManager.execute_read(rasql_query)
  193.         ndarray = data.to_array()
  194.         avg_ndarray = ndarray.mean()
  195.         self.assertEqual(avg_ndarray, 1.0)
  196.         print "Done"
  197.  
  198.     def test_07_read_specific_with_star_interval(self):
  199.         rasql_query = 'select c[0:2, 0:*] from ' + self.collName + ' as c'
  200.         data = ConfigManager.execute_read(rasql_query)
  201.         ndarray = data.to_array()
  202.         avg_ndarray = ndarray.mean()
  203.         self.assertEqual(avg_ndarray, 1.0)
  204.         print "Done"
  205.  
  206.     def test_08_array_constants(self):
  207.         # TODO - ticket:1674 - error parsing from array constants
  208.         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'
  209.         #data = ConfigManager.execute_read(rasql_query)
  210.         #ndarray = data.to_array()
  211.         #print ndarray
  212.         self.assertEqual(None, None)
  213.         # print "Done"
  214.    
  215.     def test_09_atomic_constant_char(self):
  216.         rasql_query = 'select 24c from ' + self.collName + ' as c'
  217.         data = ConfigManager.execute_read(rasql_query)
  218.         self.assertEqual(data, [24])
  219.         print "Done"
  220.  
  221.     def test_10_atomic_scientific_number(self):
  222.         rasql_query = 'select 4e-5D from ' + self.collName + ' as c'
  223.         data = ConfigManager.execute_read(rasql_query)
  224.         print data
  225.  
  226.     def test_11_encode_png(self):
  227.         #rasql_query = 'select encode(c, "png") from test_mr as c'
  228.         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'
  229.         data = ConfigManager.execute_read(rasql_query)
  230.         print data.to_array().shape
  231.         data.to_array().astype('uint8').tofile(ConfigManager.script_dir + "/test.tiff")
  232.  
  233.     def test_12_no_encode(self):
  234.         rasql_query = 'select c from ' + self.collName + ' as c'
  235.         data = ConfigManager.execute_read(rasql_query)
  236.         print data.to_array()
  237.        
  238.    
  239.     """ def test_30_drop_collection(self):
  240.        rasql_query = "drop collection " + self.collName
  241.        res = ConfigManager.execute_write(rasql_query)
  242.        self.assertEqual(res, None)
  243.        print "Done" """
  244.  
  245. if __name__ == '__main__':
  246.     test_suite = unittest.defaultTestLoader.discover('.', 'test_crud.py')
  247.     test_runner = unittest.TextTestRunner(resultclass=unittest.TextTestResult)
  248.     result = test_runner.run(test_suite)
  249.     os._exit(not result.wasSuccessful())
Add Comment
Please, Sign In to add comment