<commit_before>import copy
import time
import subprocess
import bliss.saga
import paramiko
import os 
class SSHJobProcess(object):
    '''A wrapper around an SSH process'''
    def __init__(self, jobdescription,  plugin, service_object):
        self.pi = plugin

        #check for things we dont support
        if jobdescription.file_transfer     != None: 
            self.pi.log_error_and_raise(bliss.saga.Error.NotImplemented,
                                        "SSH Job adaptor doesn't support the file transfer attribute!") 

        if jobdescription.project     != None: 
            self.pi.log_error_and_raise(bliss.saga.Error.NotImplemented,
                                        "SSH Job adaptor doesn't support the project attribute!") 
        
        if jobdescription.queue     != None: 
            self.pi.log_error_and_raise(bliss.saga.Error.NotImplemented,
                                        "SSH Job adaptor doesn't support the queue attribute!") 

        if jobdescription.wall_time_limit     != None: 
            self.pi.log_error_and_raise(bliss.saga.Error.NotImplemented,
                                        "SSH Job adaptor doesn't support the wall_time_limit attribute!") 
              
        if jobdescription.contact     != None: 
            self.pi.log_error_and_raise(bliss.saga.Error.NotImplemented,
                                        "SSH Job adaptor doesn't support the contact attribute!")

        if jobdescription.total_cpu_count     != None: 
            self.pi.log_error_and_raise(bliss.saga.Error.NotImplemented,
                                        "SSH Job adaptor doesn't support the total_cpu_count attribute!")

        if jobdescription.number_of_processes     != None: 
            self.pi.log_error_and_raise(bliss.saga.Error.NotImplemented,
                                        "SSH Job adaptor doesn't support the number_of_processes attribute!")

        if jobdescription.spmd_variation     != None: 
            self.pi.log_error_and_raise(bliss.saga.Error.NotImplemented,
                                        "SSH Job adaptor doesn't support the spmd_variation attribute!")    

        self.executable  = jobdescription.executable
        self.arguments   = jobdescription.arguments
        self.environment = jobdescription.environment
        self.working_directory=jobdescription.working_directory
        self.so = service_object
        self.sshclient = None
        self.sshchannel = None
        self.pid = None
        self.returncode = None
        self.state = bliss.saga.job.Job.New
        self._job_output=None
        self._job_error=None

    def __del__(self):
        if self._job_output is not None:
            self._job_output.close()
        if self._job_error is not None:
            self._job_error.close()
        if self.sshchannel is not None:
            self.pi.log_debug("Closing SSH channel")   
            self.sshchannel.close()
        if self.sshclient is not None:
            self.pi.log_debug("Closing SSH client")
            self.sshclient.close()


    def run(self, jd, url):
        #check if there are things in the job descriptor that we don't support

        #load up ssh config file
        self.config = paramiko.SSHConfig()
        config_file = os.path.expanduser(os.path.join("~", ".ssh", "config"))
        self.pi.log_info("Loading SSH configuration file: %s" % config_file)
        self.config.parse(open(config_file))

        usable_ctx = None

        #grab an SSH context if one exists
        for ctx in self.so.session.contexts:
            if ctx.type is bliss.saga.Context.SSH:
                usable_ctx = ctx
                self.pi.log_debug("Found SSH context to use!")
                break

        #look for a username and key to use in the context
        username = None
        userkey = None
        if usable_ctx is not None:
            if usable_ctx.userid is not None:
                username = usable_ctx.userid
            if usable_ctx.userkey is not None:
                userkey = usable_ctx.userkey

        #overwrite the context username/password with our url-provided username/password
        #(if they exist)
        if url.username:
            username = url.username
        if url.password:
            password = <PASSWORD>

        #set missing host key policy to automatically add it
        self.sshclient=paramiko.SSHClient()
        self.sshclient.set_missing_host_key_policy(paramiko.AutoAddPolicy())

        #have paramiko load host keys from .ssh directory
        self.sshclient.load_host_keys(os.path.expanduser(os.path.join("~", ".ssh", "known_hosts")))    
          
        #if the context provides a userkey, let us know that we're using it
        if not userkey:
            self.pi.log_info("Using default ssh key")    
        else:
            self.pi.log_info("Using context-provided ssh key")

        #if the context provided a username, let us know that we're using it
        if not username:
            self.pi.log_info("Using default username")
        else:
            if url.username:
                self.pi.log_info("Using username from URL")
            else:
                self.pi.log_info("Using context-provided username")

        hn = url.host

        #see if we have any information on the host from the ssh config
        # that must be loaded
        try:
            temp = self.config.lookup(hn)['hostname']
            self.pi.log_debug("Translating provided hostname %s to %s" % (hn, temp))
            hn = temp
        except Exception, ex:
            self.pi.log_debug("No hostname lookup for %s" % hn)

        self.pi.log_info("Connecting to host %s" % hn)

        #set up our ssh channel
        self.sshclient.connect(hn, username=username, key_filename=userkey)
        self.sshchannel = self.sshclient.get_transport().open_session()

        #parse environment variables
        envline="env "
        for k in self.environment.keys():
            envline += k+"="+self.environment[k]+" "
        envline += " /bin/sh -c "
        
        #set up our commandline
        cmdline = ""

        #do we have a working directory to cd into?
        if(self.working_directory):
            cmdline+="cd "+self.working_directory + " && "
        cmdline+= str(self.executable)

        #make a list of arguments
        args = ""
        if self.arguments is not None:
            for arg in self.arguments:
                cmdline += " %s" % arg <commit_msg>SSH adaptor:  doesn't fail when there is no ~/.ssh/config file<commit_after>import copy
