Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_paramiko_connection(_):
proxmox = ProxmoxAPI('proxmox', user='root', backend='ssh_paramiko', port=123)
session = proxmox._store['session']
eq_(session.ssh_client.connect.call_args[0], ('proxmox',))
eq_(session.ssh_client.connect.call_args[1], {'username': 'root',
'allow_agent': True,
'key_filename': None,
'look_for_keys': True,
'timeout': 5,
'password': None,
'port': 123})
def setUp(self, _):
super(TestParamikoSuiteWithSudo, self).__init__(sudo=True)
self.proxmox = ProxmoxAPI('proxmox', user='root', backend='ssh_paramiko', port=123, sudo=True)
self.client = self.proxmox._store['session'].ssh_client
self.session = self.client.get_transport().open_session()
self._set_stderr('200 OK')
self._set_stdout('')
def test_https_connection(req_session):
response = {'ticket': 'ticket',
'CSRFPreventionToken': 'CSRFPreventionToken'}
req_session.request.return_value = response
ProxmoxAPI('proxmox', user='root@pam', password='secret', port=123, verify_ssl=False)
call = req_session.return_value.request.call_args[1]
eq_(call['url'], 'https://proxmox:123/api2/json/access/ticket')
eq_(call['data'], {'username': 'root@pam', 'password': 'secret'})
eq_(call['method'], 'post')
eq_(call['verify'], False)
def test_https_connection_wth_port_in_host(req_session):
response = {'ticket': 'ticket',
'CSRFPreventionToken': 'CSRFPreventionToken'}
req_session.request.return_value = response
ProxmoxAPI('proxmox:123', user='root@pam', password='secret', port=124, verify_ssl=False)
call = req_session.return_value.request.call_args[1]
eq_(call['url'], 'https://proxmox:123/api2/json/access/ticket')
eq_(call['data'], {'username': 'root@pam', 'password': 'secret'})
eq_(call['method'], 'post')
eq_(call['verify'], False)
if userresource is None:
raise ValueError("No proxmox user found!! Please use proxmoxutil command to update user credentials")
user = userresource.properties[PROPERTIES_USER]
password = userresource.properties[PROPERTIES_PASSWORD]
authrealm = userresource.properties[PROPERTIES_AUTHREALM]
puser = user+'@'+authrealm
primary = proxmoxutil.listprimary()
if primary is None:
raise ValueError("Primary proxmox hypervisor not found!! Please use proxmoxutil command to update primary hypervisor")
hypervisor = primary.properties[PROPERTIES_HYPERVISOR]
print "Authenticating "+puser +" on "+ hypervisor
proxmox = ProxmoxAPI(hypervisor, user=puser, password=password, verify_ssl=False)
node = proxmox.nodes(hostresource.properties[HYPERVISOR])
hostname = hostresource.properties[HOSTNAME]
vmid = int(hostresource.properties[HOSTID])
ostemplate = str(hostresource.properties[PROPERTIES_OSTEMPLATE])
cpulimit = int(hostresource.properties[PROPERTIES_CPULIMIT])
cpuunits = int(hostresource.properties[PROPERTIES_CPUUNITS])
memory = int(hostresource.properties[PROPERTIES_MEMORY])
swap = int(hostresource.properties[PROPERTIES_SWAP])
storage = hostresource.properties[PROPERTIES_STORAGE]
disk = int(hostresource.properties[PROPERTIES_DISK])
disksize="%dG"%(disk)
interfaces = hostresource.properties[INTERFACES]
i=0
netconfig = dict()
def setUp(self, _):
self.proxmox = ProxmoxAPI('proxmox', user='root', backend='ssh_paramiko', port=123)
self.client = self.proxmox._store['session'].ssh_client
self.session = self.client.get_transport().open_session()
self._set_stderr('200 OK')
self._set_stdout('')
swap = module.params['swap']
storage = module.params['storage']
hostname = module.params['hostname']
if module.params['ostemplate'] is not None:
template_store = module.params['ostemplate'].split(":")[0]
timeout = module.params['timeout']
# If password not set get it from PROXMOX_PASSWORD env
if not api_password:
try:
api_password = os.environ['PROXMOX_PASSWORD']
except KeyError as e:
module.fail_json(msg='You should set api_password param or use PROXMOX_PASSWORD environment variable')
try:
proxmox = ProxmoxAPI(api_host, user=api_user, password=api_password, verify_ssl=validate_certs)
global VZ_TYPE
VZ_TYPE = 'openvz' if float(proxmox.version.get()['version']) < 4.0 else 'lxc'
except Exception as e:
module.fail_json(msg='authorization on proxmox cluster failed with exception: %s' % e)
# If vmid not set get the Next VM id from ProxmoxAPI
# If hostname is set get the VM id from ProxmoxAPI
if not vmid and state == 'present':
vmid = get_nextvmid(module, proxmox)
elif not vmid and hostname:
hosts = get_vmid(proxmox, hostname)
if len(hosts) == 0:
module.fail_json(msg="Vmid could not be fetched => Hostname doesn't exist (action: %s)" % state)
vmid = hosts[0]
elif not vmid:
api_host = module.params['api_host']
api_password = module.params['api_password']
validate_certs = module.params['validate_certs']
node = module.params['node']
storage = module.params['storage']
timeout = module.params['timeout']
# If password not set get it from PROXMOX_PASSWORD env
if not api_password:
try:
api_password = os.environ['PROXMOX_PASSWORD']
except KeyError as e:
module.fail_json(msg='You should set api_password param or use PROXMOX_PASSWORD environment variable')
try:
proxmox = ProxmoxAPI(api_host, user=api_user, password=api_password, verify_ssl=validate_certs)
except Exception as e:
module.fail_json(msg='authorization on proxmox cluster failed with exception: %s' % e)
if state == 'present':
try:
content_type = module.params['content_type']
src = module.params['src']
template = os.path.basename(src)
if get_template(proxmox, node, storage, content_type, template) and not module.params['force']:
module.exit_json(changed=False, msg='template with volid=%s:%s/%s is already exists' % (storage, content_type, template))
elif not src:
module.fail_json(msg='src param to uploading template file is mandatory')
elif not (os.path.exists(src) and os.path.isfile(src)):
module.fail_json(msg='template file on path %s not exists' % src)
parser.add_argument('--storage', '-s', type=str,
default='dir', choices=['dir', 'lvm'],
help='Virtual machine storage backend')
args = parser.parse_args()
vmid = str(args.vmid)
name = args.name
flavor_type = args.flavor
storage_type = args.storage
# customize flavor
instance, flavor = instance_customize(instance, vmid, name, flavor_type)
# proxmoxer initialize
proxmox_api = ProxmoxAPI(proxmox['host'], user=proxmox['user'],
password=proxmox['password'],
verify_ssl=proxmox['verify_ssl'])
node = proxmox_api.nodes(proxmox['node'])
# create kvm machine
node.qemu.create(vmid=vmid, name=name, sockets=flavor['sockets'],
cores=flavor['cores'], balloon=flavor['balloon'],
memory=flavor['memory'], net0=instance['net'])
# seeding
seed(vmid, instance, proxmox)
# set seed iso
node.qemu(vmid).config.set(virtio1=instance['hd_seed'])