PK-sGN:  pylxd/profiles.py# Copyright (c) 2015 Canonical Ltd # # Licensed 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. import json from pylxd import base class LXDProfile(base.LXDBase): def profile_list(self): '''List profiles on the LXD daemon as an array.''' (state, data) = self.connection.get_object('GET', '/1.0/profiles') return [profiles.split('/1.0/profiles/')[-1] for profiles in data['metadata']] def profile_create(self, profile): '''Create an LXD Profile''' return self.connection.get_status('POST', '/1.0/profiles', json.dumps(profile)) def profile_show(self, profile): '''Display the LXD profile''' return self.connection.get_object('GET', '/1.0/profiles/%s' % profile) def profile_defined(self, profile): '''Check for an LXD profile''' return self.connection.get_status('GET', '/1.0/profiles/%s' % profile) def profile_update(self, profile, config): '''Update the LXD profile (not implemented)''' return self.connection.get_status('PUT', '/1.0/profiles/%s' % profile, json.dumps(config)) def profile_rename(self, profile, config): '''Rename the LXD profile''' return self.connection.get_status('POST', '/1.0/profiles/%s' % profile, json.dumps(config)) def profile_delete(self, profile): '''Delete the LXD profile''' return self.connection.get_status('DELETE', '/1.0/profiles/%s' % profile) PK-sGwpylxd/hosts.py# Copyright (c) 2015 Canonical Ltd # # Licensed 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 print_function from pylxd import base from pylxd import exceptions class LXDHost(base.LXDBase): def host_ping(self): try: return self.connection.get_status('GET', '/1.0') except Exception as e: msg = 'LXD service is unavailable. %s' % e raise exceptions.PyLXDException(msg) def host_info(self): (state, data) = self.connection.get_object('GET', '/1.0') return { 'lxd_api_compat_level': self.get_lxd_api_compat(data.get('metadata')), 'lxd_trusted_host': self.get_lxd_host_trust(data.get('metadata')), 'lxd_backing_fs': self.get_lxd_backing_fs(data.get('metadata')), 'lxd_driver': self.get_lxd_driver(data.get('metadata')), 'lxd_version': self.get_lxd_version(data.get('metadata')), 'lxc_version': self.get_lxc_version(data.get('metadata')), 'kernel_version': self.get_kernel_version(data.get('metadata')) } def get_lxd_api_compat(self, data): try: if data is None: (state, data) = self.connection.get_object('GET', '/1.0') data = data.get('metadata') return data['api_compat'] except exceptions.PyLXDException as e: print('Handling run-time error: {}'.format(e)) def get_lxd_host_trust(self, data): try: if data is None: (state, data) = self.connection.get_object('GET', '/1.0') data = data.get('metadata') return True if data['auth'] == 'trusted' else False except exceptions.PyLXDException as e: print('Handling run-time error: {}'.format(e)) def get_lxd_backing_fs(self, data): try: if data is None: (state, data) = self.connection.get_object('GET', '/1.0') data = data.get('metadata') return data['environment']['backing_fs'] except exceptions.PyLXDException as e: print('Handling run-time error: {}'.format(e)) def get_lxd_driver(self, data): try: if data is None: (state, data) = self.connection.get_object('GET', '/1.0') data = data.get('metadata') return data['environment']['driver'] except exceptions.PyLXDException as e: print('Handling run-time error: {}'.format(e)) def get_lxc_version(self, data): try: if data is None: (state, data) = self.connection.get_object('GET', '/1.0') data = data.get('metadata') return data['environment']['lxc_version'] except exceptions.PyLXDException as e: print('Handling run-time error: {}'.format(e)) def get_lxd_version(self, data): try: if data is None: (state, data) = self.connection.get_object('GET', '/1.0') data = data.get('metadata') return float(data['environment']['lxd_version']) except exceptions.PyLXDException as e: print('Handling run-time error: {}'.format(e)) def get_kernel_version(self, data): try: if data is None: (state, data) = self.connection.get_object('GET', '/1.0') data = data.get('metadata') return data['environment']['kernel_version'] except exceptions.PyLXDException as e: print('Handling run-time error: {}'.format(e)) PK5sGvֳ{C C pylxd/operation.py# Copyright (c) 2015 Canonical Ltd # # Licensed 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 dateutil.parser import parse as parse_date from pylxd import base class LXDOperation(base.LXDBase): def operation_list(self): (state, data) = self.connection.get_object('GET', '/1.0/operations') return [operation.split('/1.0/operations/')[-1] for operation in data['metadata']] def operation_show(self, operation): (state, data) = self.connection.get_object('GET', '/1.0/operations/%s' % operation) return { 'operation_create_time': self.operation_create_time(operation, data.get('metadata')), 'operation_update_time': self.operation_update_time(operation, data.get('metadata')), 'operation_status_code': self.operation_status_code(operation, data.get('metadata')) } def operation_info(self, operation): return self.connection.get_object('GET', '/1.0/operations/%s' % operation) def operation_create_time(self, operation, data): if data is None: (state, data) = self.connection.get_object( 'GET', '/1.0/operations/%s' % operation) data = data.get('metadata') return parse_date(data['created_at']).strftime('%Y-%m-%d %H:%M:%S') def operation_update_time(self, operation, data): if data is None: (state, data) = self.connection.get_object( 'GET', '/1.0/operations/%s' % operation) data = data.get('metadata') return parse_date(data['updated_at']).strftime('%Y-%m-%d %H:%M:%S') def operation_status_code(self, operation, data): if data is None: (state, data) = self.connection.get_object( 'GET', '/1.0/operations/%s' % operation) data = data.get('metadata') return data['status'] def operation_wait(self, operation, status_code, timeout): return self.connection.get_status( 'GET', '/1.0/operations/%s/wait?status_code=%s&timeout=%s' % (operation, status_code, timeout)) def operation_stream(self, operation, operation_secret): return self.connection.get_ws( 'GET', '/1.0/operations/%s/websocket?secret=%s' % (operation, operation_secret)) def operation_delete(self, operation): return self.connection.get_status('DELETE', '/1.0/operations/%s' % operation) PK-sGA_6#6#pylxd/image.py# Copyright (c) 2015 Canonical Ltd # # Licensed 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 print_function import datetime import json from six.moves import urllib from pylxd import base from pylxd import connection from pylxd import exceptions image_architecture = { 0: 'Unknown', 1: 'i686', 2: 'x86_64', 3: 'armv7l', 4: 'aarch64', 5: 'ppc', 6: 'ppc64', 7: 'ppc64le' } class LXDImage(base.LXDBase): def __init__(self, conn=None): self.connection = conn or connection.LXDConnection() # list images def image_list(self): try: (state, data) = self.connection.get_object('GET', '/1.0/images') return [image.split('/1.0/images/')[-1] for image in data['metadata']] except Exception as e: print("Unable to fetch image info - {}".format(e)) raise def image_defined(self, image): try: (state, data) = self.connection.get_object('GET', '/1.0/images/%s' % image) except exceptions.APIError as ex: if ex.status_code == 404: return False else: raise else: return True def image_list_by_key(self, params): try: (state, data) = self.connection.get_object( 'GET', '/1.0/images', urllib.parse.urlencode(params)) return [image.split('/1.0/images/')[-1] for image in data['metadata']] except Exception as e: print("Unable to fetch image info - {}".format(e)) raise # image info def image_info(self, image): try: (state, data) = self.connection.get_object('GET', '/1.0/images/%s' % image) image = { 'image_upload_date': self.get_image_date(image, data.get('metadata'), 'uploaded_at'), 'image_created_date': self.get_image_date(image, data.get('metadata'), 'created_at'), 'image_expires_date': self.get_image_date(image, data.get('metadata'), 'expires_at'), 'image_public': self.get_image_permission( image, data.get('metadata')), 'image_size': '%sMB' % self.get_image_size( image, data.get('metadata')), 'image_fingerprint': self.get_image_fingerprint( image, data.get('metadata')), 'image_architecture': self.get_image_architecture( image, data.get('metadata')), } return image except Exception as e: print("Unable to fetch image info - {}".format(e)) raise def get_image_date(self, image, data, key): try: if data is None: (state, data) = self.connection.get_object( 'GET', '/1.0/images/%s' % image) data = data.get('metadata') if data[key] != 0: return datetime.datetime.fromtimestamp( data[key]).strftime('%Y-%m-%d %H:%M:%S') else: return 'Unknown' except Exception as e: print("Unable to fetch image info - {}".format(e)) raise def get_image_permission(self, image, data): try: if data is None: (state, data) = self.connection.get_object( 'GET', '/1.0/images/%s' % image) data = data.get('metadata') return True if data['public'] == 1 else False except Exception as e: print("Unable to fetch image info - {}".format(e)) raise def get_image_size(self, image, data): try: if data is None: (state, data) = self.connection.get_object( 'GET', '/1.0/images/%s' % image) data = data.get('metadata') image_size = data['size'] if image_size <= 0: raise exceptions.ImageInvalidSize() return image_size // 1024 ** 2 except Exception as e: print("Unable to fetch image info - {}".format(e)) raise def get_image_fingerprint(self, image, data): try: if data is None: (state, data) = self.connection.get_object( 'GET', '/1.0/images/%s' % image) data = data.get('metadata') return data['fingerprint'] except Exception as e: print("Unable to fetch image info - {}".format(e)) raise def get_image_architecture(self, image, data): try: if data is None: (state, data) = self.connection.get_object( 'GET', '/1.0/images/%s' % image) data = data.get('metadata') return image_architecture[data['architecture']] except Exception as e: print("Unable to fetch image info - {}".format(e)) raise # image operations def image_upload(self, path=None, data=None, headers={}): data = data or open(path, 'rb').read() try: return self.connection.get_object('POST', '/1.0/images', data, headers) except Exception as e: print("Unable to upload image - {}".format(e)) raise def image_delete(self, image): try: return self.connection.get_status('DELETE', '/1.0/images/%s' % image) except Exception as e: print("Unable to delete image - {}".format(e)) raise def image_export(self, image): try: return self.connection.get_raw('GET', '/1.0/images/%s/export' % image) except Exception as e: print("Unable to export image - {}".format(e)) raise def image_update(self, image, data): try: return self.connection.get_status('PUT', '/1.0/images/%s' % image, json.dumps(data)) except Exception as e: print("Unable to update image - {}".format(e)) raise def image_rename(self, image, data): try: return self.connection.get_status('POST', '/1.0/images/%s' % image, json.dumps(data)) except Exception as e: print("Unable to rename image - {}".format(e)) raise class LXDAlias(base.LXDBase): def alias_list(self): (state, data) = self.connection.get_object( 'GET', '/1.0/images/aliases') return [alias.split('/1.0/images/aliases/')[-1] for alias in data['metadata']] def alias_defined(self, alias): return self.connection.get_status('GET', '/1.0/images/aliases/%s' % alias) def alias_show(self, alias): return self.connection.get_object('GET', '/1.0/images/aliases/%s' % alias) def alias_update(self, alias, data): return self.connection.get_status('PUT', '/1.0/images/aliases/%s' % alias, json.dumps(data)) def alias_rename(self, alias, data): return self.connection.get_status('POST', '/1.0/images/aliases/%s' % alias, json.dumps(data)) def alias_create(self, data): return self.connection.get_status('POST', '/1.0/images/aliases', json.dumps(data)) def alias_delete(self, alias): return self.connection.get_status('DELETE', '/1.0/images/aliases/%s' % alias) PK-sG 6pylxd/exceptions.py# Copyright (c) 2015 Canonical Ltd # # Licensed 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. class PyLXDException(Exception): pass class ContainerUnDefined(PyLXDException): pass class UntrustedHost(PyLXDException): pass class ContainerProfileCreateFail(PyLXDException): pass class ContainerProfileDeleteFail(PyLXDException): pass class ImageInvalidSize(PyLXDException): pass class APIError(PyLXDException): def __init__(self, error, status_code): msg = 'Error %s - %s.' % (status_code, error) super(APIError, self).__init__(msg) self.status_code = status_code self.error = error PK-sG6  pylxd/network.py# Copyright (c) 2015 Canonical Ltd # # Licensed 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 pylxd import base class LXDNetwork(base.LXDBase): def network_list(self): (state, data) = self.connection.get_object('GET', '/1.0/networks') return [network.split('/1.0/networks/')[-1] for network in data['metadata']] def network_show(self, network): '''Show details of the LXD network''' (state, data) = self.connection.get_object('GET', '/1.0/networks/%s' % network) return { 'network_name': self.show_network_name(network, data.get('metadata')), 'network_type': self.show_network_type(network, data.get('metadata')), 'network_members': self.show_network_members(network, data.get('metadata')) } def show_network_name(self, network, data): '''Show the LXD network name''' if data is None: (state, data) = self.connection.get_object( 'GET', '/1.0/networks/%s' % network) data = data.get('metadata') return data['name'] def show_network_type(self, network, data): if data is None: (state, data) = self.connection.get_object( 'GET', '/1.0/networks/%s' % network) data = data.get('metadata') return data['type'] def show_network_members(self, network, data): if data is None: (state, data) = self.connection.get_object( 'GET', '/1.0/networks/%s' % network) data = data.get('metadata') return [network_members.split('/1.0/networks/')[-1] for network_members in data['members']] PK5sG*\;;pylxd/connection.py# Copyright (c) 2015 Canonical Ltd # # Licensed 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. import json import os import socket import ssl from pylxd import exceptions from pylxd import utils from six.moves import http_client class UnixHTTPConnection(http_client.HTTPConnection): def __init__(self, path, host='localhost', port=None, strict=None, timeout=None): http_client.HTTPConnection.__init__(self, host, port=port, strict=strict, timeout=timeout) self.path = path def connect(self): sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) sock.connect(self.path) self.sock = sock class HTTPSConnection(http_client.HTTPConnection): default_port = 8443 def __init__(self, *args, **kwargs): http_client.HTTPConnection.__init__(self, *args, **kwargs) def connect(self): sock = socket.create_connection((self.host, self.port), self.timeout, self.source_address) if self._tunnel_host: self.sock = sock self._tunnel() (cert_file, key_file) = self._get_ssl_certs() self.sock = ssl.wrap_socket(sock, certfile=cert_file, keyfile=key_file, ssl_version=ssl.PROTOCOL_TLSv1_2) @staticmethod def _get_ssl_certs(): return (os.path.join(os.environ['HOME'], '.config/lxc/client.crt'), os.path.join(os.environ['HOME'], '.config/lxc/client.key')) class LXDConnection(object): def __init__(self, host=None, port=8443): if host: self.host = host self.port = port self.unix_socket = None else: if 'LXD_DIR' in os.environ: self.unix_socket = os.path.join(os.environ['LXD_DIR'], 'unix.socket') else: self.unix_socket = '/var/lib/lxd/unix.socket' self.host, self.port = None, None self.connection = None def get_connection(self): if self.host: return HTTPSConnection(self.host, self.port) return UnixHTTPConnection(self.unix_socket) def get_object(self, *args, **kwargs): self.connection = self.get_connection() self.connection.request(*args, **kwargs) response = self.connection.getresponse() state = response.status data = json.loads(response.read()) if not data: msg = "Null Data" raise exceptions.PyLXDException(msg) elif state == 200 or (state == 202 and data.get('status_code') == 100): return state, data else: utils.get_lxd_error(state, data) def get_status(self, *args, **kwargs): status = False self.connection = self.get_connection() self.connection.request(*args, **kwargs) response = self.connection.getresponse() state = response.status data = json.loads(response.read()) if not data: msg = "Null Data" raise exceptions.PyLXDException(msg) elif data.get('error'): utils.get_lxd_error(state, data) elif state == 200 or (state == 202 and data.get('status_code') == 100): status = True return status def get_raw(self, *args, **kwargs): self.connection = self.get_connection() self.connection.request(*args, **kwargs) response = self.connection.getresponse() body = response.read() if not body: msg = "Null Body" raise exceptions.PyLXDException(msg) elif response.status == 200: return body else: msg = "Failed to get raw response" raise exceptions.PyLXDException(msg) def get_ws(self, *args, **kwargs): self.connection = self.get_connection() self.connection.request(*args, **kwargs) response = self.connection.getresponse() return response.status PK5sGiKS S pylxd/container.py# Copyright (c) 2015 Canonical Ltd # # Licensed 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. import json from pylxd import base class LXDContainer(base.LXDBase): # containers: def container_list(self): (state, data) = self.connection.get_object('GET', '/1.0/containers') return [container.split('/1.0/containers/')[-1] for container in data['metadata']] def container_running(self, container): (state, data) = self.connection.get_object( 'GET', '/1.0/containers/%s/state' % container) data = data.get('metadata') container_running = False if data['status'] in ['RUNNING', 'STARTING', 'FREEZING', 'FROZEN', 'THAWED']: container_running = True return container_running def container_init(self, container): return self.connection.get_object('POST', '/1.0/containers', json.dumps(container)) def container_update(self, container, config): return self.connection.get_object('PUT', '/1.0/containers/%s' % container, json.dumps(config)) def container_defined(self, container): return self.connection.get_status('GET', '/1.0/containers/%s/state' % container) def container_state(self, container): return self.connection.get_object( 'GET', '/1.0/containers/%s/state' % container) def container_start(self, container, timeout): action = {'action': 'start', 'force': True, 'timeout': timeout} return self.connection.get_object('PUT', '/1.0/containers/%s/state' % container, json.dumps(action)) def container_stop(self, container, timeout): action = {'action': 'stop', 'force': True, 'timeout': timeout} return self.connection.get_object('PUT', '/1.0/containers/%s/state' % container, json.dumps(action)) def container_suspend(self, container, timeout): action = {'action': 'freeze', 'force': True, 'timeout': timeout} return self.connection.get_object('PUT', '/1.0/containers/%s/state' % container, json.dumps(action)) def container_resume(self, container, timeout): action = {'action': 'unfreeze', 'force': True, 'timeout': timeout} return self.connection.get_object('PUT', '/1.0/containers/%s/state' % container, json.dumps(action)) def container_reboot(self, container, timeout): action = {'action': 'restart', 'force': True, 'timeout': timeout} return self.connection.get_object('PUT', '/1.0/containers/%s/state' % container, json.dumps(action)) def container_destroy(self, container): return self.connection.get_object('DELETE', '/1.0/containers/%s' % container) def get_container_log(self, container): (state, data) = self.connection.get_object( 'GET', '/1.0/containers/%s?log=true' % container) return data['metadata']['log'] def get_container_config(self, container): (state, data) = self.connection.get_object( 'GET', '/1.0/containers/%s?log=false' % container) return data['metadata'] def get_container_websocket(self, container): return self.connection.get_status( 'GET', '/1.0/operations/%s/websocket?secret=%s' % (container['operation'], container['fs'])) def container_info(self, container): (state, data) = self.connection.get_object( 'GET', '/1.0/containers/%s/state' % container) return data['metadata'] def container_migrate(self, container): action = {'migration': True} (state, data) = self.connection.get_object( 'POST', '/1.0/containers/%s' % container, json.dumps(action)) return_data = { 'operation': str(data['operation'].split('/1.0/operations/')[-1]), } return_data.update(data['metadata']) return return_data def container_migrate_sync(self, operation_id, container_secret): return self.connection.get_ws( 'GET', '/1.0/operations/%s/websocket?secret=%s' % (operation_id, container_secret)) def container_local_copy(self, container): return self.connection.get_object( 'POST', '/1.0/containers', json.dumps(container)) def container_local_move(self, instance, config): return self.connection.get_object( 'POST', '/1.0/containers/%s' % instance, json.dumps(config)) # file operations def get_container_file(self, container, filename): return self.connection.get_raw( 'GET', '/1.0/containers/%s/files?path=%s' % (container, filename)) def put_container_file(self, container, src_file, dst_file, uid, gid, mode): with open(src_file, 'rb') as f: data = f.read() return self.connection.get_object( 'POST', '/1.0/containers/%s/files?path=%s' % (container, dst_file), body=data, headers={'X-LXD-uid': uid, 'X-LXD-gid': gid, 'X-LXD-mode': mode}) def container_publish(self, container): return self.connection.get_object('POST', '/1.0/images', json.dumps(container)) # misc operations def run_command(self, container, args, interactive, web_sockets, env): env = env or {} data = {'command': args, 'interactive': interactive, 'wait-for-websocket': web_sockets, 'environment': env} return self.connection.get_object('POST', '/1.0/containers/%s/exec' % container, json.dumps(data)) # snapshots def snapshot_list(self, container): (state, data) = self.connection.get_object( 'GET', '/1.0/containers/%s/snapshots' % container) return [snapshot.split('/1.0/containers/%s/snapshots/%s/' % (container, container))[-1] for snapshot in data['metadata']] def snapshot_create(self, container, config): return self.connection.get_object('POST', '/1.0/containers/%s/snapshots' % container, json.dumps(config)) def snapshot_info(self, container, snapshot): return self.connection.get_object('GET', '/1.0/containers/%s/snapshots/%s' % (container, snapshot)) def snapshot_rename(self, container, snapshot, config): return self.connection.get_object('POST', '/1.0/containers/%s/snapshots/%s' % (container, snapshot), json.dumps(config)) def snapshot_delete(self, container, snapshot): return self.connection.get_object('DELETE', '/1.0/containers/%s/snapshots/%s' % (container, snapshot)) PK-sGɇ9nnpylxd/utils.py# Copyright (c) 2015 Canonical Ltd # # Licensed 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 pylxd import exceptions def wait_for_container(name, timeout): pass def block_container(): pass def get_lxd_error(state, data): status_code = data.get('error_code') error = data.get('error') raise exceptions.APIError(error, status_code) PK-sG6(( pylxd/api.py # Copyright (c) 2015 Canonical Ltd # # Licensed 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 pylxd import certificate from pylxd import connection from pylxd import container from pylxd import hosts from pylxd import image from pylxd import network from pylxd import operation from pylxd import profiles class API(object): def __init__(self, host=None, port=8443): conn = self.connection = connection.LXDConnection(host=host, port=port) self.hosts = hosts.LXDHost(conn) self.image = image.LXDImage(conn) self.alias = image.LXDAlias(conn) self.network = network.LXDNetwork(conn) self.operation = operation.LXDOperation(conn) self.profiles = profiles.LXDProfile(conn) self.certificate = certificate.LXDCertificate(conn) self.container = container.LXDContainer(conn) # host def host_ping(self): return self.hosts.host_ping() def host_info(self): return self.hosts.host_info() def get_lxd_api_compat(self, data=None): return self.hosts.get_lxd_api_compat(data) def get_lxd_host_trust(self, data=None): return self.hosts.get_lxd_host_trust(data) def get_lxd_backing_fs(self, data=None): return self.hosts.get_lxd_backing_fs(data) def get_lxd_driver(self, data=None): return self.hosts.get_lxd_driver(data) def get_lxc_version(self, data=None): return self.hosts.get_lxc_version(data) def get_lxd_version(self, data=None): return self.hosts.get_lxd_version(data) def get_kernel_version(self, data=None): return self.hosts.get_kernel_version(data) # images def image_list(self): return self.image.image_list() def image_defined(self, image): return self.image.image_defined(image) def image_search(self, params): return self.image.image_list_by_key(params) def image_info(self, image): return self.image.image_info(image) def image_upload_date(self, image, data=None): return self.image.get_image_date(image, data, 'uploaded_at') def image_create_date(self, image, data=None): return self.image.get_image_date(image, data, 'created_at') def image_expire_date(self, image, data=None): return self.image.get_image_date(image, data, 'expires_at') def image_upload(self, path=None, data=None, headers={}): return self.image.image_upload(path=path, data=data, headers=headers) def image_delete(self, image): return self.image.image_delete(image) def image_export(self, image): return self.image.image_export(image) def image_update(self, image, data): return self.image.image_update(image, data) def image_rename(self, image, data): return self.image.image_rename(image, data) # alias def alias_list(self): return self.alias.alias_list() def alias_defined(self, alias): return self.alias.alias_defined(alias) def alias_create(self, data): return self.alias.alias_create(data) def alias_update(self, alias, data): return self.alias.alias_update(alias, data) def alias_show(self, alias): return self.alias.alias_show(alias) def alias_rename(self, alias, data): return self.alias.alias_rename(alias, data) def alias_delete(self, alias): return self.alias.alias_delete(alias) # containers: def container_list(self): return self.container.container_list() def container_defined(self, container): return self.container.container_defined(container) def container_running(self, container): return self.container.container_running(container) def container_init(self, container): return self.container.container_init(container) def container_update(self, container, config): return self.container.container_update(container, config) def container_state(self, container): return self.container.container_state(container) def container_start(self, container, timeout): return self.container.container_start(container, timeout) def container_stop(self, container, timeout): return self.container.container_stop(container, timeout) def container_suspend(self, container, timeout): return self.container.container_suspend(container, timeout) def container_resume(self, container, timeout): return self.container.container_resume(container, timeout) def container_reboot(self, container, timeout): return self.container.container_reboot(container, timeout) def container_destroy(self, container): return self.container.container_destroy(container) def get_container_log(self, container): return self.container.get_container_log(container) def get_container_config(self, container): return self.container.get_container_config(container) def get_container_websocket(self, container): return self.container.get_container_websocket(container) def container_info(self, container): return self.container.container_info(container) def container_local_copy(self, container): return self.container.container_local_copy(container) def container_local_move(self, instance, container): return self.container.container_local_move(instance, container) # file operations def get_container_file(self, container, filename): return self.container.get_container_file(container, filename) def container_publish(self, container): return self.container.container_publish(container) def put_container_file(self, container, src_file, dst_file, uid=0, gid=0, mode=0o644): return self.container.put_container_file( container, src_file, dst_file, uid, gid, mode) # snapshots def container_snapshot_list(self, container): return self.container.snapshot_list(container) def container_snapshot_create(self, container, config): return self.container.snapshot_create(container, config) def container_snapshot_info(self, container, snapshot): return self.container.snapshot_info(container, snapshot) def container_snapshot_rename(self, container, snapshot, config): return self.container.snapshot_rename(container, snapshot, config) def container_snapshot_delete(self, container, snapshot): return self.container.snapshot_delete(container, snapshot) def container_migrate(self, container): return self.container.container_migrate(container) def container_migrate_sync(self, operation_id, container_secret): return self.container.container_migrate_sync( operation_id, container_secret) # misc container def container_run_command(self, container, args, interactive=False, web_sockets=False, env=None): return self.container.run_command(container, args, interactive, web_sockets, env) # certificates def certificate_list(self): return self.certificate.certificate_list() def certificate_show(self, fingerprint): return self.certificate.certificate_show(fingerprint) def certificate_delete(self, fingerprint): return self.certificate.certificate_delete(fingerprint) def certificate_create(self, fingerprint): return self.certificate.certificate_create(fingerprint) # profiles def profile_create(self, profile): '''Create LXD profile''' return self.profiles.profile_create(profile) def profile_show(self, profile): '''Show LXD profile''' return self.profiles.profile_show(profile) def profile_defined(self, profile): '''Check to see if profile is defined''' return self.profiles.profile_defined(profile) def profile_list(self): '''List LXD profiles''' return self.profiles.profile_list() def profile_update(self, profile, config): '''Update LXD profile''' return self.profiles.profile_update(profile, config) def profile_rename(self, profile, config): '''Rename LXD profile''' return self.profiles.profile_rename(profile, config) def profile_delete(self, profile): '''Delete LXD profile''' return self.profiles.profile_delete(profile) # lxd operations def list_operations(self): return self.operation.operation_list() def wait_container_operation(self, operation, status_code, timeout): return self.operation.operation_wait(operation, status_code, timeout) def operation_delete(self, operation): return self.operation.operation_delete(operation) def operation_info(self, operation): return self.operation.operation_info(operation) def operation_show_create_time(self, operation, data=None): return self.operation.operation_create_time(operation, data) def operation_show_update_time(self, operation, data=None): return self.operation.operation_update_time(operation, data) def operation_show_status(self, operation, data=None): return self.operation.operation_status_code(operation, data) def operation_stream(self, operation, operation_secret): return self.operation.operation_stream(operation, operation_secret) # networks def network_list(self): return self.network.network_list() def network_show(self, network): return self.network.network_show(network) def network_show_name(self, network, data=None): return self.network.show_network_name(network, data) def network_show_type(self, network, data=None): return self.network.show_network_type(network, data) def network_show_members(self, network, data=None): return self.network.show_network_members(network, data) PK-sG'Ypylxd/__init__.py# -*- coding: utf-8 -*- # Licensed 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. import pbr.version __version__ = pbr.version.VersionInfo('pylxd').version_string() PK-sGd pylxd/base.py# Copyright (c) 2015 Canonical Ltd # # Licensed 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 print_function from pylxd import connection class LXDBase(object): def __init__(self, conn=None): self.connection = conn or connection.LXDConnection() PK-sG3ipylxd/certificate.py# Copyright (c) 2015 Canonical Ltd # # Licensed 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. import json from pylxd import base class LXDCertificate(base.LXDBase): def certificate_list(self): (state, data) = self.connection.get_object('GET', '/1.0/certificates') return [certificate.split('/1.0/certificates/')[-1] for certificate in data['metadata']] def certificate_show(self, fingerprint): return self.connection.get_object('GET', '/1.0/certificates/%s' % fingerprint) def certificate_create(self, certificate): return self.connection.get_status('POST', '/1.0/certificates', json.dumps(certificate)) def certificate_delete(self, fingerprint): return self.connection.get_status('DELETE', '/1.0/certificates/%s' % fingerprint) PK-sG}G1ʼpylxd/tests/test_network.py# Copyright (c) 2015 Canonical Ltd # # Licensed 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 ddt import ddt import mock from pylxd import connection from pylxd.tests import annotated_data from pylxd.tests import fake_api from pylxd.tests import LXDAPITestBase @ddt @mock.patch.object(connection.LXDConnection, 'get_object', return_value=(200, fake_api.fake_network())) class LXDAPINetworkTest(LXDAPITestBase): def test_list_networks(self, ms): ms.return_value = ('200', fake_api.fake_network_list()) self.assertEqual( ['lxcbr0'], self.lxd.network_list()) ms.assert_called_with('GET', '/1.0/networks') def test_network_show(self, ms): self.assertEqual({ 'network_name': 'lxcbr0', 'network_type': 'bridge', 'network_members': ['/1.0/containers/trusty-1'], }, self.lxd.network_show('lxcbr0')) ms.assert_called_with('GET', '/1.0/networks/lxcbr0') @annotated_data( ('name', 'lxcbr0'), ('type', 'bridge'), ('members', ['/1.0/containers/trusty-1']), ) def test_network_data(self, method, expected, ms): self.assertEqual( expected, getattr(self.lxd, 'network_show_' + method)('lxcbr0')) ms.assert_called_with('GET', '/1.0/networks/lxcbr0') PK5sG""pylxd/tests/test_container.py# Copyright (c) 2015 Canonical Ltd # # Licensed 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 collections import OrderedDict from ddt import data from ddt import ddt import json import mock import tempfile from pylxd import connection from pylxd.tests import annotated_data from pylxd.tests import fake_api from pylxd.tests import LXDAPITestBase @ddt @mock.patch.object(connection.LXDConnection, 'get_object', return_value=('200', fake_api.fake_operation())) class LXDAPIContainerTestObject(LXDAPITestBase): def test_list_containers(self, ms): ms.return_value = ('200', fake_api.fake_container_list()) self.assertEqual( ['trusty-1'], self.lxd.container_list()) ms.assert_called_once_with('GET', '/1.0/containers') @annotated_data( ('STOPPED', False), ('STOPPING', False), ('ABORTING', False), ('RUNNING', True), ('STARTING', True), ('FREEZING', True), ('FROZEN', True), ('THAWED', True), ) def test_container_running(self, status, running, ms): with mock.patch.object(connection.LXDConnection, 'get_object') as ms: ms.return_value = ('200', fake_api.fake_container_state(status)) self.assertEqual(running, self.lxd.container_running('trusty-1')) ms.assert_called_once_with('GET', '/1.0/containers/trusty-1/state') def test_container_init(self, ms): self.assertEqual(ms.return_value, self.lxd.container_init('fake')) ms.assert_called_once_with('POST', '/1.0/containers', '"fake"') def test_container_update(self, ms): self.assertEqual(ms.return_value, self.lxd.container_update('trusty-1', 'fake')) ms.assert_called_once_with('PUT', '/1.0/containers/trusty-1', '"fake"') def test_container_state(self, ms): ms.return_value = ('200', fake_api.fake_container_state('RUNNING')) self.assertEqual(ms.return_value, self.lxd.container_state('trusty-1')) ms.assert_called_with('GET', '/1.0/containers/trusty-1/state') @annotated_data( ('start', 'start'), ('stop', 'stop'), ('suspend', 'freeze'), ('resume', 'unfreeze'), ('reboot', 'restart'), ) def test_container_actions(self, method, action, ms): self.assertEqual( ms.return_value, getattr(self.lxd, 'container_' + method)('trusty-1', 30)) ms.assert_called_once_with('PUT', '/1.0/containers/trusty-1/state', json.dumps({'action': action, 'force': True, 'timeout': 30, })) def test_container_destroy(self, ms): self.assertEqual( ms.return_value, self.lxd.container_destroy('trusty-1')) ms.assert_called_once_with('DELETE', '/1.0/containers/trusty-1') def test_container_log(self, ms): ms.return_value = ('200', fake_api.fake_container_log()) self.assertEqual( 'fake log', self.lxd.get_container_log('trusty-1')) ms.assert_called_once_with('GET', '/1.0/containers/trusty-1?log=true') def test_container_config(self, ms): ms.return_value = ('200', fake_api.fake_container_state('fake')) self.assertEqual( {'status': 'fake'}, self.lxd.get_container_config('trusty-1')) ms.assert_called_once_with('GET', '/1.0/containers/trusty-1?log=false') def test_container_info(self, ms): ms.return_value = ('200', fake_api.fake_container_state('fake')) self.assertEqual( {'status': 'fake'}, self.lxd.container_info('trusty-1')) ms.assert_called_once_with('GET', '/1.0/containers/trusty-1/state') def test_container_migrate(self, ms): ms.return_value = ('200', fake_api.fake_container_migrate()) self.assertEqual( {'control': 'fake_control', 'criu': 'fake_criu', 'fs': 'fake_fs', 'operation': '1234'}, self.lxd.container_migrate('trusty-1')) ms.assert_called_once_with('POST', '/1.0/containers/trusty-1', '{"migration": true}') def test_container_publish(self, ms): ms.return_value = ('200', fake_api.fake_operation()) self.assertEqual( ms.return_value, self.lxd.container_publish('trusty-1')) ms.assert_called_once_with('POST', '/1.0/images', '"trusty-1"') def test_container_put_file(self, ms): temp_file = tempfile.NamedTemporaryFile() ms.return_value = ('200', fake_api.fake_standard_return()) self.assertEqual( ms.return_value, self.lxd.put_container_file('trusty-1', temp_file.name, 'dst_file')) ms.assert_called_once_with( 'POST', '/1.0/containers/trusty-1/files?path=dst_file', body=b'', headers={'X-LXD-gid': 0, 'X-LXD-mode': 0o644, 'X-LXD-uid': 0}) def test_list_snapshots(self, ms): ms.return_value = ('200', fake_api.fake_snapshots_list()) self.assertEqual( ['/1.0/containers/trusty-1/snapshots/first'], self.lxd.container_snapshot_list('trusty-1')) ms.assert_called_once_with('GET', '/1.0/containers/trusty-1/snapshots') @annotated_data( ('create', 'POST', '', ('fake config',), ('"fake config"',)), ('info', 'GET', '/first', ('first',), ()), ('rename', 'POST', '/first', ('first', 'fake config'), ('"fake config"',)), ('delete', 'DELETE', '/first', ('first',), ()), ) def test_snapshot_operations(self, method, http, path, args, call_args, ms): self.assertEqual( ms.return_value, getattr(self.lxd, 'container_snapshot_' + method)('trusty-1', *args)) ms.assert_called_once_with(http, '/1.0/containers/trusty-1/snapshots' + path, *call_args) def test_container_run_command(self, ms): data = OrderedDict(( ('command', ['/fake/command']), ('interactive', False), ('wait-for-websocket', False), ('environment', {'FAKE_ENV': 'fake'}) )) self.assertEqual( ms.return_value, self.lxd.container_run_command('trusty-1', *data.values())) self.assertEqual(1, ms.call_count) self.assertEqual( ms.call_args[0][:2], ('POST', '/1.0/containers/trusty-1/exec')) self.assertEqual( json.loads(ms.call_args[0][2]), dict(data) ) @ddt @mock.patch.object(connection.LXDConnection, 'get_status') class LXDAPIContainerTestStatus(LXDAPITestBase): @data(True, False) def test_container_defined(self, defined, ms): ms.return_value = defined self.assertEqual(defined, self.lxd.container_defined('trusty-1')) ms.assert_called_once_with('GET', '/1.0/containers/trusty-1/state') @ddt @mock.patch.object(connection.LXDConnection, 'get_raw', return_value='fake contents') class LXDAPIContainerTestRaw(LXDAPITestBase): def test_container_file(self, ms): self.assertEqual( 'fake contents', self.lxd.get_container_file('trusty-1', '/file/name')) ms.assert_called_once_with( 'GET', '/1.0/containers/trusty-1/files?path=/file/name') PK-sG pylxd/tests/test_host.py# Copyright (c) 2015 Canonical Ltd # # Licensed 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 ddt import data from ddt import ddt import mock from pylxd import connection from pylxd import exceptions from pylxd.tests import annotated_data from pylxd.tests import fake_api from pylxd.tests import LXDAPITestBase @ddt @mock.patch.object(connection.LXDConnection, 'get_object', return_value=('200', fake_api.fake_host())) class LXDAPIHostTestObject(LXDAPITestBase): def test_get_host_info(self, ms): result = self.lxd.host_info() self.assertEqual(result, { 'lxd_api_compat_level': 1, 'lxd_trusted_host': True, 'lxd_backing_fs': 'ext4', 'lxd_driver': 'lxc', 'lxd_version': 0.12, 'lxc_version': '1.1.2', 'kernel_version': '3.19.0-22-generic', }) ms.assert_called_once_with('GET', '/1.0') host_data = ( ('lxd_api_compat', 1), ('lxd_host_trust', True), ('lxd_backing_fs', 'ext4'), ('lxd_driver', 'lxc'), ('lxc_version', '1.1.2'), ('lxd_version', 0.12), ('kernel_version', '3.19.0-22-generic'), ) @annotated_data(*host_data) def test_get_host_data(self, method, expected, ms): result = getattr(self.lxd, 'get_' + method)(data=None) self.assertEqual(expected, result) ms.assert_called_once_with('GET', '/1.0') @annotated_data(*host_data) def test_get_host_data_fail(self, method, expected, ms): ms.side_effect = exceptions.PyLXDException result = getattr(self.lxd, 'get_' + method)(data=None) self.assertEqual(None, result) ms.assert_called_once_with('GET', '/1.0') @ddt @mock.patch.object(connection.LXDConnection, 'get_status') class LXDAPIHostTestStatus(LXDAPITestBase): @data(True, False) def test_get_host_ping(self, value, ms): ms.return_value = value self.assertEqual(value, self.lxd.host_ping()) ms.assert_called_once_with('GET', '/1.0') def test_get_host_ping_fail(self, ms): ms.side_effect = Exception self.assertRaises(exceptions.PyLXDException, self.lxd.host_ping) ms.assert_called_once_with('GET', '/1.0') PK-sG`pylxd/tests/utils.py# Copyright (c) 2015 Canonical Ltd # # Licensed 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 pylxd import api from pylxd import exceptions as lxd_exceptions def upload_image(image): alias = '{}/{}/{}/{}'.format(image['os'], image['release'], image['arch'], image['variant']) lxd = api.API() imgs = api.API(host='images.linuxcontainers.org') d = imgs.alias_show(alias) meta = d[1]['metadata'] tgt = meta['target'] try: lxd.alias_update(meta) except lxd_exceptions.APIError as ex: if ex.status_code == 404: lxd.alias_create(meta) return tgt def delete_image(image): lxd = api.API() lxd.image_delete(image) PK-sG $$pylxd/tests/test_image.py# Copyright (c) 2015 Canonical Ltd # # Licensed 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. import datetime from ddt import ddt import mock from six.moves import builtins from six.moves import cStringIO import unittest from pylxd import connection from pylxd import exceptions from pylxd import image from pylxd.tests import annotated_data from pylxd.tests import fake_api from pylxd.tests import LXDAPITestBase @ddt @mock.patch.object(connection.LXDConnection, 'get_object', return_value=('200', fake_api.fake_image_info())) class LXDAPIImageTestObject(LXDAPITestBase): list_data = ( ('list', (), ()), ('search', ({'foo': 'bar'},), ('foo=bar',)), ) @annotated_data(*list_data) def test_list_images(self, method, args, call_args, ms): ms.return_value = ('200', fake_api.fake_image_list()) self.assertEqual( ['trusty'], getattr(self.lxd, 'image_' + method)(*args)) ms.assert_called_once_with('GET', '/1.0/images', *call_args) @annotated_data(*list_data) def test_list_images_fail(self, method, args, call_args, ms): ms.side_effect = exceptions.PyLXDException self.assertRaises(exceptions.PyLXDException, getattr(self.lxd, 'image_' + method), *args) ms.assert_called_once_with('GET', '/1.0/images', *call_args) @annotated_data( (True, (('200', fake_api.fake_image_info()),)), (False, exceptions.APIError("404", 404)), ) def test_image_defined(self, expected, side_effect, ms): ms.side_effect = side_effect self.assertEqual(expected, self.lxd.image_defined('test-image')) ms.assert_called_once_with('GET', '/1.0/images/test-image') @annotated_data( ('APIError', exceptions.APIError("500", 500), exceptions.APIError), ('PyLXDException', exceptions.PyLXDException, exceptions.PyLXDException) ) def test_image_defined_fail(self, tag, side_effect, expected, ms): ms.side_effect = side_effect self.assertRaises(expected, self.lxd.image_defined, ('test-image',)) ms.assert_called_once_with('GET', '/1.0/images/test-image') def test_image_info(self, ms): self.assertEqual({ 'image_upload_date': (datetime.datetime .fromtimestamp(1435669853) .strftime('%Y-%m-%d %H:%M:%S')), 'image_created_date': 'Unknown', 'image_expires_date': 'Unknown', 'image_public': False, 'image_size': '63MB', 'image_fingerprint': '04aac4257341478b49c25d22cea8a6ce' '0489dc6c42d835367945e7596368a37f', 'image_architecture': 'x86_64', }, self.lxd.image_info('test-image')) ms.assert_called_once_with('GET', '/1.0/images/test-image') def test_image_info_fail(self, ms): ms.side_effect = exceptions.PyLXDException self.assertRaises(exceptions.PyLXDException, self.lxd.image_info, ('test-image',)) ms.assert_called_once_with('GET', '/1.0/images/test-image') dates_data = ( ('upload', (datetime.datetime.fromtimestamp(1435669853) .strftime('%Y-%m-%d %H:%M:%S'))), ('create', 'Unknown'), ('expire', 'Unknown'), ) @annotated_data(*dates_data) def test_image_date(self, method, expected, ms): self.assertEqual(expected, getattr( self.lxd, 'image_{}_date'.format(method))('test-image', data=None)) ms.assert_called_once_with('GET', '/1.0/images/test-image') @annotated_data(*dates_data) def test_image_date_fail(self, method, expected, ms): ms.side_effect = exceptions.PyLXDException self.assertRaises(exceptions.PyLXDException, getattr( self.lxd, 'image_{}_date'.format(method)), 'test-image', data=None) ms.assert_called_once_with('GET', '/1.0/images/test-image') @ddt @mock.patch.object(connection.LXDConnection, 'get_status', return_value=True) class LXDAPIImageTestStatus(LXDAPITestBase): operations_data = ( ('delete', 'DELETE', '/test-image', ('test-image',), ()), ('update', 'PUT', '/test-image', ('test-image', 'fake',), ('"fake"',)), ('rename', 'POST', '/test-image', ('test-image', 'fake',), ('"fake"',)), ) @annotated_data(*operations_data) def test_image_operations(self, method, http, path, args, call_args, ms): self.assertTrue( getattr(self.lxd, 'image_' + method)(*args)) ms.assert_called_once_with( http, '/1.0/images' + path, *call_args ) @annotated_data(*operations_data) def test_image_operations_fail(self, method, http, path, args, call_args, ms): ms.side_effect = exceptions.PyLXDException self.assertRaises(exceptions.PyLXDException, getattr(self.lxd, 'image_' + method), *args) ms.assert_called_once_with( http, '/1.0/images' + path, *call_args ) @mock.patch.object(connection.LXDConnection, 'get_object', return_value=('200', fake_api.fake_image_info())) class LXDAPAPIImageTestUpload(LXDAPITestBase): @mock.patch.object(builtins, 'open', return_value=cStringIO('fake')) def test_image_upload_file(self, mo, ms): self.assertTrue(self.lxd.image_upload(path='/fake/path')) mo.assert_called_once_with('/fake/path', 'rb') ms.assert_called_once_with('POST', '/1.0/images', 'fake', {}) @mock.patch.object(connection.LXDConnection, 'get_raw') class LXDAPIImageTestRaw(LXDAPITestBase): def test_image_export(self, ms): ms.return_value = 'fake contents' self.assertEqual('fake contents', self.lxd.image_export('fake')) ms.assert_called_once_with('GET', '/1.0/images/fake/export') def test_image_export_fail(self, ms): ms.side_effect = exceptions.PyLXDException self.assertRaises(exceptions.PyLXDException, self.lxd.image_export, 'fake') ms.assert_called_once_with('GET', '/1.0/images/fake/export') @ddt @mock.patch.object(connection.LXDConnection, 'get_object', return_value=(200, fake_api.fake_image_info())) class LXDAPIImageInfoTest(unittest.TestCase): def setUp(self): super(LXDAPIImageInfoTest, self).setUp() self.image = image.LXDImage() info_list = ( ('permission', False), ('size', 63), ('fingerprint', '04aac4257341478b49c25d22cea8a6ce' '0489dc6c42d835367945e7596368a37f'), ('architecture', 'x86_64'), ) @annotated_data(*info_list) def test_info_no_data(self, method, expected, mc): self.assertEqual(expected, (getattr(self.image, 'get_image_' + method) ('test-image', data=None))) mc.assert_called_once_with('GET', '/1.0/images/test-image') @annotated_data(*info_list) def test_info_no_data_fail(self, method, expected, mc): mc.side_effect = exceptions.PyLXDException self.assertRaises(exceptions.PyLXDException, getattr(self.image, 'get_image_' + method), 'test-image', data=None) @annotated_data( ('permission_true', 'permission', {'public': 0}, False), ('permission_false', 'permission', {'public': 1}, True), ('size', 'size', {'size': 52428800}, 50), ('fingerprint', 'fingerprint', {'fingerprint': 'AAAA'}, 'AAAA'), *[('architecture_' + v, 'architecture', {'architecture': k}, v) for k, v in image.image_architecture.items()] ) def test_info_data(self, tag, method, metadata, expected, mc): self.assertEqual( expected, getattr(self.image, 'get_image_' + method) ('test-image', data=metadata)) self.assertFalse(mc.called) @annotated_data( ('permission', 'permission', {}, KeyError), ('size', 'size', {'size': 0}, exceptions.ImageInvalidSize), ('size', 'size', {'size': -1}, exceptions.ImageInvalidSize), ('fingerprint', 'fingerprint', {}, KeyError), ('architecture', 'architecture', {}, KeyError), ('architecture_invalid', 'architecture', {'architecture': -1}, KeyError) ) def test_info_data_fail(self, tag, method, metadata, expected, mc): self.assertRaises(expected, getattr(self.image, 'get_image_' + method), 'test-image', data=metadata) self.assertFalse(mc.called) PK-sGɼbpylxd/tests/test_certificate.py# Copyright (c) 2015 Canonical Ltd # # Licensed 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 ddt import ddt import json import mock from pylxd import connection from pylxd.tests import annotated_data from pylxd.tests import fake_api from pylxd.tests import LXDAPITestBase @ddt class LXDAPICertificateTest(LXDAPITestBase): def test_list_certificates(self): with mock.patch.object(connection.LXDConnection, 'get_object') as ms: ms.return_value = ('200', fake_api.fake_certificate_list()) self.assertEqual( ['ABCDEF01'], self.lxd.certificate_list()) ms.assert_called_with('GET', '/1.0/certificates') def test_certificate_show(self): with mock.patch.object(connection.LXDConnection, 'get_object') as ms: ms.return_value = ('200', fake_api.fake_certificate()) self.assertEqual( ms.return_value, self.lxd.certificate_show('ABCDEF01')) ms.assert_called_with('GET', '/1.0/certificates/ABCDEF01') @annotated_data( ('delete', 'DELETE', '/ABCDEF01'), ('create', 'POST', '', (json.dumps('ABCDEF01'),)), ) def test_certificate_operations(self, method, http, path, call_args=()): with mock.patch.object(connection.LXDConnection, 'get_status') as ms: ms.return_value = True self.assertTrue( getattr(self.lxd, 'certificate_' + method)('ABCDEF01')) ms.assert_called_with(http, '/1.0/certificates' + path, *call_args) PK-sG@ipylxd/tests/fake_api.py# Copyright (c) 2015 Canonical Ltd # # Licensed 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. def fake_standard_return(): return { "type": "sync", "status": "Success", "status_code": 200, "metadata": {} } def fake_host(): return { "type": "sync", "status": "Success", "status_code": 200, "metadata": { "api_compat": 1, "auth": "trusted", "config": {}, "environment": { "backing_fs": "ext4", "driver": "lxc", "kernel_version": "3.19.0-22-generic", "lxc_version": "1.1.2", "lxd_version": "0.12" } } } def fake_image_list_empty(): return { "type": "sync", "status": "Success", "status_code": 200, "metadata": [] } def fake_image_list(): return { "type": "sync", "status": "Success", "status_code": 200, "metadata": ['/1.0/images/trusty'] } def fake_image_info(): return { "type": "sync", "status": "Success", "status_code": 200, "metadata": { "aliases": [ { "target": "ubuntu", "description": "ubuntu" } ], "architecture": 2, "fingerprint": "04aac4257341478b49c25d22cea8a6ce" "0489dc6c42d835367945e7596368a37f", "filename": "", "properties": {}, "public": 0, "size": 67043148, "created_at": 0, "expires_at": 0, "uploaded_at": 1435669853 } } def fake_alias(): return { "type": "sync", "status": "Success", "status_code": 200, "metadata": { "target": "ubuntu", "description": "ubuntu" } } def fake_alias_list(): return { "type": "sync", "status": "Success", "status_code": 200, "metadata": [ "/1.0/images/aliases/ubuntu" ] } def fake_container_list(): return { "type": "sync", "status": "Success", "status_code": 200, "metadata": [ "/1.0/containers/trusty-1" ] } def fake_container_state(status): return { "type": "sync", "status": "Success", "status_code": 200, "metadata": { "status": status } } def fake_container_log(): return { "type": "sync", "status": "Success", "status_code": 200, "metadata": { "log": "fake log" } } def fake_container_migrate(): return { "type": "sync", "status": "Success", "status_code": 200, "operation": "/1.0/operations/1234", "metadata": { "control": "fake_control", "fs": "fake_fs", "criu": "fake_criu", } } def fake_snapshots_list(): return { "type": "sync", "status": "Success", "status_code": 200, "metadata": [ "/1.0/containers/trusty-1/snapshots/first" ] } def fake_certificate_list(): return { "type": "sync", "status": "Success", "status_code": 200, "metadata": [ "/1.0/certificates/ABCDEF01" ] } def fake_certificate(): return { "type": "sync", "status": "Success", "status_code": 200, "metadata": { "type": "client", "certificate": "ABCDEF01" } } def fake_profile_list(): return { "type": "sync", "status": "Success", "status_code": 200, "metadata": [ "/1.0/profiles/fake-profile" ] } def fake_profile(): return { "type": "sync", "status": "Success", "status_code": 200, "metadata": { "name": "fake-profile", "config": { "resources.memory": "2GB", "network.0.bridge": "lxcbr0" } } } def fake_operation_list(): return { "type": "sync", "status": "Success", "status_code": 200, "metadata": [ "/1.0/operations/1234" ] } def fake_operation(): return { "type": "async", "status": "OK", "status_code": 100, "operation": "/1.0/operation/1234", "metadata": { "created_at": "2015-06-09T19:07:24.379615253-06:00", "updated_at": "2015-06-09T19:07:23.379615253-06:00", "status": "Running", "status_code": 103, "resources": { "containers": ["/1.0/containers/1"] }, "metadata": {}, "may_cancel": True } } def fake_network_list(): return { "type": "sync", "status": "Success", "status_code": 200, "metadata": [ "/1.0/networks/lxcbr0" ] } def fake_network(): return { "type": "async", "status": "OK", "status_code": 100, "operation": "/1.0/operation/1234", "metadata": { "name": "lxcbr0", "type": "bridge", "members": ["/1.0/containers/trusty-1"] } } PK-sG̡>>pylxd/tests/__init__.py# Copyright (c) 2015 Canonical Ltd # # Licensed 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 ddt import data from ddt import unpack import unittest from pylxd import api class LXDAPITestBase(unittest.TestCase): def setUp(self): super(LXDAPITestBase, self).setUp() self.lxd = api.API() def annotated_data(*args): class List(list): pass new_args = [] for arg in args: new_arg = List(arg) new_arg.__name__ = arg[0] new_args.append(new_arg) return lambda func: data(*new_args)(unpack(func)) PK-sGE  pylxd/tests/test_profiles.py# Copyright (c) 2015 Canonical Ltd # # Licensed 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 ddt import data from ddt import ddt import mock from pylxd import connection from pylxd.tests import annotated_data from pylxd.tests import fake_api from pylxd.tests import LXDAPITestBase @mock.patch.object(connection.LXDConnection, 'get_object', return_value=(200, fake_api.fake_profile())) class LXDAPIProfilesTestObject(LXDAPITestBase): def test_list_profiles(self, ms): ms.return_value = ('200', fake_api.fake_profile_list()) self.assertEqual( ['fake-profile'], self.lxd.profile_list()) ms.assert_called_with('GET', '/1.0/profiles') def test_profile_show(self, ms): self.assertEqual( ms.return_value, self.lxd.profile_show('fake-profile')) ms.assert_called_with('GET', '/1.0/profiles/fake-profile') @ddt @mock.patch.object(connection.LXDConnection, 'get_status', return_value=True) class LXDAPIProfilesTestStatus(LXDAPITestBase): @data(True, False) def test_profile_defined(self, defined, ms): ms.return_value = defined self.assertEqual(defined, self.lxd.profile_defined('fake-profile')) ms.assert_called_with('GET', '/1.0/profiles/fake-profile') @annotated_data( ('create', 'POST', '', ('fake config',), ('"fake config"',)), ('update', 'PUT', '/fake-profile', ('fake-profile', 'fake config',), ('"fake config"',)), ('rename', 'POST', '/fake-profile', ('fake-profile', 'fake config',), ('"fake config"',)), ('delete', 'DELETE', '/fake-profile', ('fake-profile',), ()), ) def test_profile_operations(self, method, http, path, args, call_args, ms): self.assertTrue( getattr(self.lxd, 'profile_' + method)(*args)) ms.assert_called_with(http, '/1.0/profiles' + path, *call_args) PK5sGX<} } pylxd/tests/test_operation.py# Copyright (c) 2015 Canonical Ltd # # Licensed 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. import datetime from ddt import ddt import mock from pylxd import connection from pylxd.tests import annotated_data from pylxd.tests import fake_api from pylxd.tests import LXDAPITestBase @ddt @mock.patch.object(connection.LXDConnection, 'get_object', return_value=(200, fake_api.fake_operation())) class LXDAPIOperationTestObject(LXDAPITestBase): def test_list_operations(self, ms): ms.return_value = ('200', fake_api.fake_operation_list()) self.assertEqual( ['1234'], self.lxd.list_operations()) ms.assert_called_with('GET', '/1.0/operations') def test_operation_info(self, ms): ms.return_value = ('200', fake_api.fake_operation()) self.assertEqual( ms.return_value, self.lxd.operation_info('1234')) ms.assert_called_with('GET', '/1.0/operations/1234') @annotated_data( ('create_time', datetime.datetime.utcfromtimestamp(1433876844) .strftime('%Y-%m-%d %H:%M:%S')), ('update_time', datetime.datetime.utcfromtimestamp(1433876843) .strftime('%Y-%m-%d %H:%M:%S')), ('status', 'Running'), ) def test_operation_show(self, method, expected, ms): self.assertEqual( expected, getattr(self.lxd, 'operation_show_' + method)('1234')) ms.assert_called_with('GET', '/1.0/operations/1234') @ddt @mock.patch.object(connection.LXDConnection, 'get_status', return_value=True) class LXDAPIOperationTestStatus(LXDAPITestBase): @annotated_data( ('operation_delete', 'DELETE', '', ()), ('wait_container_operation', 'GET', '/wait?status_code=200&timeout=30', ('200', '30')), ) def test_operation_actions(self, method, http, path, args, ms): self.assertTrue( getattr(self.lxd, method)('1234', *args)) ms.assert_called_with(http, '/1.0/operations/1234' + path) PK5sG07pylxd/tests/test_connection.py# Copyright (c) 2015 Canonical Ltd # # Licensed 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 ddt import ddt import inspect import mock from six.moves import cStringIO from six.moves import http_client import socket import ssl import unittest from pylxd import connection from pylxd import exceptions from pylxd.tests import annotated_data @ddt class LXDInitConnectionTest(unittest.TestCase): @mock.patch('socket.socket') @mock.patch.object(http_client.HTTPConnection, '__init__') def test_http_connection(self, mc, ms): conn = connection.UnixHTTPConnection('/', 'host', 1234) mc.assert_called_once_with( conn, 'host', port=1234, strict=None, timeout=None) conn.connect() ms.assert_called_once_with(socket.AF_UNIX, socket.SOCK_STREAM) ms.return_value.connect.assert_called_once_with('/') @mock.patch('os.environ', {'HOME': '/home/foo'}) @mock.patch('ssl.wrap_socket') @mock.patch('socket.create_connection') def test_https_connection(self, ms, ml): conn = connection.HTTPSConnection('host', 1234) with mock.patch.object(conn, '_tunnel') as mc: conn.connect() self.assertFalse(mc.called) ms.assert_called_once_with( ('host', 1234), socket._GLOBAL_DEFAULT_TIMEOUT, None) ml.assert_called_once_with( ms.return_value, certfile='/home/foo/.config/lxc/client.crt', keyfile='/home/foo/.config/lxc/client.key', ssl_version=ssl.PROTOCOL_TLSv1_2, ) @mock.patch('os.environ', {'HOME': '/home/foo'}) @mock.patch('ssl.wrap_socket') @mock.patch('socket.create_connection') def test_https_proxy_connection(self, ms, ml): conn = connection.HTTPSConnection('host', 1234) conn._tunnel_host = 'host' with mock.patch.object(conn, '_tunnel') as mc: conn.connect() self.assertTrue(mc.called) ms.assert_called_once_with( ('host', 1234), socket._GLOBAL_DEFAULT_TIMEOUT, None) ml.assert_called_once_with( ms.return_value, certfile='/home/foo/.config/lxc/client.crt', keyfile='/home/foo/.config/lxc/client.key', ssl_version=ssl.PROTOCOL_TLSv1_2) @mock.patch('pylxd.connection.HTTPSConnection') @mock.patch('pylxd.connection.UnixHTTPConnection') @annotated_data( ('unix', (None,), {}, '/var/lib/lxd/unix.socket'), ('unix_path', (None,), {'LXD_DIR': '/fake/'}, '/fake/unix.socket'), ('https', ('host',), {}, ''), ('https_port', ('host', 1234), {}, ''), ) def test_get_connection(self, mode, args, env, path, mc, ms): with mock.patch('os.environ', env): conn = connection.LXDConnection(*args).get_connection() if mode.startswith('unix'): self.assertEqual(mc.return_value, conn) mc.assert_called_once_with(path) elif mode.startswith('https'): self.assertEqual(ms.return_value, conn) ms.assert_called_once_with( args[0], len(args) == 2 and args[1] or 8443) class FakeResponse(object): def __init__(self, status, data): self.status = status self.read = cStringIO(data).read @ddt @mock.patch('pylxd.connection.LXDConnection.get_connection') class LXDConnectionTest(unittest.TestCase): def setUp(self): super(LXDConnectionTest, self).setUp() self.conn = connection.LXDConnection() @annotated_data( ('null', (200, '{}'), exceptions.PyLXDException), ('200', (200, '{"foo": "bar"}'), (200, {'foo': 'bar'})), ('202', (202, '{"status_code": 100}'), (202, {'status_code': 100})), ('500', (500, '{"foo": "bar"}'), exceptions.APIError), ) def test_get_object(self, tag, effect, result, mg): mg.return_value.getresponse.return_value = FakeResponse(*effect) if inspect.isclass(result): self.assertRaises(result, self.conn.get_object) else: self.assertEqual(result, self.conn.get_object()) @annotated_data( ('null', (200, '{}'), exceptions.PyLXDException), ('200', (200, '{"foo": "bar"}'), True), ('202', (202, '{"status_code": 100}'), True), ('200', (200, '{"error": "bar"}'), exceptions.APIError), ('500', (500, '{"foo": "bar"}'), False), ) def test_get_status(self, tag, effect, result, mg): mg.return_value.getresponse.return_value = FakeResponse(*effect) if inspect.isclass(result): self.assertRaises(result, self.conn.get_status) else: self.assertEqual(result, self.conn.get_status()) @annotated_data( ('null', (200, ''), exceptions.PyLXDException), ('200', (200, '{"foo": "bar"}'), '{"foo": "bar"}'), ('500', (500, '{"foo": "bar"}'), exceptions.PyLXDException), ) def test_get_raw(self, tag, effect, result, mg): mg.return_value.getresponse.return_value = FakeResponse(*effect) if inspect.isclass(result): self.assertRaises(result, self.conn.get_raw) else: self.assertEqual(result, self.conn.get_raw()) PK-sGK9= = pylxd/tests/test_image_alias.py# Copyright (c) 2015 Canonical Ltd # # Licensed 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 ddt import data from ddt import ddt import mock from pylxd import connection from pylxd.tests import annotated_data from pylxd.tests import fake_api from pylxd.tests import LXDAPITestBase @ddt @mock.patch.object(connection.LXDConnection, 'get_object') class LXDAPIImageAliasTestObject(LXDAPITestBase): def test_alias_list(self, ms): ms.return_value = ('200', fake_api.fake_alias_list()) self.assertEqual(['ubuntu'], self.lxd.alias_list()) ms.assert_called_once_with('GET', '/1.0/images/aliases') def test_alias_show(self, ms): ms.return_value = ('200', fake_api.fake_alias()) self.assertEqual( fake_api.fake_alias(), self.lxd.alias_show('fake')[1]) ms.assert_called_once_with('GET', '/1.0/images/aliases/fake') @ddt @mock.patch.object(connection.LXDConnection, 'get_status') class LXDAPIImageAliasTestStatus(LXDAPITestBase): @data(True, False) def test_alias_defined(self, expected, ms): ms.return_value = expected self.assertEqual(expected, self.lxd.alias_defined('fake')) ms.assert_called_once_with('GET', '/1.0/images/aliases/fake') @annotated_data( ('create', 'POST', '', ('fake',), ('"fake"',)), ('update', 'PUT', '/test-alias', ('test-alias', 'fake',), ('"fake"',)), ('rename', 'POST', '/test-alias', ('test-alias', 'fake',), ('"fake"',)), ('delete', 'DELETE', '/test-alias', ('test-alias',), ()), ) def test_alias_operations(self, method, http, path, args, call_args, ms): self.assertTrue(getattr(self.lxd, 'alias_' + method)(*args)) ms.assert_called_once_with( http, '/1.0/images/aliases' + path, *call_args ) PKsGtp66&pylxd-0.18.0.dist-info/DESCRIPTION.rst# pylxd [![Build Status](https://travis-ci.org/lxc/pylxd.svg?branch=master)](https://travis-ci.org/lxc/pylxd) A Python library for interacting with the LXD REST API. ## Getting started with pylxd If you're running on Ubuntu Wily or greater: sudo apt-get install python-pylxd lxd otherwise you can track LXD development on other Ubuntu releases: sudo add-apt-repository ppa:ubuntu-lxc/lxd-git-master && sudo apt-get update sudo apt-get install lxd and install pylxd using pip: pip install pylxd ## First steps Once you have pylxd installed, you're ready to start interacting with LXD: ```python import uuid from pylxd import api # Let's pick a random name, avoiding clashes CONTAINER_NAME = str(uuid.uuid1()) lxd = api.API() try: lxd.container_defined(CONTAINER_NAME) except Exception as e: print("Container does not exist: %s" % e) config = {'name': CONTAINER_NAME, 'source': {'type': 'none'}} lxd.container_init(config) if lxd.container_defined(CONTAINER_NAME): print("Container is running") else: print("Whoops - please report a bug!") containers = lxd.container_list() for x in containers: lxd.container_destroy(x) ``` ## Bug reports Bug reports can be filed at https://github.com/lxc/pylxd/issues/new ## Support and discussions We use the LXC mailing-lists for developer and user discussions, you can find and subscribe to those at: https://lists.linuxcontainers.org If you prefer live discussions, some of us also hang out in [#lxcontainers](http://webchat.freenode.net/?channels=#lxcontainers) on irc.freenode.net. PKsG$pylxd-0.18.0.dist-info/metadata.json{"classifiers": ["Environment :: OpenStack", "Intended Audience :: Information Technology", "Intended Audience :: System Administrators", "License :: OSI Approved :: Apache Software License", "Operating System :: POSIX :: Linux", "Programming Language :: Python", "Programming Language :: Python :: 2", "Programming Language :: Python :: 2.7", "Programming Language :: Python :: 2.6", "Programming Language :: Python :: 3", "Programming Language :: Python :: 3.3", "Programming Language :: Python :: 3.4"], "extensions": {"python.details": {"contacts": [{"email": "chuck.short@canonical.com", "name": "Chuck Short", "role": "author"}], "document_names": {"description": "DESCRIPTION.rst"}, "project_urls": {"Home": "http://www.linuxcontainers.org"}}}, "extras": [], "generator": "bdist_wheel (0.26.0)", "metadata_version": "2.0", "name": "pylxd", "run_requires": [{"requires": ["Babel (>=1.3)", "pbr (<2.0,>=1.6)", "python-dateutil (>=2.4.2)", "six (>=1.9.0)"]}], "summary": "python library for lxd", "test_requires": [{"requires": ["coverage (>=3.6)", "ddt (>=0.7.0)", "discover", "hacking (<0.11,>=0.10.0)", "os-testr", "oslosphinx (>=2.5.0)", "oslotest (>=1.10.0)", "python-subunit (>=0.0.18)", "sphinx (!=1.2.0,!=1.3b1,<1.3,>=1.1.2)"]}], "version": "0.18.0"}PKsGZÞJ//pylxd-0.18.0.dist-info/pbr.json{"is_release": false, "git_version": "0411d6f"}PKsG2t$pylxd-0.18.0.dist-info/top_level.txtpylxd PKsG''\\pylxd-0.18.0.dist-info/WHEELWheel-Version: 1.0 Generator: bdist_wheel (0.26.0) Root-Is-Purelib: true Tag: py2-none-any PKsG pylxd-0.18.0.dist-info/METADATAMetadata-Version: 2.0 Name: pylxd Version: 0.18.0 Summary: python library for lxd Home-page: http://www.linuxcontainers.org Author: Chuck Short Author-email: chuck.short@canonical.com License: UNKNOWN Platform: UNKNOWN Classifier: Environment :: OpenStack Classifier: Intended Audience :: Information Technology Classifier: Intended Audience :: System Administrators Classifier: License :: OSI Approved :: Apache Software License Classifier: Operating System :: POSIX :: Linux Classifier: Programming Language :: Python Classifier: Programming Language :: Python :: 2 Classifier: Programming Language :: Python :: 2.7 Classifier: Programming Language :: Python :: 2.6 Classifier: Programming Language :: Python :: 3 Classifier: Programming Language :: Python :: 3.3 Classifier: Programming Language :: Python :: 3.4 Requires-Dist: Babel (>=1.3) Requires-Dist: pbr (<2.0,>=1.6) Requires-Dist: python-dateutil (>=2.4.2) Requires-Dist: six (>=1.9.0) # pylxd [![Build Status](https://travis-ci.org/lxc/pylxd.svg?branch=master)](https://travis-ci.org/lxc/pylxd) A Python library for interacting with the LXD REST API. ## Getting started with pylxd If you're running on Ubuntu Wily or greater: sudo apt-get install python-pylxd lxd otherwise you can track LXD development on other Ubuntu releases: sudo add-apt-repository ppa:ubuntu-lxc/lxd-git-master && sudo apt-get update sudo apt-get install lxd and install pylxd using pip: pip install pylxd ## First steps Once you have pylxd installed, you're ready to start interacting with LXD: ```python import uuid from pylxd import api # Let's pick a random name, avoiding clashes CONTAINER_NAME = str(uuid.uuid1()) lxd = api.API() try: lxd.container_defined(CONTAINER_NAME) except Exception as e: print("Container does not exist: %s" % e) config = {'name': CONTAINER_NAME, 'source': {'type': 'none'}} lxd.container_init(config) if lxd.container_defined(CONTAINER_NAME): print("Container is running") else: print("Whoops - please report a bug!") containers = lxd.container_list() for x in containers: lxd.container_destroy(x) ``` ## Bug reports Bug reports can be filed at https://github.com/lxc/pylxd/issues/new ## Support and discussions We use the LXC mailing-lists for developer and user discussions, you can find and subscribe to those at: https://lists.linuxcontainers.org If you prefer live discussions, some of us also hang out in [#lxcontainers](http://webchat.freenode.net/?channels=#lxcontainers) on irc.freenode.net. PKsG' pylxd-0.18.0.dist-info/RECORDpylxd/__init__.py,sha256=kKkulnVj0846EvxqDw0rRfEiao45xUhzIyn_Vc23WhU,655 pylxd/api.py,sha256=vVAb1H0DaCztiQcA8C5NkQcYGjevzAsKfVB4herh8cQ,10483 pylxd/base.py,sha256=QgqIKxe5bTsR5QZEv374BE0pP1v-aBR7Nimp0lG_qSE,799 pylxd/certificate.py,sha256=tUwkRXsOX7wGcMsQ_Bia9ag3MBAun8pbGCwp4Ovf4yQ,1450 pylxd/connection.py,sha256=B2J613f9h5wkVLwAM4PiRiIInU8BB_vsonaolwsMYp8,4667 pylxd/container.py,sha256=BXAFls7zCMowMEEtcQJBouh7PH4BEMbFytcYrJdHNYs,8275 pylxd/exceptions.py,sha256=GDxuCzz6rzbIaC9NOKE8yGc9z_n_O5BT2-PTk0cuFQo,1170 pylxd/hosts.py,sha256=9XBeVS19RIbpmehg1X0ZCHKBowEc-omTbo4EqxsvDrY,4239 pylxd/image.py,sha256=sUxMGd2LwCpFJE2xyo5NHuQHvRNAk4xrxIGEQbEPLwg,9014 pylxd/network.py,sha256=49PvYimpXbk3Iv-ALdfo6TEIy4OXKog32tIR78wjmK8,2308 pylxd/operation.py,sha256=H_u3LPMUaZpxugXYiojwJOawJBWQaU2Uq0mMjoPoAh4,3139 pylxd/profiles.py,sha256=OCQaMD3k_-VvLshbI4W4eQD0t9cTidzGKbstvnl5cV0,2306 pylxd/utils.py,sha256=Xy7mGhOWvi7RxH0x5pGGvp4W9hjtJEDSM0CAZ5hha8s,878 pylxd/tests/__init__.py,sha256=z9J1No3TEPsiNWfeo9KXlw99x11-liDNk1RNoxf3qNU,1086 pylxd/tests/fake_api.py,sha256=kw7eOqUz_xTl9MTEIwj3s7KRXm3tkUsjtLN6Z26Uv40,6015 pylxd/tests/test_certificate.py,sha256=n0ZUhP_uoR-qQpOSKTKOrgPzp7Bo4eMJ-mPHpqmxQWY,2197 pylxd/tests/test_connection.py,sha256=meX6YrS-1yQctzr5KYw3vXQAJDwMMNi85219DTTRqpI,5820 pylxd/tests/test_container.py,sha256=PXPYmcVL6jEdYrmJV6V6kOhHgPsrkYhRmYkW9AQF0Y0,8893 pylxd/tests/test_host.py,sha256=6LPkgAZd54ZInTo8GczxzyfP-U_hTrJwUpyRDmPpu2I,2778 pylxd/tests/test_image.py,sha256=NA5gUw24L5j94bIPKHOBXTA-tvxzHPK8dG4rbrnpnvs,9468 pylxd/tests/test_image_alias.py,sha256=7MTcwsWsV5XKmS-eGrRSsvpqa75AoTskI7qY-dROEls,2365 pylxd/tests/test_network.py,sha256=8kiO_NNWtnMrMG-pb6we3w0738NW259stxlZq8C2MeU,1980 pylxd/tests/test_operation.py,sha256=LKtfZKmfUzE-YQGqsL_AOiDkfN0To6MWFG16DYCmFI4,2685 pylxd/tests/test_profiles.py,sha256=SaXcslUVkqV_mXJ8GX6sTYw1TZMmuVeQvVRra29aJIw,2579 pylxd/tests/utils.py,sha256=b5b4SELcoyhSMPMMjQf_oN5AbjgtBSk0UkvX9pfg7P0,1304 pylxd-0.18.0.dist-info/DESCRIPTION.rst,sha256=-HKDxKEhR2bwShujrUkOM1SfsZVIIF9Q_Ovh4AZn-Mw,1590 pylxd-0.18.0.dist-info/METADATA,sha256=Xew4O7FyY4WLZlUpxDYpYXSGTFtNayxOMjoZ2KheylU,2538 pylxd-0.18.0.dist-info/RECORD,, pylxd-0.18.0.dist-info/WHEEL,sha256=JTb7YztR8fkPg6aSjc571Q4eiVHCwmUDlX8PhuuqIIE,92 pylxd-0.18.0.dist-info/metadata.json,sha256=LicpPDMvGS22gPVUwRUa9HkJ-UIKzHH4IKtLQlxx-jA,1262 pylxd-0.18.0.dist-info/pbr.json,sha256=Cb43zI3yYF04RxjyA51fQdiLNysLlAJJVo21KvB21G4,47 pylxd-0.18.0.dist-info/top_level.txt,sha256=6zCViCDxq9jv3964X80UWF_KBHfBxinPALQtY4Mssuc,6 PK-sGN:  pylxd/profiles.pyPK-sGw1 pylxd/hosts.pyPK5sGvֳ{C C pylxd/operation.pyPK-sGA_6#6#_&pylxd/image.pyPK-sG 6Ipylxd/exceptions.pyPK-sG6  Npylxd/network.pyPK5sG*\;;Wpylxd/connection.pyPK5sGiKS S "jpylxd/container.pyPK-sGɇ9nnpylxd/utils.pyPK-sG6(( ?pylxd/api.pyPK-sG'Y\pylxd/__init__.pyPK-sGd pylxd/base.pyPK-sG3idpylxd/certificate.pyPK-sG}G1ʼ@pylxd/tests/test_network.pyPK5sG""5pylxd/tests/test_container.pyPK-sG -pylxd/tests/test_host.pyPK-sG`=pylxd/tests/utils.pyPK-sG $$pylxd/tests/test_image.pyPK-sGɼb#pylxd/tests/test_certificate.pyPK-sG@i,pylxd/tests/fake_api.pyPK-sG̡>>@Dpylxd/tests/__init__.pyPK-sGE  Hpylxd/tests/test_profiles.pyPK5sGX<} } Spylxd/tests/test_operation.pyPK5sG07]pylxd/tests/test_connection.pyPK-sGK9= = tpylxd/tests/test_image_alias.pyPKsGtp66&*~pylxd-0.18.0.dist-info/DESCRIPTION.rstPKsG$pylxd-0.18.0.dist-info/metadata.jsonPKsGZÞJ//ԉpylxd-0.18.0.dist-info/pbr.jsonPKsG2t$@pylxd-0.18.0.dist-info/top_level.txtPKsG''\\pylxd-0.18.0.dist-info/WHEELPKsG pylxd-0.18.0.dist-info/METADATAPKsG' Epylxd-0.18.0.dist-info/RECORDPK y