diff options
Diffstat (limited to 'roles/lib_openshift/src')
-rw-r--r-- | roles/lib_openshift/src/ansible/oc_pvc.py | 35 | ||||
-rw-r--r-- | roles/lib_openshift/src/ansible/oc_volume.py | 41 | ||||
-rw-r--r-- | roles/lib_openshift/src/class/oc_pvc.py | 167 | ||||
-rw-r--r-- | roles/lib_openshift/src/class/oc_volume.py | 195 | ||||
-rw-r--r-- | roles/lib_openshift/src/doc/pvc | 76 | ||||
-rw-r--r-- | roles/lib_openshift/src/doc/volume | 105 | ||||
-rw-r--r-- | roles/lib_openshift/src/lib/pvc.py | 167 | ||||
-rw-r--r-- | roles/lib_openshift/src/lib/volume.py | 7 | ||||
-rw-r--r-- | roles/lib_openshift/src/sources.yml | 23 | ||||
-rwxr-xr-x | roles/lib_openshift/src/test/unit/test_oc_pvc.py | 366 | ||||
-rwxr-xr-x | roles/lib_openshift/src/test/unit/test_oc_volume.py | 633 |
11 files changed, 1814 insertions, 1 deletions
diff --git a/roles/lib_openshift/src/ansible/oc_pvc.py b/roles/lib_openshift/src/ansible/oc_pvc.py new file mode 100644 index 000000000..a5181e281 --- /dev/null +++ b/roles/lib_openshift/src/ansible/oc_pvc.py @@ -0,0 +1,35 @@ +# pylint: skip-file +# flake8: noqa + +#pylint: disable=too-many-branches +def main(): + ''' + ansible oc module for pvc + ''' + + module = AnsibleModule( + argument_spec=dict( + kubeconfig=dict(default='/etc/origin/master/admin.kubeconfig', type='str'), + state=dict(default='present', type='str', + choices=['present', 'absent', 'list']), + debug=dict(default=False, type='bool'), + name=dict(default=None, required=True, type='str'), + namespace=dict(default=None, required=True, type='str'), + volume_capacity=dict(default='1G', type='str'), + access_modes=dict(default='ReadWriteOnce', + choices=['ReadWriteOnce', 'ReadOnlyMany', 'ReadWriteMany'], + type='str'), + ), + supports_check_mode=True, + ) + + rval = OCPVC.run_ansible(module.params, module.check_mode) + + if 'failed' in rval: + module.fail_json(**rval) + + return module.exit_json(**rval) + + +if __name__ == '__main__': + main() diff --git a/roles/lib_openshift/src/ansible/oc_volume.py b/roles/lib_openshift/src/ansible/oc_volume.py new file mode 100644 index 000000000..660376d2f --- /dev/null +++ b/roles/lib_openshift/src/ansible/oc_volume.py @@ -0,0 +1,41 @@ +# pylint: skip-file +# flake8: noqa + +def main(): + ''' + ansible oc module for volumes + ''' + + module = AnsibleModule( + argument_spec=dict( + kubeconfig=dict(default='/etc/origin/master/admin.kubeconfig', type='str'), + state=dict(default='present', type='str', + choices=['present', 'absent', 'list']), + debug=dict(default=False, type='bool'), + kind=dict(default='dc', choices=['dc', 'rc', 'pods'], type='str'), + namespace=dict(default='default', type='str'), + vol_name=dict(default=None, type='str'), + name=dict(default=None, type='str'), + mount_type=dict(default=None, + choices=['emptydir', 'hostpath', 'secret', 'pvc', 'configmap'], + type='str'), + mount_path=dict(default=None, type='str'), + # secrets require a name + secret_name=dict(default=None, type='str'), + # pvc requires a size + claim_size=dict(default=None, type='str'), + claim_name=dict(default=None, type='str'), + # configmap requires a name + configmap_name=dict(default=None, type='str'), + ), + supports_check_mode=True, + ) + rval = OCVolume.run_ansible(module.params, module.check_mode) + if 'failed' in rval: + module.fail_json(**rval) + + module.exit_json(**rval) + + +if __name__ == '__main__': + main() diff --git a/roles/lib_openshift/src/class/oc_pvc.py b/roles/lib_openshift/src/class/oc_pvc.py new file mode 100644 index 000000000..c73abc47c --- /dev/null +++ b/roles/lib_openshift/src/class/oc_pvc.py @@ -0,0 +1,167 @@ +# pylint: skip-file +# flake8: noqa + + +# pylint: disable=too-many-instance-attributes +class OCPVC(OpenShiftCLI): + ''' Class to wrap the oc command line tools ''' + kind = 'pvc' + + # pylint allows 5 + # pylint: disable=too-many-arguments + def __init__(self, + config, + verbose=False): + ''' Constructor for OCVolume ''' + super(OCPVC, self).__init__(config.namespace, config.kubeconfig) + self.config = config + self.namespace = config.namespace + self._pvc = None + + @property + def pvc(self): + ''' property function pvc''' + if not self._pvc: + self.get() + return self._pvc + + @pvc.setter + def pvc(self, data): + ''' setter function for yedit var ''' + self._pvc = data + + def bound(self): + '''return whether the pvc is bound''' + if self.pvc.get_volume_name(): + return True + + return False + + def exists(self): + ''' return whether a pvc exists ''' + if self.pvc: + return True + + return False + + def get(self): + '''return pvc information ''' + result = self._get(self.kind, self.config.name) + if result['returncode'] == 0: + self.pvc = PersistentVolumeClaim(content=result['results'][0]) + elif '\"%s\" not found' % self.config.name in result['stderr']: + result['returncode'] = 0 + result['results'] = [{}] + + return result + + def delete(self): + '''delete the object''' + return self._delete(self.kind, self.config.name) + + def create(self): + '''create the object''' + return self._create_from_content(self.config.name, self.config.data) + + def update(self): + '''update the object''' + # need to update the tls information and the service name + return self._replace_content(self.kind, self.config.name, self.config.data) + + def needs_update(self): + ''' verify an update is needed ''' + if self.pvc.get_volume_name() or self.pvc.is_bound(): + return False + + skip = [] + return not Utils.check_def_equal(self.config.data, self.pvc.yaml_dict, skip_keys=skip, debug=True) + + # pylint: disable=too-many-branches,too-many-return-statements + @staticmethod + def run_ansible(params, check_mode): + '''run the idempotent ansible code''' + pconfig = PersistentVolumeClaimConfig(params['name'], + params['namespace'], + params['kubeconfig'], + params['access_modes'], + params['volume_capacity'], + ) + oc_pvc = OCPVC(pconfig, verbose=params['debug']) + + state = params['state'] + + api_rval = oc_pvc.get() + if api_rval['returncode'] != 0: + return {'failed': True, 'msg': api_rval} + + ##### + # Get + ##### + if state == 'list': + return {'changed': False, 'results': api_rval['results'], 'state': state} + + ######## + # Delete + ######## + if state == 'absent': + if oc_pvc.exists(): + + if check_mode: + return {'changed': False, 'msg': 'CHECK_MODE: Would have performed a delete.'} + + api_rval = oc_pvc.delete() + + if api_rval['returncode'] != 0: + return {'failed': True, 'msg': api_rval} + + return {'changed': True, 'results': api_rval, 'state': state} + + return {'changed': False, 'state': state} + + if state == 'present': + ######## + # Create + ######## + if not oc_pvc.exists(): + + if check_mode: + return {'changed': True, 'msg': 'CHECK_MODE: Would have performed a create.'} + + # Create it here + api_rval = oc_pvc.create() + + if api_rval['returncode'] != 0: + return {'failed': True, 'msg': api_rval} + + # return the created object + api_rval = oc_pvc.get() + + if api_rval['returncode'] != 0: + return {'failed': True, 'msg': api_rval} + + return {'changed': True, 'results': api_rval, 'state': state} + + ######## + # Update + ######## + if oc_pvc.pvc.is_bound() or oc_pvc.pvc.get_volume_name(): + api_rval['msg'] = '##### - This volume is currently bound. Will not update - ####' + return {'changed': False, 'results': api_rval, 'state': state} + + if oc_pvc.needs_update(): + api_rval = oc_pvc.update() + + if api_rval['returncode'] != 0: + return {'failed': True, 'msg': api_rval} + + # return the created object + api_rval = oc_pvc.get() + + if api_rval['returncode'] != 0: + return {'failed': True, 'msg': api_rval} + + return {'changed': True, 'results': api_rval, 'state': state} + + return {'changed': False, 'results': api_rval, 'state': state} + + return {'failed': True, 'msg': 'Unknown state passed. {}'.format(state)} diff --git a/roles/lib_openshift/src/class/oc_volume.py b/roles/lib_openshift/src/class/oc_volume.py new file mode 100644 index 000000000..5211a1afd --- /dev/null +++ b/roles/lib_openshift/src/class/oc_volume.py @@ -0,0 +1,195 @@ +# pylint: skip-file +# flake8: noqa + + +# pylint: disable=too-many-instance-attributes +class OCVolume(OpenShiftCLI): + ''' Class to wrap the oc command line tools ''' + volume_mounts_path = {"pod": "spec.containers[0].volumeMounts", + "dc": "spec.template.spec.containers[0].volumeMounts", + "rc": "spec.template.spec.containers[0].volumeMounts", + } + volumes_path = {"pod": "spec.volumes", + "dc": "spec.template.spec.volumes", + "rc": "spec.template.spec.volumes", + } + + # pylint allows 5 + # pylint: disable=too-many-arguments + def __init__(self, + kind, + resource_name, + namespace, + vol_name, + mount_path, + mount_type, + secret_name, + claim_size, + claim_name, + configmap_name, + kubeconfig='/etc/origin/master/admin.kubeconfig', + verbose=False): + ''' Constructor for OCVolume ''' + super(OCVolume, self).__init__(namespace, kubeconfig) + self.kind = kind + self.volume_info = {'name': vol_name, + 'secret_name': secret_name, + 'path': mount_path, + 'type': mount_type, + 'claimSize': claim_size, + 'claimName': claim_name, + 'configmap_name': configmap_name} + self.volume, self.volume_mount = Volume.create_volume_structure(self.volume_info) + self.name = resource_name + self.namespace = namespace + self.kubeconfig = kubeconfig + self.verbose = verbose + self._resource = None + + @property + def resource(self): + ''' property function for resource var ''' + if not self._resource: + self.get() + return self._resource + + @resource.setter + def resource(self, data): + ''' setter function for resource var ''' + self._resource = data + + def exists(self): + ''' return whether a volume exists ''' + volume_mount_found = False + volume_found = self.resource.exists_volume(self.volume) + if not self.volume_mount and volume_found: + return True + + if self.volume_mount: + volume_mount_found = self.resource.exists_volume_mount(self.volume_mount) + + if volume_found and self.volume_mount and volume_mount_found: + return True + + return False + + def get(self): + '''return volume information ''' + vol = self._get(self.kind, self.name) + if vol['returncode'] == 0: + if self.kind == 'dc': + self.resource = DeploymentConfig(content=vol['results'][0]) + vol['results'] = self.resource.get_volumes() + + return vol + + def delete(self): + '''remove a volume''' + self.resource.delete_volume_by_name(self.volume) + return self._replace_content(self.kind, self.name, self.resource.yaml_dict) + + def put(self): + '''place volume into dc ''' + self.resource.update_volume(self.volume) + self.resource.get_volumes() + self.resource.update_volume_mount(self.volume_mount) + return self._replace_content(self.kind, self.name, self.resource.yaml_dict) + + def needs_update(self): + ''' verify an update is needed ''' + return self.resource.needs_update_volume(self.volume, self.volume_mount) + + # pylint: disable=too-many-branches,too-many-return-statements + @staticmethod + def run_ansible(params, check_mode=False): + '''run the idempotent ansible code''' + oc_volume = OCVolume(params['kind'], + params['name'], + params['namespace'], + params['vol_name'], + params['mount_path'], + params['mount_type'], + # secrets + params['secret_name'], + # pvc + params['claim_size'], + params['claim_name'], + # configmap + params['configmap_name'], + kubeconfig=params['kubeconfig'], + verbose=params['debug']) + + state = params['state'] + + api_rval = oc_volume.get() + + if api_rval['returncode'] != 0: + return {'failed': True, 'msg': api_rval} + + ##### + # Get + ##### + if state == 'list': + return {'changed': False, 'results': api_rval['results'], 'state': state} + + ######## + # Delete + ######## + if state == 'absent': + if oc_volume.exists(): + + if check_mode: + return {'changed': False, 'msg': 'CHECK_MODE: Would have performed a delete.'} + + api_rval = oc_volume.delete() + + if api_rval['returncode'] != 0: + return {'failed': True, 'msg': api_rval} + + return {'changed': True, 'results': api_rval, 'state': state} + + return {'changed': False, 'state': state} + + if state == 'present': + ######## + # Create + ######## + if not oc_volume.exists(): + + if check_mode: + exit_json(changed=False, msg='Would have performed a create.') + + # Create it here + api_rval = oc_volume.put() + + if api_rval['returncode'] != 0: + return {'failed': True, 'msg': api_rval} + + # return the created object + api_rval = oc_volume.get() + + if api_rval['returncode'] != 0: + return {'failed': True, 'msg': api_rval} + + return {'changed': True, 'results': api_rval, 'state': state} + + ######## + # Update + ######## + if oc_volume.needs_update(): + api_rval = oc_volume.put() + + if api_rval['returncode'] != 0: + return {'failed': True, 'msg': api_rval} + + # return the created object + api_rval = oc_volume.get() + + if api_rval['returncode'] != 0: + return {'failed': True, 'msg': api_rval} + + return {'changed': True, 'results': api_rval, state: state} + + return {'changed': False, 'results': api_rval, state: state} + + return {'failed': True, 'msg': 'Unknown state passed. {}'.format(state)} diff --git a/roles/lib_openshift/src/doc/pvc b/roles/lib_openshift/src/doc/pvc new file mode 100644 index 000000000..9240f2a0f --- /dev/null +++ b/roles/lib_openshift/src/doc/pvc @@ -0,0 +1,76 @@ +# flake8: noqa +# pylint: skip-file + +DOCUMENTATION = ''' +--- +module: oc_pvc +short_description: Modify, and idempotently manage openshift persistent volume claims +description: + - Modify openshift persistent volume claims programmatically. +options: + state: + description: + - Supported states, present, absent, list + - present - will ensure object is created or updated to the value specified + - list - will return a pvc + - absent - will remove a pvc + required: False + default: present + choices: ["present", 'absent', 'list'] + aliases: [] + kubeconfig: + description: + - The path for the kubeconfig file to use for authentication + required: false + default: /etc/origin/master/admin.kubeconfig + aliases: [] + debug: + description: + - Turn on debug output. + required: false + default: False + aliases: [] + name: + description: + - Name of the object that is being queried. + required: false + default: None + aliases: [] + namespace: + description: + - The namespace where the object lives. + required: false + default: str + aliases: [] + volume_capacity: + description: + - The requested volume capacity + required: False + default: 1G + aliases: [] + access_modes: + description: + - The access modes allowed for the pvc + - Expects a list + required: False + default: ReadWriteOnce + choices: + - ReadWriteOnce + - ReadOnlyMany + - ReadWriteMany + aliases: [] +author: +- "Kenny Woodson <kwoodson@redhat.com>" +extends_documentation_fragment: [] +''' + +EXAMPLES = ''' +- name: create a pvc + oc_pvc: + namespace: awesomeapp + name: dbstorage + access_modes: + - ReadWriteOnce + volume_capacity: 5G + register: pvcout +''' diff --git a/roles/lib_openshift/src/doc/volume b/roles/lib_openshift/src/doc/volume new file mode 100644 index 000000000..1d04afeef --- /dev/null +++ b/roles/lib_openshift/src/doc/volume @@ -0,0 +1,105 @@ +# flake8: noqa +# pylint: skip-file + +DOCUMENTATION = ''' +--- +module: oc_volume +short_description: Create, modify, and idempotently manage openshift volumes. +description: + - Modify openshift volumes programmatically. +options: + state: + description: + - State controls the action that will be taken with resource + - 'present' will create or update and object to the desired state + - 'absent' will ensure volumes are removed + - 'list' will read the volumes + default: present + choices: ["present", "absent", "list"] + aliases: [] + kubeconfig: + description: + - The path for the kubeconfig file to use for authentication + required: false + default: /etc/origin/master/admin.kubeconfig + aliases: [] + debug: + description: + - Turn on debug output. + required: false + default: False + aliases: [] + namespace: + description: + - The name of the namespace where the object lives + required: false + default: default + aliases: [] + kind: + description: + - The kind of object that can be managed. + default: dc + choices: + - dc + - rc + - pods + aliases: [] + mount_type: + description: + - The type of volume to be used + required: false + default: None + choices: + - emptydir + - hostpath + - secret + - pvc + - configmap + aliases: [] + mount_path: + description: + - The path to where the mount will be attached + required: false + default: None + aliases: [] + secret_name: + description: + - The name of the secret. Used when mount_type is secret. + required: false + default: None + aliases: [] + claim_size: + description: + - The size in GB of the pv claim. e.g. 100G + required: false + default: None + aliases: [] + claim_name: + description: + - The name of the pv claim + required: false + default: None + aliases: [] + configmap_name: + description: + - The name of the configmap + required: false + default: None + aliases: [] +author: +- "Kenny Woodson <kwoodson@redhat.com>" +extends_documentation_fragment: [] +''' + +EXAMPLES = ''' +- name: attach storage volumes to deploymentconfig + oc_volume: + namespace: logging + kind: dc + name: name_of_the_dc + mount_type: pvc + claim_name: loggingclaim + claim_size: 100G + vol_name: logging-storage + run_once: true +''' diff --git a/roles/lib_openshift/src/lib/pvc.py b/roles/lib_openshift/src/lib/pvc.py new file mode 100644 index 000000000..929b50990 --- /dev/null +++ b/roles/lib_openshift/src/lib/pvc.py @@ -0,0 +1,167 @@ +# pylint: skip-file +# flake8: noqa + + +# pylint: disable=too-many-instance-attributes +class PersistentVolumeClaimConfig(object): + ''' Handle pvc options ''' + # pylint: disable=too-many-arguments + def __init__(self, + sname, + namespace, + kubeconfig, + access_modes=None, + vol_capacity='1G'): + ''' constructor for handling pvc options ''' + self.kubeconfig = kubeconfig + self.name = sname + self.namespace = namespace + self.access_modes = access_modes + self.vol_capacity = vol_capacity + self.data = {} + + self.create_dict() + + def create_dict(self): + ''' return a service as a dict ''' + # version + self.data['apiVersion'] = 'v1' + # kind + self.data['kind'] = 'PersistentVolumeClaim' + # metadata + self.data['metadata'] = {} + self.data['metadata']['name'] = self.name + # spec + self.data['spec'] = {} + self.data['spec']['accessModes'] = ['ReadWriteOnce'] + if self.access_modes: + self.data['spec']['accessModes'] = self.access_modes + + # storage capacity + self.data['spec']['resources'] = {} + self.data['spec']['resources']['requests'] = {} + self.data['spec']['resources']['requests']['storage'] = self.vol_capacity + + +# pylint: disable=too-many-instance-attributes,too-many-public-methods +class PersistentVolumeClaim(Yedit): + ''' Class to wrap the oc command line tools ''' + access_modes_path = "spec.accessModes" + volume_capacity_path = "spec.requests.storage" + volume_name_path = "spec.volumeName" + bound_path = "status.phase" + kind = 'PersistentVolumeClaim' + + def __init__(self, content): + '''RoleBinding constructor''' + super(PersistentVolumeClaim, self).__init__(content=content) + self._access_modes = None + self._volume_capacity = None + self._volume_name = None + + @property + def volume_name(self): + ''' volume_name property ''' + if self._volume_name is None: + self._volume_name = self.get_volume_name() + return self._volume_name + + @volume_name.setter + def volume_name(self, data): + ''' volume_name property setter''' + self._volume_name = data + + @property + def access_modes(self): + ''' access_modes property ''' + if self._access_modes is None: + self._access_modes = self.get_access_modes() + if not isinstance(self._access_modes, list): + self._access_modes = list(self._access_modes) + + return self._access_modes + + @access_modes.setter + def access_modes(self, data): + ''' access_modes property setter''' + if not isinstance(data, list): + data = list(data) + + self._access_modes = data + + @property + def volume_capacity(self): + ''' volume_capacity property ''' + if self._volume_capacity is None: + self._volume_capacity = self.get_volume_capacity() + return self._volume_capacity + + @volume_capacity.setter + def volume_capacity(self, data): + ''' volume_capacity property setter''' + self._volume_capacity = data + + def get_access_modes(self): + '''get access_modes''' + return self.get(PersistentVolumeClaim.access_modes_path) or [] + + def get_volume_capacity(self): + '''get volume_capacity''' + return self.get(PersistentVolumeClaim.volume_capacity_path) or [] + + def get_volume_name(self): + '''get volume_name''' + return self.get(PersistentVolumeClaim.volume_name_path) or [] + + def is_bound(self): + '''return whether volume is bound''' + return self.get(PersistentVolumeClaim.bound_path) or [] + + #### ADD ##### + def add_access_mode(self, inc_mode): + ''' add an access_mode''' + if self.access_modes: + self.access_modes.append(inc_mode) + else: + self.put(PersistentVolumeClaim.access_modes_path, [inc_mode]) + + return True + + #### /ADD ##### + + #### Remove ##### + def remove_access_mode(self, inc_mode): + ''' remove an access_mode''' + try: + self.access_modes.remove(inc_mode) + except ValueError as _: + return False + + return True + + #### /REMOVE ##### + + #### UPDATE ##### + def update_access_mode(self, inc_mode): + ''' update an access_mode''' + try: + index = self.access_modes.index(inc_mode) + except ValueError as _: + return self.add_access_mode(inc_mode) + + self.access_modes[index] = inc_mode + + return True + + #### /UPDATE ##### + + #### FIND #### + def find_access_mode(self, inc_mode): + ''' find a user ''' + index = None + try: + index = self.access_modes.index(inc_mode) + except ValueError as _: + return index + + return index diff --git a/roles/lib_openshift/src/lib/volume.py b/roles/lib_openshift/src/lib/volume.py index e0abb1d1b..c049c8b49 100644 --- a/roles/lib_openshift/src/lib/volume.py +++ b/roles/lib_openshift/src/lib/volume.py @@ -2,7 +2,7 @@ # flake8: noqa class Volume(object): - ''' Class to model an openshift volume object''' + ''' Class to represent an openshift volume object''' volume_mounts_path = {"pod": "spec.containers[0].volumeMounts", "dc": "spec.template.spec.containers[0].volumeMounts", "rc": "spec.template.spec.containers[0].volumeMounts", @@ -34,5 +34,10 @@ class Volume(object): elif volume_type == 'hostpath': volume['hostPath'] = {} volume['hostPath']['path'] = volume_info['path'] + elif volume_type == 'configmap': + volume['configMap'] = {} + volume['configMap']['name'] = volume_info['configmap_name'] + volume_mount = {'mountPath': volume_info['path'], + 'name': volume_info['name']} return (volume, volume_mount) diff --git a/roles/lib_openshift/src/sources.yml b/roles/lib_openshift/src/sources.yml index 91ee86750..0dba6016b 100644 --- a/roles/lib_openshift/src/sources.yml +++ b/roles/lib_openshift/src/sources.yml @@ -152,6 +152,17 @@ oc_project.py: - class/oc_project.py - ansible/oc_project.py +oc_pvc.py: +- doc/generated +- doc/license +- lib/import.py +- doc/pvc +- ../../lib_utils/src/class/yedit.py +- lib/base.py +- lib/pvc.py +- class/oc_pvc.py +- ansible/oc_pvc.py + oc_route.py: - doc/generated - doc/license @@ -229,6 +240,18 @@ oc_version.py: - class/oc_version.py - ansible/oc_version.py +oc_volume.py: +- doc/generated +- doc/license +- lib/import.py +- doc/volume +- ../../lib_utils/src/class/yedit.py +- lib/base.py +- lib/deploymentconfig.py +- lib/volume.py +- class/oc_volume.py +- ansible/oc_volume.py + oc_objectvalidator.py: - doc/generated - doc/license diff --git a/roles/lib_openshift/src/test/unit/test_oc_pvc.py b/roles/lib_openshift/src/test/unit/test_oc_pvc.py new file mode 100755 index 000000000..82187917d --- /dev/null +++ b/roles/lib_openshift/src/test/unit/test_oc_pvc.py @@ -0,0 +1,366 @@ +''' + Unit tests for oc pvc +''' + +import copy +import os +import six +import sys +import unittest +import mock + +# Removing invalid variable names for tests so that I can +# keep them brief +# pylint: disable=invalid-name,no-name-in-module +# Disable import-error b/c our libraries aren't loaded in jenkins +# pylint: disable=import-error,wrong-import-position +# place class in our python path +module_path = os.path.join('/'.join(os.path.realpath(__file__).split('/')[:-4]), 'library') # noqa: E501 +sys.path.insert(0, module_path) +from oc_pvc import OCPVC, locate_oc_binary # noqa: E402 + + +class OCPVCTest(unittest.TestCase): + ''' + Test class for OCPVC + ''' + params = {'kubeconfig': '/etc/origin/master/admin.kubeconfig', + 'state': 'present', + 'debug': False, + 'name': 'mypvc', + 'namespace': 'test', + 'volume_capacity': '1G', + 'access_modes': 'ReadWriteMany'} + + @mock.patch('oc_pvc.Utils.create_tmpfile_copy') + @mock.patch('oc_pvc.OCPVC._run') + def test_create_pvc(self, mock_run, mock_tmpfile_copy): + ''' Testing a pvc create ''' + params = copy.deepcopy(OCPVCTest.params) + + pvc = '''{"kind": "PersistentVolumeClaim", + "apiVersion": "v1", + "metadata": { + "name": "mypvc", + "namespace": "test", + "selfLink": "/api/v1/namespaces/test/persistentvolumeclaims/mypvc", + "uid": "77597898-d8d8-11e6-aea5-0e3c0c633889", + "resourceVersion": "126510787", + "creationTimestamp": "2017-01-12T15:04:50Z", + "labels": { + "mypvc": "database" + }, + "annotations": { + "pv.kubernetes.io/bind-completed": "yes", + "pv.kubernetes.io/bound-by-controller": "yes", + "v1.2-volume.experimental.kubernetes.io/provisioning-required": "volume.experimental.kubernetes.io/provisioning-completed" + } + }, + "spec": { + "accessModes": [ + "ReadWriteOnce" + ], + "resources": { + "requests": { + "storage": "1Gi" + } + }, + "volumeName": "pv-aws-ow5vl" + }, + "status": { + "phase": "Bound", + "accessModes": [ + "ReadWriteOnce" + ], + "capacity": { + "storage": "1Gi" + } + } + }''' + + mock_run.side_effect = [ + (1, '', 'Error from server: persistentvolumeclaims "mypvc" not found'), + (1, '', 'Error from server: persistentvolumeclaims "mypvc" not found'), + (0, '', ''), + (0, pvc, ''), + ] + + mock_tmpfile_copy.side_effect = [ + '/tmp/mocked_kubeconfig', + ] + + results = OCPVC.run_ansible(params, False) + + self.assertTrue(results['changed']) + self.assertEqual(results['results']['results'][0]['metadata']['name'], 'mypvc') + + @mock.patch('oc_pvc.Utils.create_tmpfile_copy') + @mock.patch('oc_pvc.OCPVC._run') + def test_update_pvc(self, mock_run, mock_tmpfile_copy): + ''' Testing a pvc create ''' + params = copy.deepcopy(OCPVCTest.params) + params['access_modes'] = 'ReadWriteMany' + + pvc = '''{"kind": "PersistentVolumeClaim", + "apiVersion": "v1", + "metadata": { + "name": "mypvc", + "namespace": "test", + "selfLink": "/api/v1/namespaces/test/persistentvolumeclaims/mypvc", + "uid": "77597898-d8d8-11e6-aea5-0e3c0c633889", + "resourceVersion": "126510787", + "creationTimestamp": "2017-01-12T15:04:50Z", + "labels": { + "mypvc": "database" + }, + "annotations": { + "pv.kubernetes.io/bind-completed": "yes", + "pv.kubernetes.io/bound-by-controller": "yes", + "v1.2-volume.experimental.kubernetes.io/provisioning-required": "volume.experimental.kubernetes.io/provisioning-completed" + } + }, + "spec": { + "accessModes": [ + "ReadWriteOnce" + ], + "resources": { + "requests": { + "storage": "1Gi" + } + }, + "volumeName": "pv-aws-ow5vl" + }, + "status": { + "phase": "Bound", + "accessModes": [ + "ReadWriteOnce" + ], + "capacity": { + "storage": "1Gi" + } + } + }''' + + mod_pvc = '''{"kind": "PersistentVolumeClaim", + "apiVersion": "v1", + "metadata": { + "name": "mypvc", + "namespace": "test", + "selfLink": "/api/v1/namespaces/test/persistentvolumeclaims/mypvc", + "uid": "77597898-d8d8-11e6-aea5-0e3c0c633889", + "resourceVersion": "126510787", + "creationTimestamp": "2017-01-12T15:04:50Z", + "labels": { + "mypvc": "database" + }, + "annotations": { + "pv.kubernetes.io/bind-completed": "yes", + "pv.kubernetes.io/bound-by-controller": "yes", + "v1.2-volume.experimental.kubernetes.io/provisioning-required": "volume.experimental.kubernetes.io/provisioning-completed" + } + }, + "spec": { + "accessModes": [ + "ReadWriteMany" + ], + "resources": { + "requests": { + "storage": "1Gi" + } + }, + "volumeName": "pv-aws-ow5vl" + }, + "status": { + "phase": "Bound", + "accessModes": [ + "ReadWriteOnce" + ], + "capacity": { + "storage": "1Gi" + } + } + }''' + + mock_run.side_effect = [ + (0, pvc, ''), + (0, pvc, ''), + (0, '', ''), + (0, mod_pvc, ''), + ] + + mock_tmpfile_copy.side_effect = [ + '/tmp/mocked_kubeconfig', + ] + + results = OCPVC.run_ansible(params, False) + + self.assertFalse(results['changed']) + self.assertEqual(results['results']['msg'], '##### - This volume is currently bound. Will not update - ####') + + @mock.patch('oc_pvc.Utils.create_tmpfile_copy') + @mock.patch('oc_pvc.OCPVC._run') + def test_delete_pvc(self, mock_run, mock_tmpfile_copy): + ''' Testing a pvc create ''' + params = copy.deepcopy(OCPVCTest.params) + params['state'] = 'absent' + + pvc = '''{"kind": "PersistentVolumeClaim", + "apiVersion": "v1", + "metadata": { + "name": "mypvc", + "namespace": "test", + "selfLink": "/api/v1/namespaces/test/persistentvolumeclaims/mypvc", + "uid": "77597898-d8d8-11e6-aea5-0e3c0c633889", + "resourceVersion": "126510787", + "creationTimestamp": "2017-01-12T15:04:50Z", + "labels": { + "mypvc": "database" + }, + "annotations": { + "pv.kubernetes.io/bind-completed": "yes", + "pv.kubernetes.io/bound-by-controller": "yes", + "v1.2-volume.experimental.kubernetes.io/provisioning-required": "volume.experimental.kubernetes.io/provisioning-completed" + } + }, + "spec": { + "accessModes": [ + "ReadWriteOnce" + ], + "resources": { + "requests": { + "storage": "1Gi" + } + }, + "volumeName": "pv-aws-ow5vl" + }, + "status": { + "phase": "Bound", + "accessModes": [ + "ReadWriteOnce" + ], + "capacity": { + "storage": "1Gi" + } + } + }''' + + mock_run.side_effect = [ + (0, pvc, ''), + (0, '', ''), + ] + + mock_tmpfile_copy.side_effect = [ + '/tmp/mocked_kubeconfig', + ] + + results = OCPVC.run_ansible(params, False) + + self.assertTrue(results['changed']) + + @unittest.skipIf(six.PY3, 'py2 test only') + @mock.patch('os.path.exists') + @mock.patch('os.environ.get') + def test_binary_lookup_fallback(self, mock_env_get, mock_path_exists): + ''' Testing binary lookup fallback ''' + + mock_env_get.side_effect = lambda _v, _d: '' + + mock_path_exists.side_effect = lambda _: False + + self.assertEqual(locate_oc_binary(), 'oc') + + @unittest.skipIf(six.PY3, 'py2 test only') + @mock.patch('os.path.exists') + @mock.patch('os.environ.get') + def test_binary_lookup_in_path(self, mock_env_get, mock_path_exists): + ''' Testing binary lookup in path ''' + + oc_bin = '/usr/bin/oc' + + mock_env_get.side_effect = lambda _v, _d: '/bin:/usr/bin' + + mock_path_exists.side_effect = lambda f: f == oc_bin + + self.assertEqual(locate_oc_binary(), oc_bin) + + @unittest.skipIf(six.PY3, 'py2 test only') + @mock.patch('os.path.exists') + @mock.patch('os.environ.get') + def test_binary_lookup_in_usr_local(self, mock_env_get, mock_path_exists): + ''' Testing binary lookup in /usr/local/bin ''' + + oc_bin = '/usr/local/bin/oc' + + mock_env_get.side_effect = lambda _v, _d: '/bin:/usr/bin' + + mock_path_exists.side_effect = lambda f: f == oc_bin + + self.assertEqual(locate_oc_binary(), oc_bin) + + @unittest.skipIf(six.PY3, 'py2 test only') + @mock.patch('os.path.exists') + @mock.patch('os.environ.get') + def test_binary_lookup_in_home(self, mock_env_get, mock_path_exists): + ''' Testing binary lookup in ~/bin ''' + + oc_bin = os.path.expanduser('~/bin/oc') + + mock_env_get.side_effect = lambda _v, _d: '/bin:/usr/bin' + + mock_path_exists.side_effect = lambda f: f == oc_bin + + self.assertEqual(locate_oc_binary(), oc_bin) + + @unittest.skipIf(six.PY2, 'py3 test only') + @mock.patch('shutil.which') + @mock.patch('os.environ.get') + def test_binary_lookup_fallback_py3(self, mock_env_get, mock_shutil_which): + ''' Testing binary lookup fallback ''' + + mock_env_get.side_effect = lambda _v, _d: '' + + mock_shutil_which.side_effect = lambda _f, path=None: None + + self.assertEqual(locate_oc_binary(), 'oc') + + @unittest.skipIf(six.PY2, 'py3 test only') + @mock.patch('shutil.which') + @mock.patch('os.environ.get') + def test_binary_lookup_in_path_py3(self, mock_env_get, mock_shutil_which): + ''' Testing binary lookup in path ''' + + oc_bin = '/usr/bin/oc' + + mock_env_get.side_effect = lambda _v, _d: '/bin:/usr/bin' + + mock_shutil_which.side_effect = lambda _f, path=None: oc_bin + + self.assertEqual(locate_oc_binary(), oc_bin) + + @unittest.skipIf(six.PY2, 'py3 test only') + @mock.patch('shutil.which') + @mock.patch('os.environ.get') + def test_binary_lookup_in_usr_local_py3(self, mock_env_get, mock_shutil_which): + ''' Testing binary lookup in /usr/local/bin ''' + + oc_bin = '/usr/local/bin/oc' + + mock_env_get.side_effect = lambda _v, _d: '/bin:/usr/bin' + + mock_shutil_which.side_effect = lambda _f, path=None: oc_bin + + self.assertEqual(locate_oc_binary(), oc_bin) + + @unittest.skipIf(six.PY2, 'py3 test only') + @mock.patch('shutil.which') + @mock.patch('os.environ.get') + def test_binary_lookup_in_home_py3(self, mock_env_get, mock_shutil_which): + ''' Testing binary lookup in ~/bin ''' + + oc_bin = os.path.expanduser('~/bin/oc') + + mock_env_get.side_effect = lambda _v, _d: '/bin:/usr/bin' + + mock_shutil_which.side_effect = lambda _f, path=None: oc_bin + + self.assertEqual(locate_oc_binary(), oc_bin) diff --git a/roles/lib_openshift/src/test/unit/test_oc_volume.py b/roles/lib_openshift/src/test/unit/test_oc_volume.py new file mode 100755 index 000000000..d91e22bc7 --- /dev/null +++ b/roles/lib_openshift/src/test/unit/test_oc_volume.py @@ -0,0 +1,633 @@ +''' + Unit tests for oc volume +''' + +import copy +import os +import six +import sys +import unittest +import mock + +# Removing invalid variable names for tests so that I can +# keep them brief +# pylint: disable=invalid-name,no-name-in-module +# Disable import-error b/c our libraries aren't loaded in jenkins +# pylint: disable=import-error +# place class in our python path +module_path = os.path.join('/'.join(os.path.realpath(__file__).split('/')[:-4]), 'library') # noqa: E501 +sys.path.insert(0, module_path) +from oc_volume import OCVolume, locate_oc_binary # noqa: E402 + + +class OCVolumeTest(unittest.TestCase): + ''' + Test class for OCVolume + ''' + params = {'name': 'oso-rhel7-zagg-web', + 'kubeconfig': '/etc/origin/master/admin.kubeconfig', + 'namespace': 'test', + 'labels': None, + 'state': 'present', + 'kind': 'dc', + 'mount_path': None, + 'secret_name': None, + 'mount_type': 'pvc', + 'claim_name': 'testclaim', + 'claim_size': '1G', + 'configmap_name': None, + 'vol_name': 'test-volume', + 'debug': False} + + @mock.patch('oc_volume.Utils.create_tmpfile_copy') + @mock.patch('oc_volume.OCVolume._run') + def test_create_pvc(self, mock_cmd, mock_tmpfile_copy): + ''' Testing a label list ''' + params = copy.deepcopy(OCVolumeTest.params) + + dc = '''{ + "kind": "DeploymentConfig", + "apiVersion": "v1", + "metadata": { + "name": "oso-rhel7-zagg-web", + "namespace": "new-monitoring", + "selfLink": "/oapi/v1/namespaces/new-monitoring/deploymentconfigs/oso-rhel7-zagg-web", + "uid": "f56e9dd2-7c13-11e6-b046-0e8844de0587", + "resourceVersion": "137095771", + "generation": 4, + "creationTimestamp": "2016-09-16T13:46:24Z", + "labels": { + "app": "oso-rhel7-ops-base", + "name": "oso-rhel7-zagg-web" + }, + "annotations": { + "openshift.io/generated-by": "OpenShiftNewApp" + } + }, + "spec": { + "strategy": { + "type": "Rolling", + "rollingParams": { + "updatePeriodSeconds": 1, + "intervalSeconds": 1, + "timeoutSeconds": 600, + "maxUnavailable": "25%", + "maxSurge": "25%" + }, + "resources": {} + }, + "triggers": [ + { + "type": "ConfigChange" + }, + { + "type": "ImageChange", + "imageChangeParams": { + "automatic": true, + "containerNames": [ + "oso-rhel7-zagg-web" + ], + "from": { + "kind": "ImageStreamTag", + "namespace": "new-monitoring", + "name": "oso-rhel7-zagg-web:latest" + }, + "lastTriggeredImage": "notused" + } + } + ], + "replicas": 10, + "test": false, + "selector": { + "deploymentconfig": "oso-rhel7-zagg-web" + }, + "template": { + "metadata": { + "creationTimestamp": null, + "labels": { + "app": "oso-rhel7-ops-base", + "deploymentconfig": "oso-rhel7-zagg-web" + }, + "annotations": { + "openshift.io/generated-by": "OpenShiftNewApp" + } + }, + "spec": { + "volumes": [ + { + "name": "monitoring-secrets", + "secret": { + "secretName": "monitoring-secrets" + } + } + ], + "containers": [ + { + "name": "oso-rhel7-zagg-web", + "image": "notused", + "resources": {}, + "volumeMounts": [ + { + "name": "monitoring-secrets", + "mountPath": "/secrets" + } + ], + "terminationMessagePath": "/dev/termination-log", + "imagePullPolicy": "Always", + "securityContext": { + "capabilities": {}, + "privileged": false + } + } + ], + "restartPolicy": "Always", + "terminationGracePeriodSeconds": 30, + "dnsPolicy": "ClusterFirst", + "securityContext": {} + } + } + } + }''' + + post_dc = '''{ + "kind": "DeploymentConfig", + "apiVersion": "v1", + "metadata": { + "name": "oso-rhel7-zagg-web", + "namespace": "new-monitoring", + "selfLink": "/oapi/v1/namespaces/new-monitoring/deploymentconfigs/oso-rhel7-zagg-web", + "uid": "f56e9dd2-7c13-11e6-b046-0e8844de0587", + "resourceVersion": "137095771", + "generation": 4, + "creationTimestamp": "2016-09-16T13:46:24Z", + "labels": { + "app": "oso-rhel7-ops-base", + "name": "oso-rhel7-zagg-web" + }, + "annotations": { + "openshift.io/generated-by": "OpenShiftNewApp" + } + }, + "spec": { + "strategy": { + "type": "Rolling", + "rollingParams": { + "updatePeriodSeconds": 1, + "intervalSeconds": 1, + "timeoutSeconds": 600, + "maxUnavailable": "25%", + "maxSurge": "25%" + }, + "resources": {} + }, + "triggers": [ + { + "type": "ConfigChange" + }, + { + "type": "ImageChange", + "imageChangeParams": { + "automatic": true, + "containerNames": [ + "oso-rhel7-zagg-web" + ], + "from": { + "kind": "ImageStreamTag", + "namespace": "new-monitoring", + "name": "oso-rhel7-zagg-web:latest" + }, + "lastTriggeredImage": "notused" + } + } + ], + "replicas": 10, + "test": false, + "selector": { + "deploymentconfig": "oso-rhel7-zagg-web" + }, + "template": { + "metadata": { + "creationTimestamp": null, + "labels": { + "app": "oso-rhel7-ops-base", + "deploymentconfig": "oso-rhel7-zagg-web" + }, + "annotations": { + "openshift.io/generated-by": "OpenShiftNewApp" + } + }, + "spec": { + "volumes": [ + { + "name": "monitoring-secrets", + "secret": { + "secretName": "monitoring-secrets" + } + }, + { + "name": "test-volume", + "persistentVolumeClaim": { + "claimName": "testclass", + "claimSize": "1G" + } + } + ], + "containers": [ + { + "name": "oso-rhel7-zagg-web", + "image": "notused", + "resources": {}, + "volumeMounts": [ + { + "name": "monitoring-secrets", + "mountPath": "/secrets" + }, + { + "name": "test-volume", + "mountPath": "/data" + } + ], + "terminationMessagePath": "/dev/termination-log", + "imagePullPolicy": "Always", + "securityContext": { + "capabilities": {}, + "privileged": false + } + } + ], + "restartPolicy": "Always", + "terminationGracePeriodSeconds": 30, + "dnsPolicy": "ClusterFirst", + "securityContext": {} + } + } + } + }''' + + mock_cmd.side_effect = [ + (0, dc, ''), + (0, dc, ''), + (0, '', ''), + (0, post_dc, ''), + ] + + mock_tmpfile_copy.side_effect = [ + '/tmp/mocked_kubeconfig', + ] + + results = OCVolume.run_ansible(params, False) + + self.assertTrue(results['changed']) + self.assertTrue(results['results']['results'][-1]['name'] == 'test-volume') + + @mock.patch('oc_volume.Utils.create_tmpfile_copy') + @mock.patch('oc_volume.OCVolume._run') + def test_create_configmap(self, mock_cmd, mock_tmpfile_copy): + ''' Testing a label list ''' + params = copy.deepcopy(OCVolumeTest.params) + params.update({'mount_path': '/configmap', + 'mount_type': 'configmap', + 'configmap_name': 'configtest', + 'vol_name': 'configvol'}) + + dc = '''{ + "kind": "DeploymentConfig", + "apiVersion": "v1", + "metadata": { + "name": "oso-rhel7-zagg-web", + "namespace": "new-monitoring", + "selfLink": "/oapi/v1/namespaces/new-monitoring/deploymentconfigs/oso-rhel7-zagg-web", + "uid": "f56e9dd2-7c13-11e6-b046-0e8844de0587", + "resourceVersion": "137095771", + "generation": 4, + "creationTimestamp": "2016-09-16T13:46:24Z", + "labels": { + "app": "oso-rhel7-ops-base", + "name": "oso-rhel7-zagg-web" + }, + "annotations": { + "openshift.io/generated-by": "OpenShiftNewApp" + } + }, + "spec": { + "strategy": { + "type": "Rolling", + "rollingParams": { + "updatePeriodSeconds": 1, + "intervalSeconds": 1, + "timeoutSeconds": 600, + "maxUnavailable": "25%", + "maxSurge": "25%" + }, + "resources": {} + }, + "triggers": [ + { + "type": "ConfigChange" + }, + { + "type": "ImageChange", + "imageChangeParams": { + "automatic": true, + "containerNames": [ + "oso-rhel7-zagg-web" + ], + "from": { + "kind": "ImageStreamTag", + "namespace": "new-monitoring", + "name": "oso-rhel7-zagg-web:latest" + }, + "lastTriggeredImage": "notused" + } + } + ], + "replicas": 10, + "test": false, + "selector": { + "deploymentconfig": "oso-rhel7-zagg-web" + }, + "template": { + "metadata": { + "creationTimestamp": null, + "labels": { + "app": "oso-rhel7-ops-base", + "deploymentconfig": "oso-rhel7-zagg-web" + }, + "annotations": { + "openshift.io/generated-by": "OpenShiftNewApp" + } + }, + "spec": { + "volumes": [ + { + "name": "monitoring-secrets", + "secret": { + "secretName": "monitoring-secrets" + } + } + ], + "containers": [ + { + "name": "oso-rhel7-zagg-web", + "image": "notused", + "resources": {}, + "volumeMounts": [ + { + "name": "monitoring-secrets", + "mountPath": "/secrets" + } + ], + "terminationMessagePath": "/dev/termination-log", + "imagePullPolicy": "Always", + "securityContext": { + "capabilities": {}, + "privileged": false + } + } + ], + "restartPolicy": "Always", + "terminationGracePeriodSeconds": 30, + "dnsPolicy": "ClusterFirst", + "securityContext": {} + } + } + } + }''' + + post_dc = '''{ + "kind": "DeploymentConfig", + "apiVersion": "v1", + "metadata": { + "name": "oso-rhel7-zagg-web", + "namespace": "new-monitoring", + "selfLink": "/oapi/v1/namespaces/new-monitoring/deploymentconfigs/oso-rhel7-zagg-web", + "uid": "f56e9dd2-7c13-11e6-b046-0e8844de0587", + "resourceVersion": "137095771", + "generation": 4, + "creationTimestamp": "2016-09-16T13:46:24Z", + "labels": { + "app": "oso-rhel7-ops-base", + "name": "oso-rhel7-zagg-web" + }, + "annotations": { + "openshift.io/generated-by": "OpenShiftNewApp" + } + }, + "spec": { + "strategy": { + "type": "Rolling", + "rollingParams": { + "updatePeriodSeconds": 1, + "intervalSeconds": 1, + "timeoutSeconds": 600, + "maxUnavailable": "25%", + "maxSurge": "25%" + }, + "resources": {} + }, + "triggers": [ + { + "type": "ConfigChange" + }, + { + "type": "ImageChange", + "imageChangeParams": { + "automatic": true, + "containerNames": [ + "oso-rhel7-zagg-web" + ], + "from": { + "kind": "ImageStreamTag", + "namespace": "new-monitoring", + "name": "oso-rhel7-zagg-web:latest" + }, + "lastTriggeredImage": "notused" + } + } + ], + "replicas": 10, + "test": false, + "selector": { + "deploymentconfig": "oso-rhel7-zagg-web" + }, + "template": { + "metadata": { + "creationTimestamp": null, + "labels": { + "app": "oso-rhel7-ops-base", + "deploymentconfig": "oso-rhel7-zagg-web" + }, + "annotations": { + "openshift.io/generated-by": "OpenShiftNewApp" + } + }, + "spec": { + "volumes": [ + { + "name": "monitoring-secrets", + "secret": { + "secretName": "monitoring-secrets" + } + }, + { + "name": "configvol", + "configMap": { + "name": "configtest" + } + } + ], + "containers": [ + { + "name": "oso-rhel7-zagg-web", + "image": "notused", + "resources": {}, + "volumeMounts": [ + { + "name": "monitoring-secrets", + "mountPath": "/secrets" + }, + { + "name": "configvol", + "mountPath": "/configmap" + } + ], + "terminationMessagePath": "/dev/termination-log", + "imagePullPolicy": "Always", + "securityContext": { + "capabilities": {}, + "privileged": false + } + } + ], + "restartPolicy": "Always", + "terminationGracePeriodSeconds": 30, + "dnsPolicy": "ClusterFirst", + "securityContext": {} + } + } + } + }''' + + mock_cmd.side_effect = [ + (0, dc, ''), + (0, dc, ''), + (0, '', ''), + (0, post_dc, ''), + ] + + mock_tmpfile_copy.side_effect = [ + '/tmp/mocked_kubeconfig', + ] + + results = OCVolume.run_ansible(params, False) + + self.assertTrue(results['changed']) + self.assertTrue(results['results']['results'][-1]['name'] == 'configvol') + + @unittest.skipIf(six.PY3, 'py2 test only') + @mock.patch('os.path.exists') + @mock.patch('os.environ.get') + def test_binary_lookup_fallback(self, mock_env_get, mock_path_exists): + ''' Testing binary lookup fallback ''' + + mock_env_get.side_effect = lambda _v, _d: '' + + mock_path_exists.side_effect = lambda _: False + + self.assertEqual(locate_oc_binary(), 'oc') + + @unittest.skipIf(six.PY3, 'py2 test only') + @mock.patch('os.path.exists') + @mock.patch('os.environ.get') + def test_binary_lookup_in_path(self, mock_env_get, mock_path_exists): + ''' Testing binary lookup in path ''' + + oc_bin = '/usr/bin/oc' + + mock_env_get.side_effect = lambda _v, _d: '/bin:/usr/bin' + + mock_path_exists.side_effect = lambda f: f == oc_bin + + self.assertEqual(locate_oc_binary(), oc_bin) + + @unittest.skipIf(six.PY3, 'py2 test only') + @mock.patch('os.path.exists') + @mock.patch('os.environ.get') + def test_binary_lookup_in_usr_local(self, mock_env_get, mock_path_exists): + ''' Testing binary lookup in /usr/local/bin ''' + + oc_bin = '/usr/local/bin/oc' + + mock_env_get.side_effect = lambda _v, _d: '/bin:/usr/bin' + + mock_path_exists.side_effect = lambda f: f == oc_bin + + self.assertEqual(locate_oc_binary(), oc_bin) + + @unittest.skipIf(six.PY3, 'py2 test only') + @mock.patch('os.path.exists') + @mock.patch('os.environ.get') + def test_binary_lookup_in_home(self, mock_env_get, mock_path_exists): + ''' Testing binary lookup in ~/bin ''' + + oc_bin = os.path.expanduser('~/bin/oc') + + mock_env_get.side_effect = lambda _v, _d: '/bin:/usr/bin' + + mock_path_exists.side_effect = lambda f: f == oc_bin + + self.assertEqual(locate_oc_binary(), oc_bin) + + @unittest.skipIf(six.PY2, 'py3 test only') + @mock.patch('shutil.which') + @mock.patch('os.environ.get') + def test_binary_lookup_fallback_py3(self, mock_env_get, mock_shutil_which): + ''' Testing binary lookup fallback ''' + + mock_env_get.side_effect = lambda _v, _d: '' + + mock_shutil_which.side_effect = lambda _f, path=None: None + + self.assertEqual(locate_oc_binary(), 'oc') + + @unittest.skipIf(six.PY2, 'py3 test only') + @mock.patch('shutil.which') + @mock.patch('os.environ.get') + def test_binary_lookup_in_path_py3(self, mock_env_get, mock_shutil_which): + ''' Testing binary lookup in path ''' + + oc_bin = '/usr/bin/oc' + + mock_env_get.side_effect = lambda _v, _d: '/bin:/usr/bin' + + mock_shutil_which.side_effect = lambda _f, path=None: oc_bin + + self.assertEqual(locate_oc_binary(), oc_bin) + + @unittest.skipIf(six.PY2, 'py3 test only') + @mock.patch('shutil.which') + @mock.patch('os.environ.get') + def test_binary_lookup_in_usr_local_py3(self, mock_env_get, mock_shutil_which): + ''' Testing binary lookup in /usr/local/bin ''' + + oc_bin = '/usr/local/bin/oc' + + mock_env_get.side_effect = lambda _v, _d: '/bin:/usr/bin' + + mock_shutil_which.side_effect = lambda _f, path=None: oc_bin + + self.assertEqual(locate_oc_binary(), oc_bin) + + @unittest.skipIf(six.PY2, 'py3 test only') + @mock.patch('shutil.which') + @mock.patch('os.environ.get') + def test_binary_lookup_in_home_py3(self, mock_env_get, mock_shutil_which): + ''' Testing binary lookup in ~/bin ''' + + oc_bin = os.path.expanduser('~/bin/oc') + + mock_env_get.side_effect = lambda _v, _d: '/bin:/usr/bin' + + mock_shutil_which.side_effect = lambda _f, path=None: oc_bin + + self.assertEqual(locate_oc_binary(), oc_bin) |