diff --git a/pylxd/__init__.py b/pylxd/__init__.py index c86f6ea4..968b9d06 100644 --- a/pylxd/__init__.py +++ b/pylxd/__init__.py @@ -10,10 +10,10 @@ # License for the specific language governing permissions and limitations # under the License. -from pkg_resources import get_distribution +from importlib.metadata import version from pylxd.client import Client, EventType __all__ = ["Client", "EventType"] -__version__ = get_distribution("pylxd").version +__version__ = version("pylxd") diff --git a/pylxd/deprecated/__init__.py b/pylxd/deprecated/__init__.py deleted file mode 100644 index e69de29b..00000000 diff --git a/pylxd/deprecated/api.py b/pylxd/deprecated/api.py deleted file mode 100644 index 63fcd20b..00000000 --- a/pylxd/deprecated/api.py +++ /dev/null @@ -1,316 +0,0 @@ -# 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 warnings - -from pylxd.deprecated import ( - certificate, - connection, - container, - hosts, - image, - network, - operation, - profiles, -) - - -class API: - def __init__(self, host=None, port=8443): - warnings.warn( - "pylxd.api.API is deprecated. Please use pylxd.Client.", DeprecationWarning - ) - 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) - - def get_host_certificate(self): - return self.hosts.get_certificate() - - def host_config(self): - return self.hosts.host_config() - - # 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) diff --git a/pylxd/deprecated/base.py b/pylxd/deprecated/base.py deleted file mode 100644 index f79fcac4..00000000 --- a/pylxd/deprecated/base.py +++ /dev/null @@ -1,20 +0,0 @@ -# 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.deprecated import connection - - -class LXDBase: - def __init__(self, conn=None): - self.connection = conn or connection.LXDConnection() diff --git a/pylxd/deprecated/certificate.py b/pylxd/deprecated/certificate.py deleted file mode 100644 index e7cb7cde..00000000 --- a/pylxd/deprecated/certificate.py +++ /dev/null @@ -1,37 +0,0 @@ -# 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.deprecated 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", f"/1.0/certificates/{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", f"/1.0/certificates/{fingerprint}") diff --git a/pylxd/deprecated/connection.py b/pylxd/deprecated/connection.py deleted file mode 100644 index dcbc46f3..00000000 --- a/pylxd/deprecated/connection.py +++ /dev/null @@ -1,201 +0,0 @@ -# Copyright (c) 2015 Canonical Ltd -# Copyright (c) 2015 Mirantis inc. -# -# 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 copy -import json -import os -import queue -import socket -import ssl -import threading -from http import client as http_client -from typing import Any, NamedTuple - -from ws4py.client import WebSocketBaseClient - -from pylxd.deprecated import exceptions, utils - -DEFAULT_TLS_VERSION = ssl.PROTOCOL_TLSv1_2 - - -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, 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=DEFAULT_TLS_VERSION - ) - - @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 _LXDResponse(NamedTuple): - status: int - body: bytes - json: Any - - -class WebSocketClient(WebSocketBaseClient): - def __init__( - self, url, protocols=None, extensions=None, ssl_options=None, headers=None - ): - """WebSocket client that executes into a eventlet green thread.""" - WebSocketBaseClient.__init__( - self, - url, - protocols, - extensions, - ssl_options=ssl_options, - headers=headers, - ) - self._th = threading.Thread(target=self.run, name="WebSocketClient") - self._th.daemon = True - - self.messages = queue.Queue() - - def handshake_ok(self): - """Starts the client's thread.""" - self._th.start() - - def received_message(self, message): - """Override the base class to store the incoming message.""" - self.messages.put(copy.deepcopy(message)) - - def closed(self, code, reason=None): - # When the connection is closed, put a StopIteration - # on the message queue to signal there's nothing left - # to wait for - self.messages.put(StopIteration) - - def receive(self): - # If the websocket was terminated and there are no messages - # left in the queue, return None immediately otherwise the client - # will block forever - if self.terminated and self.messages.empty(): - return None - message = self.messages.get() - if message is StopIteration: - return None - return message - - -class LXDConnection: - 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 _request(self, *args, **kwargs): - if self.connection is None: - self.connection = self.get_connection() - self.connection.request(*args, **kwargs) - response = self.connection.getresponse() - - status = response.status - raw_body = response.read() - try: - body = json.loads(raw_body.decode()) - except ValueError: - body = None - - return _LXDResponse(status, raw_body, body) - - def get_connection(self): - if self.host: - return HTTPSConnection(self.host, self.port) - return UnixHTTPConnection(self.unix_socket) - - def get_object(self, *args, **kwargs): - response = self._request(*args, **kwargs) - - if not response.json: - raise exceptions.PyLXDException("Null Data") - elif response.status == 200 or ( - response.status == 202 and response.json.get("status_code") == 100 - ): - return response.status, response.json - else: - utils.get_lxd_error(response.status, response.json) - - def get_status(self, *args, **kwargs): - response = self._request(*args, **kwargs) - - if not response.json: - raise exceptions.PyLXDException("Null Data") - elif response.json.get("error"): - utils.get_lxd_error(response.status, response.json) - elif response.status == 200 or ( - response.status == 202 and response.json.get("status_code") == 100 - ): - return True - return False - - def get_raw(self, *args, **kwargs): - response = self._request(*args, **kwargs) - - if not response.body: - raise exceptions.PyLXDException("Null Body") - elif response.status == 200: - return response.body - else: - raise exceptions.PyLXDException("Failed to get raw response") - - def get_ws(self, path): - if self.unix_socket: - connection_string = f"ws+unix://{self.unix_socket}" - else: - connection_string = f"wss://{self.host}:{self.port}" - - ws = WebSocketClient(connection_string) - ws.resource = path - ws.connect() - - return ws diff --git a/pylxd/deprecated/container.py b/pylxd/deprecated/container.py deleted file mode 100644 index 60a1eae8..00000000 --- a/pylxd/deprecated/container.py +++ /dev/null @@ -1,211 +0,0 @@ -# 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.deprecated import base, exceptions - - -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", f"/1.0/containers/{container}/state" - ) - data = data.get("metadata") - container_running = False - if data["status"].upper() 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", f"/1.0/containers/{container}", json.dumps(config) - ) - - def container_defined(self, container): - _, data = self.connection.get_object("GET", "/1.0/containers") - try: - containers = data["metadata"] - except KeyError: - raise exceptions.PyLXDException("no metadata in GET containers?") - - container_url = f"/1.0/containers/{container}" - for ct in containers: - if ct == container_url: - return True - return False - - def container_state(self, container): - return self.connection.get_object("GET", f"/1.0/containers/{container}/state") - - def container_start(self, container, timeout): - action = {"action": "start", "force": True, "timeout": timeout} - return self.connection.get_object( - "PUT", f"/1.0/containers/{container}/state", json.dumps(action) - ) - - def container_stop(self, container, timeout): - action = {"action": "stop", "force": True, "timeout": timeout} - return self.connection.get_object( - "PUT", f"/1.0/containers/{container}/state", json.dumps(action) - ) - - def container_suspend(self, container, timeout): - action = {"action": "freeze", "force": True, "timeout": timeout} - return self.connection.get_object( - "PUT", f"/1.0/containers/{container}/state", json.dumps(action) - ) - - def container_resume(self, container, timeout): - action = {"action": "unfreeze", "force": True, "timeout": timeout} - return self.connection.get_object( - "PUT", f"/1.0/containers/{container}/state", json.dumps(action) - ) - - def container_reboot(self, container, timeout): - action = {"action": "restart", "force": True, "timeout": timeout} - return self.connection.get_object( - "PUT", f"/1.0/containers/{container}/state", json.dumps(action) - ) - - def container_destroy(self, container): - return self.connection.get_object("DELETE", f"/1.0/containers/{container}") - - def get_container_log(self, container): - (state, data) = self.connection.get_object( - "GET", f"/1.0/containers/{container}?log=true" - ) - return data["metadata"]["log"] - - def get_container_config(self, container): - (state, data) = self.connection.get_object( - "GET", f"/1.0/containers/{container}?log=false" - ) - return data["metadata"] - - def get_container_websocket(self, container): - return self.connection.get_status( - "GET", - f"/1.0/operations/{container['operation']}/websocket?secret={container['fs']}", - ) - - def container_info(self, container): - (state, data) = self.connection.get_object( - "GET", f"/1.0/containers/{container}/state" - ) - return data["metadata"] - - def container_migrate(self, container): - action = {"migration": True} - return self.connection.get_object( - "POST", f"/1.0/containers/{container}", json.dumps(action) - ) - - def container_migrate_sync(self, operation_id, container_secret): - return self.connection.get_ws( - f"/1.0/operations/{operation_id}/websocket?secret={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", f"/1.0/containers/{instance}", json.dumps(config) - ) - - # file operations - def get_container_file(self, container, filename): - return self.connection.get_raw( - "GET", f"/1.0/containers/{container}/files?path={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", - f"/1.0/containers/{container}/files?path={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", f"/1.0/containers/{container}/exec", json.dumps(data) - ) - - # snapshots - def snapshot_list(self, container): - (state, data) = self.connection.get_object( - "GET", f"/1.0/containers/{container}/snapshots" - ) - return [ - snapshot.split(f"/1.0/containers/{container}/snapshots/{container}/")[-1] - for snapshot in data["metadata"] - ] - - def snapshot_create(self, container, config): - return self.connection.get_object( - "POST", f"/1.0/containers/{container}/snapshots", json.dumps(config) - ) - - def snapshot_info(self, container, snapshot): - return self.connection.get_object( - "GET", f"/1.0/containers/{container}/snapshots/{snapshot}" - ) - - def snapshot_rename(self, container, snapshot, config): - return self.connection.get_object( - "POST", - f"/1.0/containers/{container}/snapshots/{snapshot}", - json.dumps(config), - ) - - def snapshot_delete(self, container, snapshot): - return self.connection.get_object( - "DELETE", f"/1.0/containers/{container}/snapshots/{snapshot}" - ) diff --git a/pylxd/deprecated/exceptions.py b/pylxd/deprecated/exceptions.py deleted file mode 100644 index 47ac59c9..00000000 --- a/pylxd/deprecated/exceptions.py +++ /dev/null @@ -1,45 +0,0 @@ -# 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 = f"Error {status_code} - {error}." - super().__init__(msg) - self.status_code = status_code - self.error = error diff --git a/pylxd/deprecated/hosts.py b/pylxd/deprecated/hosts.py deleted file mode 100644 index 65301d93..00000000 --- a/pylxd/deprecated/hosts.py +++ /dev/null @@ -1,115 +0,0 @@ -# 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.deprecated import base, exceptions - - -class LXDHost(base.LXDBase): - def host_ping(self): - try: - return self.connection.get_status("GET", "/1.0") - except Exception as e: - msg = f"LXD service is unavailable. {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(f"Handling run-time error: {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(f"Handling run-time error: {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(f"Handling run-time error: {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(f"Handling run-time error: {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(f"Handling run-time error: {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(f"Handling run-time error: {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(f"Handling run-time error: {e}") - - def get_certificate(self): - try: - (state, data) = self.connection.get_object("GET", "/1.0") - data = data.get("metadata") - return data["environment"]["certificate"] - except exceptions.PyLXDException as e: - print(f"Handling run-time error: {e}") - - def host_config(self): - try: - (state, data) = self.connection.get_object("GET", "/1.0") - return data.get("metadata") - except exceptions.PyLXDException as e: - print(f"Handling run-time error: {e}") diff --git a/pylxd/deprecated/image.py b/pylxd/deprecated/image.py deleted file mode 100644 index 624ad80d..00000000 --- a/pylxd/deprecated/image.py +++ /dev/null @@ -1,233 +0,0 @@ -# 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 -import json -import urllib - -from pylxd.deprecated import base, connection, 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(f"Unable to fetch image info - {e}") - raise - - def image_defined(self, image): - try: - (state, data) = self.connection.get_object("GET", f"/1.0/images/{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(f"Unable to fetch image info - {e}") - raise - - # image info - def image_info(self, image): - try: - (state, data) = self.connection.get_object("GET", f"/1.0/images/{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": f"{self.get_image_size(image, data.get('metadata'))}MB", - "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(f"Unable to fetch image info - {e}") - raise - - def get_image_date(self, image, data, key): - try: - if data is None: - (state, data) = self.connection.get_object( - "GET", f"/1.0/images/{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(f"Unable to fetch image info - {e}") - raise - - def get_image_permission(self, image, data): - try: - if data is None: - (state, data) = self.connection.get_object( - "GET", f"/1.0/images/{image}" - ) - data = data.get("metadata") - return True if data["public"] == 1 else False - except Exception as e: - print(f"Unable to fetch image info - {e}") - raise - - def get_image_size(self, image, data): - try: - if data is None: - (state, data) = self.connection.get_object( - "GET", f"/1.0/images/{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(f"Unable to fetch image info - {e}") - raise - - def get_image_fingerprint(self, image, data): - try: - if data is None: - (state, data) = self.connection.get_object( - "GET", f"/1.0/images/{image}" - ) - data = data.get("metadata") - return data["fingerprint"] - except Exception as e: - print(f"Unable to fetch image info - {e}") - raise - - def get_image_architecture(self, image, data): - try: - if data is None: - (state, data) = self.connection.get_object( - "GET", f"/1.0/images/{image}" - ) - data = data.get("metadata") - return image_architecture[data["architecture"]] - except Exception as e: - print(f"Unable to fetch image info - {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(f"Unable to upload image - {e}") - raise - - def image_delete(self, image): - try: - return self.connection.get_status("DELETE", f"/1.0/images/{image}") - except Exception as e: - print(f"Unable to delete image - {e}") - raise - - def image_export(self, image): - try: - return self.connection.get_raw("GET", f"/1.0/images/{image}/export") - except Exception as e: - print(f"Unable to export image - {e}") - raise - - def image_update(self, image, data): - try: - return self.connection.get_status( - "PUT", f"/1.0/images/{image}", json.dumps(data) - ) - except Exception as e: - print(f"Unable to update image - {e}") - raise - - def image_rename(self, image, data): - try: - return self.connection.get_status( - "POST", f"/1.0/images/{image}", json.dumps(data) - ) - except Exception as e: - print(f"Unable to rename image - {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", f"/1.0/images/aliases/{alias}") - - def alias_show(self, alias): - return self.connection.get_object("GET", f"/1.0/images/aliases/{alias}") - - def alias_update(self, alias, data): - return self.connection.get_status( - "PUT", f"/1.0/images/aliases/{alias}", json.dumps(data) - ) - - def alias_rename(self, alias, data): - return self.connection.get_status( - "POST", f"/1.0/images/aliases/{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", f"/1.0/images/aliases/{alias}") diff --git a/pylxd/deprecated/network.py b/pylxd/deprecated/network.py deleted file mode 100644 index e892207b..00000000 --- a/pylxd/deprecated/network.py +++ /dev/null @@ -1,58 +0,0 @@ -# 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.deprecated 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", f"/1.0/networks/{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", f"/1.0/networks/{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", f"/1.0/networks/{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", f"/1.0/networks/{network}" - ) - data = data.get("metadata") - return [ - network_members.split("/1.0/networks/")[-1] - for network_members in data["members"] - ] diff --git a/pylxd/deprecated/operation.py b/pylxd/deprecated/operation.py deleted file mode 100644 index 344036b1..00000000 --- a/pylxd/deprecated/operation.py +++ /dev/null @@ -1,77 +0,0 @@ -# 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.deprecated import base - - -class LXDOperation(base.LXDBase): - def operation_list(self): - (state, data) = self.connection.get_object("GET", "/1.0/operations") - return data["metadata"] - - def operation_show(self, operation): - (state, data) = self.connection.get_object("GET", 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", operation) - - def operation_create_time(self, operation, data): - if data is None: - (state, data) = self.connection.get_object("GET", 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", 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", operation) - data = data.get("metadata") - return data["status"] - - def operation_wait(self, operation, status_code, timeout): - if timeout == -1: - return self.connection.get_status( - "GET", f"{operation}/wait?status_code={status_code}" - ) - else: - return self.connection.get_status( - "GET", - f"{operation}/wait?status_code={status_code}&timeout={timeout}", - ) - - def operation_stream(self, operation, operation_secret): - return self.connection.get_ws( - f"{operation}/websocket?secret={operation_secret}" - ) - - def operation_delete(self, operation): - return self.connection.get_status("DELETE", operation) diff --git a/pylxd/deprecated/profiles.py b/pylxd/deprecated/profiles.py deleted file mode 100644 index d59ec7c0..00000000 --- a/pylxd/deprecated/profiles.py +++ /dev/null @@ -1,51 +0,0 @@ -# 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.deprecated 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", f"/1.0/profiles/{profile}") - - def profile_defined(self, profile): - """Check for an LXD profile""" - return self.connection.get_status("GET", f"/1.0/profiles/{profile}") - - def profile_update(self, profile, config): - """Update the LXD profile (not implemented)""" - return self.connection.get_status( - "PUT", f"/1.0/profiles/{profile}", json.dumps(config) - ) - - def profile_rename(self, profile, config): - """Rename the LXD profile""" - return self.connection.get_status( - "POST", f"/1.0/profiles/{profile}", json.dumps(config) - ) - - def profile_delete(self, profile): - """Delete the LXD profile""" - return self.connection.get_status("DELETE", f"/1.0/profiles/{profile}") diff --git a/pylxd/deprecated/tests/__init__.py b/pylxd/deprecated/tests/__init__.py deleted file mode 100644 index 12fca5ff..00000000 --- a/pylxd/deprecated/tests/__init__.py +++ /dev/null @@ -1,39 +0,0 @@ -# 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 unittest - -from ddt import data, unpack - -from pylxd.deprecated import api - - -class LXDAPITestBase(unittest.TestCase): - def setUp(self): - super().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)) diff --git a/pylxd/deprecated/tests/fake_api.py b/pylxd/deprecated/tests/fake_api.py deleted file mode 100644 index 1f6bc5c7..00000000 --- a/pylxd/deprecated/tests/fake_api.py +++ /dev/null @@ -1,228 +0,0 @@ -# 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"], - }, - } diff --git a/pylxd/deprecated/tests/test_certificate.py b/pylxd/deprecated/tests/test_certificate.py deleted file mode 100644 index e4081902..00000000 --- a/pylxd/deprecated/tests/test_certificate.py +++ /dev/null @@ -1,46 +0,0 @@ -# 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 unittest import mock - -from ddt import ddt - -from pylxd.deprecated import connection -from pylxd.deprecated.tests import LXDAPITestBase, annotated_data, fake_api - - -@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) diff --git a/pylxd/deprecated/tests/test_connection.py b/pylxd/deprecated/tests/test_connection.py deleted file mode 100644 index 44d267ed..00000000 --- a/pylxd/deprecated/tests/test_connection.py +++ /dev/null @@ -1,156 +0,0 @@ -# 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 inspect -import socket -from http import client as http_client -from io import BytesIO -from unittest import TestCase, mock - -from ddt import ddt - -from pylxd.deprecated import connection, exceptions -from pylxd.deprecated.tests import annotated_data - - -@ddt -class LXDInitConnectionTest(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, 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=connection.DEFAULT_TLS_VERSION, - ) - - @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=connection.DEFAULT_TLS_VERSION, - ) - - @mock.patch("pylxd.deprecated.connection.HTTPSConnection") - @mock.patch("pylxd.deprecated.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: - def __init__(self, status, data): - self.status = status - self.read = BytesIO(bytes(data, "utf-8")).read - - -@ddt -@mock.patch("pylxd.deprecated.connection.LXDConnection.get_connection") -class LXDConnectionTest(TestCase): - def setUp(self): - super().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"}'), b'{"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()) - - @mock.patch("pylxd.deprecated.connection.WebSocketClient") - @annotated_data( - ("fake_host", "wss://fake_host:8443"), - (None, "ws+unix:///var/lib/lxd/unix.socket"), - ) - def test_get_ws(self, host, result, mock_ws, _): - conn = connection.LXDConnection(host) - - conn.get_ws("/fake/path") - - mock_ws.assert_has_calls([mock.call(result), mock.call().connect()]) diff --git a/pylxd/deprecated/tests/test_container.py b/pylxd/deprecated/tests/test_container.py deleted file mode 100644 index f848afac..00000000 --- a/pylxd/deprecated/tests/test_container.py +++ /dev/null @@ -1,213 +0,0 @@ -# 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 tempfile -from collections import OrderedDict -from unittest import mock - -from ddt import ddt - -from pylxd.deprecated import connection -from pylxd.deprecated.tests import LXDAPITestBase, annotated_data, fake_api - - -@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( - ( - "200", - { - "type": "sync", - "status": "Success", - "metadata": { - "criu": "fake_criu", - "fs": "fake_fs", - "control": "fake_control", - }, - "operation": "/1.0/operations/1234", - "status_code": 200, - }, - ), - 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_object", - return_value=("200", fake_api.fake_container_list()), -) -class LXDAPIContainerTestStatus(LXDAPITestBase): - def test_container_defined(self, ms): - self.assertTrue(self.lxd.container_defined("trusty-1")) - ms.assert_called_once_with("GET", "/1.0/containers") - - -@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" - ) diff --git a/pylxd/deprecated/tests/test_host.py b/pylxd/deprecated/tests/test_host.py deleted file mode 100644 index 1220ff38..00000000 --- a/pylxd/deprecated/tests/test_host.py +++ /dev/null @@ -1,80 +0,0 @@ -# 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 unittest import mock - -from ddt import data, ddt - -from pylxd.deprecated import connection, exceptions -from pylxd.deprecated.tests import LXDAPITestBase, annotated_data, fake_api - - -@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") diff --git a/pylxd/deprecated/tests/test_image.py b/pylxd/deprecated/tests/test_image.py deleted file mode 100644 index 981431a6..00000000 --- a/pylxd/deprecated/tests/test_image.py +++ /dev/null @@ -1,262 +0,0 @@ -# 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 builtins -import datetime -from io import StringIO -from unittest import TestCase, mock - -from ddt import ddt - -from pylxd.deprecated import connection, exceptions, image -from pylxd.deprecated.tests import LXDAPITestBase, annotated_data, fake_api - - -@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, f"image_{method}_date")("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, f"image_{method}_date"), - "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=StringIO("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(TestCase): - def setUp(self): - super().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) diff --git a/pylxd/deprecated/tests/test_image_alias.py b/pylxd/deprecated/tests/test_image_alias.py deleted file mode 100644 index 41afa03b..00000000 --- a/pylxd/deprecated/tests/test_image_alias.py +++ /dev/null @@ -1,72 +0,0 @@ -# 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 unittest import mock - -from ddt import data, ddt - -from pylxd.deprecated import connection -from pylxd.deprecated.tests import LXDAPITestBase, annotated_data, fake_api - - -@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) diff --git a/pylxd/deprecated/tests/test_network.py b/pylxd/deprecated/tests/test_network.py deleted file mode 100644 index e3823d25..00000000 --- a/pylxd/deprecated/tests/test_network.py +++ /dev/null @@ -1,53 +0,0 @@ -# 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 unittest import mock - -from ddt import ddt - -from pylxd.deprecated import connection -from pylxd.deprecated.tests import LXDAPITestBase, annotated_data, fake_api - - -@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") diff --git a/pylxd/deprecated/tests/test_operation.py b/pylxd/deprecated/tests/test_operation.py deleted file mode 100644 index 27febf68..00000000 --- a/pylxd/deprecated/tests/test_operation.py +++ /dev/null @@ -1,78 +0,0 @@ -# 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 unittest import mock - -from ddt import ddt - -from pylxd.deprecated import connection -from pylxd.deprecated.tests import LXDAPITestBase, annotated_data, fake_api - - -@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(["/1.0/operations/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("/1.0/operations/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): - call = getattr(self.lxd, "operation_show_" + method) - self.assertEqual(expected, call("/1.0/operations/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)("/1.0/operations/1234", *args)) - ms.assert_called_with(http, "/1.0/operations/1234" + path) diff --git a/pylxd/deprecated/tests/test_profiles.py b/pylxd/deprecated/tests/test_profiles.py deleted file mode 100644 index d8c74fcf..00000000 --- a/pylxd/deprecated/tests/test_profiles.py +++ /dev/null @@ -1,72 +0,0 @@ -# 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 unittest import mock - -from ddt import data, ddt - -from pylxd.deprecated import connection -from pylxd.deprecated.tests import LXDAPITestBase, annotated_data, fake_api - - -@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) diff --git a/pylxd/deprecated/utils.py b/pylxd/deprecated/utils.py deleted file mode 100644 index 811252cd..00000000 --- a/pylxd/deprecated/utils.py +++ /dev/null @@ -1,29 +0,0 @@ -# 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.deprecated 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) diff --git a/pylxd/deprecation.py b/pylxd/deprecation.py deleted file mode 100644 index 8115d4cc..00000000 --- a/pylxd/deprecation.py +++ /dev/null @@ -1,26 +0,0 @@ -import warnings - -warnings.simplefilter("once", DeprecationWarning) - - -class deprecated: # pragma: no cover - """A decorator for warning about deprecation warnings. - - The decorator takes an optional message argument. This message can - be used to direct the user to a new API or specify when it will - be removed. - """ - - DEFAULT_MESSAGE = "{} is deprecated and will be removed soon." - - def __init__(self, message=None): - self.message = message - - def __call__(self, f): - def wrapped(*args, **kwargs): - if self.message is None: - self.message = self.DEFAULT_MESSAGE.format(f.__name__) - warnings.warn(self.message, DeprecationWarning) - return f(*args, **kwargs) - - return wrapped