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()