[ccplugin, tests] Anonymize Assessments's identifiers. Fix absolute path when moving data files
authorVladimir Popescu <vladimir.popescu@logilab.fr>
Wed, 17 Jul 2013 12:12:21 +0200
changeset 284 6b889bd4625d
parent 283 2920bacd4257
child 285 bc1b051a853c
[ccplugin, tests] Anonymize Assessments's identifiers. Fix absolute path when moving data files
ccplugin.py
test/unittest_remake-uuid.py
--- a/ccplugin.py	Wed Jul 17 10:37:26 2013 +0200
+++ b/ccplugin.py	Wed Jul 17 12:12:21 2013 +0200
@@ -21,7 +21,7 @@
 from os import path as osp
 import shutil
 import logging
-import csv 
+import csv
 from cubicweb import AuthenticationError
 from cubicweb import cwconfig
 from cubicweb.utils import make_uid
@@ -50,26 +50,35 @@
     if logger:
         logger.info('update_ident_fpath: processing subject %d...\n' % subject_eid)
     session.execute('SET X identifier %(i)s WHERE X eid %(e)s', {'i': uuid, 'e': subject_eid})
-    rset = session.execute('Any X WHERE X concerns S, S eid %(e)s, '
-                           'X is QuestionnaireRun', {'e': subject_eid}, build_descr=False)
-    for concerned_eid in rset:
-        concerned_eid = concerned_eid[0]
+    rset = session.execute('(Any X WHERE X concerns S, S eid %(e)s, '
+                           'X is QuestionnaireRun) UNION '
+                           '(Any X WHERE S concerned_by X, X is Assessment, '
+                            'S eid %(e)s)',
+                           {'e': subject_eid}, build_descr=False)
+    for concerned_or_assessment_eid in rset:
+        concerned_or_assessment_eid = concerned_or_assessment_eid[0]
         if logger:
-            logger.info('update_ident_fpath: processing related entity %d\n' % concerned_eid)
-        uuid = make_uid() + '_' + unicode(concerned_eid)
-        session.execute('SET X identifier %(i)s WHERE X eid %(e)s', {'i': uuid, 'e': concerned_eid})
+            logger.info('update_ident_fpath: processing related entity %d\n'
+                        % concerned_or_assessment_eid)
+        uuid = make_uid() + '_' + unicode(concerned_or_assessment_eid)
+        session.execute('SET X identifier %(i)s WHERE X eid %(e)s',
+                        {'i': uuid, 'e': concerned_or_assessment_eid})
     rset = session.execute('Any X, F WHERE X concerns S, S eid %(e)s, X filepath F, '
-                           'X is IN (Scan, GenomicMeasure)', {'e': subject_eid}, build_descr=False)
+                           'X is IN (Scan, GenomicMeasure)',
+                           {'e': subject_eid}, build_descr=False)
     for concerned_eid, concerned_fpath in rset:
         if logger:
             logger.info('update_ident_fpath: processing related entity %d\n' % concerned_eid)
         uuid = make_uid() + '_' + unicode(concerned_eid)
         session.execute('SET X identifier %(i)s WHERE X eid %(e)s', {'i': uuid, 'e': concerned_eid})
         new_filepath = uuid + '_' + osp.basename(concerned_fpath)
-        copy_file(concerned_fpath, new_filepath, logger)
+        full_fpath = session.entity_from_eid(concerned_eid).full_filepath
+        base_fpath = full_fpath.split(concerned_fpath)[0]
+        dest_fpath = osp.join(base_fpath, new_filepath)
+        copy_file(full_fpath, dest_fpath, logger)
         if delete:
             try:
-                os.remove(concerned_fpath)
+                os.remove(full_fpath)
             except OSError as oserr:
                 if logger:
                     logger.error('update_ident_fpath: %s\n' % oserr)
--- a/test/unittest_remake-uuid.py	Wed Jul 17 10:37:26 2013 +0200
+++ b/test/unittest_remake-uuid.py	Wed Jul 17 12:12:21 2013 +0200
@@ -15,7 +15,7 @@
         super(TestBrainomicsRemakeUid, self).setup_database()
         ce = self.request().create_entity
         self.rql = self.session.execute
-        _study = ce('Study', name=u'Test study', data_filepath=u'/tmp/test_data',
+        _study = ce('Study', name=u'Test study', data_filepath=u'/tmp',
                     description=u'Test study with dummy data', keywords=u'test;dummy')
         self.subject_1 = ce('Subject', identifier=u'demo_subject_1', gender=u'female',
                             handedness=u'right')
@@ -37,40 +37,62 @@
                           label=u'anatomy', type=u'normalized T1', format=u'nii.hz',
                           completed=True, valid=True, filepath=u'3785/anat.nii.gz',
                           concerns=self.subject_2, related_study=_study)
