bbohard
/
eole-galaxy
1
0
Fork 0
Code Issues 1 Pull Requests Projects Releases Wiki Activity
eole-galaxy/cadoles/eole/plugins/modules/instance.py

381 lines
14 KiB
Python
Raw Normal View History

Quelques procédures de plus
2021-08-30 10:06:42 +02:00
#!/usr/bin/python
# Copyright: (c) 2018, Terry Jones <terry.jones@example.org>
# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
from __future__ import (absolute_import, division, print_function)
__metaclass__ = type
DOCUMENTATION = r'''
---
module: instance
short_description: This is a module to automate EOLE module instance
# If this is part of a collection, you need to use semantic versioning,
# i.e. the version is of the form "2.5.0" and not "2.4".
version_added: "1.0.0"
description: This is my longer description explaining my test module.
options:
module:
description: EOLE module type
required: true
type: str
variables:
description: dictionnary of variables overloading default instance values.
required: false
type: dict
# Specify this value according to your collection
# in format of namespace.collection.doc_fragment_name
extends_documentation_fragment:
- cadoles.eole.instance
author:
- Cadoles
'''
EXAMPLES = r'''
# Pass in a message
- name: Test with a message
cadoles.eole.instance:
module: eolebase
variables:
root_password: eole
eole_password: eole
'''
RETURN = r'''
# These are examples of possible return values, and in general should use other names for return values.
original_message:
description: The original name param that was passed in.
type: str
returned: always
sample: 'hello world'
message:
description: The output message that the test module generates.
type: str
returned: always
sample: 'goodbye'
'''
from ansible.module_utils.basic import AnsibleModule
import pexpect
import time
import sys
import re
ansi_escape = re.compile(r'\x1B(?:[@-Z\\-_]|\[[0-?]*[ -/]*[@-~])')
class ExpectationCollection:
def __init__(self):
self.expectations_lookup = {}
def add_expectation(self, expectation):
if expectation.get_pattern() in self.expectations_lookup:
if expectation.context in [exp.context for exp in self.expectations_lookup[expectation.get_pattern()]]:
print(f'Can not add {expectation} to collection')
return False
self.expectations_lookup[expectation.get_pattern()] = self.expectations_lookup.setdefault(expectation.get_pattern(), []) + [expectation]
return True
def add_expectation_from_descr(self, expectation, response=None, name=None):
expectation = Expectation(expectation, response=response, name=name)
self.add_expectation(expectation)
def get_patterns(self):
return [pexpect.EOF, pexpect.TIMEOUT] + list(self.expectations_lookup.keys())
def get_expectations_by_pattern(self, pattern):
return self.expectations_lookup.get(pattern, [])
def get_expectations_by_name(self, name):
return [expectation for expectation in self.get_expectations(recursive=True) if expectation.name == name]
def get_exposed_expectation_names(self):
return [expectation.name for expectation in self.get_expectations(recursive=True) if expectation.name]
def set_expectation_response_by_name(self, name, response):
for expectation in self.get_expectations_by_name(name):
expectation.set_response(response)
def get_expectations(self, recursive=False):
expectations = [exp for exps in self.expectations_lookup.values() for exp in exps]
if recursive:
expectations.extend([exp for exps in expectations for exp in exps.get_following_expectations()])
pass
return expectations
def count_expectations(self):
return sum([exp.count_expectations() for exp in self.get_expectations()])
def merge(self, collection):
for expectation in collection.get_expectations():
self.add_expectation(expectation)
class Expectation:
def __init__(self, multiline_pattern, response=None, name=None):
self.name = name
self.context = [p.strip() for p in multiline_pattern.strip().split('\n') if p.strip()]
self.pattern = self.context[-1]
self.response = response
self.next = None
def get_pattern(self):
return re.compile(f'(.*){re.escape(self.pattern)}(.*)')
def set_next_expectation(self, expectation):
self.next = expectation
def set_response(self, response):
print(f'Setting {response} for {self.pattern}')
self.response = response
def expect(self, spawned):
print(f'-> expecting next "{self.pattern}"')
p = spawned.expect([pexpect.EOF, pexpect.TIMEOUT, self.get_pattern()])
if p not in [0, 1]:
self.answer(spawned)
else:
print(f'-> before: {spawned.before}\n-> after: {spawned.after}\n-> {self.pattern}')
def answer(self, spawned):
print(f'-> answering "{self.response}" to "{spawned.after}"')
if self.response is not None:
spawned.sendline(self.response)
if self.next:
self.next.expect(spawned)
def is_the_one(self, context):
#print(f'testing if it is the one {self.get_pattern()} for {context}')
context = [l.strip() for l in context.strip().split('\r\n') if l.strip()]
if not context:
print('No context provided')
return False
context.reverse()
for index, c in enumerate(self.context[len(self.context)-2::-1]):
if c != ansi_escape.sub('', context[index]):
return False
return True
def count_expectations(self):
count = 1
if self.next:
count += self.next.count_expectations()
return count
def get_following_expectations(self):
if self.next:
return [self.next] + self.next.get_following_expectations()
return []
expectations = ExpectationCollection()
supported_modules = {}
# eolebase
eolebase_expectations = ExpectationCollection()
exp3 = Expectation("""######################################################
# Changement du mot de passe pour l’utilisateur root #
######################################################
Nouveau mot de passe:""", response='eole', name='root_password')
exp3_1 = Expectation("""Confirmation du mot de passe:""", response='eole', name='root_password')
exp3.set_next_expectation(exp3_1)
eolebase_expectations.add_expectation(exp3)
exp4 = Expectation("""######################################################
# Changement du mot de passe pour l’utilisateur eole #
######################################################
Nouveau mot de passe:""", response='eole', name='eole_password')
exp4_1 = Expectation("""Confirmation du mot de passe:""", response='eole', name='eole_password')
exp4_2 = Expectation("""Créer un nouvel administrateur eole2 ? [oui/non]
[non] :""", response='')
exp4.set_next_expectation(exp4_1)
exp4_1.set_next_expectation(exp4_2)
eolebase_expectations.add_expectation(exp4)
exp5 = Expectation("""Une mise à jour est recommandée
Voulez-vous effectuer une mise à jour via le réseau maintenant ? [oui/non]
[oui] :""", response='non', name='maj_fin_instance')
eolebase_expectations.add_expectation(exp5)
supported_modules['eolebase'] = eolebase_expectations
# zephir
zephir_expectations = ExpectationCollection()
exp1 = Expectation("""#############################################################################
# Initialisation du mot de passe de l'administrateur de base (admin_zephir) #
#############################################################################
Mot de passe :""", response='eole', name='admin_zephir_password')
exp1_1 = Expectation("""Confirmation du mot de passe :""", response='eole', name='admin_zephir_password')
exp1.set_next_expectation(exp1_1)
zephir_expectations.add_expectation(exp1)
exp2 = Expectation("""Vous pouvez maintenant créer des utilisateurs si vous le souhaitez
Attribuez leur des droits sur l'application via l'interface web
nom d'utilisateur a créer (rien pour terminer) : """, response='admin', name='other_zephir_account')
exp2_1 = Expectation("""Mot de passe du nouvel utilisateur : """, response='eole', name='other_zephir_account_password')
exp2_2 = Expectation("""Saisissez à nouveau le mot de passe : """, response='eole', name='other_zephir_account_password')
exp2_3 = Expectation("""nom d'utilisateur a créer (rien pour terminer) : """, response='', name='empty_zephir_account')
exp2_2.set_next_expectation(exp2_3)
exp2_1.set_next_expectation(exp2_2)
exp2.set_next_expectation(exp2_1)
zephir_expectations.add_expectation(exp2)
zephir_expectations.add_expectation_from_descr("""* Vérification des données (md5) : Eole 2.6.1""")
zephir_expectations.add_expectation_from_descr("""* Vérification des données (md5) : Eole 2.6.2""")
zephir_expectations.add_expectation_from_descr("""* Vérification des données (md5) : Eole 2.7.0""")
zephir_expectations.add_expectation_from_descr("""* Vérification des données (md5) : Eole 2.7.1""")
zephir_expectations.add_expectation_from_descr("""* Vérification des données (md5) : Eole 2.7.2""")
zephir_expectations.add_expectation_from_descr("""* Vérification des données (md5) : Eole 2.8.0""")
zephir_expectations.add_expectation_from_descr("""Start Systemd services""")
supported_modules['zephir'] = zephir_expectations
# amon
amon_expectations = ExpectationCollection()
# scribe
scribe_expectations = ExpectationCollection()
# seth
seth_expectations = ExpectationCollection()
module_expectations = [
eolebase_expectations,
amon_expectations,
zephir_expectations,
scribe_expectations,
seth_expectations,
]
for module_expectation in module_expectations:
expectations.merge(module_expectation)
cmd = ['/usr/bin/instance']
patterns = expectations.get_patterns()
nb_expectations = expectations.count_expectations()
def run_module():
# define available arguments/parameters a user can pass to the module
module_args = dict(
module=dict(type='str', required=True),
variables=dict(type='dict', required=False),
)
# seed the result dict in the object
# we primarily care about changed and state
# changed is if this module effectively modified the target
# state will include any data that you want your module to pass back
# for consumption, for example, in a subsequent task
result = dict(
changed=False,
module='',
msg='',
)
# the AnsibleModule object will be our abstraction working with Ansible
# this includes instantiation, a couple of common attr would be the
# args/params passed to the execution, as well as if the module
# supports check mode
module = AnsibleModule(
argument_spec=module_args,
supports_check_mode=True
)
# if the user is working with this module in only check mode we do not
# want to make any changes to the environment, just return the current
# state with no modifications
if module.check_mode:
if module.params['module'] in supported_modules:
result['module'] = module.params["module"]
if module.params.get('variables', None) and not set(module.params['variables'].keys()).issubset(set(expectations.get_exposed_expectation_names())):
unknown_variables = list(set(module.params['variables'].keys()).difference(set(expectations.get_exposed_expectation_names())))
result['msg'] += f"Variables {unknown_variables} not available\n"
else:
for variable in module.params.get('variables', {}).keys():
result['msg'] += f"Overloading variable {variable}\n"
else:
result['msg'] += f'Module {module.module} not supported\n'
module.exit_json(**result)
if module.params['module'] not in supported_modules:
result['msg'] += f"Unsupported module {module.params['module']}\n"
module.fail_json(**result)
else:
result['module'] = module.params['module']
if module.params.get('variables', None):
if not set(module.params['variables'].keys()).issubset(set(expectations.get_exposed_expectation_names())):
unknown_variables = list(set(module.params['variables'].keys()).difference(set(expectations.get_exposed_expectation_names())))
result['msg'] += f"Variables {unknown_variables} not available\n"
module.fail_json(**result)
else:
for expectation_name, response in module.params['variables'].items():
expectations.set_expectation_response_by_name(expectation_name, response)
instance_process = pexpect.spawn(cmd, encoding='utf-8', timeout=60)
some_index = 0
while some_index < nb_expectations:
p = instance_process.expect(patterns)
if p == 0:
break
if p == 1:
print(f'Some missing expectations for {instance_process.before}{instance_process.after}')
break
pattern = patterns[p]
for expectation in expectations.get_expectations_by_pattern(patterns[p]):
if expectation.is_the_one(instance_process.before):
expectation.answer(instance_process)
break
some_index += 1
result['changed'] = True
result['msg'] += f"Module {result['module']} instanciated"
# manipulate or modify the state as needed (this is going to be the
# part where your module will do what it needs to do)
# use whatever logic you need to determine whether or not this module
# made any modifications to your target
# during the execution of the module, if there is an exception or a
# conditional state that effectively causes a failure, run
# AnsibleModule.fail_json() to pass in the message and the result
# in the event of a successful module execution, you will want to
# simple AnsibleModule.exit_json(), passing the key/value results
module.exit_json(**result)
def main():
run_module()
if __name__ == '__main__':
main()