[ccplugin] Remove the remake-uid command, as it has been superseded by newer developments (closes #3400780)
authorVladimir Popescu <vladimir.popescu@logilab.fr>
Fri, 03 Jan 2014 13:57:11 +0100
changeset 650 f70e9473e57f
parent 649 767b35392314
child 651 eece52372752
[ccplugin] Remove the remake-uid command, as it has been superseded by newer developments (closes #3400780)
ccplugin.py
test/unittest_remake-uuid.py
--- a/ccplugin.py	Mon Jan 20 14:40:57 2014 +0000
+++ /dev/null	Thu Jan 01 00:00:00 1970 +0000
@@ -1,179 +0,0 @@
-# -*- coding: utf-8 -*-
-# copyright 2013 LOGILAB S.A. (Paris, FRANCE), all rights reserved.
-# copyright 2013 CEA (Saclay, FRANCE), all rights reserved.
-# contact http://www.logilab.fr -- mailto:contact@logilab.fr
-#
-# This program is free software: you can redistribute it and/or modify it under
-# the terms of the GNU Lesser General Public License as published by the Free
-# Software Foundation, either version 2.1 of the License, or (at your option)
-# any later version.
-#
-# This program is distributed in the hope that it will be useful, but WITHOUT
-# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
-# FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public License for more
-# details.
-#
-# You should have received a copy of the GNU Lesser General Public License along
-# with this program. If not, see <http://www.gnu.org/licenses/>.
-
-
-import os
-from os import path as osp
-import shutil
-import logging
-import csv
-from cubicweb import AuthenticationError
-from cubicweb import cwconfig
-from cubicweb.utils import make_uid
-from cubicweb.server.utils import manager_userpasswd
-from cubicweb.dbapi import in_memory_repo_cnx
-from cubicweb.toolsutils import Command
-from cubicweb.cwctl import CWCTL, check_options_consistency
-
-
-def copy_file(old_fpath, new_fpath, logger=None):
-    """Safe copy of a file from one path to another
-    """
-    if osp.isfile(old_fpath):
-        if logger:
-            logger.info('copy_file: Attempting to copy file %s to %s...\n'
-                        % (old_fpath, new_fpath))
-        try:
-            shutil.copyfile(old_fpath, new_fpath)
-        except IOError as ioer:
-            logger.error('copy_file: %s\n' % ioer)
-            raise
-
-def update_ident_fpath(session, subject_eid, uuid, delete=False, logger=None):
-    """Update the identifier and paths of a subject
-    """
-    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) 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_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)
-    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)
-        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(full_fpath)
-            except OSError as oserr:
-                if logger:
-                    logger.error('update_ident_fpath: %s\n' % oserr)
-        session.execute('SET X filepath %(i)s WHERE X eid %(e)s', {'i': new_filepath, 'e': concerned_eid})
-
-
-def get_ids_uids_from_csv(csv_filename, logger=None):
-    try:
-        csvf = open(csv_filename, 'r')
-    except IOError:
-        if logger:
-            logger.error('get_ids_uids_from_csv: External UUID CSV file not found: %s\n' % csv_filename)
-        raise
-    with csvf:
-        dialect = csv.Sniffer().sniff(csvf.readline())
-        csvf.seek(0)
-        return dict(([unicode(x) for x in row] for row in csv.reader(csvf, dialect)))
-
-
-
-class BrainomicsRemakeUidCommand(Command):
-    """Change subject uids for anonymization.
-
-    <instance>
-      the identifier of the instance to anonymize
-    """
-    name = 'remake-uid'
-    arguments = '<instance>'
-    min_args = 1
-    options = (
-        ('log',
-         {'short': 'l', 'action': 'store_true', 'default': False,
-          'help': 'logging mode: logs all command operations'}),
-        ('delete',
-         {'short': 'd', 'action': 'store_true', 'default': False,
-          'help': ('delete files from the old path, '
-                   'after creating the files at the new path')}),
-        ('external-uids',
-         {'short': 'u', 'type': 'string',
-          'help': 'specify external UID file'}),
-        )
-
-    def _init_cw_connection(self, appid):
-        logger = logging.getLogger('brainomics')
-        config = cwconfig.instance_configuration(appid)
-        sourcescfg = config.sources()
-        config.set_sources_mode(('system',))
-        cnx = repo = None
-        while cnx is None:
-            try:
-                login = sourcescfg['admin']['login']
-                pwd = sourcescfg['admin']['password']
-            except KeyError:
-                login, pwd = manager_userpasswd()
-            try:
-                repo, cnx = in_memory_repo_cnx(config, login=login, password=pwd)
-            except AuthenticationError:
-                logger.warning('remake-uuid: wrong user/password')
-            else:
-                break
-        session = repo._get_session(cnx.sessionid)
-        return cnx, session
-
-    def run(self, args):
-        logger = logging.getLogger('brainomics')
-        if len(args) > 0:
-            appid = args.pop(0)
-            cw_cnx, session = self._init_cw_connection(appid)
-            session.set_cnxset()
-            log = self.config.log
-            delete = self.config.delete
-            if self.config.external_uids:
-                csv_filename = self.config.external_uids
-                ids_uids = get_ids_uids_from_csv(csv_filename, logger)
-            if log:
-                log_file = logging.FileHandler(appid + '.log')
-                logger.addHandler(log_file)
-            rset = session.execute('Any X, I WHERE X is Subject, X identifier I', build_descr=False)
-            for subject_eid, subject_identifier in rset:
-                if not self.config.external_uids:
-                    uuid = unicode(make_uid())
-                else:
-                    try:
-                        uuid = ids_uids[subject_identifier]
-                    except KeyError:
-                        if logger:
-                            logger.error('remake-uid: New identifier not found in %s '
-                                         'for identifier %s; keeping the old one \n'
-                                         % (csv_filename, subject_identifier))
-                        session.rollback()
-                        raise
-                update_ident_fpath(session, subject_eid, uuid, delete, logger)
-            session.commit()
-        else:
-            logger.error("remake-uuid: Please specify Brainomics instance")
-
-
-CWCTL.register(BrainomicsRemakeUidCommand)
--- a/test/unittest_remake-uuid.py	Mon Jan 20 14:40:57 2014 +0000
+++ /dev/null	Thu Jan 01 00:00:00 1970 +0000
@@ -1,112 +0,0 @@
-import os
-import os.path as osp
-import tempfile
-import logging
-import csv
-from cubicweb.utils import make_uid
-from cubicweb.devtools.testlib import CubicWebTC
-from cubes.brainomics.ccplugin import copy_file, update_ident_fpath, get_ids_uids_from_csv
-
-
-
-class TestBrainomicsRemakeUid(CubicWebTC):
-
-    def setup_database(self):
-        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',
-                    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')
-        self.subject_2 = ce('Subject', identifier=u'demo_subject_2', gender=u'male',
-                            handedness=u'ambidextrous')
-        self.scan_11 = ce('Scan', identifier=u'anat_1709',
-                          label=u'anatomy', type=u'normalized T1', format=u'nii.hz',
-                          completed=True, valid=True, filepath=u'1709/anat.nii.gz',
-                          concerns=self.subject_1, related_study=_study)
-        self.scan_12 = ce('Scan', identifier=u'anat_1899',
-                          label=u'anatomy', type=u'normalized T1', format=u'nii.hz',
-                          completed=True, valid=True, filepath=u'1899/anat.nii.gz',
-                          concerns=self.subject_1, related_study=_study)
-        self.scan_21 = ce('Scan', identifier=u'anat_2325',
-                          label=u'anatomy', type=u'normalized T1', format=u'nii.hz',
-                          completed=True, valid=True, filepath=u'2325/anat.nii.gz',
-                          concerns=self.subject_2, related_study=_study)
-        self.scan_22 = ce('Scan', identifier=u'anat_3785',
-                          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, 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_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(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(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()), logger=self.log)
-        new_idents = self.rql(ident_query)
-        self.assertNotEqual(old_idents, new_idents)
-
-    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 = 'external_uuids.csv'
-        with open(test_csv_fname, 'wb') as csvf:
-            csvw = csv.writer(csvf, delimiter='\t')
-            for old_identifier in identifiers_rset: 
-                old_new_uuids[old_identifier[0]] = u'test_' + unicode(make_uid())
-                csvw.writerow([old_identifier[0], old_new_uuids[old_identifier[0]]])
-        # Verify that the new UUIDs are correctly retrieved from the CSV
-        ids_uids = get_ids_uids_from_csv(test_csv_fname, logger=self.log)
-        for old_identifier in identifiers_rset:
-            self.assertEqual(old_new_uuids[old_identifier[0]], 
-                             ids_uids[old_identifier[0]])
-
-
-
-
-if __name__ == '__main__':
-    from logilab.common.testlib import unittest_main
-    unittest_main()