<s> """<STR_LIT>""" <EOL> from __future__ import division <EOL> import os <EOL> import tempfile <EOL> import unittest <EOL> from vistrails . core . utils import VistrailsInternalError , abstract <EOL> from vistrails . core . utils . enum import enum <EOL> from vistrails . core . utils . uxml import named_elements , XMLWrapper <EOL> ConnectionType = enum ( '<STR_LIT>' , <EOL> [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] , <EOL> "<STR_LIT>" ) <EOL> class ExtConnection ( object ) : <EOL> """<STR_LIT>""" <EOL> parseDispatch = { } <EOL> def __init__ ( self , id = - <NUM_LIT:1> , name = '<STR_LIT>' , host = '<STR_LIT>' , type = ConnectionType . Unknown ) : <EOL> """<STR_LIT>""" <EOL> self . id = id <EOL> self . name = name <EOL> self . host = host <EOL> self . type = type <EOL> def serialize ( self , dom , element ) : <EOL> abstract ( ) <EOL> @ staticmethod <EOL> def parse ( element ) : <EOL> type = str ( element . getAttribute ( '<STR_LIT:type>' ) ) <EOL> return ExtConnection . parseDispatch [ type ] ( element ) <EOL> class DBConnection ( ExtConnection ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , id = - <NUM_LIT:1> , name = '<STR_LIT>' , host = '<STR_LIT>' , port = <NUM_LIT:0> , user = '<STR_LIT>' , passwd = '<STR_LIT>' , <EOL> database = '<STR_LIT>' , dbtype = '<STR_LIT>' ) : <EOL> """<STR_LIT>""" <EOL> ExtConnection . __init__ ( self , id , name , host , ConnectionType . DB ) <EOL> self . port = port <EOL> self . user = user <EOL> self . passwd = passwd <EOL> self . database = database <EOL> self . dbtype = dbtype <EOL> def serialize ( self , dom , element ) : <EOL> """<STR_LIT>""" <EOL> conn = dom . createElement ( '<STR_LIT>' ) <EOL> conn . setAttribute ( '<STR_LIT:id>' , str ( self . id ) ) <EOL> conn . setAttribute ( '<STR_LIT:name>' , str ( self . name ) ) <EOL> conn . setAttribute ( '<STR_LIT:host>' , str ( self . host ) ) <EOL> conn . setAttribute ( '<STR_LIT:port>' , str ( self . port ) ) <EOL> conn . setAttribute ( '<STR_LIT:user>' , str ( self . user ) ) <EOL> conn . setAttribute ( '<STR_LIT>' , str ( self . passwd ) ) <EOL> conn . setAttribute ( '<STR_LIT>' , str ( self . database ) ) <EOL> conn . setAttribute ( '<STR_LIT:type>' , str ( self . type ) ) <EOL> conn . setAttribute ( '<STR_LIT>' , str ( self . dbtype ) ) <EOL> element . appendChild ( conn ) <EOL> def __str__ ( self ) : <EOL> """<STR_LIT>""" <EOL> return """<STR_LIT>""" % ( <EOL> self . id , <EOL> self . name , <EOL> self . type , <EOL> self . host , <EOL> self . user , <EOL> self . database , <EOL> self . dbtype ) <EOL> @ staticmethod <EOL> def parse ( element ) : <EOL> """<STR_LIT>""" <EOL> conn = DBConnection ( ) <EOL> conn . type = ConnectionType . DB <EOL> conn . id = int ( element . getAttribute ( '<STR_LIT:id>' ) ) <EOL> conn . name = str ( element . getAttribute ( '<STR_LIT:name>' ) ) <EOL> conn . host = str ( element . getAttribute ( '<STR_LIT:host>' ) ) <EOL> conn . port = int ( element . getAttribute ( '<STR_LIT:port>' ) ) <EOL> conn . user = str ( element . getAttribute ( '<STR_LIT:user>' ) ) <EOL> conn . passwd = str ( element . getAttribute ( '<STR_LIT>' ) ) <EOL> conn . database = str ( element . getAttribute ( '<STR_LIT>' ) ) <EOL> conn . dbtype = str ( element . getAttribute ( '<STR_LIT>' ) ) <EOL> return conn <EOL> def __eq__ ( self , other ) : <EOL> """<STR_LIT>""" <EOL> if other is None : <EOL> return False <EOL> if self . type != other . type : <EOL> return False <EOL> if self . id != other . id : <EOL> return False <EOL> if self . name != other . name : <EOL> return False <EOL> if self . host != other . host : <EOL> return False <EOL> if self . port != other . port : <EOL> return False <EOL> if self . user != other . user : <EOL> return False <EOL> if self . passwd != other . passwd : <EOL> return False <EOL> if self . database != other . database : <EOL> return False <EOL> if self . dbtype != other . dbtype : <EOL> return False <EOL> return True <EOL> def __ne__ ( self , other ) : <EOL> return not ( self == other ) <EOL> ExtConnection . parseDispatch [ str ( ConnectionType . DB ) ] = DBConnection . parse <EOL> class ExtConnectionList ( XMLWrapper ) : <EOL> """<STR_LIT>""" <EOL> _instance = None <EOL> @ staticmethod <EOL> def getInstance ( * args , ** kwargs ) : <EOL> if ExtConnectionList . _instance is None : <EOL> obj = ExtConnectionList ( * args , ** kwargs ) <EOL> ExtConnectionList . _instance = obj <EOL> return ExtConnectionList . _instance <EOL> def __init__ ( self , filename = '<STR_LIT>' ) : <EOL> """<STR_LIT>""" <EOL> if not ExtConnectionList . _instance : <EOL> self . __connections = { } <EOL> self . changed = False <EOL> self . current_id = <NUM_LIT:1> <EOL> self . filename = filename <EOL> self . load_connections ( ) <EOL> ExtConnectionList . _instance = self <EOL> else : <EOL> raise RuntimeError , '<STR_LIT>' <EOL> def load_connections ( self ) : <EOL> """<STR_LIT>""" <EOL> if os . path . exists ( self . filename ) : <EOL> self . parse ( self . filename ) <EOL> def add_connection ( self , conn ) : <EOL> """<STR_LIT>""" <EOL> if self . __connections . has_key ( conn . id ) : <EOL> msg = "<STR_LIT>" % conn . name <EOL> raise VistrailsInternalError ( msg ) <EOL> self . __connections [ conn . id ] = conn <EOL> self . current_id = max ( self . current_id , conn . id + <NUM_LIT:1> ) <EOL> self . serialize ( ) <EOL> def get_connection ( self , id ) : <EOL> """<STR_LIT>""" <EOL> if self . __connections . has_key ( id ) : <EOL> return self . __connections [ id ] <EOL> else : <EOL> return None <EOL> def has_connection ( self , id ) : <EOL> """<STR_LIT>""" <EOL> return self . __connections . has_key ( id ) <EOL> def find_db_connection ( self , host , port , db ) : <EOL> """<STR_LIT>""" <EOL> for conn in self . __connections . itervalues ( ) : <EOL> if conn . host == host and conn . port == port and conn . database == db : <EOL> return conn . id <EOL> return - <NUM_LIT:1> <EOL> def set_connection ( self , id , conn ) : <EOL> """<STR_LIT>""" <EOL> if self . __connections . has_key ( id ) : <EOL> self . __connections [ id ] = conn <EOL> self . serialize ( ) <EOL> def remove_connection ( self , id ) : <EOL> """<STR_LIT>""" <EOL> if self . __connections . has_key ( id ) : <EOL> del self . __connections [ id ] <EOL> self . serialize ( ) <EOL> def clear ( self ) : <EOL> """<STR_LIT>""" <EOL> self . __connections . clear ( ) <EOL> self . current_id = <NUM_LIT:1> <EOL> def count ( self ) : <EOL> """<STR_LIT>""" <EOL> return len ( self . __connections ) <EOL> def items ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . __connections . items ( ) <EOL> def parse ( self , filename ) : <EOL> """<STR_LIT>""" <EOL> self . open_file ( filename ) <EOL> root = self . dom . documentElement <EOL> for element in named_elements ( root , '<STR_LIT>' ) : <EOL> self . add_connection ( ExtConnection . parse ( element ) ) <EOL> self . refresh_current_id ( ) <EOL> def serialize ( self ) : <EOL> """<STR_LIT>""" <EOL> dom = self . create_document ( '<STR_LIT>' ) <EOL> root = dom . documentElement <EOL> for conn in self . __connections . values ( ) : <EOL> conn . serialize ( dom , root ) <EOL> self . write_document ( root , self . filename ) <EOL> def refresh_current_id ( self ) : <EOL> """<STR_LIT>""" <EOL> self . current_id = max ( [ <NUM_LIT:0> ] + self . __connections . keys ( ) ) + <NUM_LIT:1> <EOL> def get_fresh_id ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . current_id <EOL> class TestConnectionList ( unittest . TestCase ) : <EOL> def test1 ( self ) : <EOL> """<STR_LIT>""" <EOL> fd , filename = tempfile . mkstemp ( prefix = '<STR_LIT>' , suffix = '<STR_LIT>' ) <EOL> os . close ( fd ) <EOL> try : <EOL> conns = ExtConnectionList . getInstance ( ) <EOL> conns . filename = filename <EOL> conns . clear ( ) <EOL> conn = DBConnection ( ) <EOL> conn . id = <NUM_LIT:1> <EOL> conn . name = '<STR_LIT:test>' <EOL> conn . host = '<STR_LIT>' <EOL> conn . port = <NUM_LIT> <EOL> conn . user = '<STR_LIT>' <EOL> conn . passwd = '<STR_LIT>' <EOL> conn . database = '<STR_LIT>' <EOL> conn . dbtype = '<STR_LIT>' <EOL> conns . add_connection ( conn ) <EOL> conns . clear ( ) <EOL> self . assertEquals ( conns . count ( ) , <NUM_LIT:0> ) <EOL> conns . load_connections ( ) <EOL> self . assertEquals ( conns . count ( ) , <NUM_LIT:1> ) <EOL> newconn = conns . get_connection ( <NUM_LIT:1> ) <EOL> self . assertEqual ( conn , newconn ) <EOL> finally : <EOL> os . unlink ( filename ) <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> unittest . main ( ) </s>
