Task #8721: adapted store and test_store to new postgres db

LTA/ltastorageoverview/lib/create_db_ltastorageoverview.sql

@@ -297,6 +297,26 @@ CREATE TRIGGER trigger_on_directory_inserted_parse_project_info
 
 --------------------------------------------------------------------------------
 
+CREATE OR REPLACE FUNCTION lta.on_site_deleted_delete_site_directories()
+  RETURNS trigger AS
+$BODY$
+BEGIN
+    DELETE FROM lta.directory WHERE id in (SELECT dir_id FROM lta.site_directory_tree WHERE site_id = OLD.ID);
+
+    RETURN OLD;
+END;
+$BODY$
+  LANGUAGE plpgsql VOLATILE
+  COST 100;
+
+CREATE TRIGGER trigger_on_site_deleted_delete_site_directories
+  BEFORE DELETE
+  ON lta.storage_site
+  FOR EACH ROW
+  EXECUTE PROCEDURE lta.on_site_deleted_delete_site_directories();
+
+--------------------------------------------------------------------------------
+
 -- END TRIGGERS
 
 

LTA/ltastorageoverview/test/test_store.py

@@ -20,40 +20,45 @@
 # $Id$
 
 import unittest
+import logging
 import datetime
 import time
 import os
 import os.path
-import tempfile
+from lofar.common import dbcredentials
 from ltastorageoverview import store
 
+logger = logging.getLogger(__name__)
 
 class TestLTAStorageDb(unittest.TestCase):
     def setUp(self):
-        tmpfile = os.path.join(tempfile.gettempdir(), 'test.sqlite')
-        self.db = store.LTAStorageDb(tmpfile, True)
-
-        self.assertTrue(os.path.exists(self.db.db_filename))
+        dbcreds = dbcredentials.DBCredentials().get('LTASO')
+        self.db = store.LTAStorageDb(dbcreds, True)
+        self.db.executeQuery('delete from lta.fileinfo;')
+        self.db.executeQuery('delete from lta.directory;')
+        self.db.executeQuery('delete from lta.directory_closure;')
+        self.db.executeQuery('delete from lta.storage_site;')
+        self.db.commit()
 
     #def tearDown(self):
         #if os.path.exists(self.db.db_filename):
             #os.remove(self.db.db_filename)
 
     def testSites(self):
-        self.db.insertSite('siteA', 'srm://siteA.org')
-        self.db.insertSite('siteB', 'srm://siteB.org')
+        siteA_id = self.db.insertSite('siteA', 'srm://siteA.org')
+        siteB_id = self.db.insertSite('siteB', 'srm://siteB.org')
 
         sites = self.db.sites()
-        siteNames = [x[1] for x in sites]
+        siteNames = [x['name'] for x in sites]
         self.assertEquals(2, len(siteNames))
         self.assertTrue('siteA' in siteNames)
         self.assertTrue('siteB' in siteNames)
 
-        site = self.db.site(1)
-        self.assertTrue('siteA' in site[1])
+        site = self.db.site(siteA_id)
+        self.assertEqual('siteA', site['name'])
 
-        site = self.db.site(2)
-        self.assertTrue('siteB' in site[1])
+        site = self.db.site(siteB_id)
+        self.assertEqual('siteB', site['name'])
 
     def testRootDirs(self):
         siteA_id = self.db.insertSite('siteA', 'srm://siteA.org')
@@ -68,11 +73,28 @@ class TestLTAStorageDb(unittest.TestCase):
 
         rootDirs = self.db.rootDirectories()
         self.assertEquals(5, len(rootDirs))
-        self.assertTrue((dirA1_id, 'rootDir1', siteA_id, 'siteA') in rootDirs)
-        self.assertTrue((dirA2_id, 'rootDir2', siteA_id, 'siteA') in rootDirs)
-        self.assertTrue((dirA3_id, 'path/to/rootDir3', siteA_id, 'siteA') in rootDirs)
-        self.assertTrue((dirB1_id, 'rootDir1', siteB_id, 'siteB') in rootDirs)
-        self.assertTrue((dirB2_id, 'path/to/otherRootDir', siteB_id, 'siteB') in rootDirs)
+
+        rootDirsDict = {rd['dir_id']:rd for rd in rootDirs}
+
+        self.assertEqual('rootDir1', rootDirsDict[dirA1_id]['dir_name'])
+        self.assertEqual(siteA_id, rootDirsDict[dirA1_id]['site_id'])
+        self.assertEqual('siteA', rootDirsDict[dirA1_id]['site_name'])
+
+        self.assertEqual('rootDir2', rootDirsDict[dirA2_id]['dir_name'])
+        self.assertEqual(siteA_id, rootDirsDict[dirA2_id]['site_id'])
+        self.assertEqual('siteA', rootDirsDict[dirA2_id]['site_name'])
+
+        self.assertEqual('path/to/rootDir3', rootDirsDict[dirA3_id]['dir_name'])
+        self.assertEqual(siteA_id, rootDirsDict[dirA3_id]['site_id'])
+        self.assertEqual('siteA', rootDirsDict[dirA3_id]['site_name'])
+
+        self.assertEqual('rootDir1', rootDirsDict[dirB1_id]['dir_name'])
+        self.assertEqual(siteB_id, rootDirsDict[dirB1_id]['site_id'])
+        self.assertEqual('siteB', rootDirsDict[dirB1_id]['site_name'])
+
+        self.assertEqual('path/to/otherRootDir', rootDirsDict[dirB2_id]['dir_name'])
+        self.assertEqual(siteB_id, rootDirsDict[dirB2_id]['site_id'])
+        self.assertEqual('siteB', rootDirsDict[dirB2_id]['site_name'])
 
     def testDirectoryTrees(self):
         siteA_id = self.db.insertSite('siteA', 'srm://siteA.org')
@@ -92,10 +114,10 @@ class TestLTAStorageDb(unittest.TestCase):
         rootDirs = self.db.rootDirectories()
         self.assertEquals(2, len(rootDirs))
 
-        for (id, name, site_id, site_name) in rootDirs:
+        for id in [x['dir_id'] for x in rootDirs]:
             subDirs = self.db.subDirectories(id, 1, False)
             for subDir in subDirs:
-                subDir_parent_id = subDir[2]
+                subDir_parent_id = subDir['parent_dir_id']
                 self.assertEquals(id, subDir_parent_id)
 
         print '\n'.join([str(x) for x in self.db.filesInTree(rootDir_id)])
@@ -128,4 +150,7 @@ class TestLTAStorageDb(unittest.TestCase):
 
 # run tests if main
 if __name__ == '__main__':
+    logging.basicConfig(format='%(asctime)s %(levelname)s %(message)s',
+                        level=logging.INFO)
+
     unittest.main()

LTA/ltastorageoverview/test/test_store.run

 #!/bin/bash
 
-python test_store.py
+source python-coverage.sh
+python_coverage_test "ltas*" test_store.py