#!/usr/bin/python # Copyright: (c) 2018, Terry Jones # 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()