本文整理汇总了Python中marvin.lib.base.PublicIPAddress类的典型用法代码示例。如果您正苦于以下问题：Python PublicIPAddress类的具体用法？Python PublicIPAddress怎么用？Python PublicIPAddress使用的例子？那么恭喜您, 这里精选的类代码示例或许可以为您提供帮助。
在下文中一共展示了PublicIPAddress类的20个代码示例，这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞，您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: test_04_publicip_per_project
def test_04_publicip_per_project(self):
"""Test Public IP limit per project
"""
# Validate the following
# 1. set max no of IPs per project to 2.
# 2. Create an account in this domain
# 3. Create 1 VM in this domain
# 4. Acquire 1 IP in the domain. IP should be successfully acquired
# 5. Try to acquire 3rd IP in this domain. It should give the user an
# appropriate error and an alert should be generated.
self.debug("Updating public IP resource limits for project: %s" % self.project.id)
# Set usage_vm=1 for Account 1
update_resource_limit(self.apiclient, 1, max=2, projectid=self.project.id) # Public Ip
self.debug("Deploying VM for Project: %s" % self.project.id)
virtual_machine_1 = VirtualMachine.create(
self.apiclient,
self.services["server"],
templateid=self.template.id,
serviceofferingid=self.service_offering.id,
projectid=self.project.id,
)
self.cleanup.append(virtual_machine_1)
# Verify VM state
self.assertEqual(virtual_machine_1.state, "Running", "Check VM state is Running or not")
networks = Network.list(self.apiclient, projectid=self.project.id, listall=True)
self.assertEqual(isinstance(networks, list), True, "Check list networks response returns a valid response")
self.assertNotEqual(len(networks), 0, "Check list networks response returns a valid network")
network = networks[0]
self.debug("Associating public IP for project: %s" % self.project.id)
public_ip_1 = PublicIPAddress.create(
self.apiclient,
zoneid=virtual_machine_1.zoneid,
services=self.services["server"],
networkid=network.id,
projectid=self.project.id,
)
self.cleanup.append(public_ip_1)
# Verify Public IP state
self.assertEqual(
public_ip_1.ipaddress.state in ["Allocated", "Allocating"],
True,
"Check Public IP state is allocated or not",
)
# Exception should be raised for second Public IP
with self.assertRaises(Exception):
PublicIPAddress.create(
self.apiclient,
zoneid=virtual_machine_1.zoneid,
services=self.services["server"],
networkid=network.id,
projectid=self.project.id,
)
return
开发者ID:vaddanak，项目名称:challenges，代码行数:56，代码来源:test_project_limits.py
示例2: isIpInDesiredState
def isIpInDesiredState(apiclient, ipaddressid, state):
""" Check if the given IP is in the correct state (given)
and return True/False accordingly"""
retriesCount = 10
ipInDesiredState = False
exceptionOccured = False
exceptionMessage = ""
try:
while retriesCount >= 0:
portableips = PublicIPAddress.list(apiclient, id=ipaddressid)
assert validateList(
portableips)[0] == PASS, "IPs list validation failed"
if str(portableips[0].state).lower() == state:
ipInDesiredState = True
break
retriesCount -= 1
time.sleep(60)
except Exception as e:
exceptionOccured = True
exceptionMessage = e
return [exceptionOccured, ipInDesiredState, e]
if not ipInDesiredState:
exceptionMessage = "Ip should be in %s state, it is in %s" %\
(state, portableips[0].state)
return [False, ipInDesiredState, exceptionMessage]
开发者ID:aali-dincloud，项目名称:cloudstack，代码行数:25，代码来源:common.py
示例3: setUp
def setUp(self):
try:
self.apiclient = self.testClient.getApiClient()
self.dbclient = self.testClient.getDbConnection()
self.account = Account.create(
self.apiclient,
self.services["account"],
domainid=self.domain.id
)
self.cleanup = [
self.account,
]
self.virtual_machine = VirtualMachine.create(
self.apiclient,
self.services["virtual_machine"],
templateid=self.template.id,
accountid=self.account.name,
domainid=self.account.domainid,
serviceofferingid=self.service_offering.id
)
self.public_ip = PublicIPAddress.create(
self.apiclient,
accountid=self.virtual_machine.account,
zoneid=self.virtual_machine.zoneid,
domainid=self.virtual_machine.domainid,
services=self.services["virtual_machine"]
)
return
except CloudstackAPIException as e:
self.tearDown()
raise e
开发者ID:CIETstudents，项目名称:cloudstack，代码行数:31，代码来源:test_vpn_users.py
示例4: acquire_Public_Ip
def acquire_Public_Ip(self):
"""Acquires the public IP"""
try:
self.debug("Acquiring public IP for account: %s" %
self.account.name)
public_ip = PublicIPAddress.create(
self.apiclient,
self.virtual_machine.account,
self.virtual_machine.zoneid,
self.virtual_machine.domainid,
self.services["virtual_machine"]
)
self.debug("Acquired public IP: %s" %
public_ip.ipaddress.ipaddress)
FireWallRule.create(
self.apiclient,
ipaddressid=public_ip.ipaddress.id,
protocol='TCP',
cidrlist=[self.services["fwrule"]["cidr"]],
startport=self.services["fwrule"]["startport"],
endport=self.services["fwrule"]["endport"]
)
return public_ip
except Exception as e:
self.fail("Failed to acquire new public IP: %s" % e)
开发者ID:K0zka，项目名称:cloudstack，代码行数:27，代码来源:test_haproxy.py
示例5: create_vm
def create_vm(self, pfrule=False, egress_policy=True, RR=False):
self.create_network_offering(egress_policy, RR)
# Creating network using the network offering created
self.debug("Creating network with network offering: %s" % self.network_offering.id)
self.network = Network.create(
self.apiclient,
self.services["network"],
accountid=self.account.name,
domainid=self.account.domainid,
networkofferingid=self.network_offering.id,
zoneid=self.zone.id,
)
self.debug("Created network with ID: %s" % self.network.id)
self.debug("Deploying instance in the account: %s" % self.account.name)
project = None
self.virtual_machine = VirtualMachine.create(
self.apiclient,
self.services["virtual_machine"],
accountid=self.account.name,
domainid=self.domain.id,
serviceofferingid=self.service_offering.id,
mode=self.zone.networktype if pfrule else "basic",
networkids=[str(self.network.id)],
projectid=project.id if project else None,
)
self.debug("Deployed instance %s in account: %s" % (self.virtual_machine.id, self.account.name))
# Checking if VM is running or not, in case it is deployed in error state, test case fails
self.vm_list = list_virtual_machines(self.apiclient, id=self.virtual_machine.id)
self.assertEqual(validateList(self.vm_list)[0], PASS, "vm list validation failed, vm list is %s" % self.vm_list)
self.assertEqual(
str(self.vm_list[0].state).lower(),
"running",
"VM state should be running, it is %s" % self.vm_list[0].state,
)
self.public_ip = PublicIPAddress.create(
self.apiclient,
accountid=self.account.name,
zoneid=self.zone.id,
domainid=self.account.domainid,
networkid=self.network.id,
)
# Open up firewall port for SSH
FireWallRule.create(
self.apiclient,
ipaddressid=self.public_ip.ipaddress.id,
protocol=self.services["natrule"]["protocol"],
cidrlist=["0.0.0.0/0"],
startport=self.services["natrule"]["publicport"],
endport=self.services["natrule"]["publicport"],
)
self.debug("Creating NAT rule for VM ID: %s" % self.virtual_machine.id)
# Create NAT rule
NATRule.create(self.apiclient, self.virtual_machine, self.services["natrule"], self.public_ip.ipaddress.id)
return
开发者ID:tianshangjun，项目名称:cloudstack，代码行数:60，代码来源:test_egress_fw_rules.py
示例6: validate_PublicIPAddress
def validate_PublicIPAddress(self, public_ip, network, static_nat=False,
vm=None):
"""Validates the Public IP Address"""
self.debug("Validating the assignment and state of public IP address "
"- %s" % public_ip.ipaddress.ipaddress)
public_ips = PublicIPAddress.list(self.api_client,
id=public_ip.ipaddress.id,
networkid=network.id,
isstaticnat=static_nat,
listall=True
)
self.assertEqual(isinstance(public_ips, list), True,
"List public IP for network should return a "
"valid list"
)
self.assertEqual(public_ips[0].ipaddress,
public_ip.ipaddress.ipaddress,
"List public IP for network should list the assigned "
"public IP address"
)
self.assertEqual(public_ips[0].state, "Allocated",
"Assigned public IP is not in the allocated state"
)
if static_nat and vm:
self.assertEqual(public_ips[0].virtualmachineid, vm.id,
"Static NAT rule is not enabled for the VM on "
"the assigned public IP"
)
self.debug("Successfully validated the assignment and state of public "
"IP address - %s" % public_ip.ipaddress.ipaddress)
开发者ID:prashanthvarma，项目名称:cloudstack，代码行数:30，代码来源:nuageTestCase.py
示例7: setUp
def setUp(self):
self.apiclient = self.testClient.getApiClient()
self.dbclient = self.testClient.getDbConnection()
self.account = Account.create(self.apiclient, self.services["account"], domainid=self.domain.id)
self.virtual_machine = VirtualMachine.create(
self.apiclient,
self.services["virtual_machine"],
templateid=self.template.id,
accountid=self.account.name,
domainid=self.account.domainid,
serviceofferingid=self.service_offering.id,
)
self.virtual_machine_2 = VirtualMachine.create(
self.apiclient,
self.services["virtual_machine"],
templateid=self.template.id,
accountid=self.account.name,
domainid=self.account.domainid,
serviceofferingid=self.service_offering.id,
)
self.public_ip = PublicIPAddress.create(
self.apiclient,
self.virtual_machine.account,
self.virtual_machine.zoneid,
self.virtual_machine.domainid,
self.services["virtual_machine"],
)
NATRule.create(
self.apiclient, self.virtual_machine, self.services["natrule"], ipaddressid=self.public_ip.ipaddress.id
)
self.cleanup = [self.account]
return
开发者ID:ktenzer，项目名称:cloudstack，代码行数:35，代码来源:test_haproxy.py
示例8: createNetworkRulesForVM
def createNetworkRulesForVM(apiclient, virtualmachine, ruletype,
account, networkruledata):
"""Acquire IP, create Firewall and NAT/StaticNAT rule
(associating it with given vm) for that IP"""
try:
public_ip = PublicIPAddress.create(
apiclient,accountid=account.name,
zoneid=virtualmachine.zoneid,domainid=account.domainid,
networkid=virtualmachine.nic[0].networkid)
FireWallRule.create(
apiclient,ipaddressid=public_ip.ipaddress.id,
protocol='TCP', cidrlist=[networkruledata["fwrule"]["cidr"]],
startport=networkruledata["fwrule"]["startport"],
endport=networkruledata["fwrule"]["endport"]
)
if ruletype == NAT_RULE:
# Create NAT rule
NATRule.create(apiclient, virtualmachine,
networkruledata["natrule"],ipaddressid=public_ip.ipaddress.id,
networkid=virtualmachine.nic[0].networkid)
elif ruletype == STATIC_NAT_RULE:
# Enable Static NAT for VM
StaticNATRule.enable(apiclient,public_ip.ipaddress.id,
virtualmachine.id, networkid=virtualmachine.nic[0].networkid)
except Exception as e:
[FAIL, e]
return [PASS, public_ip]
开发者ID:aali-dincloud，项目名称:cloudstack，代码行数:30，代码来源:common.py
示例9: acquire_Public_Ip
def acquire_Public_Ip(self):
"""Acquires the public IP"""
try:
self.debug("Acquiring public IP for account: %s" %
self.account.name)
public_ip = PublicIPAddress.create(
self.apiclient,
self.virtual_machine.account,
self.virtual_machine.zoneid,
self.virtual_machine.domainid,
self.services["virtual_machine"]
)
self.debug("Acquired public IP: %s" %
public_ip.ipaddress.ipaddress)
self.debug("Configuring NAT rule for the acquired public ip")
NATRule.create(
self.apiclient,
self.virtual_machine,
self.services["natrule"],
ipaddressid=public_ip.ipaddress.id
)
return public_ip
except Exception as e:
self.fail("Failed to acquire new public IP: %s" % e)
开发者ID:diejiazhao，项目名称:cloudstack，代码行数:28，代码来源:test_haproxy.py
示例10: acquire_PublicIPAddress
def acquire_PublicIPAddress(self, network, vpc=None):
self.debug("Associating public IP for network with ID - %s" % network.id)
public_ip = PublicIPAddress.create(self.api_client,
accountid=self.account.name,
zoneid=self.zone.id,
domainid=self.account.domainid,
networkid=network.id if vpc is None else None,
vpcid=vpc.id if vpc else self.vpc.id if hasattr(self, "vpc") else None
)
self.debug("Associated public IP address - %s with network with ID - %s" %
(public_ip.ipaddress.ipaddress, network.id))
return public_ip
开发者ID:CIETstudents，项目名称:cloudstack，代码行数:12，代码来源:nuageTestCase.py
示例11: acquire_publicip
def acquire_publicip(self, network):
self.logger.debug('Associating public IP for network: %s' % network.name)
public_ip = PublicIPAddress.create(
self.api_client,
accountid='admin',
zoneid=self.zone.id,
domainid=self.domain.id,
networkids=[str(network.id)]
)
self.logger.debug('Associated %s with network %s' % (public_ip.ipaddress.ipaddress, network.id))
self.test_cleanup.append(public_ip)
return public_ip
开发者ID:EdwardBetts，项目名称:blackhole，代码行数:12，代码来源:test_nicira.py
示例12: deploy_isolatednetwork_publicipaddress
def deploy_isolatednetwork_publicipaddress(self, ipaddress_data, virtualmachines_data, network):
self.logger.debug('>>> ISOLATED NETWORK PUBLIC IP ADDRESS => Creating...')
publicipaddress = PublicIPAddress.create(
api_client=self.api_client,
data=ipaddress_data,
network=network
)
self.logger.debug('>>> ISOLATED NETWORK PUBLIC IP ADDRESS => Created! TODO: MISSING FIELDS!')
self.deploy_firewallrules(ipaddress_data, publicipaddress)
self.deploy_portforwards(ipaddress_data['portforwards'], virtualmachines_data, None, publicipaddress)
开发者ID:MissionCriticalCloud，项目名称:cosmic，代码行数:12，代码来源:testScenarioManager.py
示例13: test_public_ip_admin_account
def test_public_ip_admin_account(self):
"""Test for Associate/Disassociate public IP address for admin account"""
# Validate the following:
# 1. listPubliIpAddresses API returns the list of acquired addresses
# 2. the returned list should contain our acquired IP address
ip_address = PublicIPAddress.create(
self.apiclient,
self.account.name,
self.zone.id,
self.account.domainid
)
list_pub_ip_addr_resp = list_publicIP(
self.apiclient,
id=ip_address.ipaddress.id
)
self.assertEqual(
isinstance(list_pub_ip_addr_resp, list),
True,
"Check list response returns a valid list"
)
# listPublicIpAddresses should return newly created public IP
self.assertNotEqual(
len(list_pub_ip_addr_resp),
0,
"Check if new IP Address is associated"
)
self.assertEqual(
list_pub_ip_addr_resp[0].id,
ip_address.ipaddress.id,
"Check Correct IP Address is returned in the List Cacls"
)
ip_address.delete(self.apiclient)
time.sleep(30)
# Validate the following:
# 1.listPublicIpAddresses should no more return the released address
list_pub_ip_addr_resp = list_publicIP(
self.apiclient,
id=ip_address.ipaddress.id
)
if list_pub_ip_addr_resp is None:
return
if (list_pub_ip_addr_resp) and (
isinstance(
list_pub_ip_addr_resp,
list)) and (
len(list_pub_ip_addr_resp) > 0):
self.fail("list public ip response is not empty")
return
开发者ID:Tosta-Mixta，项目名称:cloudstack，代码行数:52，代码来源:test_network.py
示例14: acquire_publicip
def acquire_publicip(self, vpc, network):
self.logger.debug("Associating public IP for network: %s" % network.name)
public_ip = PublicIPAddress.create(
self.apiclient,
accountid=self.account.name,
zoneid=self.zone.id,
domainid=self.account.domainid,
networkid=network.id,
vpcid=vpc.id,
)
self.assertIsNotNone(public_ip, "Failed to acquire public IP")
self.logger.debug("Associated %s with network %s" % (public_ip.ipaddress.ipaddress, network.id))
return public_ip
开发者ID:shapeblue，项目名称:Trillian，代码行数:13，代码来源:test_internal_lb.py
示例15: acquire_publicip
def acquire_publicip(self, network):
self.debug("Associating public IP for network: %s" % network.name)
public_ip = PublicIPAddress.create(self.apiclient,
accountid=self.account.name,
zoneid=self.zone.id,
domainid=self.account.domainid,
networkid=network.id,
vpcid=self.vpc.id
)
self.debug("Associated %s with network %s" % (public_ip.ipaddress.ipaddress,
network.id
))
return public_ip
开发者ID:abnercosta，项目名称:cloudstack，代码行数:13，代码来源:test_vpc_network_pfrules.py
示例16: test_public_ip_user_account
def test_public_ip_user_account(self):
"""Test for Associate/Disassociate public IP address for user account"""
# Validate the following:
# 1. listPubliIpAddresses API returns the list of acquired addresses
# 2. the returned list should contain our acquired IP address
ip_address = PublicIPAddress.create(
self.apiclient,
self.user.name,
self.zone.id,
self.user.domainid
)
# listPublicIpAddresses should return newly created public IP
list_pub_ip_addr_resp = list_publicIP(
self.apiclient,
id=ip_address.ipaddress.id
)
self.assertEqual(
isinstance(list_pub_ip_addr_resp, list),
True,
"Check list response returns a valid list"
)
self.assertNotEqual(
len(list_pub_ip_addr_resp),
0,
"Check if new IP Address is associated"
)
self.assertEqual(
list_pub_ip_addr_resp[0].id,
ip_address.ipaddress.id,
"Check Correct IP Address is returned in the List Call"
)
ip_address.delete(self.apiclient)
list_pub_ip_addr_resp = list_publicIP(
self.apiclient,
id=ip_address.ipaddress.id
)
self.assertEqual(
list_pub_ip_addr_resp,
None,
"Check if disassociated IP Address is no longer available"
)
return
开发者ID:Tosta-Mixta，项目名称:cloudstack，代码行数:48，代码来源:test_network.py
示例17: setUpClass
def setUpClass(cls):
cls.testClient = super(TestNATRules, cls).getClsTestClient()
cls.api_client = cls.testClient.getApiClient()
cls.services = Services().services
# Get Zone, Domain and templates
cls.domain = get_domain(cls.api_client)
cls.zone = get_zone(cls.api_client, cls.testClient.getZoneForTests())
cls.services['mode'] = cls.zone.networktype
template = get_template(
cls.api_client,
cls.zone.id,
cls.services["ostype"]
)
#Create an account, network, VM and IP addresses
cls.account = Account.create(
cls.api_client,
cls.services["account"],
admin=True,
domainid=cls.domain.id
)
cls.services["virtual_machine"]["zoneid"] = cls.zone.id
cls.service_offering = ServiceOffering.create(
cls.api_client,
cls.services["service_offering"]
)
cls.virtual_machine = VirtualMachine.create(
cls.api_client,
cls.services["virtual_machine"],
templateid=template.id,
accountid=cls.account.name,
domainid=cls.account.domainid,
serviceofferingid=cls.service_offering.id
)
cls.public_ip = PublicIPAddress.create(
cls.api_client,
accountid=cls.account.name,
zoneid=cls.zone.id,
domainid=cls.account.domainid,
services=cls.services["virtual_machine"]
)
cls._cleanup = [
cls.virtual_machine,
cls.account,
cls.service_offering
]
开发者ID:MountHuang，项目名称:cloudstack，代码行数:48，代码来源:test_blocker_bugs.py
示例18: deploy_publicipaddress
def deploy_publicipaddress(self, publicipaddress_data, virtualmachines_data, vpc):
self.logger.debug('>>> PUBLIC IP ADDRESS => Creating...')
publicipaddress = PublicIPAddress.create(
api_client=self.api_client,
data=publicipaddress_data,
vpc=vpc
)
ipaddress = publicipaddress.ipaddress
self.logger.debug('>>> PUBLIC IP ADDRESS => ID: %s => IP: %s => State: %s => Source NAT: %s '
'=> Static NAT: %s => ACL: %s => VLAN: %s => Physical Network: %s => Network: %s '
'=> VPC: %s => Domain: %s', ipaddress.id, ipaddress.ipaddress, ipaddress.state,
ipaddress.issourcenat, ipaddress.isstaticnat, ipaddress.aclid, ipaddress.vlanid,
ipaddress.physicalnetworkid, ipaddress.networkid, ipaddress.vpcid, ipaddress.domainid)
self.deploy_portforwards(publicipaddress_data['portforwards'], virtualmachines_data, vpc, publicipaddress)
开发者ID:MissionCriticalCloud，项目名称:cosmic，代码行数:16，代码来源:testScenarioManager.py
示例19: setUp
def setUp(self):
self.apiclient = self.testClient.getApiClient()
self.dbclient = self.testClient.getDbConnection()
self.account = Account.create(
self.apiclient,
self.services["account"],
domainid=self.domain.id
)
self.virtual_machine = VirtualMachine.create(
self.apiclient,
self.services["virtual_machine"],
templateid=self.template.id,
accountid=self.account.name,
domainid=self.account.domainid,
serviceofferingid=self.service_offering.id
)
self.virtual_machine_2 = VirtualMachine.create(
self.apiclient,
self.services["virtual_machine"],
templateid=self.template.id,
accountid=self.account.name,
domainid=self.account.domainid,
serviceofferingid=self.service_offering.id
)
self.public_ip = PublicIPAddress.create(
self.apiclient,
self.virtual_machine.account,
self.virtual_machine.zoneid,
self.virtual_machine.domainid,
self.services["virtual_machine"]
)
FireWallRule.create(
self.apiclient,
ipaddressid=self.public_ip.ipaddress.id,
protocol='TCP',
cidrlist=[self.services["fwrule"]["cidr"]],
startport=self.services["fwrule"]["startport"],
endport=self.services["fwrule"]["endport"]
)
self.cleanup = [self.account, ]
return
开发者ID:K0zka，项目名称:cloudstack，代码行数:42，代码来源:test_haproxy.py
示例20: validate_network_rules
def validate_network_rules(self):
""" Validate network rules
"""
vms = VirtualMachine.list(
self.api_client,
account=self.account.name,
domainid=self.account.domainid,
listall=True
)
public_ips = PublicIPAddress.list(
self.api_client,
account=self.account.name,
domainid=self.account.domainid,
listall=True
)
for vm, public_ip in zip(vms, public_ips):
try:
ssh_1 = vm.get_ssh_client(
ipaddress=public_ip.ipaddress.ipaddress)
self.debug("SSH into VM is successfully")
self.debug(
"Verifying if we can ping to outside world from VM?")
# Ping to outsite world
res = ssh_1.execute("ping -c 1 www.google.com")
# res = 64 bytes from maa03s17-in-f20.1e100.net (74.125.236.212):
# icmp_req=1 ttl=57 time=25.9 ms
# --- www.l.google.com ping statistics ---
# 1 packets transmitted, 1 received, 0% packet loss, time 0ms
# rtt min/avg/max/mdev = 25.970/25.970/25.970/0.000 ms
except Exception as e:
self.fail("Failed to SSH into VM - %s, %s" %
(public_ip.ipaddress.ipaddress, e))
result = str(res)
self.assertEqual(
result.count("1 received"),
1,
"Ping to outside world from VM should be successful"
)
开发者ID:EdwardBetts，项目名称:blackhole，代码行数:40，代码来源:test_vpc_routers.py
注：本文中的marvin.lib.base.PublicIPAddress类示例由纯净天空整理自Github/MSDocs等源码及文档管理平台，相关代码片段筛选自各路编程大神贡献的开源项目，源码版权归原作者所有，传播和使用请参考对应项目的License；未经允许，请勿转载。 |
客服电话
APP下载
官方微信
请发表评论