files: # Licensed to the Apache Software Foundation (ASF) under one or more
  # contributor license agreements.  See the NOTICE file distributed with
  # this work for additional information regarding copyright ownership.
  # The ASF licenses this file to You under the Apache License, Version 2.0
  # (the "License"); you may not use this file except in compliance with
  # the License.  You may obtain a copy of the License at
  #
  #     http://www.apache.org/licenses/LICENSE-2.0
  #
  # Unless required by applicable law or agreed to in writing, software
  # distributed under the License is distributed on an "AS IS" BASIS,
  # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  # See the License for the specific language governing permissions and
  # limitations under the License.
  
  """
  Provides generic deployment steps for machines post boot.
  """
  
  from __future__ import with_statement
  
  from typing import Union
  from typing import Optional
  from typing import List
  from typing import IO
  from typing import cast
  
  import os
  import binascii
  
  from libcloud.utils.py3 import basestring, PY3
  from libcloud.compute.base import Node
  from libcloud.compute.ssh import BaseSSHClient
  
  
  class Deployment(object):
      """
      Base class for deployment tasks.
      """
  
      def run(self, node, client):
          # type: (Node, BaseSSHClient) -> Node
          """
          Runs this deployment task on node using the client provided.
  
          :type node: :class:`Node`
          :keyword node: Node to operate one
  
          :type client: :class:`BaseSSHClient`
          :keyword client: Connected SSH client to use.
  
          :return: :class:`Node`
          """
          raise NotImplementedError(
              'run not implemented for this deployment')
  
      def _get_string_value(self, argument_name, argument_value):
          if not isinstance(argument_value, basestring) and \
             not hasattr(argument_value, 'read'):
              raise TypeError('%s argument must be a string or a file-like '
                              'object' % (argument_name))
  
          if hasattr(argument_value, 'read'):
              argument_value = argument_value.read()
  
          return argument_value
  
  
  class SSHKeyDeployment(Deployment):
      """
      Installs a public SSH Key onto a server.
      """
  
      def __init__(self, key):
          # type: (Union[str, IO]) -> None
          """
          :type key: ``str`` or :class:`File` object
          :keyword key: Contents of the public key write or a file object which
                        can be read.
          """
          self.key = self._get_string_value(argument_name='key',
                                            argument_value=key)
  
      def run(self, node, client):
          # type: (Node, BaseSSHClient) -> Node
          """
          Installs SSH key into ``.ssh/authorized_keys``
  
          See also :class:`Deployment.run`
          """
          client.put(".ssh/authorized_keys", contents=self.key, mode='a')
          return node
  
  
  class FileDeployment(Deployment):
      """
      Installs a file on the server.
      """
  
      def __init__(self, source, target):
          # type: (str, str) -> None
          """
          :type source: ``str``
          :keyword source: Local path of file to be installed
  
          :type target: ``str``
          :keyword target: Path to install file on node
          """
          self.source = source
          self.target = target
  
      def run(self, node, client):
          # type: (Node, BaseSSHClient) -> Node
          """
          Upload the file, retaining permissions.
  
          See also :class:`Deployment.run`
          """
          perms = int(oct(os.stat(self.source).st_mode)[4:], 8)
  
          with open(self.source, 'rb') as fp:
              content = fp.read()
  
          client.put(path=self.target, chmod=perms,
                     contents=content)
          return node
  
  
  class ScriptDeployment(Deployment):
      """
      Runs an arbitrary shell script on the server.
  
      This step works by first writing the content of the shell script (script
      argument) in a *.sh file on a remote server and then running that file.
  
      If you are running a non-shell script, make sure to put the appropriate
      shebang to the top of the script. You are also advised to do that even if
      you are running a plan shell script.
      """
  
      def __init__(self,
                   script,  # type: str
                   args=None,  # type: Optional[List[str]]
                   name=None,  # type: Optional[str]
                   delete=False,  # type bool
                   timeout=None  # type: Optional[float]
                   ):
          # type: (...) -> None
          """
          :type script: ``str``
          :keyword script: Contents of the script to run.
  
          :type args: ``list``
          :keyword args: Optional command line arguments which get passed to the
                         deployment script file.
  
          :type name: ``str``
          :keyword name: Name of the script to upload it as, if not specified,
                         a random name will be chosen.
  
          :type delete: ``bool``
          :keyword delete: Whether to delete the script on completion.
  
          :param timeout: Optional run timeout for this command.
          :type timeout: ``float``
          """
          script = self._get_string_value(argument_name='script',
                                          argument_value=script)
  
          self.script = script
          self.args = args or []
          self.stdout = None  # type: Optional[str]
          self.stderr = None  # type: Optional[str]
          self.exit_status = None  # type: Optional[int]
          self.delete = delete
          self.timeout = timeout
          self.name = name  # type: Optional[str]
  
          if self.name is None:
              # File is put under user's home directory
              # (~/libcloud_deployment_<random_string>.sh)
              random_string = ''  # type: Union[str, bytes]
              random_string = binascii.hexlify(os.urandom(4))
              random_string = cast(bytes,
  
  [... the rest is omitted ...]
files: libcloud/compute/deployment.py
files: # Licensed to the Apache Software Foundation (ASF) under one or more
  # contributor license agreements.  See the NOTICE file distributed with
  # this work for additional information regarding copyright ownership.
  # The ASF licenses this file to You under the Apache License, Version 2.0
  # (the "License"); you may not use this file except in compliance with
  # the License.  You may obtain a copy of the License at
  #
  #     http://www.apache.org/licenses/LICENSE-2.0
  #
  # Unless required by applicable law or agreed to in writing, software
  # distributed under the License is distributed on an "AS IS" BASIS,
  # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  # See the License for the specific language governing permissions and
  # limitations under the License.
  
  """
  Provides generic deployment steps for machines post boot.
  """
  
  from __future__ import with_statement
  
  from typing import Union
  from typing import Optional
  from typing import List
  from typing import IO
  from typing import cast
  
  import os
  import binascii
  
  from libcloud.utils.py3 import basestring, PY3
  from libcloud.compute.base import Node
  from libcloud.compute.ssh import BaseSSHClient
  
  
  class Deployment(object):
      """
      Base class for deployment tasks.
      """
  
      def run(self, node, client):
          # type: (Node, BaseSSHClient) -> Node
          """
          Runs this deployment task on node using the client provided.
  
          :type node: :class:`Node`
          :keyword node: Node to operate one
  
          :type client: :class:`BaseSSHClient`
          :keyword client: Connected SSH client to use.
  
          :return: :class:`Node`
          """
          raise NotImplementedError(
              'run not implemented for this deployment')
  
      def _get_string_value(self, argument_name, argument_value):
          if not isinstance(argument_value, basestring) and \
             not hasattr(argument_value, 'read'):
              raise TypeError('%s argument must be a string or a file-like '
                              'object' % (argument_name))
  
          if hasattr(argument_value, 'read'):
              argument_value = argument_value.read()
  
          return argument_value
  
  
  class SSHKeyDeployment(Deployment):
      """
      Installs a public SSH Key onto a server.
      """
  
      def __init__(self, key):
          # type: (Union[str, IO]) -> None
          """
          :type key: ``str`` or :class:`File` object
          :keyword key: Contents of the public key write or a file object which
                        can be read.
          """
          self.key = self._get_string_value(argument_name='key',
                                            argument_value=key)
  
      def run(self, node, client):
          # type: (Node, BaseSSHClient) -> Node
          """
          Installs SSH key into ``.ssh/authorized_keys``
  
          See also :class:`Deployment.run`
          """
          client.put(".ssh/authorized_keys", contents=self.key, mode='a')
          return node
  
  
  class FileDeployment(Deployment):
      """
      Installs a file on the server.
      """
  
      def __init__(self, source, target):
          # type: (str, str) -> None
          """
          :type source: ``str``
          :keyword source: Local path of file to be installed
  
          :type target: ``str``
          :keyword target: Path to install file on node
          """
          self.source = source
          self.target = target
  
      def run(self, node, client):
          # type: (Node, BaseSSHClient) -> Node
          """
          Upload the file, retaining permissions.
  
          See also :class:`Deployment.run`
          """
          perms = int(oct(os.stat(self.source).st_mode)[4:], 8)
  
          with open(self.source, 'rb') as fp:
              content = fp.read()
  
          client.put(path=self.target, chmod=perms,
                     contents=content)
          return node
  
  
  class ScriptDeployment(Deployment):
      """
      Runs an arbitrary shell script on the server.
  
      This step works by first writing the content of the shell script (script
      argument) in a *.sh file on a remote server and then running that file.
  
      If you are running a non-shell script, make sure to put the appropriate
      shebang to the top of the script. You are also advised to do that even if
      you are running a plan shell script.
      """
  
      def __init__(self, script, args=None, name=None, delete=False):
          # type: (str, Optional[List[str]], Optional[str], bool) -> None
          """
          :type script: ``str``
          :keyword script: Contents of the script to run.
  
          :type args: ``list``
          :keyword args: Optional command line arguments which get passed to the
                         deployment script file.
  
          :type name: ``str``
          :keyword name: Name of the script to upload it as, if not specified,
                         a random name will be chosen.
  
          :type delete: ``bool``
          :keyword delete: Whether to delete the script on completion.
          """
          script = self._get_string_value(argument_name='script',
                                          argument_value=script)
  
          self.script = script
          self.args = args or []
          self.stdout = None  # type: Optional[str]
          self.stderr = None  # type: Optional[str]
          self.exit_status = None  # type: Optional[int]
          self.delete = delete
          self.name = name  # type: Optional[str]
  
          if self.name is None:
              # File is put under user's home directory
              # (~/libcloud_deployment_<random_string>.sh)
              random_string = ''  # type: Union[str, bytes]
              random_string = binascii.hexlify(os.urandom(4))
              random_string = cast(bytes, random_string)
              random_string = random_string.decode('ascii')
              self.name = 'libcloud_deployment_%s.sh' % (random_string)
  
      def run(self, node, client):
          # type: (Node, BaseSSHClient) -> Node
          """
         
  
  [... the rest is omitted ...]
files: libcloud/compute/deployment.py
files: # Licensed to the Apache Software Foundation (ASF) under one or more
  # contributor license agreements.  See the NOTICE file distributed with
  # this work for additional information regarding copyright ownership.
  # The ASF licenses this file to You under the Apache License, Version 2.0
  # (the "License"); you may not use this file except in compliance with
  # the License.  You may obtain a copy of the License at
  #
  #     http://www.apache.org/licenses/LICENSE-2.0
  #
  # Unless required by applicable law or agreed to in writing, software
  # distributed under the License is distributed on an "AS IS" BASIS,
  # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  # See the License for the specific language governing permissions and
  # limitations under the License.
  
  """
  Wraps multiple ways to communicate over SSH.
  """
  
  from typing import Type
  from typing import Optional
  from typing import Tuple
  from typing import List
  from typing import Union
  from typing import cast
  
  have_paramiko = False
  
  try:
      import paramiko
      have_paramiko = True
  except ImportError:
      pass
  
  # Depending on your version of Paramiko, it may cause a deprecation
  # warning on Python 2.6.
  # Ref: https://bugs.launchpad.net/paramiko/+bug/392973
  
  import os
  import time
  import subprocess
  import logging
  import warnings
  
  from os.path import split as psplit
  from os.path import join as pjoin
  
  from libcloud.utils.logging import ExtraLogFormatter
  from libcloud.utils.py3 import StringIO
  from libcloud.utils.py3 import b
  
  __all__ = [
      'BaseSSHClient',
      'ParamikoSSHClient',
      'ShellOutSSHClient',
  
      'SSHCommandTimeoutError'
  ]
  
  SUPPORTED_KEY_TYPES_URL = 'https://libcloud.readthedocs.io/en/latest/compute/deployment.html#supported-private-ssh-key-types' # NOQA
  
  
  class SSHCommandTimeoutError(Exception):
      """
      Exception which is raised when an SSH command times out.
      """
      def __init__(self, cmd, timeout):
          # type: (str, float) -> None
          self.cmd = cmd
          self.timeout = timeout
          self.message = 'Command didn\'t finish in %s seconds' % (timeout)
          super(SSHCommandTimeoutError, self).__init__(self.message)
  
      def __repr__(self):
          return ('<SSHCommandTimeoutError: cmd="%s",timeout=%s)>' %
                  (self.cmd, self.timeout))
  
      def __str__(self):
          return self.__repr__()
  
  
  class BaseSSHClient(object):
      """
      Base class representing a connection over SSH/SCP to a remote node.
      """
  
      def __init__(self,
                   hostname,  # type: str
                   port=22,  # type: int
                   username='root',  # type: str
                   password=None,  # type: Optional[str]
                   key=None,  # type: Optional[str]
                   key_files=None,  # type: Optional[Union[str, List[str]]]
                   timeout=None  # type: Optional[float]
                   ):
          """
          :type hostname: ``str``
          :keyword hostname: Hostname or IP address to connect to.
  
          :type port: ``int``
          :keyword port: TCP port to communicate on, defaults to 22.
  
          :type username: ``str``
          :keyword username: Username to use, defaults to root.
  
          :type password: ``str``
          :keyword password: Password to authenticate with or a password used
                             to unlock a private key if a password protected key
                             is used.
  
          :param key: Deprecated in favor of ``key_files`` argument.
  
          :type key_files: ``str`` or ``list``
          :keyword key_files: A list of paths to the private key files to use.
          """
          if key is not None:
              message = ('You are using deprecated "key" argument which has '
                         'been replaced with "key_files" argument')
              warnings.warn(message, DeprecationWarning)
  
              # key_files has precedent
              key_files = key if not key_files else key_files
  
          self.hostname = hostname
          self.port = port
          self.username = username
          self.password = password
          self.key_files = key_files
          self.timeout = timeout
  
      def connect(self):
          # type: () -> bool
          """
          Connect to the remote node over SSH.
  
          :return: True if the connection has been successfully established,
                   False otherwise.
          :rtype: ``bool``
          """
          raise NotImplementedError(
              'connect not implemented for this ssh client')
  
      def put(self, path, contents=None, chmod=None, mode='w'):
          # type: (str, Optional[Union[str, bytes]], Optional[int], str) -> str
          """
          Upload a file to the remote node.
  
          :type path: ``str``
          :keyword path: File path on the remote node.
  
          :type contents: ``str``
          :keyword contents: File Contents.
  
          :type chmod: ``int``
          :keyword chmod: chmod file to this after creation.
  
          :type mode: ``str``
          :keyword mode: Mode in which the file is opened.
  
          :return: Full path to the location where a file has been saved.
          :rtype: ``str``
          """
          raise NotImplementedError(
              'put not implemented for this ssh client')
  
      def delete(self, path):
          # type: (str) -> bool
          """
          Delete/Unlink a file on the remote node.
  
          :type path: ``str``
          :keyword path: File path on the remote node.
  
          :return: True if the file has been successfully deleted, False
                   otherwise.
          :rtype: ``bool``
          """
          raise NotImplementedError(
              'delete not implemented for this ssh client')
  
      def run(self, cmd, timeout=None):
          # type: (str, Optional[float]) -> Tuple[str, str, int]
          """
         
  
  [... the rest is omitted ...]
files: libcloud/compute/ssh.py
files: # Licensed to the Apache Software Foundation (ASF) under one or more
  # contributor license agreements.  See the NOTICE file distributed with
  # this work for additional information regarding copyright ownership.
  # The ASF licenses this file to You under the Apache License, Version 2.0
  # (the "License"); you may not use this file except in compliance with
  # the License.  You may obtain a copy of the License at
  #
  #     http://www.apache.org/licenses/LICENSE-2.0
  #
  # Unless required by applicable law or agreed to in writing, software
  # distributed under the License is distributed on an "AS IS" BASIS,
  # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  # See the License for the specific language governing permissions and
  # limitations under the License.
  
  """
  Wraps multiple ways to communicate over SSH.
  """
  
  from typing import Type
  from typing import Optional
  from typing import Tuple
  from typing import List
  from typing import Union
  from typing import cast
  
  have_paramiko = False
  
  try:
      import paramiko
      have_paramiko = True
  except ImportError:
      pass
  
  # Depending on your version of Paramiko, it may cause a deprecation
  # warning on Python 2.6.
  # Ref: https://bugs.launchpad.net/paramiko/+bug/392973
  
  import os
  import time
  import subprocess
  import logging
  import warnings
  
  from os.path import split as psplit
  from os.path import join as pjoin
  
  from libcloud.utils.logging import ExtraLogFormatter
  from libcloud.utils.py3 import StringIO
  from libcloud.utils.py3 import b
  
  __all__ = [
      'BaseSSHClient',
      'ParamikoSSHClient',
      'ShellOutSSHClient',
  
      'SSHCommandTimeoutError'
  ]
  
  SUPPORTED_KEY_TYPES_URL = 'https://libcloud.readthedocs.io/en/latest/compute/deployment.html#supported-private-ssh-key-types' # NOQA
  
  
  class SSHCommandTimeoutError(Exception):
      """
      Exception which is raised when an SSH command times out.
      """
      def __init__(self, cmd, timeout):
          # type: (str, float) -> None
          self.cmd = cmd
          self.timeout = timeout
          self.message = 'Command didn\'t finish in %s seconds' % (timeout)
          super(SSHCommandTimeoutError, self).__init__(self.message)
  
      def __repr__(self):
          return ('<SSHCommandTimeoutError: cmd="%s",timeout=%s)>' %
                  (self.cmd, self.timeout))
  
      def __str__(self):
          return self.__repr__()
  
  
  class BaseSSHClient(object):
      """
      Base class representing a connection over SSH/SCP to a remote node.
      """
  
      def __init__(self,
                   hostname,  # type: str
                   port=22,  # type: int
                   username='root',  # type: str
                   password=None,  # type: Optional[str]
                   key=None,  # type: Optional[str]
                   key_files=None,  # type: Optional[Union[str, List[str]]]
                   timeout=None  # type: Optional[float]
                   ):
          """
          :type hostname: ``str``
          :keyword hostname: Hostname or IP address to connect to.
  
          :type port: ``int``
          :keyword port: TCP port to communicate on, defaults to 22.
  
          :type username: ``str``
          :keyword username: Username to use, defaults to root.
  
          :type password: ``str``
          :keyword password: Password to authenticate with or a password used
                             to unlock a private key if a password protected key
                             is used.
  
          :param key: Deprecated in favor of ``key_files`` argument.
  
          :type key_files: ``str`` or ``list``
          :keyword key_files: A list of paths to the private key files to use.
          """
          if key is not None:
              message = ('You are using deprecated "key" argument which has '
                         'been replaced with "key_files" argument')
              warnings.warn(message, DeprecationWarning)
  
              # key_files has precedent
              key_files = key if not key_files else key_files
  
          self.hostname = hostname
          self.port = port
          self.username = username
          self.password = password
          self.key_files = key_files
          self.timeout = timeout
  
      def connect(self):
          # type: () -> bool
          """
          Connect to the remote node over SSH.
  
          :return: True if the connection has been successfully established,
                   False otherwise.
          :rtype: ``bool``
          """
          raise NotImplementedError(
              'connect not implemented for this ssh client')
  
      def put(self, path, contents=None, chmod=None, mode='w'):
          # type: (str, Optional[Union[str, bytes]], Optional[int], str) -> str
          """
          Upload a file to the remote node.
  
          :type path: ``str``
          :keyword path: File path on the remote node.
  
          :type contents: ``str``
          :keyword contents: File Contents.
  
          :type chmod: ``int``
          :keyword chmod: chmod file to this after creation.
  
          :type mode: ``str``
          :keyword mode: Mode in which the file is opened.
  
          :return: Full path to the location where a file has been saved.
          :rtype: ``str``
          """
          raise NotImplementedError(
              'put not implemented for this ssh client')
  
      def delete(self, path):
          # type: (str) -> bool
          """
          Delete/Unlink a file on the remote node.
  
          :type path: ``str``
          :keyword path: File path on the remote node.
  
          :return: True if the file has been successfully deleted, False
                   otherwise.
          :rtype: ``bool``
          """
          raise NotImplementedError(
              'delete not implemented for this ssh client')
  
      def run(self, cmd):
          # type: (str) -> Tuple[str, str, int]
          """
          Run a command on a remote node
  
  [... the rest is omitted ...]
files: libcloud/compute/ssh.py
files: # -*- coding: utf-8 -*-
  # Licensed to the Apache Software Foundation (ASF) under one or moreÂ§
  # contributor license agreements.  See the NOTICE file distributed with
  # this work for additional information regarding copyright ownership.
  # The ASF licenses this file to You under the Apache License, Version 2.0
  # (the "License"); you may not use this file except in compliance with
  # the License.  You may obtain a copy of the License at
  #
  #     http://www.apache.org/licenses/LICENSE-2.0
  #
  # Unless required by applicable law or agreed to in writing, software
  # distributed under the License is distributed on an "AS IS" BASIS,
  # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  # See the License for the specific language governing permissions and
  # limitations under the License.
  
  from __future__ import with_statement
  
  import os
  import sys
  import time
  import unittest
  
  from libcloud.utils.py3 import httplib
  from libcloud.utils.py3 import u
  from libcloud.utils.py3 import PY3
  from libcloud.utils.py3 import assertRaisesRegex
  
  from libcloud.compute.deployment import MultiStepDeployment, Deployment
  from libcloud.compute.deployment import SSHKeyDeployment, ScriptDeployment
  from libcloud.compute.deployment import ScriptFileDeployment, FileDeployment
  from libcloud.compute.base import Node
  from libcloud.compute.base import NodeAuthPassword
  from libcloud.compute.types import NodeState, DeploymentError, LibcloudError
  from libcloud.compute.ssh import BaseSSHClient
  from libcloud.compute.ssh import have_paramiko
  from libcloud.compute.ssh import SSHCommandTimeoutError
  from libcloud.compute.drivers.rackspace import RackspaceFirstGenNodeDriver as Rackspace
  
  from libcloud.test import MockHttp, XML_HEADERS
  from libcloud.test.file_fixtures import ComputeFileFixtures
  from mock import Mock, patch
  
  from libcloud.test.secrets import RACKSPACE_PARAMS
  
  # Keyword arguments which are specific to deploy_node() method, but not
  # create_node()
  DEPLOY_NODE_KWARGS = ['deploy', 'ssh_username', 'ssh_alternate_usernames',
                        'ssh_port', 'ssh_timeout', 'ssh_key', 'timeout',
                        'max_tries', 'ssh_interface']
  
  FILE_PATH = '{0}home{0}ubuntu{0}relative.sh'.format(os.path.sep)
  
  
  class MockDeployment(Deployment):
  
      def run(self, node, client):
          return node
  
  
  class MockClient(BaseSSHClient):
  
      def __init__(self, throw_on_timeout=False, *args, **kwargs):
          self.stdout = ''
          self.stderr = ''
          self.exit_status = 0
          self.throw_on_timeout = throw_on_timeout
  
      def put(self, path, contents, chmod=755, mode='w'):
          return contents
  
      def run(self, cmd, timeout=None):
          if self.throw_on_timeout and timeout is not None:
              raise ValueError("timeout")
  
          return self.stdout, self.stderr, self.exit_status
  
      def delete(self, name):
          return True
  
  
  class DeploymentTests(unittest.TestCase):
  
      def setUp(self):
          Rackspace.connectionCls.conn_class = RackspaceMockHttp
          RackspaceMockHttp.type = None
          self.driver = Rackspace(*RACKSPACE_PARAMS)
          # normally authentication happens lazily, but we force it here
          self.driver.connection._populate_hosts_and_request_paths()
          self.driver.features = {'create_node': ['generates_password']}
          self.node = Node(id=12345, name='test', state=NodeState.RUNNING,
                           public_ips=['1.2.3.4'], private_ips=['1.2.3.5'],
                           driver=Rackspace)
          self.node2 = Node(id=123456, name='test', state=NodeState.RUNNING,
                            public_ips=['1.2.3.4'], private_ips=['1.2.3.5'],
                            driver=Rackspace)
  
      def test_multi_step_deployment(self):
          msd = MultiStepDeployment()
          self.assertEqual(len(msd.steps), 0)
  
          msd.add(MockDeployment())
          self.assertEqual(len(msd.steps), 1)
  
          self.assertEqual(self.node, msd.run(node=self.node, client=None))
  
      def test_ssh_key_deployment(self):
          sshd = SSHKeyDeployment(key='1234')
  
          self.assertEqual(self.node, sshd.run(node=self.node,
                                               client=MockClient(hostname='localhost')))
  
      def test_file_deployment(self):
          # use this file (__file__) for obtaining permissions
          target = os.path.join('/tmp', os.path.basename(__file__))
          fd = FileDeployment(__file__, target)
          self.assertEqual(target, fd.target)
          self.assertEqual(__file__, fd.source)
          self.assertEqual(self.node, fd.run(
              node=self.node, client=MockClient(hostname='localhost')))
  
      def test_script_deployment(self):
          sd1 = ScriptDeployment(script='foobar', delete=True)
          sd2 = ScriptDeployment(script='foobar', delete=False)
          sd3 = ScriptDeployment(
              script='foobar', delete=False, name='foobarname')
          sd4 = ScriptDeployment(
              script='foobar', delete=False, name='foobarname', timeout=10)
  
          self.assertTrue(sd1.name.find('deployment') != '1')
          self.assertEqual(sd3.name, 'foobarname')
          self.assertEqual(sd3.timeout, None)
          self.assertEqual(sd4.timeout, 10)
  
          self.assertEqual(self.node, sd1.run(node=self.node,
                                              client=MockClient(hostname='localhost')))
          self.assertEqual(self.node, sd2.run(node=self.node,
                                              client=MockClient(hostname='localhost')))
          self.assertEqual(self.node, sd3.run(node=self.node,
                                              client=MockClient(hostname
  
  [... the rest is omitted ...]
files: libcloud/test/compute/test_deployment.py
files: # -*- coding: utf-8 -*-
  # Licensed to the Apache Software Foundation (ASF) under one or moreÂ§
  # contributor license agreements.  See the NOTICE file distributed with
  # this work for additional information regarding copyright ownership.
  # The ASF licenses this file to You under the Apache License, Version 2.0
  # (the "License"); you may not use this file except in compliance with
  # the License.  You may obtain a copy of the License at
  #
  #     http://www.apache.org/licenses/LICENSE-2.0
  #
  # Unless required by applicable law or agreed to in writing, software
  # distributed under the License is distributed on an "AS IS" BASIS,
  # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  # See the License for the specific language governing permissions and
  # limitations under the License.
  
  from __future__ import with_statement
  
  import os
  import sys
  import time
  import unittest
  
  from libcloud.utils.py3 import httplib
  from libcloud.utils.py3 import u
  from libcloud.utils.py3 import PY3
  from libcloud.utils.py3 import assertRaisesRegex
  
  from libcloud.compute.deployment import MultiStepDeployment, Deployment
  from libcloud.compute.deployment import SSHKeyDeployment, ScriptDeployment
  from libcloud.compute.deployment import ScriptFileDeployment, FileDeployment
  from libcloud.compute.base import Node
  from libcloud.compute.base import NodeAuthPassword
  from libcloud.compute.types import NodeState, DeploymentError, LibcloudError
  from libcloud.compute.ssh import BaseSSHClient
  from libcloud.compute.ssh import have_paramiko
  from libcloud.compute.ssh import SSHCommandTimeoutError
  from libcloud.compute.drivers.rackspace import RackspaceFirstGenNodeDriver as Rackspace
  
  from libcloud.test import MockHttp, XML_HEADERS
  from libcloud.test.file_fixtures import ComputeFileFixtures
  from mock import Mock, patch
  
  from libcloud.test.secrets import RACKSPACE_PARAMS
  
  # Keyword arguments which are specific to deploy_node() method, but not
  # create_node()
  DEPLOY_NODE_KWARGS = ['deploy', 'ssh_username', 'ssh_alternate_usernames',
                        'ssh_port', 'ssh_timeout', 'ssh_key', 'timeout',
                        'max_tries', 'ssh_interface']
  
  FILE_PATH = '{0}home{0}ubuntu{0}relative.sh'.format(os.path.sep)
  
  
  class MockDeployment(Deployment):
  
      def run(self, node, client):
          return node
  
  
  class MockClient(BaseSSHClient):
  
      def __init__(self, *args, **kwargs):
          self.stdout = ''
          self.stderr = ''
          self.exit_status = 0
  
      def put(self, path, contents, chmod=755, mode='w'):
          return contents
  
      def run(self, name):
          return self.stdout, self.stderr, self.exit_status
  
      def delete(self, name):
          return True
  
  
  class DeploymentTests(unittest.TestCase):
  
      def setUp(self):
          Rackspace.connectionCls.conn_class = RackspaceMockHttp
          RackspaceMockHttp.type = None
          self.driver = Rackspace(*RACKSPACE_PARAMS)
          # normally authentication happens lazily, but we force it here
          self.driver.connection._populate_hosts_and_request_paths()
          self.driver.features = {'create_node': ['generates_password']}
          self.node = Node(id=12345, name='test', state=NodeState.RUNNING,
                           public_ips=['1.2.3.4'], private_ips=['1.2.3.5'],
                           driver=Rackspace)
          self.node2 = Node(id=123456, name='test', state=NodeState.RUNNING,
                            public_ips=['1.2.3.4'], private_ips=['1.2.3.5'],
                            driver=Rackspace)
  
      def test_multi_step_deployment(self):
          msd = MultiStepDeployment()
          self.assertEqual(len(msd.steps), 0)
  
          msd.add(MockDeployment())
          self.assertEqual(len(msd.steps), 1)
  
          self.assertEqual(self.node, msd.run(node=self.node, client=None))
  
      def test_ssh_key_deployment(self):
          sshd = SSHKeyDeployment(key='1234')
  
          self.assertEqual(self.node, sshd.run(node=self.node,
                                               client=MockClient(hostname='localhost')))
  
      def test_file_deployment(self):
          # use this file (__file__) for obtaining permissions
          target = os.path.join('/tmp', os.path.basename(__file__))
          fd = FileDeployment(__file__, target)
          self.assertEqual(target, fd.target)
          self.assertEqual(__file__, fd.source)
          self.assertEqual(self.node, fd.run(
              node=self.node, client=MockClient(hostname='localhost')))
  
      def test_script_deployment(self):
          sd1 = ScriptDeployment(script='foobar', delete=True)
          sd2 = ScriptDeployment(script='foobar', delete=False)
          sd3 = ScriptDeployment(
              script='foobar', delete=False, name='foobarname')
  
          self.assertTrue(sd1.name.find('deployment') != '1')
          self.assertEqual(sd3.name, 'foobarname')
  
          self.assertEqual(self.node, sd1.run(node=self.node,
                                              client=MockClient(hostname='localhost')))
          self.assertEqual(self.node, sd2.run(node=self.node,
                                              client=MockClient(hostname='localhost')))
  
      def test_script_file_deployment(self):
          file_path = os.path.abspath(__file__)
          with open(file_path, 'rb') as fp:
              content = fp.read()
  
          if PY3:
              content = content.decode('utf-8')
  
          sfd1 = ScriptFileDeployment(script_file=file_path)
          self.assertEqual(sfd1.script, content)
  
      def test_script_deployment_relative_path(self):
          client = Mock()
          client.put.return_value = FILE_PATH
          client.run.return
  
  [... the rest is omitted ...]
files: libcloud/test/compute/test_deployment.py