clean code: fix sanity

pull/25/head
Markus Bergholz 8 months ago
parent d85967820e
commit b058704701
No known key found for this signature in database
GPG Key ID: B45724801354B174
  1. 4
      Makefile
  2. 4
      plugins/callback/talk.py
  3. 19
      plugins/lookup/generate_password.py
  4. 19
      plugins/lookup/passwords.py
  5. 64
      plugins/module_utils/nextcloud.py
  6. 5
      plugins/modules/file.py
  7. 2
      plugins/modules/file_info.py
  8. 2
      plugins/modules/info.py
  9. 7
      plugins/modules/password.py
  10. 2
      plugins/modules/password_info.py
  11. 2
      plugins/modules/talk.py
  12. 9
      plugins/modules/user_info.py

@ -14,6 +14,9 @@ install: ## install collection localy
remove: ## remove collection localy
rm -rf markuman* ~/.ansible/collections/ansible_collections/markuman/nextcloud
sanity: ## sanity checks
ansible-test sanity --python 3.8 --docker default
syntax: ## test compile
python -m py_compile plugins/lookup/*.py
python -m py_compile plugins/module_utils/nextcloud.py
@ -25,6 +28,7 @@ fullround: ## everything
$(MAKE) remove
$(MAKE) build
$(MAKE) install
$(MAKE) sanity
$(MAKE) test
round: ## short round

@ -26,8 +26,8 @@ class CallbackModule(CallbackBase):
CALLBACK_NEEDS_WHITELIST = True
def __init__(self):
self.nc = NextcloudHandler({})
params = {}
self.nc = NextcloudHandler(params, self._display.warning)
self.channel = os.environ.get('NEXTCLOUD_TALK_CALLBACK_CHANNEL')
self.headers = {
'Accept': 'application/json',

@ -1,15 +1,11 @@
from __future__ import (absolute_import, division, print_function)
__metaclass__ = type
from ansible.errors import AnsibleError, AnsibleParserError
from ansible.errors import AnsibleError
from ansible.plugins.lookup import LookupBase
from ansible_collections.markuman.nextcloud.plugins.module_utils.nextcloud import NextcloudHandler
try:
import requests
except ImportError:
raise AnsibleError("Please install requests library.")
ANSIBLE_METADATA = {
'metadata_version': '1.1',
@ -52,6 +48,17 @@ EXAMPLES = """
class LookupModule(LookupBase):
def run(self, terms, variables, **kwargs):
params = dict()
if kwargs.get('host'):
params['host'] = kwargs.get('host')
if kwargs.get('api_token'):
params['api_token'] = kwargs.get('api_token')
if kwargs.get('user'):
params['user'] = kwargs.get('user')
if kwargs.get('details') is not None:
params['details'] = kwargs.get('details')
if kwargs.get('ssl_mode') is not None:
params['ssl_mode'] = kwargs.get('ssl_mode')
nc = NextcloudHandler(kwargs)
nc = NextcloudHandler(params, AnsibleError)
return nc.fetch_generated_password()

@ -1,15 +1,15 @@
from __future__ import (absolute_import, division, print_function)
__metaclass__ = type
from ansible.errors import AnsibleError, AnsibleParserError
from ansible.plugins.lookup import LookupBase
from ansible.errors import AnsibleError
from ansible_collections.markuman.nextcloud.plugins.module_utils.nextcloud import NextcloudHandler
try:
import requests
HAS_REQUESTS_LIB = True
except ImportError:
raise AnsibleError("Please install requests library.")
HAS_REQUESTS_LIB = False
ANSIBLE_METADATA = {
'metadata_version': '1.1',
@ -55,7 +55,18 @@ EXAMPLES = """
class LookupModule(LookupBase):
def run(self, terms, variables, **kwargs):
params = dict()
if kwargs.get('host'):
params['host'] = kwargs.get('host')
if kwargs.get('api_token'):
params['api_token'] = kwargs.get('api_token')
if kwargs.get('user'):
params['user'] = kwargs.get('user')
if kwargs.get('details') is not None:
params['details'] = kwargs.get('details')
if kwargs.get('ssl_mode') is not None:
params['ssl_mode'] = kwargs.get('ssl_mode')
nc = NextcloudHandler(kwargs)
nc = NextcloudHandler(params, AnsibleError)
for term in terms:
return nc.get_password(term)

@ -3,17 +3,15 @@ __metaclass__ = type
import os
import json
import traceback
from ansible.errors import AnsibleError
from xml.dom import minidom
from ansible.module_utils.basic import missing_required_lib
try:
import requests
HAS_REQUESTS_LIB = True
except ImportError:
HAS_REQUESTS_LIB = False
IMPORT_ERROR = traceback.format_exc()
else:
HAS_REQUESTS_LIB = True
dios_mio = """<?xml version="1.0"?>
@ -35,8 +33,15 @@ dios_mio = """<?xml version="1.0"?>
"""
def status_code_error(status):
raise AnsibleError('Nextcloud returned with status code {SC}'.format(SC=status))
class NextcloudErrorHandler:
def __init__(self, fail_json):
self.fail = fail_json
def status_code_error(self, status):
try:
self.fail(msg='Nextcloud returned with status code {SC}'.format(SC=status))
except Exception:
self.fail('Nextcloud returned with status code {SC}'.format(SC=status))
def parameter_spects(spec_arguments):
@ -51,31 +56,32 @@ def parameter_spects(spec_arguments):
class NextcloudHandler:
def __init__(self, kwargs={}):
def __init__(self, params, fail_json):
self.exit = NextcloudErrorHandler(fail_json)
self.HTTP = 'https'
self.ssl = True
if kwargs.get('ssl_mode') == 'http':
if params.get('ssl_mode') == 'http':
self.HTTP = 'http'
elif kwargs.get('ssl_mode') == 'skip':
elif params.get('ssl_mode') == 'skip':
self.ssl = False
elif os.environ.get('NEXTCLOUD_SSL_MODE') == 'http':
self.HTTP = 'http'
elif os.environ.get('NEXTCLOUD_SSL_MODE') == 'skip':
self.ssl = False
self.details = kwargs.get('details') or False
self.details = params.get('details') or False
self.HOST = kwargs.get('host') or os.environ.get('NEXTCLOUD_HOST')
self.HOST = params.get('host') or os.environ.get('NEXTCLOUD_HOST')
if self.HOST is None:
raise AnsibleError('Unable to continue. No Nextcloud Host is given.')
self.exit.status_code_error('Unable to continue. No Nextcloud Host is given.')
self.USER = kwargs.get('user') or os.environ.get('NEXTCLOUD_USER')
self.USER = params.get('user') or os.environ.get('NEXTCLOUD_USER')
if self.USER is None:
raise AnsibleError('Unable to continue. No Nextcloud User is given.')
self.exit.status_code_error('Unable to continue. No Nextcloud User is given.')
self.TOKEN = kwargs.get('api_token') or os.environ.get('NEXTCLOUD_TOKEN')
self.TOKEN = params.get('api_token') or os.environ.get('NEXTCLOUD_TOKEN')
if self.TOKEN is None:
raise AnsibleError('Unable to continue. No Nextcloud Token is given.')
self.exit.status_code_error('Unable to continue. No Nextcloud Token is given.')
self.headers = {
'Accept': 'application/json',
@ -91,9 +97,9 @@ class NextcloudHandler:
if r.status_code == 200:
return r
elif r.status_code == 404:
raise AnsibleError('File {FILE} does not exist'.format(FILE=path))
self.exit.status_code_error('File {FILE} does not exist'.format(FILE=path))
else:
status_code_error(r.status_code)
self.exit.status_code_error(r.status_code)
def propfind(self, path):
s = requests.Session()
@ -118,7 +124,7 @@ class NextcloudHandler:
'owner': dom.getElementsByTagName('oc:owner-display-name')[0].firstChild.data,
'href': dom.getElementsByTagName('d:href')[0].firstChild.data
}
except:
except Exception:
# I guess it's folder, because it has no content_type
return {
'last_modified': dom.getElementsByTagName('d:getlastmodified')[0].firstChild.data,
@ -134,7 +140,7 @@ class NextcloudHandler:
return {}
else:
status_code_error(r.status_code)
self.exit.status_code_error(r.status_code)
def put(self, path, src):
r = requests.put(
@ -145,7 +151,7 @@ class NextcloudHandler:
if r.status_code in [200, 201, 204]:
return r, True
else:
status_code_error(r.status_code)
self.exit.status_code_error(r.status_code)
def delete(self, path):
r = requests.delete(
@ -158,7 +164,7 @@ class NextcloudHandler:
elif r.status_code == 404:
return r, False
else:
status_code_error(r.status_code)
self.exit.status_code_error(r.status_code)
def talk(self, message, channel):
body = {
@ -179,21 +185,21 @@ class NextcloudHandler:
if r.status_code == 201:
return r, True
else:
status_code_error(r.status_code)
self.exit.status_code_error(r.status_code)
def list_passwords(self):
r = self.get("index.php/apps/passwords/api/1.0/password/list")
if r.status_code == 200:
return r.json()
else:
status_code_error(r.status_code)
self.exit.status_code_error(r.status_code)
def list_passwords_folders(self):
r = self.get("index.php/apps/passwords/api/1.0/folder/list")
if r.status_code == 200:
return r.json()
else:
status_code_error(r.status_code)
self.exit.status_code_error(r.status_code)
def create_passwords_folder(self, name):
post_obj = {
@ -211,7 +217,7 @@ class NextcloudHandler:
if r.status_code == 201:
return r.json()
else:
status_code_error(r.status_code)
self.exit.status_code_error(r.status_code)
def get_passwords_folder(self, name):
for folder in self.list_passwords_folders():
@ -235,7 +241,7 @@ class NextcloudHandler:
if r.status_code == 200:
return [r.json().get('password')]
else:
status_code_error(r.status_code)
self.exit.status_code_error(r.status_code)
def create_password(self, post_obj):
r = requests.post(
@ -249,7 +255,7 @@ class NextcloudHandler:
if r.status_code == 201:
return r.json()
else:
status_code_error(r.status_code)
self.exit.status_code_error(r.status_code)
def delete_password(self, post_obj):
r = requests.delete(
@ -263,7 +269,7 @@ class NextcloudHandler:
if r.status_code == 200:
return r.json()
else:
status_code_error(r.status_code)
self.exit.status_code_error(r.status_code)
def update_password(self, post_obj):
r = requests.patch(
@ -277,7 +283,7 @@ class NextcloudHandler:
if r.status_code == 200:
return r.json()
else:
status_code_error(r.status_code)
self.exit.status_code_error(r.status_code)
def user(self):
return self.USER

@ -80,7 +80,6 @@ EXAMPLES = '''
'''
from ansible.module_utils.basic import AnsibleModule
from ansible.errors import AnsibleError
from ansible_collections.markuman.nextcloud.plugins.module_utils.nextcloud import NextcloudHandler
from ansible_collections.markuman.nextcloud.plugins.module_utils.nextcloud import parameter_spects
import os.path
@ -111,7 +110,7 @@ def main():
)
)
nc = NextcloudHandler(module.params)
nc = NextcloudHandler(module.params, module.fail_json)
mode = module.params.get("mode")
source = module.params.get("source")
@ -126,7 +125,7 @@ def main():
facts = nc.propfind("remote.php/dav/files/{USER}/{SRC}".format(USER=nc.user(), SRC=source))
if destination is None:
raise AnsibleError('No destination is given')
module.fail_json(msg='No destination is given')
if facts == {}:
message = "Requested file does not exist."

@ -53,7 +53,7 @@ def main():
))
)
nc = NextcloudHandler(module.params)
nc = NextcloudHandler(module.params, module.fail_json)
source = module.params.get("source")

@ -36,7 +36,7 @@ def main():
argument_spec=parameter_spects(dict())
)
nc = NextcloudHandler(module.params)
nc = NextcloudHandler(module.params, module.fail_json)
retval = nc.get('/ocs/v1.php/cloud/capabilities').json()
module.exit_json(nextcloud=retval.get('ocs').get('data'))

@ -79,7 +79,6 @@ EXAMPLES = '''
from ansible.module_utils.basic import AnsibleModule
from ansible_collections.markuman.nextcloud.plugins.module_utils.nextcloud import NextcloudHandler
from ansible_collections.markuman.nextcloud.plugins.module_utils.nextcloud import parameter_spects
from ansible.errors import AnsibleError
def main():
@ -99,7 +98,7 @@ def main():
))
)
module.params["details"] = True
nc = NextcloudHandler(module.params)
nc = NextcloudHandler(module.params, module.fail_json)
name = module.params.get('name')
password = module.params.get('password') or nc.fetch_generated_password()
@ -178,7 +177,7 @@ def main():
module.exit_json(changed=True, password=retval)
else:
raise AnsibleError('More than one password identifies. Cannot continue')
module.fail_json(msg='More than one password identifies. Cannot continue')
elif state == 'absent':
if len(retval) == 1:
@ -192,7 +191,7 @@ def main():
elif len(retval) == 0:
module.exit_json(changed=False, password={})
else:
raise AnsibleError('More than one password identifies. Cannot continue')
module.fail_json(msg='More than one password identifies. Cannot continue')
if __name__ == '__main__':

@ -42,7 +42,7 @@ def main():
))
)
module.params["details"] = True
nc = NextcloudHandler(module.params)
nc = NextcloudHandler(module.params, module.fail_json)
if module.params.get('name'):
retval = nc.get_password(module.params.get('name'))

@ -52,7 +52,7 @@ def main():
))
)
nc = NextcloudHandler(module.params)
nc = NextcloudHandler(module.params, module.fail_json)
message = module.params.get("msg")
channel = module.params.get("channel")

@ -56,18 +56,15 @@ def main():
))
)
nc = NextcloudHandler(module.params)
nc = NextcloudHandler(module.params, module.fail_json)
username = module.params.get('username')
if username:
retval = nc.get(f'/ocs/v1.php/cloud/users/{username}').json()
module.exit_json(
users=[username],
user_data=retval.get('ocs', {}).get('data')
)
module.exit_json(users=[username], user_data=retval.get('ocs', {}).get('data'))
else:
retval = nc.get('/ocs/v1.php/cloud/users').json()
module.exit_json(users=retval.get('ocs', {}).get('data', {}).get('users', []),user_data={})
module.exit_json(users=retval.get('ocs', {}).get('data', {}).get('users', []), user_data={})
if __name__ == '__main__':

Loading…
Cancel
Save