本文整理汇总了Python中rhsm.connection.UEPConnection类的典型用法代码示例。如果您正苦于以下问题:Python UEPConnection类的具体用法?Python UEPConnection怎么用?Python UEPConnection使用的例子?那么恭喜您, 这里精选的类代码示例或许可以为您提供帮助。
在下文中一共展示了UEPConnection类的20个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: BindRequestTests
class BindRequestTests(unittest.TestCase):
def setUp(self):
self.cp = UEPConnection(username="admin", password="admin", insecure=True)
consumerInfo = self.cp.registerConsumer("test-consumer", "system", owner="admin")
self.consumer_uuid = consumerInfo['uuid']
@patch.object(Restlib, 'validateResponse')
@patch('rhsm.connection.drift_check', return_value=False)
@patch('httplib.HTTPSConnection', auto_spec=True)
def test_bind_no_args(self, mock_conn, mock_drift, mock_validate):
self.cp.bind(self.consumer_uuid)
# verify we called request() with kwargs that include 'body' as None
# Specifically, we are checking that passing in "" to post_request, as
# it does by default, results in None here. bin() passes no args there
# so we use the default, "". See bz #907536
for (name, args, kwargs) in mock_conn.mock_calls:
if name == '().request':
self.assertEqual(None, kwargs['body'])
@patch.object(Restlib, 'validateResponse')
@patch('rhsm.connection.drift_check', return_value=False)
@patch('httplib.HTTPSConnection', auto_spec=True)
def test_bind_by_pool(self, mock_conn, mock_drift, mock_validate):
# this test is just to verify we make the httplib connection with
# right args, we don't validate the bind here
self.cp.bindByEntitlementPool(self.consumer_uuid, '123121111', '1')
for (name, args, kwargs) in mock_conn.mock_calls:
if name == '().request':
self.assertEqual(None, kwargs['body'])
开发者ID:Januson,项目名称:subscription-manager,代码行数:32,代码来源:connection_tests.py
示例2: HypervisorCheckinTests
class HypervisorCheckinTests(unittest.TestCase):
def setUp(self):
self.cp = UEPConnection(username="admin", password="admin",
insecure=True)
def test_hypervisor_checkin_can_pass_empty_map_and_updates_nothing(self):
response = self.cp.hypervisorCheckIn("admin", "", {})
if self.cp.has_capability('hypervisors_async'):
self.assertEqual(response['resultData'], None)
else:
self.assertEqual(len(response['failedUpdate']), 0)
self.assertEqual(len(response['updated']), 0)
self.assertEqual(len(response['created']), 0)
开发者ID:Januson,项目名称:subscription-manager,代码行数:14,代码来源:connection_tests.py
示例3: RestlibTests
class RestlibTests(unittest.TestCase):
def setUp(self):
# Get handle to Restlib
self.conn = UEPConnection().conn
self.request_type = "GET"
self.handler = "https://server/path"
def _validate_response(self, response):
# wrapper to specify request_type and handler
return self.conn.validateResponse(response,
request_type=self.request_type,
handler=self.handler)
def test_invalid_credentitals_thrown_on_401_with_empty_body(self):
mock_response = {"status": 401}
self.assertRaises(UnauthorizedException, self._validate_response,
mock_response)
def test_standard_error_handling_on_401_with_defined_body(self):
self._run_standard_error_handling_test(401)
def test_standard_error_handling_on_401_with_invalid_json_body(self):
self._run_standard_error_handling_test_invalid_json(401, UnauthorizedException)
def test_invalid_credentitals_thrown_on_403_with_empty_body(self):
mock_response = {"status": 403}
self.assertRaises(ForbiddenException, self._validate_response,
mock_response)
def test_standard_error_handling_on_403_with_defined_body(self):
self._run_standard_error_handling_test(403)
def test_standard_error_handling_on_403_with_invalid_json_body(self):
self._run_standard_error_handling_test_invalid_json(403, ForbiddenException)
def _run_standard_error_handling_test_invalid_json(self, expected_error_code,
expected_exception):
mock_response = {"status": expected_error_code,
"content": '<this is not valid json>>'}
self._check_for_remote_server_exception(expected_error_code,
expected_exception,
mock_response)
def _run_standard_error_handling_test(self, expected_error):
expected_error = "My Expected Error."
mock_response = {"status": expected_error,
"content": '{"displayMessage":"%s"}' % expected_error}
try:
self._validate_response(mock_response)
self.fail("An exception should have been thrown.")
except Exception, ex:
self.assertTrue(isinstance(ex, RestlibException))
self.assertEquals(expected_error, ex.code)
self.assertEqual(expected_error, str(ex))
开发者ID:barnabycourt,项目名称:python-rhsm,代码行数:57,代码来源:connection-tests.py
示例4: OwnerInfoTests
class OwnerInfoTests(unittest.TestCase):
def setUp(self):
self.cp = UEPConnection(username="admin", password="admin",
insecure=True)
self.owner_key = "test_owner_%d" % (random.randint(1, 5000))
self.cp.conn.request_post('/owners', {'key': self.owner_key,
'displayName': self.owner_key})
def test_get_owner_info(self):
owner_info = self.cp.getOwnerInfo(self.owner_key)
self.assertTrue(owner_info is not None)
开发者ID:Januson,项目名称:subscription-manager,代码行数:11,代码来源:connection_tests.py
示例5: DatetimeFormattingTests
class DatetimeFormattingTests(unittest.TestCase):
def setUp(self):
# NOTE: this won't actually work, idea for this suite of unit tests
# is to mock the actual server responses and just test logic in the
# UEPConnection:
self.cp = UEPConnection(username="dummy", password="dummy",
handler="/Test/", insecure=True)
def tearDown(self):
locale.resetlocale()
def test_date_formatted_properly_with_japanese_locale(self):
locale.setlocale(locale.LC_ALL, 'ja_JP.UTF8')
expected_headers = {
'If-Modified-Since': 'Fri, 13 Feb 2009 23:31:30 GMT'
}
timestamp = 1234567890
self.cp.conn = Mock()
self.cp.getAccessibleContent(consumerId='bob', if_modified_since=datetime.datetime.fromtimestamp(timestamp))
self.cp.conn.request_get.assert_called_with('/consumers/bob/accessible_content', headers=expected_headers)
开发者ID:Lorquas,项目名称:subscription-manager,代码行数:20,代码来源:connection-tests.py
示例6: ConnectionTests
class ConnectionTests(unittest.TestCase):
def setUp(self):
# NOTE: this won't actually work, idea for this suite of unit tests
# is to mock the actual server responses and just test logic in the
# UEPConnection:
self.cp = UEPConnection(username="dummy", password="dummy",
insecure=True)
def test_get_environment_by_name_requires_owner(self):
self.assertRaises(Exception, self.cp.getEnvironment, None, {"name": "env name"})
def test_get_environment_urlencoding(self):
self.cp.conn = Mock()
self.cp.conn.request_get = Mock(return_value=[])
self.cp.getEnvironment(owner_key="myorg", name="env name__++=*&")
self.cp.conn.request_get.assert_called_with(
"/owners/myorg/environments?name=env+name__%2B%2B%3D%2A%26")
def test_entitle_date(self):
self.cp.conn = Mock()
self.cp.conn.request_post = Mock(return_value=[])
self.cp.bind("abcd", date(2011, 9, 2))
self.cp.conn.request_post.assert_called_with(
"/consumers/abcd/entitlements?entitle_date=2011-09-02")
def test_no_entitle_date(self):
self.cp.conn = Mock()
self.cp.conn.request_post = Mock(return_value=[])
self.cp.bind("abcd")
self.cp.conn.request_post.assert_called_with("/consumers/abcd/entitlements")
开发者ID:wottop,项目名称:python-rhsm,代码行数:31,代码来源:connection-tests.py
示例7: is_valid_server_info
def is_valid_server_info(hostname, port, prefix):
"""
Check if we can communicate with a subscription service at the given
location.
Returns true or false.
May throw a MissingCaCertException if the CA certificate has not been
imported yet, which may be relevant to the caller.
"""
# Proxy info should already be in config file and used by default:
try:
conn = UEPConnection(host=hostname, ssl_port=int(port), handler=prefix)
conn.ping()
return True
except RestlibException, e:
# If we're getting Unauthorized that's a good indication this is a
# valid subscription service:
if e.code == 401:
return True
else:
log.exception(e)
return False
开发者ID:tkolhar,项目名称:subscription-manager,代码行数:23,代码来源:utils.py
示例8: setUp
def setUp(self):
# Try to remove all environment variables to not influence unit test
try:
os.environ.pop('no_proxy')
os.environ.pop('NO_PROXY')
os.environ.pop('HTTPS_PROXY')
except KeyError:
pass
# NOTE: this won't actually work, idea for this suite of unit tests
# is to mock the actual server responses and just test logic in the
# UEPConnection:
self.cp = UEPConnection(username="dummy", password="dummy",
handler="/Test/", insecure=True)
self.temp_ent_dir = mkdtemp()
开发者ID:Lorquas,项目名称:subscription-manager,代码行数:14,代码来源:connection-tests.py
示例9: setUp
def setUp(self):
self.cp = UEPConnection(username="admin", password="admin", insecure=True)
self.consumer = self.cp.registerConsumer("test-consumer", "system", owner="admin")
self.consumer_uuid = self.consumer['uuid']
# This product is present in the Candlepin test data
self.cp.bindByProduct(self.consumer_uuid, ["awesomeos-instancebased"])
entitlements = self.cp.getEntitlementList(self.consumer_uuid)
self.assertTrue(len(entitlements) > 0)
self.entitlement = entitlements[0]
self.entitlement_id = self.entitlement['id']
开发者ID:Januson,项目名称:subscription-manager,代码行数:14,代码来源:connection_tests.py
示例10: CandlepinConnection
class CandlepinConnection():
def getOwners(self):
return self.cp.getOwnerList(self.username)
def __init__(self):
self.conn_username = "admin"
self.owner = "admin"
self.cp = UEPConnection(username=self.conn_username, password="admin",
host="localhost", ssl_port=8443,
handler="/candlepin", insecure=True)
def createConsumer(self, name, facts, installed_products):
consumer = self.cp.registerConsumer(name=name, facts=facts, owner=self.owner, installed_products=installed_products)
print "created consumer with uuid %s. binding.." % consumer['uuid']
self.cp.bind(consumer['uuid'], entitle_date=datetime.now())
print "bind complete"
self.cp.checkin(consumer['uuid'])
return consumer['uuid']
def updateConsumer(self, uuid, facts, installed_products):
self.cp.updateConsumer(uuid, facts=facts, installed_products=installed_products)
self.cp.checkin(uuid)
开发者ID:beav,项目名称:spacewalk-splice-tool,代码行数:23,代码来源:cp_client.py
示例11: ConnectionTests
class ConnectionTests(unittest.TestCase):
def setUp(self):
# NOTE: this won't actually work, idea for this suite of unit tests
# is to mock the actual server responses and just test logic in the
# UEPConnection:
self.cp = UEPConnection(username="dummy", password="dummy",
handler="/Test/", insecure=True)
def test_load_manager_capabilities(self):
expected_capabilities = ['hypervisors_async', 'cores']
proper_status = {'version':'1',
'result':True,
'managerCapabilities':expected_capabilities}
improper_status = dict.copy(proper_status)
# Remove the managerCapabilities key from the dict
del improper_status['managerCapabilities']
self.cp.conn = Mock()
# The first call will return the proper_status, the second, the improper
# status
original_getStatus = self.cp.getStatus
self.cp.getStatus = Mock(side_effect=[proper_status,
improper_status])
actual_capabilities = self.cp._load_manager_capabilities()
self.assertEquals(sorted(actual_capabilities),
sorted(expected_capabilities))
self.assertEquals([], self.cp._load_manager_capabilities())
self.cp.getStatus = original_getStatus
def test_get_environment_by_name_requires_owner(self):
self.assertRaises(Exception, self.cp.getEnvironment, None, {"name": "env name"})
def test_get_environment_urlencoding(self):
self.cp.conn = Mock()
self.cp.conn.request_get = Mock(return_value=[])
self.cp.getEnvironment(owner_key="myorg", name="env name__++=*&")
self.cp.conn.request_get.assert_called_with(
"/owners/myorg/environments?name=env+name__%2B%2B%3D%2A%26")
def test_entitle_date(self):
self.cp.conn = Mock()
self.cp.conn.request_post = Mock(return_value=[])
self.cp.bind("abcd", date(2011, 9, 2))
self.cp.conn.request_post.assert_called_with(
"/consumers/abcd/entitlements?entitle_date=2011-09-02")
def test_no_entitle_date(self):
self.cp.conn = Mock()
self.cp.conn.request_post = Mock(return_value=[])
self.cp.bind("abcd")
self.cp.conn.request_post.assert_called_with("/consumers/abcd/entitlements")
def test_clean_up_prefix(self):
self.assertTrue(self.cp.handler == "/Test")
def test_https_proxy_info_allcaps(self):
with patch.dict('os.environ', {'HTTPS_PROXY': 'http://u:[email protected]:4444'}):
uep = UEPConnection(username="dummy", password="dummy",
handler="/Test/", insecure=True)
self.assertEquals("u", uep.proxy_user)
self.assertEquals("p", uep.proxy_password)
self.assertEquals("host", uep.proxy_hostname)
self.assertEquals(int("4444"), uep.proxy_port)
def test_order(self):
# should follow the order: HTTPS, https, HTTP, http
with patch.dict('os.environ', {'HTTPS_PROXY': 'http://u:[email protected]:4444', 'http_proxy': 'http://notme:[email protected]:2222'}):
uep = UEPConnection(username="dummy", password="dummy",
handler="/Test/", insecure=True)
self.assertEquals("u", uep.proxy_user)
self.assertEquals("p", uep.proxy_password)
self.assertEquals("host", uep.proxy_hostname)
self.assertEquals(int("4444"), uep.proxy_port)
def test_no_port(self):
with patch.dict('os.environ', {'HTTPS_PROXY': 'http://u:[email protected]'}):
uep = UEPConnection(username="dummy", password="dummy",
handler="/Test/", insecure=True)
self.assertEquals("u", uep.proxy_user)
self.assertEquals("p", uep.proxy_password)
self.assertEquals("host", uep.proxy_hostname)
self.assertEquals(3128, uep.proxy_port)
def test_no_user_or_password(self):
with patch.dict('os.environ', {'HTTPS_PROXY': 'http://host:1111'}):
uep = UEPConnection(username="dummy", password="dummy",
handler="/Test/", insecure=True)
self.assertEquals(None, uep.proxy_user)
self.assertEquals(None, uep.proxy_password)
self.assertEquals("host", uep.proxy_hostname)
self.assertEquals(int("1111"), uep.proxy_port)
def test_sanitizeGuestIds_supports_strs(self):
self.cp.supports_resource = Mock(return_value=True)
guestIds = ['test' + str(i) for i in range(3)]
resultGuestIds = self.cp.sanitizeGuestIds(guestIds)
# When strings are given, they should always be unchanged
self.assertEquals(guestIds, resultGuestIds)
def test_sanitizeGuestIds_no_support_strs(self):
#.........这里部分代码省略.........
开发者ID:belonesox,项目名称:python-rhsm,代码行数:101,代码来源:connection-tests.py
示例12: ConnectionTests
class ConnectionTests(unittest.TestCase):
def setUp(self):
self.cp = UEPConnection(username="admin", password="admin",
insecure=True)
consumerInfo = self.cp.registerConsumer("test-consumer", "system", owner="admin")
self.consumer_uuid = consumerInfo['uuid']
def test_supports_resource(self):
self.assertTrue(self.cp.supports_resource('consumers'))
self.assertTrue(self.cp.supports_resource('admin'))
self.assertFalse(self.cp.supports_resource('boogity'))
def test_update_consumer_can_update_guests_with_empty_list(self):
self.cp.updateConsumer(self.consumer_uuid, guest_uuids=[])
def test_update_consumer_can_update_facts_with_empty_dict(self):
self.cp.updateConsumer(self.consumer_uuid, facts={})
def test_update_consumer_can_update_installed_products_with_empty_list(self):
self.cp.updateConsumer(self.consumer_uuid, installed_products=[])
def tearDown(self):
self.cp.unregisterConsumer(self.consumer_uuid)
开发者ID:mhrivnak,项目名称:python-rhsm,代码行数:25,代码来源:connection-tests.py
示例13: ConnectionTests
class ConnectionTests(unittest.TestCase):
def setUp(self):
self.cp = UEPConnection(username="admin", password="admin", insecure=True)
self.consumer = self.cp.registerConsumer("test-consumer", "system", owner="admin")
self.consumer_uuid = self.consumer['uuid']
def test_supports_resource(self):
self.assertTrue(self.cp.supports_resource('consumers'))
self.assertTrue(self.cp.supports_resource('admin'))
self.assertFalse(self.cp.supports_resource('boogity'))
def test_has_capability(self):
self.cp.capabilities = ['cores', 'hypervisors_async']
self.assertTrue(self.cp.has_capability('cores'))
self.assertFalse(self.cp.has_capability('boogityboo'))
def test_update_consumer_can_update_guests_with_empty_list(self):
self.cp.updateConsumer(self.consumer_uuid, guest_uuids=[])
def test_update_consumer_can_update_facts_with_empty_dict(self):
self.cp.updateConsumer(self.consumer_uuid, facts={})
def test_update_consumer_can_update_installed_products_with_empty_list(self):
self.cp.updateConsumer(self.consumer_uuid, installed_products=[])
def test_update_consumer_sets_hypervisor_id(self):
testing_hypervisor_id = random_string("testHypervisor")
self.cp.updateConsumer(self.consumer_uuid, hypervisor_id=testing_hypervisor_id)
hypervisor_id = self.cp.getConsumer(self.consumer_uuid)['hypervisorId']
# Hypervisor ID should be set and lower case
expected = testing_hypervisor_id.lower()
self.assertEqual(expected, hypervisor_id['hypervisorId'])
def test_create_consumer_sets_hypervisor_id(self):
testing_hypervisor_id = random_string("someId")
consumerInfo = self.cp.registerConsumer("other-test-consumer",
"system", owner="admin", hypervisor_id=testing_hypervisor_id)
# Unregister before making assertions, that way it should always happen
self.cp.unregisterConsumer(consumerInfo['uuid'])
# Hypervisor ID should be set and lower case
expected = testing_hypervisor_id.lower()
self.assertEqual(expected, consumerInfo['hypervisorId']['hypervisorId'])
def test_add_single_guest_id_string(self):
testing_guest_id = random_string("guestid")
self.cp.addOrUpdateGuestId(self.consumer_uuid, testing_guest_id)
guestIds = self.cp.getGuestIds(self.consumer_uuid)
self.assertEqual(1, len(guestIds))
def test_add_single_guest_id(self):
testing_guest_id = random_string("guestid")
guest_id_object = {"guestId": testing_guest_id, "attributes": {"some attr": "some value"}}
self.cp.addOrUpdateGuestId(self.consumer_uuid, guest_id_object)
guestId = self.cp.getGuestId(self.consumer_uuid, testing_guest_id)
# This check seems silly...
self.assertEqual(testing_guest_id, guestId['guestId'])
self.assertEqual('some value', guestId['attributes']['some attr'])
def test_remove_single_guest_id(self):
testing_guest_id = random_string("guestid")
self.cp.addOrUpdateGuestId(self.consumer_uuid, testing_guest_id)
guestIds = self.cp.getGuestIds(self.consumer_uuid)
self.assertEqual(1, len(guestIds))
# Delete the guestId
self.cp.removeGuestId(self.consumer_uuid, testing_guest_id)
# Check that no guestIds exist anymore
guestIds = self.cp.getGuestIds(self.consumer_uuid)
self.assertEqual(0, len(guestIds))
def test_update_single_guest_id(self):
testing_guest_id = random_string("guestid")
guest_id_object = {"guestId": testing_guest_id, "attributes": {"some attr": "some value"}}
self.cp.addOrUpdateGuestId(self.consumer_uuid, guest_id_object)
guestId = self.cp.getGuestId(self.consumer_uuid, testing_guest_id)
# check the guestId was created and has the expected attribute
self.assertEqual('some value', guestId['attributes']['some attr'])
guest_id_object['attributes']['some attr'] = 'crazy new value'
self.cp.addOrUpdateGuestId(self.consumer_uuid, guest_id_object)
guestId = self.cp.getGuestId(self.consumer_uuid, testing_guest_id)
# Verify there's still only one guestId
guestIds = self.cp.getGuestIds(self.consumer_uuid)
self.assertEqual(1, len(guestIds))
# Check that the attribute has changed
self.assertEqual('crazy new value', guestId['attributes']['some attr'])
def test_get_owner_hypervisors(self):
testing_hypervisor_id = random_string("testHypervisor")
self.cp.updateConsumer(self.consumer_uuid, hypervisor_id=testing_hypervisor_id)
self.cp.getConsumer(self.consumer_uuid)['hypervisorId']
hypervisors = self.cp.getOwnerHypervisors("admin", [testing_hypervisor_id])
#.........这里部分代码省略.........
开发者ID:Januson,项目名称:subscription-manager,代码行数:101,代码来源:connection_tests.py
示例14: ConnectionTests
class ConnectionTests(unittest.TestCase):
def setUp(self):
# Try to remove all environment variables to not influence unit test
try:
os.environ.pop('no_proxy')
os.environ.pop('NO_PROXY')
os.environ.pop('HTTPS_PROXY')
except KeyError:
pass
# NOTE: this won't actually work, idea for this suite of unit tests
# is to mock the actual server responses and just test logic in the
# UEPConnection:
self.cp = UEPConnection(username="dummy", password="dummy",
handler="/Test/", insecure=True)
self.temp_ent_dir = mkdtemp()
def tearDown(self):
shutil.rmtree(self.temp_ent_dir)
def test_accepts_a_timeout(self):
self.cp = UEPConnection(username="dummy", password="dummy",
handler="/Test/", insecure=True, timeout=3)
def test_load_manager_capabilities(self):
expected_capabilities = ['hypervisors_async', 'cores']
proper_status = {'version': '1',
'result': True,
'managerCapabilities': expected_capabilities}
improper_status = dict.copy(proper_status)
# Remove the managerCapabilities key from the dict
del improper_status['managerCapabilities']
self.cp.conn = Mock()
# The first call will return the proper_status, the second, the improper
# status
original_getStatus = self.cp.getStatus
self.cp.getStatus = Mock(side_effect=[proper_status,
improper_status])
actual_capabilities = self.cp._load_manager_capabilities()
self.assertEqual(sorted(actual_capabilities),
sorted(expected_capabilities))
self.assertEqual([], self.cp._load_manager_capabilities())
self.cp.getStatus = original_getStatus
def test_get_environment_by_name_requires_owner(self):
self.assertRaises(Exception, self.cp.getEnvironment, None, {"name": "env name"})
def test_get_environment_urlencoding(self):
self.cp.conn = Mock()
self.cp.conn.request_get = Mock(return_value=[])
self.cp.getEnvironment(owner_key="myorg", name="env name__++=*&")
self.cp.conn.request_get.assert_called_with(
"/owners/myorg/environments?name=env+name__%2B%2B%3D%2A%26")
def test_entitle_date(self):
self.cp.conn = Mock()
self.cp.conn.request_post = Mock(return_value=[])
self.cp.bind("abcd", date(2011, 9, 2))
self.cp.conn.request_post.assert_called_with(
"/consumers/abcd/entitlements?entitle_date=2011-09-02")
def test_no_entitle_date(self):
self.cp.conn = Mock()
self.cp.conn.request_post = Mock(return_value=[])
self.cp.bind("abcd")
self.cp.conn.request_post.assert_called_with("/consumers/abcd/entitlements")
def test_clean_up_prefix(self):
self.assertTrue(self.cp.handler == "/Test")
def test_https_proxy_info_allcaps(self):
with patch.dict('os.environ', {'HTTPS_PROXY': 'http://u:[email protected]:4444'}):
uep = UEPConnection(username="dummy", password="dummy",
handler="/Test/", insecure=True)
self.assertEqual("u", uep.proxy_user)
self.assertEqual("p", uep.proxy_password)
self.assertEqual("host", uep.proxy_hostname)
self.assertEqual(int("4444"), uep.proxy_port)
def test_order(self):
# should follow the order: HTTPS, https, HTTP, http
with patch.dict('os.environ', {'HTTPS_PROXY': 'http://u:[email protected]:4444',
'http_proxy': 'http://notme:[email protected]:2222'}):
uep = UEPConnection(username="dummy", password="dummy",
handler="/Test/", insecure=True)
self.assertEqual("u", uep.proxy_user)
self.assertEqual("p", uep.proxy_password)
self.assertEqual("host", uep.proxy_hostname)
self.assertEqual(int("4444"), uep.proxy_port)
def test_no_port(self):
with patch.dict('os.environ', {'HTTPS_PROXY': 'http://u:[email protected]'}):
uep = UEPConnection(username="dummy", password="dummy",
handler="/Test/", insecure=True)
self.assertEqual("u", uep.proxy_user)
self.assertEqual("p", uep.proxy_password)
self.assertEqual("host", uep.proxy_hostname)
self.assertEqual(3128, uep.proxy_port)
def test_no_user_or_password(self):
with patch.dict('os.environ', {'HTTPS_PROXY': 'http://host:1111'}):
#.........这里部分代码省略.........
开发者ID:Lorquas,项目名称:subscription-manager,代码行数:101,代码来源:connection-tests.py
示例15: OptionParser
#!/usr/bin/env python
# usage:
# vsphere-virt-who-simulator.py -o ACME_Corporation -e Dev host1:guest1,guest2 host3:guest3,guest4
from rhsm.connection import UEPConnection
from optparse import OptionParser
parser = OptionParser()
parser.add_option("-o", "--org", default="ACME_Corporation")
parser.add_option("-e", "--env", default="Dev")
[options, args] = parser.parse_args()
conn = UEPConnection(cert_file="/etc/pki/consumer/cert.pem",key_file="/etc/pki/consumer/key.pem", insecure=True)
# takes array in format ["host1:guest1,guest2","host2:guest3,guest4"]
# returns dict {"host1": ["guest1","guest2"], "host2": ["guest3","guest4"]}
mapping = dict([[host, guests.split(",")] for
[host, guests] in [arg.split(":")
for arg in args]])
print conn.hypervisorCheckIn(options.org, options.env, mapping)
开发者ID:AdamSaleh,项目名称:katello,代码行数:23,代码来源:vsphere-virt-who-simulator.py
示例16: setUp
def setUp(self):
# NOTE: this won't actually work, idea for this suite of unit tests
# is to mock the actual server responses and just test logic in the
# UEPConnection:
self.cp = UEPConnection(username="dummy", password="dummy",
insecure=True)
开发者ID:wottop,项目名称:python-rhsm,代码行数:6,代码来源:connection-tests.py
示例17: EntitlementRegenerationTests
class EntitlementRegenerationTests(unittest.TestCase):
def setUp(self):
self.cp = UEPConnection(username="admin", password="admin", insecure=True)
self.consumer = self.cp.registerConsumer("test-consumer", "system", owner="admin")
self.consumer_uuid = self.consumer['uuid']
# This product is present in the Candlepin test data
self.cp.bindByProduct(self.consumer_uuid, ["awesomeos-instancebased"])
entitlements = self.cp.getEntitlementList(self.consumer_uuid)
self.assertTrue(len(entitlements) > 0)
self.entitlement = entitlements[0]
self.entitlement_id = self.entitlement['id']
def test_regenerate_entitlements_default(self):
result = self.cp.regenEntitlementCertificates(self.consumer_uuid)
self.assertTrue(result)
def test_regenerate_entitlements_lazy(self):
result = self.cp.regenEntitlementCertificates(self.consumer_uuid, True)
self.assertTrue(result)
def test_regenerate_entitlements_eager(self):
result = self.cp.regenEntitlementCertificates(self.consumer_uuid, False)
self.assertTrue(result)
def test_regenerate_entitlements_bad_uuid(self):
result = self.cp.regenEntitlementCertificates("bad_consumer_uuid")
self.assertFalse(result)
def test_regenerate_entitlement_default(self):
result = self.cp.regenEntitlementCertificate(self.consumer_uuid, self.entitlement_id)
self.assertTrue(result)
def test_regenerate_entitlement_lazy(self):
result = self.cp.regenEntitlementCertificate(self.consumer_uuid, self.entitlement_id, True)
self.assertTrue(result)
def test_regenerate_entitlement_eager(self):
result = self.cp.regenEntitlementCertificate(self.consumer_uuid, self.entitlement_id, False)
self.assertTrue(result)
def test_regenerate_entitlement_bad_consumer_uuid(self):
result = self.cp.regenEntitlementCertificate("bad_consumer_uuid", self.entitlement_id)
self.assertFalse(result)
def test_regenerate_entitlement_bad_entitlement_id(self):
result = self.cp.regenEntitlementCertificate(self.consumer_uuid, "bad_entitlement_id")
self.assertFalse(result)
def tearDown(self):
self.cp.unregisterConsumer(self.consumer_uuid)
开发者ID:Januson,项目名称:subscription-manager,代码行数:54,代码来源:connection_tests.py
示例18: __init__
def __init__(self):
self.conn_username = "admin"
self.owner = "admin"
self.cp = UEPConnection(username=self.conn_username, password="admin",
host="localhost", ssl_port=8443,
handler="/candlepin", insecure=True)
开发者ID:beav,项目名称:spacewalk-splice-tool,代码行数:6,代码来源:cp_client.py
示例19: test_accepts_a_timeout
def test_accepts_a_timeout(self):
self.cp = UEPConnection(username="dummy", password="dummy",
handler="/Test/", insecure=True, timeout=3)
开发者ID:Lorquas,项目名称:subscription-manager,代码行数:3,代码来源:connection-tests.py
示例20: __init__
def __init__(self):
key = ConsumerIdentity.keypath()
cert = ConsumerIdentity.certpath()
UEPConnection.__init__(self, key_file=key, cert_file=cert)
开发者ID:Katello,项目名称:katello-agent,代码行数:4,代码来源:plugin.py
注:本文中的rhsm.connection.UEPConnection类示例由纯净天空整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。 |
请发表评论