+        _center = ce('Center', identifier=u'testing_center', name=u'Testing Center')
+        self.assessment = ce('Assessment', identifier=u'assessment_3125',
+                             related_study=_study, reverse_holds=_center,
+                             reverse_concerned_by=self.subject_1)
         self.log = logging.getLogger('brainomics test: remake-uuid')
         fh = logging.FileHandler('brainomics_test.log')
         self.log.addHandler(fh)
 
     def test_copy_file(self):
-        for fpath in (scan.filepath for scan in (self.scan_11, self.scan_12, self.scan_21, self.scan_22)):
-            dest_path = unicode(make_uid()) + '_' + osp.basename(fpath)
+        for fpath, full_fpath in ((scan.filepath, scan.full_filepath)
+                                  for scan in (self.scan_11, self.scan_12, self.scan_21, self.scan_22)):
+            dest_path = osp.join(full_fpath.split(fpath)[0], 
+                                 unicode(make_uid()) + '_' + osp.basename(fpath))
             with tempfile.NamedTemporaryFile(suffix=osp.basename(fpath)) as f:
                 copy_file(f.name, dest_path, logger=self.log)
             if osp.isfile(dest_path):
                 os.remove(dest_path)
 
-    def test_update_ident_fpath_nodelete(self):
-        old_fpaths = self.rql('Any F WHERE X is IN (Scan, GenomicMeasure), X filepath F', build_descr=False)
+    def test_update_fpath_nodelete(self):
+        fpath_query = 'Any F WHERE X is IN (Scan, GenomicMeasure), X filepath F'
+        old_fpaths = self.rql(fpath_query, build_descr=False)
         for s in self.rql('Any X WHERE X is Subject'):
             update_ident_fpath(self.session, s[0], unicode(make_uid()), logger=self.log)
-        new_fpaths = self.rql('Any F WHERE X is IN (Scan, GenomicMeasure), X filepath F', build_descr=False)
+        new_fpaths = self.rql(fpath_query, build_descr=False)
+        self.assertNotEqual(old_fpaths, new_fpaths)
 
+    def test_update_fpath_delete(self):
+        fpath_query = 'Any F WHERE X is IN (Scan, GenomicMeasure), X filepath F'
+        old_fpaths = self.rql(fpath_query, build_descr=False)
+        for s in self.rql('Any X WHERE X is Subject'):
+            update_ident_fpath(self.session, s[0], unicode(make_uid()), delete=True, logger=self.log)
+        new_fpaths = self.rql(fpath_query, build_descr=False)
         self.assertNotEqual(old_fpaths, new_fpaths)
 
-    def test_update_ident_fpath_delete(self):
-        old_fpaths = self.rql('Any F WHERE X is IN (Scan, GenomicMeasure), X filepath F', build_descr=False)
+    def test_update_ident(self):
+        ident_query = 'Any I WHERE X is IN (Scan, GenomicMeasure, QuestionnaireRun), X identifier I'
+        old_idents = self.rql(ident_query, build_descr=False)
         for s in self.rql('Any X WHERE X is Subject'):
-            update_ident_fpath(self.session, s[0], unicode(make_uid()), delete=True, logger=self.log)
-        new_fpaths = self.rql('Any F WHERE X is IN (Scan, GenomicMeasure), X filepath F', build_descr=False)
+            update_ident_fpath(self.session, s[0], unicode(make_uid()), logger=self.log)
+        new_idents = self.rql(ident_query)
+        self.assertNotEqual(old_idents, new_idents)
 
-        self.assertNotEqual(old_fpaths, new_fpaths)
+    def test_update_ident_assessment(self):
+        assess_ident_query = 'Any I WHERE X is Assessment, X identifier I'
+        old_idents = self.rql(assess_ident_query, build_descr=False)
+        for s in self.rql('Any X WHERE X is Subject'):
+            update_ident_fpath(self.session, s[0], unicode(make_uid()), logger=self.log)
+        new_idents = self.rql(assess_ident_query, build_descr=False)
+        self.assertNotEqual(old_idents, new_idents)
 
     def test_get_uid_from_csv(self):
         old_new_uuids = {}
         identifiers_rset = self.rql('Any I WHERE X is Subject, X identifier I', 
                                     build_descr=False)
         # Generate test CSV
-        test_csv_fname = 'test_external_uuids.csv'
+        test_csv_fname = 'external_uuids.csv'
         with open(test_csv_fname, 'wb') as csvf:
             csvw = csv.writer(csvf, delimiter='\t')
             for old_identifier in identifiers_rset: 
@@ -85,9 +107,6 @@
 
 
 
-
-
-
 if __name__ == '__main__':
     from logilab.common.testlib import unittest_main
     unittest_main()