def test_get_database_user_directory(self):
     m = ParamManager(self.settings, self.request.session)
     d = m.get_database_user_directory()
     assert os.path.isdir(d)
     del self.request.session['username']
     d = m.get_database_user_directory()
     assert os.path.isdir(d)

    def __init__(self, settings, session):
        ParamManager.__init__(self, settings, session)
        self.log = logging.getLogger(__name__)
        self.databasename = "jobs.db"
        self.pathdb = self.get_database_user_directory()+"/"+self.databasename

        self.log.info(" ==> "+ self.pathdb +"<==");

        conn = sqlite3.connect("file:"+self.pathdb,uri=True)
        c = conn.cursor()
        reqSql = '''CREATE TABLE IF NOT EXISTS jobs
             (
             jobID INTEGER PRIMARY KEY AUTOINCREMENT,
             type text,
             state text,
             start int,
             end int ,
             data text,
             file text,
             preview string,
             requestGraph string,
             variates string,
             nr int
             )'''

        c.execute(reqSql)
        conn.commit()
        conn.close()

 def test_get_param(self):
     m = ParamManager(self.settings, self.request.session)
     m.set_param("test","testValue")
     d = m.get_param("test")
     assert d == "testValue"
     d = m.get_param("test2")
     assert d == ""

    def __init__(self, relation_type, identifier, label, identifier_prefix,rdfs_domain, prefixDomain, rdfs_range, prefixRange):

        type_range =  identifier
        idx = identifier.find("@")
        if idx > 0:
            uridi = identifier[0:idx]
        else:
            uridi = identifier

        if label == "":
            if idx > 0:
                self.label = identifier[0:idx]
            else:
                self.label = identifier
        else:
            self.label = label

        self.uri = ParamManager.encode_to_rdf_uri(uridi,prefix="askomics:")

        self.rdfs_range = rdfs_range

        if relation_type.startswith("entity"):
            self.relation_type = "owl:ObjectProperty"

        elif relation_type == "goterm":
            self.relation_type = "owl:ObjectProperty"
            self.rdfs_range = "owl:Class"
        else:
            self.relation_type = "owl:DatatypeProperty"

        self.rdfs_domain = ParamManager.encode_to_rdf_uri(rdfs_domain,prefixDomain)
        self.log = logging.getLogger(__name__)

    def test_encode_to_rdf_uri(self):
        r = ParamManager.encode_to_rdf_uri("A",prefix="http://helloworld/test/")
        assert r == "<http://helloworld/test/A>"
        r = ParamManager.encode_to_rdf_uri("A",prefix="<http://helloworld/test/>")
        assert r == "<http://helloworld/test/A>"

        r = ParamManager.encode_to_rdf_uri("A<A>A",prefix="<http://helloworld/test/>")
        assert r == "<http://helloworld/test/A_s3_3CA_s3_3EA>"

 def test_get_turtle_template(self):
     m = ParamManager(self.settings, self.request.session)
     try:
         m.get_turtle_template(None)
         assert False
     except Exception as e:
         assert True
     m.get_turtle_template(":a :b :c.")

    def upload_file_into_history(self, filename):
        """Upload a file (present in user upload directory) into the testing history

        :param filename: the file to upload
        """

        param_manager = ParamManager(self.settings, self.request.session)
        src_file = param_manager.get_upload_directory()
        self.galaxy_instance.tools.upload_file(src_file + filename, self.history_id, file_name=filename)

    def test_get_rdf_user_directory(self):
        m = ParamManager(self.settings, self.request.session)
        d = m.get_rdf_user_directory()
        assert os.path.isdir(d)
        shutil.rmtree(d)

        del self.request.session['username']
        d = m.get_rdf_user_directory()
        assert os.path.isdir(d)

    def statistics(self):
        """ Get information about triplet store """
        self.log.debug("== STATS ==")
        data = {}
        pm = ParamManager(self.settings, self.request.session)

        sqb = SparqlQueryBuilder(self.settings, self.request.session)
        ql = QueryLauncher(self.settings, self.request.session)
        tse = TripleStoreExplorer(self.settings, self.request.session)

        results = ql.process_query(sqb.get_statistics_number_of_triples().query)
        data["ntriples"] = results[0]["no"]

        results = ql.process_query(sqb.get_statistics_number_of_entities().query)
        data["nentities"] = results[0]["no"]

        results = ql.process_query(sqb.get_statistics_distinct_classes().query)
        data["nclasses"] = results[0]["no"]

        # Get the list of classes
        res_list_classes = ql.process_query(sqb.get_statistics_list_classes().query)

        data["class"] = {}
        for obj in res_list_classes:
            class_name = pm.remove_prefix(obj['class'])
            data["class"][class_name] = {}

        # Get the number of instances by class
        res_nb_instances = ql.process_query(sqb.get_statistics_nb_instances_by_classe().query)

        for obj in res_nb_instances:
            if 'class' in obj:
                class_name = pm.remove_prefix(obj['class'])
                data["class"][class_name]["count"] = obj['count']

        # Get details on relations for each classes
        for obj in res_list_classes:
            if 'class' in obj:
                class_name = pm.remove_prefix(obj['class'])
                uri = obj['class']

                shortcuts_list = tse.has_setting(uri, 'shortcut')

                src = Node(class_name, # We don't care about counter in stats
                    uri,
                    class_name,
                    shortcuts_list)

                attributes, nodes, links = tse.get_neighbours_for_node(src, None)

                data["class"][class_name]["attributes"] = [a.to_dict() for a in attributes]
                data["class"][class_name]["neighbours"] = [n.to_dict() for n in nodes]
                data["class"][class_name]["relations"] = [l.to_dict() for l in links]

        return data

    def test_send_to_history(self):
        """Test the send_to_history method"""

        galaxy_connector = GalaxyConnector(self.settings, self.request.session, self.galaxy['url'], self.galaxy['key'])

        param_manager = ParamManager(self.settings, self.request.session)
        src_file = param_manager.get_upload_directory()
        filepath = src_file + 'play_instrument.tsv'

        galaxy_connector.send_to_history(filepath, 'play_instrument.tsv', 'tabular')

        assert self.interface_galaxy.check_dataset_presence('play_instrument.tsv') is True

    def __init__(self, settings, session,
                 name = None, endpoint = None,
                 username=None, password=None,
                 urlupdate=None):
        ParamManager.__init__(self, settings, session)
        self.log = logging.getLogger(__name__)

        self.name = name
        self.endpoint = endpoint
        self.username = username
        self.password = password
        self.urlupdate = urlupdate
        self.allowUpdate = False

    def __init__(self, settings, session, name = None, endpoint = None ,username=None, password=None,urlupdate=None,auth='Basic'):
        ParamManager.__init__(self, settings, session)
        self.log = logging.getLogger(__name__)

        self.name = name
        self.endpoint = endpoint
        self.username = username
        self.password = password
        self.urlupdate = urlupdate
        self.auth = auth
        self.allowUpdate = False

        if self.auth != 'Basic' and self.auth != 'Digest':
            raise ValueError("Invalid Auth parameter :"+self.auth)

    def __init__(self, settings, session):
        '''
            Manage Modules Askomics
        '''
        ParamManager.__init__(self, settings, session)

        '''
            All modules have to be composed with thes keys
        '''
        self.latt = ['module','comment','version','owl','rdf']

        self.moduleFiles = {}
        self.graph_modules="askomics:graph:module"
        self.modulesdir='askomics/static/modules/'
        self.data = {}

    def saveStartSparqlJob(self,typeJob,requestGraph="{}",variates="{}"):

        conn = sqlite3.connect(self.pathdb,uri=True)
        c = conn.cursor()

        requestGraph = urllib.parse.quote(requestGraph)
        variates = ParamManager.encode(str(variates))

        reqSql = "INSERT INTO jobs VALUES ("\
                + "NULL,"     \
                +"'"+typeJob+"'," \
                + "'Wait',"     \
                + "strftime('%s','now'),"\
                + "0,"\
                + "NULL,"\
                + "''," \
                + "''," \
                + "'"+requestGraph+"'," \
                + "'"+variates+"'," \
                + "-1" \
                + ");"

        c.execute(reqSql)
        ID = c.lastrowid

        conn.commit()
        conn.close()
        return ID

    def updateEndSparqlJob(self,jobid,state,nr=-1, data=None, file=None):
        import json

        conn = sqlite3.connect(self.pathdb,uri=True)
        c = conn.cursor()

        d = 'NULL'

        if data:
            d = "'"+ParamManager.encode(json.dumps(data, ensure_ascii=False))+"'"

        f = 'NULL'
        if file:
            f = "'"+file+"'"

        reqSql = "UPDATE jobs SET "\
                + " state = '"+ state +"'," \
                + " end = strftime('%s','now'),"\
                + " nr = "+str(nr)+","\
                + " data = "+ d +"," \
                + " file = "+ f \
                + " WHERE jobID = "+str(jobid)

        c.execute(reqSql)
        conn.commit()
        conn.close()

    def __init__(self, settings, session, type_dict=None, delims=None):
        #FIXME: Can we get dict()s from config ?
        self.type_dict = {
            'Numeric' : 'xsd:decimal',
            'Text'    : 'xsd:string',
            'Category': ':',
            'Entity'  : ':'} if type_dict is None else type_dict

        self.delims = {
            'Numeric' : ('', ''),
            'Text'    : ('"', '"'),
            'Category': (':', ''),
            'Entity'  : (':', '')} if delims is None else delims

        ParamManager.__init__(self, settings, session)
        self.log = logging.getLogger(__name__)

    def source_file_overview(self):
        """ Get the first lines of the tabulated files to convert """
        data = {}
        sfc = SourceFileConvertor(self.settings, self.request.session)

        pm = ParamManager(self.settings, self.request.session)

        source_files_first_lines = sfc.get_first_lines(int(self.settings["askomics.overview_lines_limit"])) # FIXME handle default value/value validation
        html_template = sfc.get_template(pm.ASKOMICS_html_template) # FIXME there must be a more elegant solution
        #turtle_template = sfc.get_template(self.settings["askomics.turtle_template"]) # FIXME there must be a more elegant solution

        turtle_template = pm.turtle_template()
        data["sourceFiles"] = {s.get_name(): s.to_dict() for s in source_files_first_lines}
        data["html_template"] = html_template
        data["turtle_template"] = turtle_template

        return data

    def __init__(self, settings, session):
        ParamManager.__init__(self, settings, session)

        self.log = logging.getLogger(__name__)

        self.ldap_server = self.settings['askomics.ldap_host']
        self.ldap_port = self.settings['askomics.ldap_port']
        # self.ldap_bind_dn = self.settings['askomics.ldap_bind_dn']
        # self.ldap_bind_passwd = self.settings['askomics.ldap_bind_passwd']
        self.ldap_user_search_base = self.settings['askomics.ldap_user_search_base']
        self.ldap_user_filter = self.settings['askomics.ldap_user_filter']
        self.ldap_username_attr = self.settings['askomics.ldap_username_attr']
        self.ldap_email_attr = self.settings['askomics.ldap_email_attr']

        self.username = None
        self.password = None
        self.email = None

    def __init__(self, settings, session, username, email, password, password2):
        ParamManager.__init__(self, settings, session)

        self.log = logging.getLogger(__name__)
        self.username = str(username)
        self.email = str(email)
        self.passwd = str(password)
        self.passwd2 = str(password2)
        self.admin = False
        self.blocked = True
        self.galaxy = False

        # concatenate askmics salt, password and random salt and hash it with sha256 function
        # see --"https://en.wikipedia.org/wiki/Salt_(cryptography)"-- for more info about salt
        alpabet = "0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ"
        self.randomsalt = ''.join(random.choice(alpabet) for i in range(20))
        salted_pw = self.settings["askomics.salt"] + self.passwd + self.randomsalt
        self.sha256_pw = hashlib.sha256(salted_pw.encode('utf8')).hexdigest()

    def __init__(self, settings, session, path, uri_set=None):

        ParamManager.__init__(self, settings, session)

        self.timestamp = datetime.datetime.now().isoformat()

        self.path = path

        self.metadatas = {}

        # The name should not contain extension as dots are not allowed in rdf names
        # self.name = os.path.splitext(os.path.basename(path))[0]
        self.name = os.path.basename(path)
        self.alphanum_name = re.sub('[^0-9a-zA-Z]+', '_', self.name)

        self.graph = 'askomics:unkown:uri:graph'
        if 'graph' in self.session:
            self.graph = self.session['graph']

        # Graph name can't contain any non alphanumeric characters. replace all with _
        self.graph = self.graph + ':' + self.alphanum_name + '_' + self.timestamp

        self.log = logging.getLogger(__name__)

        self.reset_cache()

        self.uri = {}
        pref_uri = "http://semanticweb.org/askomics/entity/"
        if self.is_defined('askomics.prefix'):
            pref_uri = self.get_param("askomics.prefix")

        self.uri = [ pref_uri for idx in range(20) ]

        if uri_set != None:
            for idx,uri in uri_set.items():
                if uri:
                    # uri have to end with # or /
                    if not uri.endswith('#') and not uri.endswith('/'):
                        uri = uri + "/"
                    self.uri[int(idx)] = uri
                else:
                    self.uri[int(idx)] = self.get_param("askomics.prefix")