import time
import subprocess
import bliss.saga
import paramiko
import os 
class SSHJobProcess(object):
    '''A wrapper around an SSH process'''
    def __init__(self, jobdescription,  plugin, service_object):
        self.pi = plugin

        #check for things we dont support
        if jobdescription.file_transfer     != None: 
            self.pi.log_error_and_raise(bliss.saga.Error.NotImplemented,
                                        "SSH Job adaptor doesn't support the file transfer attribute!") 

        if jobdescription.project     != None: 
            self.pi.log_error_and_raise(bliss.saga.Error.NotImplemented,
                                        "SSH Job adaptor doesn't support the project attribute!") 
        
        if jobdescription.queue     != None: 
            self.pi.log_error_and_raise(bliss.saga.Error.NotImplemented,
                                        "SSH Job adaptor doesn't support the queue attribute!") 

        if jobdescription.wall_time_limit     != None: 
            self.pi.log_error_and_raise(bliss.saga.Error.NotImplemented,
                                        "SSH Job adaptor doesn't support the wall_time_limit attribute!") 
              
        if jobdescription.contact     != None: 
            self.pi.log_error_and_raise(bliss.saga.Error.NotImplemented,
                                        "SSH Job adaptor doesn't support the contact attribute!")

        if jobdescription.total_cpu_count     != None: 
            self.pi.log_error_and_raise(bliss.saga.Error.NotImplemented,
                                        "SSH Job adaptor doesn't support the total_cpu_count attribute!")

        if jobdescription.number_of_processes     != None: 
            self.pi.log_error_and_raise(bliss.saga.Error.NotImplemented,
                                        "SSH Job adaptor doesn't support the number_of_processes attribute!")

        if jobdescription.spmd_variation     != None: 
            self.pi.log_error_and_raise(bliss.saga.Error.NotImplemented,
                                        "SSH Job adaptor doesn't support the spmd_variation attribute!")    

        self.executable  = jobdescription.executable
        self.arguments   = jobdescription.arguments
        self.environment = jobdescription.environment
        self.working_directory=jobdescription.working_directory
        self.so = service_object
        self.sshclient = None
        self.sshchannel = None
        self.pid = None
        self.returncode = None
        self.state = bliss.saga.job.Job.New
        self._job_output=None
        self._job_error=None

    def __del__(self):
        if self._job_output is not None:
            self._job_output.close()
        if self._job_error is not None:
            self._job_error.close()
        if self.sshchannel is not None:
            self.pi.log_debug("Closing SSH channel")   
            self.sshchannel.close()
        if self.sshclient is not None:
            self.pi.log_debug("Closing SSH client")
            self.sshclient.close()


    def run(self, jd, url):
        #check if there are things in the job descriptor that we don't support

        #load up ssh config file
        self.config = paramiko.SSHConfig()
        try:
            config_file = os.path.expanduser(os.path.join("~", ".ssh", "config"))
            self.pi.log_info("Attempting to load SSH configuration file: %s" % config_file)
            self.config.parse(open(config_file))
        except:
            self.pi.log_info("Couldn't open SSH configuration file: %s" % config_file)

        #see if we have any information on the host from the ssh config
        # that must be loaded
        hn = url.host
        try:
            temp = self.config.lookup(hn)['hostname']
            self.pi.log_debug("Translating provided hostname %s to %s" % (hn, temp))
            hn = temp
        except Exception, ex:
            self.pi.log_debug("No hostname lookup for %s" % hn)


        usable_ctx = None

        #grab an SSH context if one exists
        for ctx in self.so.session.contexts:
            if ctx.type is bliss.saga.Context.SSH:
                usable_ctx = ctx
                self.pi.log_debug("Found SSH context to use!")
                break

        #look for a username and key to use in the context
        username = None
        userkey = None
        if usable_ctx is not None:
            if usable_ctx.userid is not None:
                username = usable_ctx.userid
            if usable_ctx.userkey is not None:
                userkey = usable_ctx.userkey

        #overwrite the context username/password with our url-provided username/password
        #(if they exist)
        if url.username:
            username = url.username
        if url.password:
            password = url.password

        #set missing host key policy to automatically add it
        self.sshclient=paramiko.SSHClient()
        self.sshclient.set_missing_host_key_policy(paramiko.AutoAddPolicy())

        #have paramiko load host keys from .ssh directory
        self.sshclient.load_host_keys(os.path.expanduser(os.path.join("~", ".ssh", "known_hosts")))    
          
        #if the context provides a userkey, let us know that we're using it
        if not userkey:
            self.pi.log_info("Using default ssh key")    
        else:
            self.pi.log_info("Using context-provided ssh key")

        #if the context provided a username, let us know that we're using it
        if not username:
            self.pi.log_info("Using default username")
        else:
            if url.username:
                self.pi.log_info("Using username from URL")
            else:
                self.pi.log_info("Using context-provided username")


        self.pi.log_info("Connecting to host %s" % hn)

        #set up our ssh channel
        self.sshclient.connect(hn, username=username, key_filename=userkey)
        self.sshchannel = self.sshclient.get_transport().open_session()

        #parse environment variables
        envline="env "
        for k in self.environment.keys():
            envline += k+"="+self.environment[k]+" "
        envline += " /bin/sh -c "
        
        #set up our commandline
        cmdline = ""

        #do we have a working directory to cd into?
        if(self.working_directory):
            cmdline+="cd "+self.working_directory + " && "
        cmdline+= str(self.executable)

        #make a list of arguments
        args = ""
        if self.arguments is not None:
            for arg in self.arguments:
                cmdline += " %s" % arg 
{'fraction_non_alphanumeric': 0.0814085573646346, 'fraction_numerical': 0.0, 'mean_word_length': 1.8646420824295011, 'pattern_counts': {'":': 0, '<': 4, '<?xml version=': 0, '>': 4, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 32, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '941459', 'n_tokens_mistral': 3678, 'n_tokens_neox': 3477, 'n_words': 978}