From 285519d2edf64785ea83dc400afd17f9a4c7ad09 Mon Sep 17 00:00:00 2001 From: Jhon Honce Date: Mon, 24 Jan 2022 11:28:22 -0700 Subject: [PATCH] Remove ApiConnection support This API was the original draft API that has been marked deprecated for the Podman3 releases, with Podman4 being released soon now is the time to remove this support. APIClient provides a compatible interface while also allowing access to libpod enhancements. Signed-off-by: Jhon Honce --- Makefile | 3 +- podman/__init__.py | 11 +- podman/api/client.py | 9 +- podman/api/typing_extensions.py | 23 +- podman/api_connection.py | 175 -------- podman/containers/__init__.py | 419 ------------------ podman/domain/containers_create.py | 7 +- podman/domain/pods_manager.py | 4 +- podman/images/__init__.py | 77 ---- podman/manifests/__init__.py | 87 ---- podman/networks/__init__.py | 76 ---- podman/pods/__init__.py | 176 -------- podman/system/__init__.py | 64 --- podman/tests/unit/containers/__init__.py | 0 .../tests/unit/containers/test_containers.py | 392 ---------------- podman/tests/unit/images/__init__.py | 0 podman/tests/unit/images/test_images.py | 139 ------ podman/tests/unit/manifests/__init__.py | 0 podman/tests/unit/manifests/test_manifests.py | 124 ------ podman/tests/unit/networks/__init__.py | 0 podman/tests/unit/networks/test_networks.py | 165 ------- podman/tests/unit/pods/__init__.py | 0 podman/tests/unit/pods/test_pods.py | 307 ------------- podman/tests/unit/system/__init__.py | 0 podman/tests/unit/system/test_system.py | 84 ---- podman/tests/unit/test_api_connection.py | 195 -------- podman/tests/unit/test_config.py | 4 +- podman/tests/unit/test_containersmanager.py | 8 +- podman/tests/unit/test_parse_utils.py | 67 +++ 29 files changed, 114 insertions(+), 2502 deletions(-) delete mode 100644 podman/api_connection.py delete mode 100644 podman/containers/__init__.py delete mode 100644 podman/images/__init__.py delete mode 100644 podman/manifests/__init__.py delete mode 100644 podman/networks/__init__.py delete mode 100644 podman/pods/__init__.py delete mode 100644 podman/system/__init__.py delete mode 100644 podman/tests/unit/containers/__init__.py delete mode 100644 podman/tests/unit/containers/test_containers.py delete mode 100644 podman/tests/unit/images/__init__.py delete mode 100644 podman/tests/unit/images/test_images.py delete mode 100644 podman/tests/unit/manifests/__init__.py delete mode 100644 podman/tests/unit/manifests/test_manifests.py delete mode 100644 podman/tests/unit/networks/__init__.py delete mode 100644 podman/tests/unit/networks/test_networks.py delete mode 100644 podman/tests/unit/pods/__init__.py delete mode 100644 podman/tests/unit/pods/test_pods.py delete mode 100644 podman/tests/unit/system/__init__.py delete mode 100644 podman/tests/unit/system/test_system.py delete mode 100644 podman/tests/unit/test_api_connection.py create mode 100644 podman/tests/unit/test_parse_utils.py diff --git a/Makefile b/Makefile index 806a6904..8ce24578 100644 --- a/Makefile +++ b/Makefile @@ -56,8 +56,7 @@ docs: cp -R docs/source/* build/docs/source sphinx-apidoc --separate --no-toc --force --templatedir build/docs/source/_templates/apidoc \ -o build/docs/source \ - podman podman/tests podman/api_connection.py podman/containers podman/images \ - podman/manifests podman/networks podman/pods podman/system + podman podman/tests sphinx-build build/docs/source build/html # .PHONY: install diff --git a/podman/__init__.py b/podman/__init__.py index bc26f542..1db26b60 100644 --- a/podman/__init__.py +++ b/podman/__init__.py @@ -3,17 +3,8 @@ assert sys.version_info >= (3, 6), "Python 3.6 or greater is required." -try: - from podman.api_connection import ApiConnection -except ImportError: - - class ApiConnection: # pylint: disable=too-few-public-methods - def __init__(self): - raise NotImplementedError("ApiConnection deprecated, please use PodmanClient().") - - from podman.client import PodmanClient, from_env from podman.version import __version__ # isort: unique-list -__all__ = ['ApiConnection', 'PodmanClient', '__version__', 'from_env'] +__all__ = ['PodmanClient', '__version__', 'from_env'] diff --git a/podman/api/client.py b/podman/api/client.py index 3cbf8bb7..560605d4 100644 --- a/podman/api/client.py +++ b/podman/api/client.py @@ -72,7 +72,14 @@ class APIClient(requests.Session): # pylint: disable=arguments-differ # pylint: disable=too-many-instance-attributes - supported_schemes: ClassVar[List[str]] = ("unix", "http+unix", "ssh", "http+ssh", "tcp", "http") + supported_schemes: ClassVar[List[str]] = ( + "unix", + "http+unix", + "ssh", + "http+ssh", + "tcp", + "http", + ) def __init__( self, diff --git a/podman/api/typing_extensions.py b/podman/api/typing_extensions.py index 21df93e7..ac00e1a7 100644 --- a/podman/api/typing_extensions.py +++ b/podman/api/typing_extensions.py @@ -884,6 +884,7 @@ def __new__(cls, *args, **kwds): return collections.deque(*args, **kwds) return _generic_new(collections.deque, cls, *args, **kwds) + else: class Deque( @@ -911,6 +912,7 @@ class ContextManager( ): __slots__ = () + else: class ContextManager(typing.Generic[T_co]): @@ -992,6 +994,7 @@ def __new__(cls, *args, **kwds): return collections.defaultdict(*args, **kwds) return _generic_new(collections.defaultdict, cls, *args, **kwds) + else: class DefaultDict( @@ -1029,6 +1032,7 @@ def __new__(cls, *args, **kwds): return collections.OrderedDict(*args, **kwds) return _generic_new(collections.OrderedDict, cls, *args, **kwds) + else: class OrderedDict( @@ -1069,6 +1073,7 @@ def __new__(cls, *args, **kwds): return collections.Counter(*args, **kwds) return _generic_new(collections.Counter, cls, *args, **kwds) + elif _geqv_defined: class Counter( @@ -1085,6 +1090,7 @@ def __new__(cls, *args, **kwds): return collections.Counter(*args, **kwds) return _generic_new(collections.Counter, cls, *args, **kwds) + else: class Counter( @@ -1311,7 +1317,10 @@ def __new__( for base in bases: if base is Generic: raise TypeError("Cannot inherit from plain Generic") - if isinstance(base, GenericMeta) and base.__origin__ in (Generic, Protocol): + if isinstance(base, GenericMeta) and base.__origin__ in ( + Generic, + Protocol, + ): if gvars is not None: raise TypeError( "Cannot inherit from Generic[...] or" @@ -1344,7 +1353,9 @@ def __new__( bases = tuple(b for b in bases if b is not Generic) namespace.update({'__origin__': origin, '__extra__': extra}) self = super(GenericMeta, cls).__new__(cls, name, bases, namespace, _root=True) - super(GenericMeta, self).__setattr__('_gorg', self if not origin else _gorg(origin)) + super(GenericMeta, self).__setattr__( + '_gorg', self if not origin else _gorg(origin) + ) self.__parameters__ = tvars self.__args__ = ( tuple( @@ -1468,7 +1479,9 @@ def __getitem__(self, params): if not isinstance(params, tuple): params = (params,) if not params and _gorg(self) is not Tuple: - raise TypeError("Parameter list to %s[...] cannot be empty" % self.__qualname__) + raise TypeError( + "Parameter list to %s[...] cannot be empty" % self.__qualname__ + ) msg = "Parameters to generic types must be types." params = tuple(_type_check(p, msg) for p in params) if self in (Generic, Protocol): @@ -2095,6 +2108,7 @@ def get_type_hints(obj, globalns=None, localns=None, include_extras=False): return hint return {k: _strip_annotations(t) for k, t in hint.items()} + elif HAVE_ANNOTATED: def _is_dunder(name): @@ -2330,6 +2344,7 @@ def TypeAlias(self, parameters): """ raise TypeError("{} is not subscriptable".format(self)) + elif sys.version_info[:2] >= (3, 7): class _TypeAliasForm(typing._SpecialForm, _root=True): @@ -2657,6 +2672,7 @@ def Concatenate(self, parameters): """ return _concatenate_getitem(self, parameters) + elif sys.version_info[:2] >= (3, 7): class _ConcatenateForm(typing._SpecialForm, _root=True): @@ -2805,6 +2821,7 @@ def is_str(val: Union[str, float]): item = typing._type_check(parameters, '{} accepts only single type.'.format(self)) return _GenericAlias(self, (item,)) + elif sys.version_info[:2] >= (3, 7): class _TypeGuardForm(typing._SpecialForm, _root=True): diff --git a/podman/api_connection.py b/podman/api_connection.py deleted file mode 100644 index 76c8b864..00000000 --- a/podman/api_connection.py +++ /dev/null @@ -1,175 +0,0 @@ -""" Provides a Connection to a Podman service.""" -import json -import logging -import socket -import urllib.parse -import warnings -from contextlib import AbstractContextManager -from http import HTTPStatus -from http.client import HTTPConnection - -from podman import containers -from podman import errors -from podman import images -from podman import system - - -class ApiConnection(HTTPConnection, AbstractContextManager): - """ - ApiConnection provides a specialized HTTPConnection - to a Podman service. - """ - - def __init__(self, url, base="/v2.0.0/libpod", *args, **kwargs): # pylint: disable-msg=W1113 - if url is None or not url: - raise ValueError("url is required for service connection.") - - super().__init__("localhost", *args, **kwargs) - supported_schemes = ("unix", "ssh") - uri = urllib.parse.urlparse(url) - if uri.scheme not in supported_schemes: - raise ValueError( - f"The scheme '{uri.scheme}' is not supported, only {supported_schemes}" - ) - self.uri = uri - self.base = base - warnings.warn("APIConnection() and supporting classes.", PendingDeprecationWarning) - - def connect(self): - """Connect to the URL given when initializing class""" - if self.uri.scheme == "unix": - sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) - sock.connect(self.uri.path) - self.sock = sock - else: - raise NotImplementedError(f"Scheme {self.uri.scheme} not yet implemented") - - def delete(self, path, params=None): - """Basic DELETE wrapper for requests - - Send a delete request with params added to the url as a query string - - :param path: url part to the call, appended to self.base - :param params: optional dictionary of query params added to the request - :return: http response object - """ - return self.request("DELETE", self.join(path, params)) - - def get(self, path, params=None): - """Basic GET wrapper for requests - - Send a get request with params added to the url as a query string - - :param path: url part to the call, appended to self.base - :param params: optional dictionary of query params added to the request - :return: http response object - """ - return self.request("GET", self.join(path, params)) - - def post(self, path, params=None, headers=None, encode=False): - """Basic POST wrapper for requests - - Send a POST request with params converted into a urlencoded form to be - sent with the post. - - :param path: url part to the call, appended to self.base - :param params: optional dictionary of query params added to the post - request as url encoded form data - :param headers: optional dictionary of request headers - :param encode: flag to indicate if you want the params to be encoded - prior to posting. When set to False, params are posted - directly as the body of the request. - :return: http response object - """ - data = params - if not headers: - headers = {} - - if encode: - if "content-type" not in set(key.lower() for key in headers) and params: - headers["content-type"] = "application/x-www-form-urlencoded" - - data = urllib.parse.urlencode(params) - return self.request('POST', self.join(path), body=data, headers=headers) - - def request(self, method, url, body=None, headers=None, *, encode_chunked=False): - """Make request to Podman service.""" - if headers is None: - headers = {} - - super().request(method, url, body, headers, encode_chunked=encode_chunked) - response = super().getresponse() - - # Errors are mapped to exceptions - if HTTPStatus.OK <= response.status < HTTPStatus.MULTIPLE_CHOICES: - pass - elif HTTPStatus.NOT_FOUND == response.status: - raise errors.NotFoundError( - "Request {}:{} failed: {}".format( # pylint: disable=consider-using-f-string - method, - url, - HTTPStatus.NOT_FOUND.description or HTTPStatus.NOT_FOUND.phrase, - ), - response, - ) - elif ( - response.status >= HTTPStatus.BAD_REQUEST - and response.status < HTTPStatus.INTERNAL_SERVER_ERROR - ): - raise errors.RequestError( - "Request {}:{} failed: {}".format( # pylint: disable=consider-using-f-string - method, - url, - response.reason or f"Response Status Code {response.status}", - ), - response, - ) - elif response.status >= HTTPStatus.INTERNAL_SERVER_ERROR: - try: - error_body = response.read() - error_message = json.loads(error_body)["message"] - except: # pylint: disable=bare-except - error_message = ( - HTTPStatus.INTERNAL_SERVER_ERROR.description - or HTTPStatus.INTERNAL_SERVER_ERROR.phrase - ) - raise errors.InternalServerError( - f"Request {method}:{url} failed: {error_message}", - response, - ) - return response - - def join(self, path, query=None): - """Create a service URL. Join base + path + query parameters""" - path = self.base + path - if query is not None: - query = urllib.parse.urlencode(query) - path = path + "?" + query - return path - - @staticmethod - def quote(value): - """Quote value for use in a URL""" - return urllib.parse.quote(value) - - @staticmethod - def raise_not_found(exc, response, exception_type=errors.ImageNotFound): - """helper function to raise a not found exception of exception_type""" - body = json.loads(response.read()) - logging.info(body["cause"]) - raise exception_type(body["message"]) from exc - - def __exit__(self, exc_type, exc_value, traceback): - super().close() - - -if __name__ == "__main__": # pragma: no cover - with ApiConnection("unix:///run/podman/podman.sock") as api: - print(system.version(api)) - print(images.list_images(api)) - print(containers.list_containers(api)) - - try: - images.inspect(api, "bozo the clown") - except errors.ImageNotFound as e: - print(e) diff --git a/podman/containers/__init__.py b/podman/containers/__init__.py deleted file mode 100644 index a89b153b..00000000 --- a/podman/containers/__init__.py +++ /dev/null @@ -1,419 +0,0 @@ -"""containers provides the operations against containers for a Podman service. -""" - -import json -from http import HTTPStatus - -from podman import errors - - -def attach(api, name): - """Attach to a container""" - raise NotImplementedError('Attach not implemented yet') - - -def changes(api, name): - """Get files added, deleted or modified in a container""" - try: - response = api.get(f'/containers/{api.quote(name)}/changes') - except errors.NotFoundError as e: - api.raise_not_found(e, e.response, errors.ContainerNotFound) - return json.loads(str(response.read(), 'utf-8')) - - -def checkpoint( - api, - name, - export_image=None, - ignore_root_fs=None, - keep=None, - leave_running=None, - tcp_established=None, -): - """Copy tar of files into a container""" - path = f'/containers/{api.quote(name)}/checkpoint' - params = {} - if export_image is not None: - params['export'] = export_image - if ignore_root_fs is not None: - params['ignoreRootFS'] = ignore_root_fs - if keep is not None: - params['keep'] = keep - if leave_running is not None: - params['leaveRunning'] = leave_running - if tcp_established is not None: - params['tcpEstablished'] = tcp_established - try: - response = api.post(path, params=params, headers={'content-type': 'application/json'}) - except errors.NotFoundError as e: - api.raise_not_found(e, e.response, errors.ContainerNotFound) - return response.read() - - -def copy(api, name, file_path, pause_container=None): - """Copy tar of files into a container""" - path = f'/containers/{api.quote(name)}/copy' - params = {'path': file_path} - if pause_container is not None: - params['pause'] = pause_container - try: - response = api.post(path, params=params, headers={'content-type': 'application/json'}) - response.read() - except errors.NotFoundError as e: - api.raise_not_found(e, e.response, errors.ContainerNotFound) - return True - - -def container_exists(api, name): - """Check if container exists""" - try: - api.get(f"/containers/{api.quote(name)}/exists") - return True - except errors.NotFoundError: - return False - - -def create(api, container_data): - """Create a container with the provided attributes - - container_data is a dictionary contining the container attributes for - creation. See documentation for specifics. - https://docs.podman.io/en/latest/_static/api.html#operation/libpodCreateContainer - """ - try: - response = api.post( - "/containers/create", - params=container_data, - headers={'content-type': 'application/json'}, - ) - response.read() - except errors.NotFoundError as e: - api.raise_not_found(e, e.response, errors.ContainerNotFound) - return response.status == HTTPStatus.CREATED - - -def export(api, name): - """Container export""" - try: - response = api.get(f"/containers/{api.quote(name)}/export") - except errors.NotFoundError as e: - api.raise_not_found(e, e.response, errors.ContainerNotFound) - return response.read() - - -def healthcheck(api, name): - """Execute container healthcheck""" - try: - response = api.get(f'/containers/{api.quote(name)}/healthcheck') - except errors.NotFoundError as e: - api.raise_not_found(e, e.response, errors.ContainerNotFound) - return json.loads(str(response.read(), 'utf-8')) - - -def init(api, name): - """Initialize a container - - Returns true if successful, false if already done - """ - try: - response = api.post(f'/containers/{api.quote(name)}/init') - response.read() - except errors.NotFoundError as e: - api.raise_not_found(e, e.response, errors.ContainerNotFound) - return response.status == HTTPStatus.NO_CONTENT - - -def inspect(api, name): - """Report on named container for a Podman service. - Name may also be a container ID. - """ - try: - response = api.get(f'/containers/{api.quote(name)}/json') - except errors.NotFoundError as e: - api.raise_not_found(e, e.response, errors.ContainerNotFound) - return json.loads(str(response.read(), 'utf-8')) - - -def kill(api, name, signal=None): - """kill named/identified container""" - path = f"/containers/{api.quote(name)}/kill" - params = {} - if signal is not None: - params = {'signal': signal} - - try: - response = api.post(path, params=params, headers={'content-type': 'application/json'}) - # returns an empty bytes object - response.read() - except errors.NotFoundError as e: - api.raise_not_found(e, e.response, errors.ContainerNotFound) - return True - - -def list_containers(api, all_=None, filters=None, limit=None, size=None, sync=None): - """List all images for a Podman service.""" - query = {} - if all_ is not None: - query["all"] = True - if filters is not None: - query["filters"] = filters - if limit is not None: - query["limit"] = limit - if size is not None: - query["size"] = size - if sync is not None: - query["sync"] = sync - response = api.get("/containers/json", query) - # observed to return None when no containers - return json.loads(str(response.read(), "utf-8")) or [] - - -def logs(api, name, follow=None, since=None, stderr=None, tail=None, timestamps=None, until=None): - """Get stdout and stderr logs""" - raise NotImplementedError('Logs not implemented yet') - - -def mount(api, name): - """Mount container to the filesystem""" - path = f"/containers/{api.quote(name)}/mount" - try: - response = api.post(path, headers={'content-type': 'application/json'}) - except errors.NotFoundError as e: - api.raise_not_found(e, e.response, errors.ContainerNotFound) - return json.loads(str(response.read(), "utf-8")) - - -def pause(api, name): - """Pause a container""" - path = f"/containers/{api.quote(name)}/pause" - try: - response = api.post(path, headers={'content-type': 'application/json'}) - response.read() - except errors.NotFoundError as e: - api.raise_not_found(e, e.response, errors.ContainerNotFound) - return response.status == HTTPStatus.NO_CONTENT - - -def prune(api, name, filters=None): - """Remove stopped containers""" - path = f"/containers/{api.quote(name)}/prune" - params = {} - if filters is not None: - params['filters'] = filters - response = api.post(path, params=params, headers={'content-type': 'application/json'}) - return json.loads(str(response.read(), "utf-8")) - - -def remove(api, name, force=None, delete_volumes=None): - """Delete container""" - path = f"/containers/{api.quote(name)}" - params = {} - if force is not None: - params['force'] = force - if delete_volumes is not None: - params['v'] = delete_volumes - - try: - response = api.delete(path, params) - # returns an empty bytes object - response.read() - except errors.NotFoundError as e: - api.raise_not_found(e, e.response, errors.ContainerNotFound) - return True - - -def resize(api, name, height, width): - """Resize container tty""" - path = f"/containers/{api.quote(name)}/resize" - params = {'h': height, 'w': width} - try: - response = api.post(path, params=params, headers={'content-type': 'application/json'}) - except errors.NotFoundError as e: - api.raise_not_found(e, e.response, errors.ContainerNotFound) - return json.loads(str(response.read(), 'utf-8')) - - -def restart(api, name, timeout=None): - """Restart container""" - path = f"/containers/{api.quote(name)}/restart" - params = {} - if timeout is not None: - params['t'] = timeout - try: - response = api.post(path, params=params, headers={'content-type': 'application/json'}) - response.read() - except errors.NotFoundError as e: - api.raise_not_found(e, e.response, errors.ContainerNotFound) - return response.status == HTTPStatus.NO_CONTENT - - -def restore( - api, - name, - ignore_root_fs=None, - ignore_static_ip=None, - ignore_static_mac=None, - import_arg=None, - keep=None, - leave_running=None, - container_name=None, - tcp_established=None, -): - """Restore container""" - path = f"/containers/{api.quote(name)}/restore" - params = {} - if ignore_root_fs is not None: - params['ignoreRootFS'] = ignore_root_fs - if ignore_static_ip is not None: - params['ignoreStaticIP'] = ignore_static_ip - if ignore_static_mac is not None: - params['ignoreStaticMAC'] = ignore_static_mac - if import_arg is not None: - params['import'] = import_arg - if keep is not None: - params['keep'] = keep - if leave_running is not None: - params['leaveRunning'] = leave_running - if container_name is not None: - params['name'] = container_name - if tcp_established is not None: - params['tcpEstablished'] = tcp_established - try: - response = api.post(path, params=params, headers={'content-type': 'application/json'}) - except errors.NotFoundError as e: - api.raise_not_found(e, e.response, errors.ContainerNotFound) - # TODO(mwhahaha): handle returned tarball better - return response.read() - - -def show_mounted(api): - """Show mounted containers""" - response = api.get('/containers/showmounted') - return json.loads(str(response.read(), 'utf-8')) - - -def start(api, name, detach_keys=None): - """Start container""" - path = f"/containers/{api.quote(name)}/start" - params = {} - if detach_keys is not None: - params['detachKeys'] = detach_keys - try: - response = api.post(path, params=params, headers={'content-type': 'application/json'}) - response.read() - except errors.NotFoundError as e: - api.raise_not_found(e, e.response, errors.ContainerNotFound) - return response.status == HTTPStatus.NO_CONTENT - - -def stats(api, containers=None, stream=True): - """Get container stats container - - When stream is set to true, the raw HTTPResponse is returned. - """ - path = "/containers/stats" - params = {'stream': stream} - if containers is not None: - params['containers'] = containers - try: - response = api.get(path, params=params) - if stream: - return response - except errors.NotFoundError as e: - api.raise_not_found(e, e.response, errors.ContainerNotFound) - return json.loads(str(response.read(), 'utf-8')) - - -def stop(api, name, timeout=None): - """Stop container""" - path = f"/containers/{api.quote(name)}/stop" - params = {} - if timeout is not None: - params['t'] = timeout - try: - response = api.post(path, params=params, headers={'content-type': 'application/json'}) - response.read() - except errors.NotFoundError as e: - api.raise_not_found(e, e.response, errors.ContainerNotFound) - return response.status == HTTPStatus.NO_CONTENT - - -def top(api, name, ps_args=None, stream=True): - """List processes in a container - - When stream is set to true, the raw HTTPResponse is returned. - """ - path = f"/containers/{api.quote(name)}/top" - params = {'stream': stream} - if ps_args is not None: - params['ps_args'] = ps_args - try: - response = api.get(path, params=params) - if stream: - return response - except errors.NotFoundError as e: - api.raise_not_found(e, e.response, errors.ContainerNotFound) - return json.loads(str(response.read(), 'utf-8')) - - -def unmount(api, name): - """Unmount container""" - path = f"/containers/{api.quote(name)}/unmount" - try: - response = api.post(path, headers={'content-type': 'application/json'}) - response.read() - except errors.NotFoundError as e: - api.raise_not_found(e, e.response, errors.ContainerNotFound) - return response.status == HTTPStatus.NO_CONTENT - - -def unpause(api, name): - """Unpause container""" - path = f"/containers/{api.quote(name)}/unpause" - try: - response = api.post(path, headers={'content-type': 'application/json'}) - response.read() - except errors.NotFoundError as e: - api.raise_not_found(e, e.response, errors.ContainerNotFound) - return response.status == HTTPStatus.NO_CONTENT - - -def wait(api, name, condition=None): - """Wait for a container state""" - path = f"/containers/{api.quote(name)}/wait" - params = {} - if condition is not None: - params['condition'] = condition - try: - response = api.post(path, params=params, headers={'content-type': 'application/json'}) - except errors.NotFoundError as e: - api.raise_not_found(e, e.response, errors.ContainerNotFound) - return json.loads(str(response.read(), 'utf-8')) - - -__all__ = [ - "attach", - "changes", - "checkpoint", - "copy", - "container_exists", - "export", - "healthcheck", - "inspect", - "kill", - "list_containers", - "logs", - "mount", - "pause", - "prune", - "remove", - "resize", - "restore", - "show_mounted", - "start", - "stop", - "top", - "unmount", - "unpause", - "wait", -] diff --git a/podman/domain/containers_create.py b/podman/domain/containers_create.py index c24b3cdf..5dd09b6e 100644 --- a/podman/domain/containers_create.py +++ b/podman/domain/containers_create.py @@ -310,7 +310,8 @@ def to_bytes(size: Union[int, str, None]) -> Union[int, None]: if search: return int(search.group(1)) * (1024 ** mapping[search.group(2)]) raise TypeError( - f"Passed string size {size} should be in format\\d+[bBkKmMgG] (e.g. '100m')" + f"Passed string size {size} should be in format\\d+[bBkKmMgG] (e.g." + " '100m')" ) from bad_size else: raise TypeError( @@ -414,7 +415,9 @@ def to_bytes(size: Union[int, str, None]) -> Union[int, None]: if "Config" in args["log_config"]: params["log_configuration"]["path"] = args["log_config"]["Config"].get("path") params["log_configuration"]["size"] = args["log_config"]["Config"].get("size") - params["log_configuration"]["options"] = args["log_config"]["Config"].get("options") + params["log_configuration"]["options"] = args["log_config"]["Config"].get( + "options" + ) args.pop("log_config") for item in args.pop("mounts", []): diff --git a/podman/domain/pods_manager.py b/podman/domain/pods_manager.py index d591984f..0051f496 100644 --- a/podman/domain/pods_manager.py +++ b/podman/domain/pods_manager.py @@ -94,7 +94,9 @@ def prune(self, filters: Optional[Dict[str, str]] = None) -> Dict[str, Any]: Raises: APIError: when service reports error """ - response = self.client.post("/pods/prune", params={"filters": api.prepare_filters(filters)}) + response = self.client.post( + "/pods/prune", params={"filters": api.prepare_filters(filters)} + ) response.raise_for_status() deleted: List[str] = [] diff --git a/podman/images/__init__.py b/podman/images/__init__.py deleted file mode 100644 index b0efc373..00000000 --- a/podman/images/__init__.py +++ /dev/null @@ -1,77 +0,0 @@ -"""images provides the operations against images for a Podman service.""" -import json -from http import HTTPStatus - -from podman import errors - - -def list_images(api): - """List all images for a Podman service.""" - response = api.get("/images/json") - return json.loads(str(response.read(), "utf-8")) - - -def inspect(api, name): - """Report on named image for a Podman service. - Name may also be an image ID. - """ - try: - response = api.get(f"/images/{api.quote(name)}/json") - except errors.NotFoundError as e: - api.raise_not_found(e, e.response) - return json.loads(str(response.read(), "utf-8")) - - -def image_exists(api, name): - """Checks if an image exists in the local store""" - try: - api.get(f"/images/{api.quote(name)}/exists") - except errors.NotFoundError: - return False - return True - - -def remove(api, name, force=None): - """Remove named/identified image from Podman storage.""" - params = {} - path = f"/images/{api.quote(name)}" - if force is not None: - params = {"force": force} - try: - response = api.delete(path, params) - except errors.NotFoundError as e: - api.raise_not_found(e, e.response) - return json.loads(str(response.read(), "utf-8")) - - -def tag_image(api, name, repo, tag): - """create an image tag using repo and tag - - :param repo: string for the image repo - :param tag: string for the image tag - :return boolean - """ - data = {"repo": repo, "tag": tag} - try: - response = api.post(f"/images/{api.quote(name)}/tag", data) - except errors.NotFoundError as e: - api.raise_image_not_found(e, e.response) - return response.status == HTTPStatus.CREATED - - -def history(api, name): - """get image history""" - try: - response = api.get(f"/images/{api.quote(name)}/history") - except errors.NotFoundError as e: - api.raise_image_not_found(e, e.response) - return json.loads(str(response.read(), "utf-8")) - - -__all__ = [ - "list_images", - "inspect", - "image_exists", - "remove", - "tag_image", -] diff --git a/podman/manifests/__init__.py b/podman/manifests/__init__.py deleted file mode 100644 index 6b8fd631..00000000 --- a/podman/manifests/__init__.py +++ /dev/null @@ -1,87 +0,0 @@ -# Copyright 2020 Red Hat, 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. -# -"""manifests provides the manifest operations for a Podman service""" -import json -from http import HTTPStatus - -from podman import errors - - -def add(api, name, manifest): - """Add image to a manifest list""" - path = f'/manifests/{api.quote(name)}/add' - try: - response = api.post(path, params=manifest, headers={'content-type': 'application/json'}) - except errors.NotFoundError as e: - api.raise_not_found(e, e.response, errors.ManifestNotFound) - return response.status == HTTPStatus.OK - - -def create(api, name, image=None, all_contents=None): - """create a manifest""" - params = {'name': name} - if image: - params['image'] = image - if all_contents: - params['all'] = all_contents - path = '/manifests/create' - try: - response = api.post(path, params=params) - except errors.NotFoundError as e: - api.raise_not_found(e, e.response, errors.ManifestNotFound) - return response.status == HTTPStatus.OK - - -def inspect(api, name): - """inspect a manifest""" - try: - response = api.get(f'/manifests/{api.quote(name)}/json') - except errors.NotFoundError as e: - api.raise_not_found(e, e.response, errors.ManifestNotFound) - return json.loads(str(response.read(), 'utf-8')) - - -def push(api, name, destination, all_images=None): - """push a manifest""" - params = {'destination': destination} - if all_images: - params['all'] = all_images - try: - response = api.post(f'/manifests/{api.quote(name)}/push', params=params) - except errors.NotFoundError as e: - api.raise_not_found(e, e.response, errors.ManifestNotFound) - return response.status == HTTPStatus.OK - - -def remove(api, name, digest=None): - """Remove manifest digest.""" - params = {} - if digest: - params['digest'] = digest - path = f'/manifests/{api.quote(name)}' - try: - response = api.delete(path, params) - except errors.NotFoundError as e: - api.raise_not_found(e, e.response, errors.ManifestNotFound) - return response.status == HTTPStatus.OK - - -__all__ = [ - "add", - "create", - "inspect", - "push", - "remove", -] diff --git a/podman/networks/__init__.py b/podman/networks/__init__.py deleted file mode 100644 index 6c9e413a..00000000 --- a/podman/networks/__init__.py +++ /dev/null @@ -1,76 +0,0 @@ -# Copyright 2020 Red Hat, 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. -# -"""network provides the network operations for a Podman service""" -import json - -from podman import errors - - -def create(api, name, network): - """create a network""" - if not isinstance(network, str): - data = json.dumps(network) - else: - data = network - path = f'/networks/create?name={api.quote(name)}' - response = api.post(path, params=data, headers={'content-type': 'application/json'}) - return json.loads(str(response.read(), 'utf-8')) - - -def inspect(api, name): - """inspect a network""" - try: - response = api.get(f'/networks/{api.quote(name)}/json') - except errors.NotFoundError as e: - api.raise_not_found(e, e.response, errors.NetworkNotFound) - return json.loads(str(response.read(), 'utf-8')) - - -def list_networks(api, filters=None): - """list networks using filters""" - filters_param = {} - if filters: - filters_param = {'filter': filters} - response = api.get('/networks/json', filters_param) - return json.loads(str(response.read(), 'utf-8')) - - -def remove(api, name, force=None): - """remove a named network""" - params = {} - path = f'/networks/{api.quote(name)}' - if force is not None: - params = {'force': force} - try: - response = api.delete(path, params) - except errors.NotFoundError as e: - api.raise_not_found(e, e.response, errors.NetworkNotFound) - return json.loads(str(response.read(), 'utf-8')) - - -def prune(api): - """prune unused networks""" - path = '/networks/prune' - response = api.post(path, headers={'content-type': 'application/json'}) - return json.loads(str(response.read(), 'utf-8')) - - -__all__ = [ - "create", - "inspect", - "list_networks", - "remove", - "prune", -] diff --git a/podman/pods/__init__.py b/podman/pods/__init__.py deleted file mode 100644 index 26e5ed9b..00000000 --- a/podman/pods/__init__.py +++ /dev/null @@ -1,176 +0,0 @@ -# Copyright 2020 Red Hat, 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. -# -"""pod provides the pod operations for a Podman service""" -import json - -from podman import errors - - -def create(api, name, pod): - """create a pod""" - if not isinstance(pod, str): - data = json.dumps(pod) - else: - data = pod - path = f'/pods/create?name={api.quote(name)}' - response = api.post(path, params=data, headers={'content-type': 'application/json'}) - return json.loads(str(response.read(), 'utf-8')) - - -def exists(api, name): - """inspect a pod""" - try: - api.get(f'/pods/{api.quote(name)}/exists') - except errors.NotFoundError: - return False - return True - - -def inspect(api, name): - """inspect a pod""" - try: - response = api.get(f'/pods/{api.quote(name)}/json') - except errors.NotFoundError as e: - api.raise_not_found(e, e.response, errors.PodNotFound) - return json.loads(str(response.read(), 'utf-8')) - - -def kill(api, name, signal=None): - """kill a pod""" - data = {} - if signal: - data['signal'] = signal - try: - response = api.post(f'/pods/{api.quote(name)}/kill', data) - except errors.NotFoundError as e: - api.raise_not_found(e, e.response, errors.PodNotFound) - return json.loads(str(response.read(), 'utf-8')) - - -def list_pods(api, filters=None): - """list pod using filter""" - filters_param = {} - if filters: - filters_param = {'filter': filters} - response = api.get('/pods/json', filters_param) - return json.loads(str(response.read(), 'utf-8')) - - -def list_processes(api, name, stream=None, ps_args=None): - """list processes from a pod""" - params = {} - # TODO(mwhahaha): test stream - if stream: - params['stream'] = stream - if ps_args: - params['ps_args'] = ps_args - response = api.get(f'/pods/{api.quote(name)}/top', params) - return json.loads(str(response.read(), 'utf-8')) - - -def pause(api, name): - """pause a pod""" - try: - response = api.post(f'/pods/{api.quote(name)}/pause') - except errors.NotFoundError as e: - api.raise_not_found(e, e.response, errors.PodNotFound) - return json.loads(str(response.read(), 'utf-8')) - - -def prune(api): - """prune pods""" - response = api.get('/pods/prune') - return json.loads(str(response.read(), 'utf-8')) - - -def remove(api, name, force=None): - """Remove named/identified image from Podman storage.""" - params = {} - path = f'/pods/{api.quote(name)}' - if force is not None: - params = {'force': force} - try: - response = api.delete(path, params) - except errors.NotFoundError as e: - api.raise_not_found(e, e.response, errors.PodNotFound) - return json.loads(str(response.read(), 'utf-8')) - - -def restart(api, name): - """restart a pod""" - try: - response = api.post(f'/pods/{api.quote(name)}/restart') - except errors.NotFoundError as e: - api.raise_not_found(e, e.response, errors.PodNotFound) - return json.loads(str(response.read(), 'utf-8')) - - -def start(api, name): - """start a pod""" - try: - response = api.post(f'/pods/{api.quote(name)}/start') - except errors.NotFoundError as e: - api.raise_not_found(e, e.response, errors.PodNotFound) - # TODO(mwhahaha): handle 304 warning - return json.loads(str(response.read(), 'utf-8')) - - -def stats(api, all_pods=True, pods=None): - """get pods stats""" - params = {'all': all_pods} - if pods: - params['namesOrIDs'] = pods - try: - response = api.post('/pods/stats', params) - except errors.NotFoundError as e: - api.raise_not_found(e, e.response, errors.PodNotFound) - return json.loads(str(response.read(), 'utf-8')) - - -def stop(api, name): - """stop a pod""" - try: - response = api.post(f'/pods/{api.quote(name)}/stop') - except errors.NotFoundError as e: - api.raise_not_found(e, e.response, errors.PodNotFound) - # TODO(mwhahaha): handle 304 warning - return json.loads(str(response.read(), 'utf-8')) - - -def unpause(api, name): - """unpause a pod""" - try: - response = api.post(f'/pods/{api.quote(name)}/unpause') - except errors.NotFoundError as e: - api.raise_not_found(e, e.response, errors.PodNotFound) - return json.loads(str(response.read(), 'utf-8')) - - -__all__ = [ - "create", - "exists", - "inspect", - "kill", - "list_pods", - "list_processes", - "pause", - "prune", - "remove", - "restart", - "stats", - "start", - "stop", - "unpause", -] diff --git a/podman/system/__init__.py b/podman/system/__init__.py deleted file mode 100644 index d6a6ba9e..00000000 --- a/podman/system/__init__.py +++ /dev/null @@ -1,64 +0,0 @@ -"""Provide system level information for the Podman service.""" -import json -import logging -from http import HTTPStatus - -from podman import errors - - -def version(api, verify_version=False): - """Obtain a dictionary of versions for the Podman components.""" - versions = {} - response = api.get("/version") - if response.status == HTTPStatus.OK: - versions = json.loads(str(response.read(), 'utf-8')) - # pylint: disable=fixme - # TODO: verify api.base and header[Api-Version] compatible - if verify_version: - pass - return versions - - -def get_info(api): - """Returns information on the system and libpod configuration""" - try: - response = api.get("/info") - except errors.NotFoundError as e: - api.raise_not_found(e, e.response) - return json.loads(str(response.read(), 'utf-8')) - - -# this **looks** a lot like the above but is not equivalent - at all ! -# the difference lies with calling api.join() -# and the output have nothing to do with one another -# xxx the naming is going to be confusing -def info(api): - """Returns information on the system and libpod configuration""" - response = api.get("/info") - return json.loads(response.read()) - - -def show_disk_usage(api): - """ - Return information about disk usage for containers, - images and volumes - """ - try: - response = api.get("/system/df") - except errors.NotFoundError as e: - api.raise_not_found(e, e.response) - return json.loads(str(response.read(), 'utf-8')) - - -def _report_not_found(e, response): - body = json.loads(response.read()) - logging.info(body["cause"]) - raise errors.ImageNotFound(body["message"]) from e - - -__all__ = [ - "version", - "get_info", - "info", - "show_disk_usage", -] diff --git a/podman/tests/unit/containers/__init__.py b/podman/tests/unit/containers/__init__.py deleted file mode 100644 index e69de29b..00000000 diff --git a/podman/tests/unit/containers/test_containers.py b/podman/tests/unit/containers/test_containers.py deleted file mode 100644 index 18465e0a..00000000 --- a/podman/tests/unit/containers/test_containers.py +++ /dev/null @@ -1,392 +0,0 @@ -"""podman.containers unit tests""" - -import unittest -import urllib.parse -from unittest import mock - -import podman.containers -import podman.errors -import podman.system - - -class TestContainers(unittest.TestCase): - """Test the containers calls.""" - - def setUp(self): - super().setUp() - self.request = mock.MagicMock() - self.response = mock.MagicMock() - self.request.return_value = self.response - self.api = mock.MagicMock() - self.api.get = self.request - self.api.post = self.request - self.api.delete = self.request - self.api.quote = urllib.parse.quote - - def test_checkpoint(self): - """test checkpoint call""" - mock_read = mock.MagicMock() - mock_read.return_value = b'' - self.response.status = 200 - self.response.read = mock_read - expected = b'' - ret = podman.containers.checkpoint(self.api, 'foo') - self.assertEqual(ret, expected) - self.request.assert_called_once_with( - "/containers/foo/checkpoint", params={}, headers={'content-type': 'application/json'} - ) - - def test_checkpoint_options(self): - """test checkpoint call with options""" - mock_read = mock.MagicMock() - mock_read.return_value = b'' - self.response.status = 200 - self.response.read = mock_read - expected = b'' - ret = podman.containers.checkpoint(self.api, 'foo', False, False, False, False, False) - self.assertEqual(ret, expected) - params = { - 'export': False, - 'ignoreRootFS': False, - 'keep': False, - 'leaveRunning': False, - 'tcpEstablished': False, - } - self.request.assert_called_once_with( - "/containers/foo/checkpoint", - params=params, - headers={'content-type': 'application/json'}, - ) - - def test_copy(self): - """test copy call""" - mock_read = mock.MagicMock() - self.response.status = 200 - self.response.read = mock_read - ret = podman.containers.copy(self.api, 'foo', '/foo.tar', False) - self.assertTrue(ret) - params = {'path': '/foo.tar', 'pause': False} - self.request.assert_called_once_with( - "/containers/foo/copy", params=params, headers={'content-type': 'application/json'} - ) - - def test_container_exists(self): - """test checkpoint call""" - self.request.side_effect = [mock.MagicMock(), podman.errors.NotFoundError('')] - ret = podman.containers.container_exists(self.api, 'foo') - self.assertTrue(ret) - ret = podman.containers.container_exists(self.api, 'foo') - self.assertFalse(ret) - calls = [ - mock.call('/containers/foo/exists'), - mock.call('/containers/foo/exists'), - ] - self.request.assert_has_calls(calls) - - def test_create(self): - """test create call""" - self.response.status = 201 - ret = podman.containers.create(self.api, {'container': 'data'}) - self.assertTrue(ret) - params = {'container': 'data'} - self.request.assert_called_once_with( - "/containers/create", params=params, headers={'content-type': 'application/json'} - ) - - def test_export(self): - """test export call""" - mock_read = mock.MagicMock() - mock_read.return_value = b'' - self.response.status = 200 - self.response.read = mock_read - ret = podman.containers.export(self.api, 'foo') - self.assertEqual(ret, b'') - self.request.assert_called_once_with("/containers/foo/export") - - def test_healthcheck(self): - """test healthcheck call""" - mock_read = mock.MagicMock() - mock_read.return_value = b'{"data": "stuff"}' - self.response.status = 200 - self.response.read = mock_read - ret = podman.containers.healthcheck(self.api, 'foo') - self.assertEqual(ret, {'data': 'stuff'}) - self.request.assert_called_once_with("/containers/foo/healthcheck") - - def test_init(self): - """test init call""" - self.response.status = 204 - ret = podman.containers.init(self.api, 'foo') - self.assertTrue(ret) - self.request.assert_called_once_with("/containers/foo/init") - - def test_inspect(self): - """test inspect call""" - mock_read = mock.MagicMock() - mock_read.return_value = b'{"Id": "foo"}' - self.response.status = 200 - self.response.read = mock_read - expected = {"Id": "foo"} - ret = podman.containers.inspect(self.api, 'foo') - self.assertEqual(ret, expected) - self.request.assert_called_once_with("/containers/foo/json") - - def test_kill(self): - """test kill call""" - mock_read = mock.MagicMock() - mock_read.return_value = b'' - self.response.status = 204 - self.response.read = mock_read - ret = podman.containers.kill(self.api, 'foo') - self.assertTrue(ret) - self.request.assert_called_once_with( - "/containers/foo/kill", params={}, headers={'content-type': 'application/json'} - ) - - def test_kill_signal(self): - """test kill call with signal""" - mock_read = mock.MagicMock() - mock_read.return_value = b'' - self.response.status = 200 - self.response.read = mock_read - ret = podman.containers.kill(self.api, 'foo', 'HUP') - self.assertTrue(ret) - self.request.assert_called_once_with( - "/containers/foo/kill", - params={'signal': 'HUP'}, - headers={'content-type': 'application/json'}, - ) - - def test_list_containers(self): - """test list call""" - mock_read = mock.MagicMock() - mock_read.return_value = b'[{"Id": "foo"}]' - self.response.status = 200 - self.response.read = mock_read - expected = [{"Id": "foo"}] - ret = podman.containers.list_containers(self.api) - self.assertEqual(ret, expected) - self.request.assert_called_once_with("/containers/json", {}) - - def test_list_containers_all(self): - """test list call""" - mock_read = mock.MagicMock() - mock_read.return_value = b'[{"Id": "foo"}]' - self.response.status = 200 - self.response.read = mock_read - expected = [{"Id": "foo"}] - ret = podman.containers.list_containers(self.api, True) - self.assertEqual(ret, expected) - self.request.assert_called_once_with("/containers/json", {"all": True}) - - def test_logs(self): - """test logs call""" - self.assertRaises(NotImplementedError, podman.containers.logs, self.api, 'foo') - - def test_mount(self): - """test mount call""" - mock_read = mock.MagicMock() - mock_read.return_value = b'"/stuff"' - self.response.status = 200 - self.response.read = mock_read - ret = podman.containers.mount(self.api, 'foo') - self.assertEqual(ret, '/stuff') - self.request.assert_called_once_with( - "/containers/foo/mount", headers={'content-type': 'application/json'} - ) - - def test_pause(self): - """test pause call""" - mock_read = mock.MagicMock() - mock_read.return_value = b'' - self.response.status = 204 - self.response.read = mock_read - ret = podman.containers.pause(self.api, 'foo') - self.assertTrue(ret) - self.request.assert_called_once_with( - "/containers/foo/pause", headers={'content-type': 'application/json'} - ) - - def test_prune(self): - """test prune call""" - mock_read = mock.MagicMock() - mock_read.return_value = b'[{"Id": "foo"}]' - self.response.status = 200 - self.response.read = mock_read - ret = podman.containers.prune(self.api, 'foo') - self.assertEqual(ret, [{"Id": "foo"}]) - self.request.assert_called_once_with( - "/containers/foo/prune", params={}, headers={'content-type': 'application/json'} - ) - - def test_remove(self): - """test remove call""" - mock_read = mock.MagicMock() - mock_read.return_value = b'' - self.response.status = 204 - self.response.read = mock_read - ret = podman.containers.remove(self.api, 'foo') - self.assertTrue(ret) - self.request.assert_called_once_with("/containers/foo", {}) - - def test_remove_options(self): - """test remove call with options""" - mock_read = mock.MagicMock() - mock_read.return_value = b'' - self.response.status = 204 - self.response.read = mock_read - ret = podman.containers.remove(self.api, 'foo', True, True) - self.assertTrue(ret) - self.request.assert_called_once_with("/containers/foo", {'force': True, 'v': True}) - - def test_resize(self): - """test resize call""" - mock_read = mock.MagicMock() - mock_read.return_value = b'{}' - self.response.status = 200 - self.response.read = mock_read - ret = podman.containers.resize(self.api, 'foo', 80, 80) - self.assertEqual(ret, {}) - self.request.assert_called_once_with( - "/containers/foo/resize", - params={'h': 80, 'w': 80}, - headers={'content-type': 'application/json'}, - ) - - def test_restart(self): - """test restart call""" - mock_read = mock.MagicMock() - mock_read.return_value = b'' - self.response.status = 204 - self.response.read = mock_read - ret = podman.containers.restart(self.api, 'foo') - self.assertTrue(ret) - self.request.assert_called_once_with( - "/containers/foo/restart", params={}, headers={'content-type': 'application/json'} - ) - - def test_restore(self): - """test restore call""" - mock_read = mock.MagicMock() - mock_read.return_value = b'' - self.response.status = 204 - self.response.read = mock_read - ret = podman.containers.restore(self.api, 'foo') - self.assertEqual(ret, b'') - self.request.assert_called_once_with( - "/containers/foo/restore", params={}, headers={'content-type': 'application/json'} - ) - - def test_show_mounted(self): - """test show mounted call""" - mock_read = mock.MagicMock() - mock_read.return_value = b'{"Id": "foo"}' - self.response.status = 200 - self.response.read = mock_read - ret = podman.containers.show_mounted(self.api) - self.assertEqual(ret, {"Id": "foo"}) - self.request.assert_called_once_with("/containers/showmounted") - - def test_start(self): - """test start call""" - mock_read = mock.MagicMock() - mock_read.return_value = b'' - self.response.status = 204 - self.response.read = mock_read - ret = podman.containers.start(self.api, 'foo') - self.assertTrue(ret) - self.request.assert_called_once_with( - "/containers/foo/start", params={}, headers={'content-type': 'application/json'} - ) - - def test_stats(self): - """test stats call""" - mock_read = mock.MagicMock() - mock_read.return_value = b'""' - self.response.status = 200 - self.response.read = mock_read - ret = podman.containers.stats(self.api, stream=False) - self.assertEqual(ret, '') - params = {'stream': False} - self.request.assert_called_once_with("/containers/stats", params=params) - - def test_stats_stream(self): - """test stream call with stream""" - mock_read = mock.MagicMock() - mock_read.return_value = b'' - self.response.status = 200 - self.response.read = mock_read - ret = podman.containers.stats(self.api, 'foo', True) - self.assertEqual(ret, self.response) - params = {'stream': True, 'containers': 'foo'} - self.request.assert_called_once_with("/containers/stats", params=params) - - def test_stop(self): - """test stop call""" - mock_read = mock.MagicMock() - mock_read.return_value = b'' - self.response.status = 204 - self.response.read = mock_read - ret = podman.containers.stop(self.api, 'foo') - self.assertTrue(ret) - self.request.assert_called_once_with( - "/containers/foo/stop", params={}, headers={'content-type': 'application/json'} - ) - - def test_top(self): - """test top call""" - mock_read = mock.MagicMock() - mock_read.return_value = b'""' - self.response.status = 200 - self.response.read = mock_read - ret = podman.containers.top(self.api, 'foo', stream=False) - self.assertEqual(ret, '') - params = {'stream': False} - self.request.assert_called_once_with("/containers/foo/top", params=params) - - def test_top_stream(self): - """test top call with stream""" - mock_read = mock.MagicMock() - mock_read.return_value = b'""' - self.response.status = 200 - self.response.read = mock_read - ret = podman.containers.top(self.api, 'foo', '-a') - self.assertEqual(ret, self.response) - params = {'stream': True, 'ps_args': '-a'} - self.request.assert_called_once_with("/containers/foo/top", params=params) - - def test_unmount(self): - """test unmount call""" - mock_read = mock.MagicMock() - mock_read.return_value = b'' - self.response.status = 204 - self.response.read = mock_read - ret = podman.containers.unmount(self.api, 'foo') - self.assertTrue(ret) - self.request.assert_called_once_with( - "/containers/foo/unmount", headers={'content-type': 'application/json'} - ) - - def test_unpause(self): - """test unpause call""" - mock_read = mock.MagicMock() - mock_read.return_value = b'' - self.response.status = 204 - self.response.read = mock_read - ret = podman.containers.unpause(self.api, 'foo') - self.assertTrue(ret) - self.request.assert_called_once_with( - "/containers/foo/unpause", headers={'content-type': 'application/json'} - ) - - def test_wait(self): - """test wait call""" - mock_read = mock.MagicMock() - mock_read.return_value = b'{"StatusCode": 0}' - self.response.status = 200 - self.response.read = mock_read - ret = podman.containers.wait(self.api, 'foo') - self.assertEqual(ret, {"StatusCode": 0}) - self.request.assert_called_once_with( - "/containers/foo/wait", params={}, headers={'content-type': 'application/json'} - ) diff --git a/podman/tests/unit/images/__init__.py b/podman/tests/unit/images/__init__.py deleted file mode 100644 index e69de29b..00000000 diff --git a/podman/tests/unit/images/test_images.py b/podman/tests/unit/images/test_images.py deleted file mode 100644 index 442ba403..00000000 --- a/podman/tests/unit/images/test_images.py +++ /dev/null @@ -1,139 +0,0 @@ -# Copyright 2020 Red Hat, 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. -# -"""podman.images unit tests""" - -import unittest -import urllib.parse -from unittest import mock - -import podman.errors -import podman.system - - -class TestImages(unittest.TestCase): - """Test the images calls.""" - - def setUp(self): - super().setUp() - self.request = mock.MagicMock() - self.response = mock.MagicMock() - self.request.return_value = self.response - self.api = mock.MagicMock() - self.api.get = self.request - self.api.post = self.request - self.api.delete = self.request - self.api.quote = urllib.parse.quote - - def test_list_images(self): - """test list call""" - mock_read = mock.MagicMock() - mock_read.return_value = b'[{"Id": "foo"}]' - self.response.status = 200 - self.response.read = mock_read - expected = [{"Id": "foo"}] - ret = podman.images.list_images(self.api) - self.assertEqual(ret, expected) - self.request.assert_called_once_with("/images/json") - - def test_inspect(self): - """test inspect call""" - mock_read = mock.MagicMock() - mock_read.return_value = b'{"Id": "foo"}' - self.response.status = 200 - self.response.read = mock_read - expected = {"Id": "foo"} - ret = podman.images.inspect(self.api, "foo") - self.assertEqual(ret, expected) - self.request.assert_called_once_with("/images/foo/json") - - def test_image_exists(self): - """test exists call""" - self.response.status = 204 - self.assertTrue(podman.images.image_exists(self.api, "foo")) - self.request.assert_called_once_with("/images/foo/exists") - - def test_image_exists_missing(self): - """test exists call with missing image""" - self.request.side_effect = podman.errors.NotFoundError("nope") - self.assertFalse(podman.images.image_exists(self.api, "foo")) - - def test_remove(self): - """test remove call""" - mock_read = mock.MagicMock() - mock_read.return_value = b'[{"deleted": "str","untagged": ["str"]}]' - self.response.status = 200 - self.response.read = mock_read - expected = [{"deleted": "str", "untagged": ["str"]}] - ret = podman.images.remove(self.api, "foo") - self.assertEqual(ret, expected) - self.api.delete.assert_called_once_with("/images/foo", {}) - - def test_remove_force(self): - """test remove call with force""" - mock_read = mock.MagicMock() - mock_read.return_value = b'[{"deleted": "str","untagged": ["str"]}]' - self.response.status = 200 - self.response.read = mock_read - expected = [{"deleted": "str", "untagged": ["str"]}] - ret = podman.images.remove(self.api, "foo", True) - self.assertEqual(ret, expected) - self.api.delete.assert_called_once_with("/images/foo", {"force": True}) - - def test_remove_missing(self): - """test remove call with missing image""" - mock_raise = mock.MagicMock() - mock_raise.side_effect = podman.errors.ImageNotFound("yikes") - self.api.raise_not_found = mock_raise - self.request.side_effect = podman.errors.NotFoundError("nope") - self.assertRaises(podman.errors.ImageNotFound, podman.images.remove, self.api, "foo") - - def test_tag_image(self): - """test tag image""" - self.response.status = 201 - self.assertTrue(podman.images.tag_image(self.api, "foo", "bar", "baz")) - - def test_tag_image_fail(self): - """test remove call with missing image""" - mock_raise = mock.MagicMock() - mock_raise.side_effect = podman.errors.ImageNotFound("yikes") - self.api.raise_image_not_found = mock_raise - self.request.side_effect = podman.errors.NotFoundError("nope") - self.assertRaises( - podman.errors.ImageNotFound, - podman.images.tag_image, - self.api, - "foo", - "bar", - "baz", - ) - - def test_history(self): - """test image history""" - mock_read = mock.MagicMock() - mock_read.return_value = b'{"Id": "a"}' - self.response.status = 200 - self.response.read = mock_read - expected = {"Id": "a"} - ret = podman.images.history(self.api, "foo") - self.assertEqual(ret, expected) - self.api.delete.assert_called_once_with("/images/foo/history") - - def test_history_missing_image(self): - """test history with missing image""" - mock_raise = mock.MagicMock() - mock_raise.side_effect = podman.errors.ImageNotFound("yikes") - self.api.raise_image_not_found = mock_raise - self.request.side_effect = podman.errors.NotFoundError("nope") - self.assertRaises(podman.errors.ImageNotFound, podman.images.history, self.api, "foo") diff --git a/podman/tests/unit/manifests/__init__.py b/podman/tests/unit/manifests/__init__.py deleted file mode 100644 index e69de29b..00000000 diff --git a/podman/tests/unit/manifests/test_manifests.py b/podman/tests/unit/manifests/test_manifests.py deleted file mode 100644 index a1d53700..00000000 --- a/podman/tests/unit/manifests/test_manifests.py +++ /dev/null @@ -1,124 +0,0 @@ -# Copyright 2020 Red Hat, 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. -# -"""podman.manifests unit tests""" - -import unittest -import urllib.parse -from unittest import mock - -import podman.errors -import podman.manifests - - -class TestManifests(unittest.TestCase): - """Test the manifest calls.""" - - def setUp(self): - super().setUp() - self.request = mock.MagicMock() - self.response = mock.MagicMock() - self.request.return_value = self.response - self.api = mock.MagicMock() - self.api.get = self.request - self.api.delete = self.request - self.api.post = self.request - self.api.quote = urllib.parse.quote - - def test_add(self): - """test add call""" - manifest = {'all': False, 'annotation': {'prop1': 'string', 'prop2': 'string'}} - self.response.status = 200 - self.assertTrue(podman.manifests.add(self.api, 'foo', manifest)) - - def test_add_missing(self): - """test add call missing manifest""" - self.request.side_effect = podman.errors.NotFoundError('yikes') - mock_raise = mock.MagicMock() - mock_raise.side_effect = podman.errors.ManifestNotFound('yikes') - self.api.raise_not_found = mock_raise - self.assertRaises(podman.errors.ManifestNotFound, podman.manifests.add, self.api, 'foo', {}) - - def test_create(self): - """test create call""" - self.response.status = 200 - self.assertTrue(podman.manifests.create(self.api, 'test', 'image', True)) - self.request.assert_called_once_with( - '/manifests/create', params={'name': 'test', 'image': 'image', 'all': True} - ) - - def test_create_fail(self): - """test create call with an error""" - self.response.status = 400 - self.request.side_effect = podman.errors.RequestError('meh', self.response) - self.assertRaises(podman.errors.RequestError, podman.manifests.create, self.api, 'test') - - def test_create_missing(self): - """test create call with missing manifest""" - self.request.side_effect = podman.errors.NotFoundError('yikes') - mock_raise = mock.MagicMock() - mock_raise.side_effect = podman.errors.ManifestNotFound('yikes') - self.api.raise_not_found = mock_raise - self.assertRaises(podman.errors.ManifestNotFound, podman.manifests.create, self.api, 'test') - - def test_inspect(self): - """test manifests inspect call""" - mock_read = mock.MagicMock() - mock_read.return_value = b'[{"Instances":["podman"]}]' - self.response.read = mock_read - ret = podman.manifests.inspect(self.api, 'podman') - self.assertEqual(ret, [{'Instances': ['podman']}]) - self.request.assert_called_once_with('/manifests/podman/json') - - def test_inspect_missing(self): - """test inspect missing pod""" - self.request.side_effect = podman.errors.NotFoundError('yikes') - mock_raise = mock.MagicMock() - mock_raise.side_effect = podman.errors.ManifestNotFound('yikes') - self.api.raise_not_found = mock_raise - self.assertRaises( - podman.errors.ManifestNotFound, podman.manifests.inspect, self.api, 'podman' - ) - - def test_push(self): - """test manifest push call""" - self.response.status = 200 - self.assertTrue(podman.manifests.push(self.api, 'foo', 'dest', True)) - self.api.post.assert_called_once_with( - '/manifests/foo/push', params={'destination': 'dest', 'all': True} - ) - - def test_push_fail(self): - """test manifest push fails""" - self.request.side_effect = podman.errors.NotFoundError('yikes') - mock_raise = mock.MagicMock() - mock_raise.side_effect = podman.errors.ManifestNotFound('yikes') - self.api.raise_not_found = mock_raise - self.assertRaises( - podman.errors.ManifestNotFound, podman.manifests.push, self.api, 'podman', 'dest' - ) - - def test_remove(self): - """test remove call""" - self.response.status = 200 - self.assertTrue(podman.manifests.remove(self.api, 'foo', 'bar')) - self.api.delete.assert_called_once_with('/manifests/foo', {'digest': 'bar'}) - - def test_remove_missing(self): - """test remove call with missing pod""" - mock_raise = mock.MagicMock() - mock_raise.side_effect = podman.errors.ManifestNotFound('yikes') - self.api.raise_not_found = mock_raise - self.request.side_effect = podman.errors.NotFoundError('nope') - self.assertRaises(podman.errors.ManifestNotFound, podman.manifests.remove, self.api, 'foo') diff --git a/podman/tests/unit/networks/__init__.py b/podman/tests/unit/networks/__init__.py deleted file mode 100644 index e69de29b..00000000 diff --git a/podman/tests/unit/networks/test_networks.py b/podman/tests/unit/networks/test_networks.py deleted file mode 100644 index 0168cf00..00000000 --- a/podman/tests/unit/networks/test_networks.py +++ /dev/null @@ -1,165 +0,0 @@ -# Copyright 2020 Red Hat, 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. -# -"""podman.networks unit tests""" - -import json -import unittest -import urllib.parse -from unittest import mock - -import podman.errors -import podman.networks - - -class TestNetwork(unittest.TestCase): - """Test the network calls.""" - - def setUp(self): - super().setUp() - self.request = mock.MagicMock() - self.response = mock.MagicMock() - self.request.return_value = self.response - self.api = mock.MagicMock() - self.api.get = self.request - self.api.delete = self.request - self.api.post = self.request - self.api.quote = urllib.parse.quote - - def test_create(self): - """test create call""" - network = { - 'DisableDNS': False, - } - mock_read = mock.MagicMock() - mock_read.return_value = b'{"Filename":"string"}' - self.response.read = mock_read - ret = podman.networks.create(self.api, 'test', network) - self.assertEqual(ret, {'Filename': 'string'}) - self.request.assert_called_once_with( - '/networks/create?name=test', - headers={'content-type': 'application/json'}, - params='{"DisableDNS": false}', - ) - - def test_create_with_string(self): - """test create call with string""" - network = { - 'DisableDNS': False, - } - mock_read = mock.MagicMock() - mock_read.return_value = b'{"Filename":"string"}' - self.response.read = mock_read - ret = podman.networks.create(self.api, 'test', json.dumps(network)) - self.assertEqual(ret, {'Filename': 'string'}) - self.request.assert_called_once_with( - '/networks/create?name=test', - headers={'content-type': 'application/json'}, - params='{"DisableDNS": false}', - ) - - def test_create_fail(self): - """test create call with an error""" - network = { - 'DisableDNS': False, - } - self.response.status = 400 - self.request.side_effect = podman.errors.RequestError('meh', self.response) - self.assertRaises( - podman.errors.RequestError, - podman.networks.create, - self.api, - 'test', - json.dumps(network), - ) - - def test_inspect(self): - """test inspect call""" - mock_read = mock.MagicMock() - mock_read.return_value = b'[{"Name":"podman"}]' - self.response.read = mock_read - ret = podman.networks.inspect(self.api, 'podman') - self.assertEqual(ret, [{'Name': 'podman'}]) - self.request.assert_called_once_with('/networks/podman/json') - - def test_inspect_missing(self): - """test inspect missing network""" - self.request.side_effect = podman.errors.NotFoundError('yikes') - mock_raise = mock.MagicMock() - mock_raise.side_effect = podman.errors.NetworkNotFound('yikes') - self.api.raise_not_found = mock_raise - self.assertRaises( - podman.errors.NetworkNotFound, podman.networks.inspect, self.api, 'podman' - ) - - def test_list_networks(self): - """test networks list call""" - mock_read = mock.MagicMock() - mock_read.return_value = b'[{"Name":"podman"}]' - self.response.read = mock_read - ret = podman.networks.list_networks(self.api) - self.assertEqual(ret, [{'Name': 'podman'}]) - self.request.assert_called_once_with('/networks/json', {}) - - def test_list_networks_filter(self): - """test networks list call with a filter""" - mock_read = mock.MagicMock() - mock_read.return_value = b'[{"Name":"podman"}]' - self.response.read = mock_read - ret = podman.networks.list_networks(self.api, 'name=podman') - self.assertEqual(ret, [{'Name': 'podman'}]) - self.request.assert_called_once_with('/networks/json', {'filter': 'name=podman'}) - - def test_remove(self): - """test remove call""" - mock_read = mock.MagicMock() - mock_read.return_value = b'[{"deleted": "str","untagged": ["str"]}]' - self.response.status = 200 - self.response.read = mock_read - expected = [{'deleted': 'str', 'untagged': ['str']}] - ret = podman.networks.remove(self.api, 'foo') - self.assertEqual(ret, expected) - self.api.delete.assert_called_once_with('/networks/foo', {}) - - def test_remove_force(self): - """test remove call with force""" - mock_read = mock.MagicMock() - mock_read.return_value = b'[{"deleted": "str","untagged": ["str"]}]' - self.response.status = 200 - self.response.read = mock_read - expected = [{'deleted': 'str', 'untagged': ['str']}] - ret = podman.networks.remove(self.api, 'foo', True) - self.assertEqual(ret, expected) - self.api.delete.assert_called_once_with('/networks/foo', {'force': True}) - - def test_remove_missing(self): - """test remove call with missing network""" - mock_raise = mock.MagicMock() - mock_raise.side_effect = podman.errors.NetworkNotFound('yikes') - self.api.raise_not_found = mock_raise - self.request.side_effect = podman.errors.NotFoundError('nope') - self.assertRaises(podman.errors.NetworkNotFound, podman.networks.remove, self.api, 'foo') - - def test_prune(self): - """test prune call""" - mock_read = mock.MagicMock() - mock_read.return_value = b'[{"Name": "net1"}, {"Name": "net2"}]' - self.response.read = mock_read - ret = podman.networks.prune(self.api) - expected = [{'Name': 'net1'}, {"Name": "net2"}] - self.assertEqual(ret, expected) - self.request.assert_called_once_with( - '/networks/prune', - headers={'content-type': 'application/json'}, - ) diff --git a/podman/tests/unit/pods/__init__.py b/podman/tests/unit/pods/__init__.py deleted file mode 100644 index e69de29b..00000000 diff --git a/podman/tests/unit/pods/test_pods.py b/podman/tests/unit/pods/test_pods.py deleted file mode 100644 index f25ff80e..00000000 --- a/podman/tests/unit/pods/test_pods.py +++ /dev/null @@ -1,307 +0,0 @@ -# Copyright 2020 Red Hat, 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. -# -"""podman.pods unit tests""" - -import json -import unittest -import urllib.parse -from unittest import mock - -import podman.errors -import podman.pods - - -class TestNetwork(unittest.TestCase): - """Test the pod calls.""" - - def setUp(self): - super().setUp() - self.request = mock.MagicMock() - self.response = mock.MagicMock() - self.request.return_value = self.response - self.api = mock.MagicMock() - self.api.get = self.request - self.api.delete = self.request - self.api.post = self.request - self.api.quote = urllib.parse.quote - - def test_create(self): - """test create call""" - pod = { - 'name': 'foo', - } - mock_read = mock.MagicMock() - mock_read.return_value = b'{"Id":"string"}' - self.response.read = mock_read - ret = podman.pods.create(self.api, 'test', pod) - self.assertEqual(ret, {'Id': 'string'}) - self.request.assert_called_once_with( - '/pods/create?name=test', - headers={'content-type': 'application/json'}, - params='{"name": "foo"}', - ) - - def test_create_fail(self): - """test create call with an error""" - pod = { - 'name': 'foo', - } - self.response.status = 400 - self.request.side_effect = podman.errors.RequestError('meh', self.response) - self.assertRaises( - podman.errors.RequestError, - podman.pods.create, - self.api, - 'test', - json.dumps(pod), - ) - - def test_exists(self): - """test pods exists call""" - self.assertTrue(podman.pods.exists(self.api, 'test')) - - def test_exists_missing(self): - """test pods exists call returns false for missing pod""" - self.request.side_effect = podman.errors.NotFoundError('yikes') - self.assertFalse(podman.pods.exists(self.api, 'test')) - - def test_inspect(self): - """test pods inspect call""" - mock_read = mock.MagicMock() - mock_read.return_value = b'[{"Name":"podman"}]' - self.response.read = mock_read - ret = podman.pods.inspect(self.api, 'podman') - self.assertEqual(ret, [{'Name': 'podman'}]) - self.request.assert_called_once_with('/pods/podman/json') - - def test_inspect_missing(self): - """test inspect missing pod""" - self.request.side_effect = podman.errors.NotFoundError('yikes') - mock_raise = mock.MagicMock() - mock_raise.side_effect = podman.errors.PodNotFound('yikes') - self.api.raise_not_found = mock_raise - self.assertRaises(podman.errors.PodNotFound, podman.pods.inspect, self.api, 'podman') - - def test_kill(self): - """test pods kill call""" - mock_read = mock.MagicMock() - mock_read.return_value = b'{"Id":"string"}' - self.response.read = mock_read - ret = podman.pods.kill(self.api, 'test', 'foo') - self.assertEqual(ret, {'Id': 'string'}) - self.request.assert_called_once_with( - '/pods/test/kill', - {"signal": "foo"}, - ) - - def test_kill_fail(self): - """test pods kill call failure""" - self.request.side_effect = podman.errors.NotFoundError('yikes') - mock_raise = mock.MagicMock() - mock_raise.side_effect = podman.errors.PodNotFound('yikes') - self.api.raise_not_found = mock_raise - self.assertRaises(podman.errors.PodNotFound, podman.pods.kill, self.api, 'podman') - - def test_list_pods(self): - """test pods list call""" - mock_read = mock.MagicMock() - mock_read.return_value = b'[{"Name":"podman"}]' - self.response.read = mock_read - ret = podman.pods.list_pods(self.api) - self.assertEqual(ret, [{'Name': 'podman'}]) - self.request.assert_called_once_with('/pods/json', {}) - - def test_list_pods_filter(self): - """test pods list call with a filter""" - mock_read = mock.MagicMock() - mock_read.return_value = b'[{"Name":"podman"}]' - self.response.read = mock_read - ret = podman.pods.list_pods(self.api, 'name=podman') - self.assertEqual(ret, [{'Name': 'podman'}]) - self.request.assert_called_once_with('/pods/json', {'filter': 'name=podman'}) - - def test_list_processes(self): - """test pods list processes""" - mock_read = mock.MagicMock() - mock_read.return_value = b'{"Processes": [[]], "Titles": []}' - self.response.read = mock_read - ret = podman.pods.list_processes(self.api, 'podman') - self.assertEqual(ret, {'Processes': [[]], 'Titles': []}) - self.request.assert_called_once_with('/pods/podman/top', {}) - - def test_list_processes_with_options(self): - """test pods list processes""" - mock_read = mock.MagicMock() - mock_read.return_value = b'{"Processes": [[]], "Titles": []}' - self.response.read = mock_read - ret = podman.pods.list_processes(self.api, 'podman', True, '-a') - self.assertEqual(ret, {'Processes': [[]], 'Titles': []}) - self.request.assert_called_once_with('/pods/podman/top', {'stream': True, 'ps_args': '-a'}) - - def test_pause(self): - """test pods pause call""" - mock_read = mock.MagicMock() - mock_read.return_value = b'{"Id": "string"}' - self.response.read = mock_read - ret = podman.pods.pause(self.api, 'podman') - self.assertEqual(ret, {'Id': "string"}) - self.request.assert_called_once_with('/pods/podman/pause') - - def test_pause_fail(self): - """test pods pause fails""" - self.request.side_effect = podman.errors.NotFoundError('yikes') - mock_raise = mock.MagicMock() - mock_raise.side_effect = podman.errors.PodNotFound('yikes') - self.api.raise_not_found = mock_raise - self.assertRaises(podman.errors.PodNotFound, podman.pods.pause, self.api, 'podman') - - def test_prune(self): - """test pods prune call""" - mock_read = mock.MagicMock() - mock_read.return_value = b'{"Id": "string"}' - self.response.read = mock_read - ret = podman.pods.prune(self.api) - self.assertEqual(ret, {'Id': "string"}) - self.request.assert_called_once_with('/pods/prune') - - def test_prune_fail(self): - """test prune call with an error""" - self.response.status = 400 - self.request.side_effect = podman.errors.RequestError('meh', self.response) - self.assertRaises(podman.errors.RequestError, podman.pods.prune, self.api) - - def test_remove(self): - """test remove call""" - mock_read = mock.MagicMock() - mock_read.return_value = b'[{"deleted": "str","untagged": ["str"]}]' - self.response.status = 200 - self.response.read = mock_read - expected = [{'deleted': 'str', 'untagged': ['str']}] - ret = podman.pods.remove(self.api, 'foo') - self.assertEqual(ret, expected) - self.api.delete.assert_called_once_with('/pods/foo', {}) - - def test_remove_force(self): - """test remove call with force""" - mock_read = mock.MagicMock() - mock_read.return_value = b'[{"deleted": "str","untagged": ["str"]}]' - self.response.status = 200 - self.response.read = mock_read - expected = [{'deleted': 'str', 'untagged': ['str']}] - ret = podman.pods.remove(self.api, 'foo', True) - self.assertEqual(ret, expected) - self.api.delete.assert_called_once_with('/pods/foo', {'force': True}) - - def test_remove_missing(self): - """test remove call with missing pod""" - mock_raise = mock.MagicMock() - mock_raise.side_effect = podman.errors.PodNotFound('yikes') - self.api.raise_not_found = mock_raise - self.request.side_effect = podman.errors.NotFoundError('nope') - self.assertRaises(podman.errors.PodNotFound, podman.pods.remove, self.api, 'foo') - - def test_restart(self): - """test pods restart call""" - mock_read = mock.MagicMock() - mock_read.return_value = b'{"Id": "string"}' - self.response.read = mock_read - ret = podman.pods.restart(self.api, 'podman') - self.assertEqual(ret, {'Id': "string"}) - self.request.assert_called_once_with('/pods/podman/restart') - - def test_restart_fail(self): - """test pods restart fail""" - self.request.side_effect = podman.errors.NotFoundError('yikes') - mock_raise = mock.MagicMock() - mock_raise.side_effect = podman.errors.PodNotFound('yikes') - self.api.raise_not_found = mock_raise - self.assertRaises(podman.errors.PodNotFound, podman.pods.restart, self.api, 'podman') - - def test_start(self): - """test pods start call""" - mock_read = mock.MagicMock() - mock_read.return_value = b'{"Id": "string"}' - self.response.read = mock_read - ret = podman.pods.start(self.api, 'podman') - self.assertEqual(ret, {'Id': "string"}) - self.request.assert_called_once_with('/pods/podman/start') - - def test_start_fail(self): - """test pods start fail""" - self.request.side_effect = podman.errors.NotFoundError('yikes') - mock_raise = mock.MagicMock() - mock_raise.side_effect = podman.errors.PodNotFound('yikes') - self.api.raise_not_found = mock_raise - self.assertRaises(podman.errors.PodNotFound, podman.pods.start, self.api, 'podman') - - def test_stats(self): - """test pods stats call""" - mock_read = mock.MagicMock() - mock_read.return_value = b'{"Processes": [[]], "Titles": []}' - self.response.read = mock_read - ret = podman.pods.stats(self.api) - self.assertEqual(ret, {'Processes': [[]], 'Titles': []}) - self.request.assert_called_once_with('/pods/stats', {'all': True}) - - def test_stats_with_names(self): - """test pods stats call""" - mock_read = mock.MagicMock() - mock_read.return_value = b'{"Processes": [[]], "Titles": []}' - self.response.read = mock_read - ret = podman.pods.stats(self.api, False, ['test']) - self.assertEqual(ret, {'Processes': [[]], 'Titles': []}) - self.request.assert_called_once_with('/pods/stats', {'all': False, 'namesOrIDs': ['test']}) - - def test_stats_fail(self): - """test pods stats fails""" - self.request.side_effect = podman.errors.NotFoundError('yikes') - mock_raise = mock.MagicMock() - mock_raise.side_effect = podman.errors.PodNotFound('yikes') - self.api.raise_not_found = mock_raise - self.assertRaises(podman.errors.PodNotFound, podman.pods.stats, self.api, 'podman') - - def test_stop(self): - """test pods stop call""" - mock_read = mock.MagicMock() - mock_read.return_value = b'{"Id": "string"}' - self.response.read = mock_read - ret = podman.pods.stop(self.api, 'podman') - self.assertEqual(ret, {'Id': "string"}) - self.request.assert_called_once_with('/pods/podman/stop') - - def test_stop_fail(self): - """test pods stop fail""" - self.request.side_effect = podman.errors.NotFoundError('yikes') - mock_raise = mock.MagicMock() - mock_raise.side_effect = podman.errors.PodNotFound('yikes') - self.api.raise_not_found = mock_raise - self.assertRaises(podman.errors.PodNotFound, podman.pods.stop, self.api, 'podman') - - def test_unpause(self): - """test pods unpause call""" - mock_read = mock.MagicMock() - mock_read.return_value = b'{"Id": "string"}' - self.response.read = mock_read - ret = podman.pods.unpause(self.api, 'podman') - self.assertEqual(ret, {'Id': "string"}) - self.request.assert_called_once_with('/pods/podman/unpause') - - def test_unpause_fail(self): - """test pods unpause fails""" - self.request.side_effect = podman.errors.NotFoundError('yikes') - mock_raise = mock.MagicMock() - mock_raise.side_effect = podman.errors.PodNotFound('yikes') - self.api.raise_not_found = mock_raise - self.assertRaises(podman.errors.PodNotFound, podman.pods.unpause, self.api, 'podman') diff --git a/podman/tests/unit/system/__init__.py b/podman/tests/unit/system/__init__.py deleted file mode 100644 index e69de29b..00000000 diff --git a/podman/tests/unit/system/test_system.py b/podman/tests/unit/system/test_system.py deleted file mode 100644 index 4c37db52..00000000 --- a/podman/tests/unit/system/test_system.py +++ /dev/null @@ -1,84 +0,0 @@ -# Copyright 2020 Red Hat, 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. -# -"""podman.system unit tests""" - -import unittest -from unittest import mock - -import podman.errors -import podman.system - - -class TestSystem(unittest.TestCase): - """Test the system calls.""" - - def setUp(self): - super().setUp() - self.request = mock.MagicMock() - self.response = mock.MagicMock() - self.request.return_value = self.response - self.api = mock.MagicMock() - self.api.get = self.request - - def test_version(self): - """test version call""" - self.response.status = 200 - mock_read = mock.MagicMock() - mock_read.return_value = b'{"Version": "2.1.1"}' - expected = {'Version': '2.1.1'} - self.response.read = mock_read - ret = podman.system.version(self.api) - self.assertEqual(ret, expected) - self.request.assert_called_once_with('/version') - - def test_version_not_ok(self): - """test version call""" - self.response.status = 404 - ret = podman.system.version(self.api) - self.assertEqual(ret, {}) - - def test_get_info(self): - """test info call""" - mock_read = mock.MagicMock() - mock_read.return_value = b'{"ID":"a"}' - self.response.read = mock_read - ret = podman.system.get_info(self.api) - self.assertEqual(ret, {'ID': 'a'}) - self.request.assert_called_once_with('/info') - - def test_get_info_fail(self): - """test info call fails""" - self.request.side_effect = podman.errors.NotFoundError('yikes') - mock_raise = mock.MagicMock() - mock_raise.side_effect = podman.errors.ImageNotFound('yikes') - self.api.raise_not_found = mock_raise - self.assertRaises(podman.errors.ImageNotFound, podman.system.get_info, self.api) - - def test_show_disk_usage(self): - """test df call""" - mock_read = mock.MagicMock() - mock_read.return_value = b'{"LayersSize":0}' - self.response.read = mock_read - ret = podman.system.show_disk_usage(self.api) - self.assertEqual(ret, {'LayersSize': 0}) - self.request.assert_called_once_with('/system/df') - - def test_show_disk_usage_fail(self): - """test df call fails""" - self.request.side_effect = podman.errors.NotFoundError('yikes') - mock_raise = mock.MagicMock() - mock_raise.side_effect = podman.errors.ImageNotFound('yikes') - self.api.raise_not_found = mock_raise - self.assertRaises(podman.errors.ImageNotFound, podman.system.show_disk_usage, self.api) diff --git a/podman/tests/unit/test_api_connection.py b/podman/tests/unit/test_api_connection.py deleted file mode 100644 index 4b6a73b0..00000000 --- a/podman/tests/unit/test_api_connection.py +++ /dev/null @@ -1,195 +0,0 @@ -# Copyright 2019 Red Hat, 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. -# -"""podman.ApiConnection unit tests""" - -import socket -import unittest -from unittest import mock - -import podman.errors -from podman import ApiConnection - - -class TestApiConnection(unittest.TestCase): - """Test the ApiConnection() object.""" - - def setUp(self): - super().setUp() - self.conn = ApiConnection('unix:///', base='/test') - - def test_missing_url(self): - """test missing url to constructor""" - self.assertRaises(ValueError, ApiConnection, None) - - def test_invalid_scheme(self): - """test invalid scheme to constructor""" - self.assertRaises(ValueError, ApiConnection, "tcp://localhost//") - - @mock.patch('http.client.HTTPConnection.close') - @mock.patch('socket.socket') - def test_connect(self, mock_sock, mock_close): - """test connect for unix""" - mock_sock_obj = mock.MagicMock() - mock_sock.return_value = mock_sock_obj - self.conn.connect() - - with self.conn: - pass # verify ContextManager code - - mock_sock.assert_called_once_with(socket.AF_UNIX, socket.SOCK_STREAM) - mock_close.assert_called_once_with() - - def test_connect_fail(self): - """test connect not implemented""" - conn = ApiConnection('ssh:///') - self.assertRaises(NotImplementedError, conn.connect) - - def test_join(self): - """test join call""" - path = '/foo' - self.assertEqual(self.conn.join(path), '/test/foo') - - def test_join_params(self): - """test join call with params""" - path = '/foo' - params = {'a': '"b"'} - self.assertEqual(self.conn.join(path, params), '/test/foo?a=%22b%22') - - def test_delete(self): - """test delete wrapper""" - mock_req = mock.MagicMock() - self.conn.request = mock_req - - self.conn.delete('/baz') - self.conn.delete('/foo', params={'a': 'b'}) - self.conn.delete('/bar', params={'a': 'b', 'c': 'd'}) - calls = [ - mock.call('DELETE', '/test/baz'), - mock.call('DELETE', '/test/foo?a=b'), - mock.call('DELETE', '/test/bar?a=b&c=d'), - ] - mock_req.assert_has_calls(calls) - - def test_get(self): - """test get wrapper""" - mock_req = mock.MagicMock() - self.conn.request = mock_req - - self.conn.get('/baz') - self.conn.get('/foo', params={'a': 'b'}) - self.conn.get('/bar', params={'a': 'b', 'c': 'd'}) - calls = [ - mock.call('GET', '/test/baz'), - mock.call('GET', '/test/foo?a=b'), - mock.call('GET', '/test/bar?a=b&c=d'), - ] - mock_req.assert_has_calls(calls) - - def test_post(self): - """test post wrapper""" - mock_req = mock.MagicMock() - self.conn.request = mock_req - - self.conn.post('/baz') - self.conn.post('/foo', params="{'a': 'b'}") - self.conn.post('/bar', params={'a': 'b', 'c': 'd'}, headers={'x': 'y'}, encode=True) - calls = [ - mock.call('POST', '/test/baz', body=None, headers={}), - mock.call('POST', '/test/foo', body="{'a': 'b'}", headers={}), - mock.call( - 'POST', - '/test/bar', - body='a=b&c=d', - headers={'x': 'y', 'content-type': 'application/x-www-form-urlencoded'}, - ), - ] - mock_req.assert_has_calls(calls) - - @mock.patch('http.client.HTTPConnection.getresponse') - @mock.patch('http.client.HTTPConnection.request') - def test_request(self, mock_request, mock_response): - """test request""" - mock_resp = mock.MagicMock() - mock_resp.status = 200 - mock_response.return_value = mock_resp - ret = self.conn.request('GET', 'unix://foo') - mock_request.assert_called_once_with('GET', 'unix://foo', None, {}, encode_chunked=False) - mock_response.assert_called_once_with() - self.assertEqual(ret, mock_resp) - - @mock.patch('http.client.HTTPConnection.getresponse') - @mock.patch('http.client.HTTPConnection.request') - def test_request_not_found(self, mock_request, mock_response): - """test request with not found response""" - mock_resp = mock.MagicMock() - mock_resp.status = 404 - mock_response.return_value = mock_resp - self.assertRaises(podman.errors.NotFoundError, self.conn.request, 'GET', 'unix://foo') - mock_request.assert_called_once_with('GET', 'unix://foo', None, {}, encode_chunked=False) - mock_response.assert_called_once_with() - - @mock.patch('http.client.HTTPConnection.getresponse') - @mock.patch('http.client.HTTPConnection.request') - def test_request_request_error(self, mock_request, mock_response): - """test request with server error response""" - mock_resp = mock.MagicMock() - mock_resp.status = 400 - mock_response.return_value = mock_resp - self.assertRaises(podman.errors.RequestError, self.conn.request, 'GET', 'unix://foo') - mock_request.assert_called_once_with('GET', 'unix://foo', None, {}, encode_chunked=False) - mock_response.assert_called_once_with() - - @mock.patch('http.client.HTTPConnection.getresponse') - @mock.patch('http.client.HTTPConnection.request') - def test_request_server_error(self, mock_request, mock_response): - """test request with server error response""" - mock_resp = mock.MagicMock() - mock_resp.status = 500 - mock_resp.read.return_value = b'{"message":"My error"}' - mock_response.return_value = mock_resp - with self.assertRaises( - podman.errors.InternalServerError, - ) as cm: - self.conn.request('GET', 'unix://foo') - mock_request.assert_called_once_with('GET', 'unix://foo', None, {}, encode_chunked=False) - mock_response.assert_called_once_with() - self.assertEqual(str(cm.exception), 'Request GET:unix://foo failed: My error') - - @mock.patch('http.client.HTTPConnection.getresponse') - @mock.patch('http.client.HTTPConnection.request') - def test_request_server_error_nomessage(self, mock_request, mock_response): - """test request with server error response, and no message""" - mock_resp = mock.MagicMock() - mock_resp.status = 500 - mock_resp.read.return_value = b'I am not json' - mock_response.return_value = mock_resp - self.assertRaises(podman.errors.InternalServerError, self.conn.request, 'GET', 'unix://foo') - mock_request.assert_called_once_with('GET', 'unix://foo', None, {}, encode_chunked=False) - mock_response.assert_called_once_with() - - def test_quote(self): - """test quote""" - ret = self.conn.quote('"') - self.assertEqual(ret, '%22') - - @mock.patch('json.loads') - def test_raise_not_found(self, mock_json): - """test raise image not found function""" - exc = Exception('meh') - mock_resp = mock.MagicMock() - mock_json.return_value = {'cause': 'c', 'message': 'msg'} - # pylint: disable=protected-access - self.assertRaises(podman.errors.ImageNotFound, self.conn.raise_not_found, exc, mock_resp) - mock_json.assert_called_once() diff --git a/podman/tests/unit/test_config.py b/podman/tests/unit/test_config.py index 7ecb475a..c7e011ad 100644 --- a/podman/tests/unit/test_config.py +++ b/podman/tests/unit/test_config.py @@ -47,7 +47,9 @@ def test_connections(self): expected = urllib.parse.urlparse("ssh://qe@localhost:2222/run/podman/podman.sock") self.assertEqual(config.active_service.url, expected) - self.assertEqual(config.services["production"].identity, Path("/home/root/.ssh/id_rsa")) + self.assertEqual( + config.services["production"].identity, Path("/home/root/.ssh/id_rsa") + ) PodmanConfigTestCase.opener.assert_called_with( Path("/home/developer/containers.conf"), encoding='utf-8' diff --git a/podman/tests/unit/test_containersmanager.py b/podman/tests/unit/test_containersmanager.py index 7187919f..e7157184 100644 --- a/podman/tests/unit/test_containersmanager.py +++ b/podman/tests/unit/test_containersmanager.py @@ -242,7 +242,9 @@ def test_run_detached(self, mock): json=FIRST_CONTAINER, ) - with patch.multiple(Container, logs=DEFAULT, wait=DEFAULT, autospec=True) as mock_container: + with patch.multiple( + Container, logs=DEFAULT, wait=DEFAULT, autospec=True + ) as mock_container: mock_container["logs"].return_value = [] mock_container["wait"].return_value = {"StatusCode": 0} @@ -275,7 +277,9 @@ def test_run(self, mock): b"This is a unittest - line 2", ) - with patch.multiple(Container, logs=DEFAULT, wait=DEFAULT, autospec=True) as mock_container: + with patch.multiple( + Container, logs=DEFAULT, wait=DEFAULT, autospec=True + ) as mock_container: mock_container["wait"].return_value = {"StatusCode": 0} with self.subTest("Results not streamed"): diff --git a/podman/tests/unit/test_parse_utils.py b/podman/tests/unit/test_parse_utils.py new file mode 100644 index 00000000..80f30b39 --- /dev/null +++ b/podman/tests/unit/test_parse_utils.py @@ -0,0 +1,67 @@ +import datetime +import ipaddress +import unittest +from typing import Any, Optional + +from dataclasses import dataclass + +from podman import api + + +class ParseUtilsTestCase(unittest.TestCase): + def test_parse_repository(self): + @dataclass + class TestCase: + name: str + input: Any + expected: Optional[str] + + cases = [ + TestCase(name="empty str", input="", expected=("", None)), + TestCase( + name="name", + input="quay.io/libpod/testimage", + expected=("quay.io/libpod/testimage", None), + ), + TestCase( + name="@digest", + input="quay.io/libpod/testimage@71f1b47263fc", + expected=("quay.io/libpod/testimage", "71f1b47263fc"), + ), + TestCase( + name=":tag", + input="quay.io/libpod/testimage:latest", + expected=("quay.io/libpod/testimage", "latest"), + ), + ] + + for case in cases: + actual = api.parse_repository(case.input) + self.assertEqual( + case.expected, + actual, + f"failed test {case.name} expected {case.expected}, actual {actual}", + ) + + def test_decode_header(self): + actual = api.decode_header("eyJIZWFkZXIiOiJ1bml0dGVzdCJ9") + self.assertDictEqual(actual, {"Header": "unittest"}) + + self.assertDictEqual(api.decode_header(None), {}) + + def test_prepare_timestamp(self): + time = datetime.datetime(2022, 1, 24, 12, 0, 0) + self.assertEqual(api.prepare_timestamp(time), 1643025600) + self.assertEqual(api.prepare_timestamp(2), 2) + + self.assertEqual(api.prepare_timestamp(None), None) + with self.assertRaises(ValueError): + api.prepare_timestamp("bad input") + + def test_prepare_cidr(self): + net = ipaddress.IPv4Network("127.0.0.0/24") + self.assertEqual(api.prepare_cidr(net), ("127.0.0.0", "////AA==")) + + +if __name__ == '__main__': + unittest.main()