files.new_contents: # 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 unittest
  import datetime
  import mock
  import pytest
  
  from libcloud.utils.iso8601 import UTC
  
  try:
      import simplejson as json
  except ImportError:
      import json
  
  from mock import Mock, patch
  import requests_mock
  
  from libcloud.utils.py3 import httplib
  from libcloud.utils.py3 import method_type
  from libcloud.utils.py3 import u
  
  from libcloud.common.base import LibcloudConnection
  from libcloud.common.exceptions import BaseHTTPError
  from libcloud.common.openstack_identity import OpenStackAuthenticationCache
  from libcloud.common.openstack_identity import AUTH_VERSIONS_WITH_EXPIRES
  from libcloud.common.types import (
      InvalidCredsError,
      MalformedResponseError,
      LibcloudError,
  )
  from libcloud.compute.types import (
      Provider,
      KeyPairDoesNotExistError,
      StorageVolumeState,
      VolumeSnapshotState,
      NodeImageMemberState,
  )
  from libcloud.compute.providers import get_driver
  from libcloud.compute.drivers.openstack import (
      OpenStack_1_0_NodeDriver,
      OpenStack_1_1_NodeDriver,
      OpenStackSecurityGroup,
      OpenStackSecurityGroupRule,
      OpenStack_1_1_FloatingIpPool,
      OpenStack_1_1_FloatingIpAddress,
      OpenStackKeyPair,
      OpenStack_1_0_Connection,
      OpenStack_2_FloatingIpPool,
      OpenStackNodeDriver,
      OpenStack_2_NodeDriver,
      OpenStack_2_PortInterfaceState,
      OpenStackNetwork,
      OpenStackException,
      OpenStack_2_ServerGroup,
  )
  from libcloud.compute.base import Node, NodeImage, NodeSize
  from libcloud.pricing import set_pricing, clear_pricing_data
  
  from libcloud.test import MockHttp, XML_HEADERS
  from libcloud.test.file_fixtures import ComputeFileFixtures, OpenStackFixtures
  from libcloud.test.compute import TestCaseMixin
  
  from libcloud.test.secrets import OPENSTACK_PARAMS
  
  BASE_DIR = os.path.abspath(os.path.split(__file__)[0])
  
  
  def test_driver_instantiation_invalid_auth():
      with pytest.raises(LibcloudError):
          d = OpenStackNodeDriver(
              "user",
              "correct_password",
              ex_force_auth_version="5.0",
              ex_force_auth_url="http://x.y.z.y:5000",
              ex_tenant_name="admin",
          )
          d.list_nodes()
  
  
  class OpenStackAuthTests(unittest.TestCase):
      def setUp(self):
          OpenStack_1_0_NodeDriver.connectionCls = OpenStack_1_0_Connection
          OpenStack_1_0_NodeDriver.connectionCls.conn_class = LibcloudConnection
  
      def test_auth_host_passed(self):
          forced_auth = "http://x.y.z.y:5000"
          d = OpenStack_1_0_NodeDriver(
              "user",
              "correct_password",
              ex_force_auth_version="2.0_password",
              ex_force_auth_url="http://x.y.z.y:5000",
              ex_tenant_name="admin",
          )
          self.assertEqual(d._ex_force_auth_url, forced_auth)
  
          with requests_mock.Mocker() as mock:
              body2 = ComputeFileFixtures("openstack").load("_v2_0__auth.json")
  
              mock.register_uri(
                  "POST",
                  "http://x.y.z.y:5000/v2.0/tokens",
                  text=body2,
                  headers={"content-type": "application/json; charset=UTF-8"},
              )
              d.connection._populate_hosts_and_request_paths()
              self.assertEqual(d.connection.host, "test_endpoint.com")
  
  
  class OpenStack_1_0_Tests(TestCaseMixin, unittest.TestCase):
      should_list_locations = False
      should_list_volumes = False
  
      driver_klass = OpenStack_1_0_NodeDriver
      driver_args = OPENSTACK_PARAMS
      driver_kwargs = {}
      # driver_kwargs = {'ex_force_auth_version': '1.0'}
  
      @classmethod
      def create_driver(self):
          if self is not OpenStack_1_0_FactoryMethodTests:
              self.driver_type = self.driver_klass
          return self.driver_type(*self.driver_args, **self.driver_kwargs)
  
      def setUp(self):
          # monkeypatch get_endpoint because the base openstack driver doesn't actually
          # work with old devstack but this class/tests are still used by the rackspace
          # driver
          def get_endpoint(*args, **kwargs):
              return "https://servers.api.rackspacecloud.com/v1.0/slug"
  
          self.driver_klass.connectionCls.get_endpoint = get_endpoint
  
          self.driver_klass.connectionCls.conn_class = OpenStackMockHttp
          self.driver_klass.connectionCls.auth_url = "https://auth.api.example.com"
  
          OpenStackMockHttp.type = None
  
          self.driver = self.create_driver()
          # normally authentication happens lazily, but we force it here
          self.driver.connection._populate_hosts_and_request_paths()
          clear_pricing_data()
  
      @patch("libcloud.common.openstack.OpenStackServiceCatalog")
      def test_populate_hosts_and_requests_path(self, _):
          tomorrow = datetime.datetime.today() + datetime.timedelta(1)
          cls = self.driver_klass.connectionCls
  
          count = 5
  
          # Test authentication and token re-use
          con = cls("username", "key")
          osa = con.get_auth_class()
  
          mocked_auth_method = Mock()
          osa.authenticate = mocked_auth_method
  
          # Valid token returned on first call, should be reused.
          for i in range(0, count):
              con._populate_hosts_and_request_paths()
  
              if i == 0:
                  osa.auth_token = "1234"
                  osa.auth_token_expires = tomorrow
  
          self.assertEqual(mocked_auth_method.call_count, 1)
  
          osa.auth_token = None
          osa.auth_token_expires = None
  
          # ex_force_auth_token provided, authenticate should never be called
          con = cls(
              "username",
              "key",
              ex_force_base_url="http://ponies",
              ex_force_auth_token="1234",
          )
          osa = con.get_auth_class()
  
          mocked_auth_method = Mock()
          osa.authenticate = mocked_auth_method
  
          for i in range(0, count):
              con._populate_hosts_and_request_paths()
  
          self.assertEqual(mocked_auth_method.call_count, 0)
  
      def test_auth_token_is_set(self):
          self.driver.connection._populate_hosts_and_request_paths()
          self.assertEqual(
              self.driver.connection.auth_token, "aaaaaaaaaaaa-bbb-cccccccccccccc"
          )
  
      def test_auth_token_expires_is_set(self):
          self.driver.connection._populate_hosts_and_request_paths()
  
          expires = self.driver.connection.auth_token_expires
          self.assertEqual(expires.isoformat(), "2999-11-23T21:00:14-06:00")
  
      def test_auth(self):
          if self.driver.connection._auth_version == "2.0":
              return
  
          OpenStackMockHttp.type = "UNAUTHORIZED"
          try:
              self.driver = self.create_driver()
              self.driver.list_nodes()
          except InvalidCredsError as e:
              self.assertEqual(True, isinstance(e, InvalidCredsError))
          else:
              self.fail("test should have thrown")
  
      def test_auth_missing_key(self):
          if self.driver.connection._auth_version == "2.0":
              return
  
          OpenStackMockHttp.type = "UNAUTHORIZED_MISSING_KEY"
          try:
              self.driver = self.create_driver()
              self.driver.list_nodes()
          except MalformedResponseError as e:
              self.assertEqual(True, isinstance(e, MalformedResponseError))
          else:
              self.fail("test should have thrown
  
  [... the rest is omitted ...]
files.new_path: libcloud/test/compute/test_openstack.py
files.old_contents: # 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 unittest
  import datetime
  import mock
  import pytest
  
  from libcloud.utils.iso8601 import UTC
  
  try:
      import simplejson as json
  except ImportError:
      import json
  
  from mock import Mock, patch
  import requests_mock
  
  from libcloud.utils.py3 import httplib
  from libcloud.utils.py3 import method_type
  from libcloud.utils.py3 import u
  
  from libcloud.common.base import LibcloudConnection
  from libcloud.common.exceptions import BaseHTTPError
  from libcloud.common.openstack_identity import OpenStackAuthenticationCache
  from libcloud.common.types import (
      InvalidCredsError,
      MalformedResponseError,
      LibcloudError,
  )
  from libcloud.compute.types import (
      Provider,
      KeyPairDoesNotExistError,
      StorageVolumeState,
      VolumeSnapshotState,
      NodeImageMemberState,
  )
  from libcloud.compute.providers import get_driver
  from libcloud.compute.drivers.openstack import (
      OpenStack_1_0_NodeDriver,
      OpenStack_1_1_NodeDriver,
      OpenStackSecurityGroup,
      OpenStackSecurityGroupRule,
      OpenStack_1_1_FloatingIpPool,
      OpenStack_1_1_FloatingIpAddress,
      OpenStackKeyPair,
      OpenStack_1_0_Connection,
      OpenStack_2_FloatingIpPool,
      OpenStackNodeDriver,
      OpenStack_2_NodeDriver,
      OpenStack_2_PortInterfaceState,
      OpenStackNetwork,
      OpenStackException,
      OpenStack_2_ServerGroup,
  )
  from libcloud.compute.base import Node, NodeImage, NodeSize
  from libcloud.pricing import set_pricing, clear_pricing_data
  
  from libcloud.test import MockHttp, XML_HEADERS
  from libcloud.test.file_fixtures import ComputeFileFixtures, OpenStackFixtures
  from libcloud.test.compute import TestCaseMixin
  
  from libcloud.test.secrets import OPENSTACK_PARAMS
  
  BASE_DIR = os.path.abspath(os.path.split(__file__)[0])
  
  
  def test_driver_instantiation_invalid_auth():
      with pytest.raises(LibcloudError):
          d = OpenStackNodeDriver(
              "user",
              "correct_password",
              ex_force_auth_version="5.0",
              ex_force_auth_url="http://x.y.z.y:5000",
              ex_tenant_name="admin",
          )
          d.list_nodes()
  
  
  class OpenStackAuthTests(unittest.TestCase):
      def setUp(self):
          OpenStack_1_0_NodeDriver.connectionCls = OpenStack_1_0_Connection
          OpenStack_1_0_NodeDriver.connectionCls.conn_class = LibcloudConnection
  
      def test_auth_host_passed(self):
          forced_auth = "http://x.y.z.y:5000"
          d = OpenStack_1_0_NodeDriver(
              "user",
              "correct_password",
              ex_force_auth_version="2.0_password",
              ex_force_auth_url="http://x.y.z.y:5000",
              ex_tenant_name="admin",
          )
          self.assertEqual(d._ex_force_auth_url, forced_auth)
  
          with requests_mock.Mocker() as mock:
              body2 = ComputeFileFixtures("openstack").load("_v2_0__auth.json")
  
              mock.register_uri(
                  "POST",
                  "http://x.y.z.y:5000/v2.0/tokens",
                  text=body2,
                  headers={"content-type": "application/json; charset=UTF-8"},
              )
              d.connection._populate_hosts_and_request_paths()
              self.assertEqual(d.connection.host, "test_endpoint.com")
  
  
  class OpenStack_1_0_Tests(TestCaseMixin, unittest.TestCase):
      should_list_locations = False
      should_list_volumes = False
  
      driver_klass = OpenStack_1_0_NodeDriver
      driver_args = OPENSTACK_PARAMS
      driver_kwargs = {}
      # driver_kwargs = {'ex_force_auth_version': '1.0'}
  
      @classmethod
      def create_driver(self):
          if self is not OpenStack_1_0_FactoryMethodTests:
              self.driver_type = self.driver_klass
          return self.driver_type(*self.driver_args, **self.driver_kwargs)
  
      def setUp(self):
          # monkeypatch get_endpoint because the base openstack driver doesn't actually
          # work with old devstack but this class/tests are still used by the rackspace
          # driver
          def get_endpoint(*args, **kwargs):
              return "https://servers.api.rackspacecloud.com/v1.0/slug"
  
          self.driver_klass.connectionCls.get_endpoint = get_endpoint
  
          self.driver_klass.connectionCls.conn_class = OpenStackMockHttp
          self.driver_klass.connectionCls.auth_url = "https://auth.api.example.com"
  
          OpenStackMockHttp.type = None
  
          self.driver = self.create_driver()
          # normally authentication happens lazily, but we force it here
          self.driver.connection._populate_hosts_and_request_paths()
          clear_pricing_data()
  
      @patch("libcloud.common.openstack.OpenStackServiceCatalog")
      def test_populate_hosts_and_requests_path(self, _):
          tomorrow = datetime.datetime.today() + datetime.timedelta(1)
          cls = self.driver_klass.connectionCls
  
          count = 5
  
          # Test authentication and token re-use
          con = cls("username", "key")
          osa = con.get_auth_class()
  
          mocked_auth_method = Mock()
          osa.authenticate = mocked_auth_method
  
          # Valid token returned on first call, should be reused.
          for i in range(0, count):
              con._populate_hosts_and_request_paths()
  
              if i == 0:
                  osa.auth_token = "1234"
                  osa.auth_token_expires = tomorrow
  
          self.assertEqual(mocked_auth_method.call_count, 1)
  
          osa.auth_token = None
          osa.auth_token_expires = None
  
          # ex_force_auth_token provided, authenticate should never be called
          con = cls(
              "username",
              "key",
              ex_force_base_url="http://ponies",
              ex_force_auth_token="1234",
          )
          osa = con.get_auth_class()
  
          mocked_auth_method = Mock()
          osa.authenticate = mocked_auth_method
  
          for i in range(0, count):
              con._populate_hosts_and_request_paths()
  
          self.assertEqual(mocked_auth_method.call_count, 0)
  
      def test_auth_token_is_set(self):
          self.driver.connection._populate_hosts_and_request_paths()
          self.assertEqual(
              self.driver.connection.auth_token, "aaaaaaaaaaaa-bbb-cccccccccccccc"
          )
  
      def test_auth_token_expires_is_set(self):
          self.driver.connection._populate_hosts_and_request_paths()
  
          expires = self.driver.connection.auth_token_expires
          self.assertEqual(expires.isoformat(), "2999-11-23T21:00:14-06:00")
  
      def test_auth(self):
          if self.driver.connection._auth_version == "2.0":
              return
  
          OpenStackMockHttp.type = "UNAUTHORIZED"
          try:
              self.driver = self.create_driver()
              self.driver.list_nodes()
          except InvalidCredsError as e:
              self.assertEqual(True, isinstance(e, InvalidCredsError))
          else:
              self.fail("test should have thrown")
  
      def test_auth_missing_key(self):
          if self.driver.connection._auth_version == "2.0":
              return
  
          OpenStackMockHttp.type = "UNAUTHORIZED_MISSING_KEY"
          try:
              self.driver = self.create_driver()
              self.driver.list_nodes()
          except MalformedResponseError as e:
              self.assertEqual(True, isinstance(e, MalformedResponseError))
          else:
              self.fail("test should have thrown")
  
      def test_auth_server_error(self):
          if self.driver.connection._auth_version ==
  
  [... the rest is omitted ...]
files.old_path: libcloud/test/compute/test_openstack.py