pexpectation

Surcouche à pexpect pour formaliser les enchaînements de questions et éviter certains problèmes d’association question, réponse quand le contexte n’est pas suffisant.

Principe

On constitue une collection de questions, réponses. Dans cette collection, certaines questions sont exposées et utilisées pour la méthode expect du module pexpect. Ce premier lot de question est associé à un contexte suffisamment large pour les discriminer entre elles. D’autres questions sont liées à ces premières questions et les réponses sont enchaînées. Pour gérer les embranchements conditionnels, on peut associer à chaque question liée un déclencheur (la réponse à la question précédente).

Composition

.
├── eole_module_expectations
│   ├── eolebase_instance_pexpectations.py
│   ├── __init__.py
│   ├── register_zephir_pexpectations.py
│   └── zephir_instance_pexpectations.py
├── __init__.py
└── pexpectation.py

Le corps du module est constitué de deux classes dans le fichier pexpectation.py

• ExpectationCollection ;
• Expectation.

Le sous-module eole_module_expectations contient des exemples d’utilisation pour des procédures classiques des modules EOLE :

• instance ;
• enregistrement_zephir.

Utilisation

Soit la transcription simplifiée de la procédure enregistrement_zephir (certains cheminement sont exclus parce qu’ils n’ont pas de sens dans ce contexte d’automatisation) :

digraph G {
    subgraph cluster_0 {
    style=filled;
    color=lightgrey;
    node [style=filled,color=white];
    edge [style=dotted];
    "Voulez-vous établir une configuration réseau minimale" -> "interface connectée sur l'extérieur" [label="oui"];
    "interface connectée sur l'extérieur" -> "adresse IP" -> "masque de sous-réseau" -> "adresse de la passerelle";
  }

  subgraph cluster_1 {
    node [style=filled,color=white];
    edge [style=dotted];
    "Serveur déjà enregistré";
    color=lightgrey;
    style=filled;
  }
  subgraph cluster_2 {
    node [style=filled,color=white];
    edge [style=dotted];
    "créer le serveur dans la base du serveur Zéphir" -> "entrez le RNE de l’établissement" [label="non"];
    "entrez le RNE de l’établissement" -> "entrez le n° identifiant le serveur l'application Zéphir" [label="rien"];
    "entrez le RNE de l’établissement" -> "choix du serveur" [label="RNE"];
    color=lightgrey;
    style=filled;
  }
  subgraph cluster_3 {
    node [style=filled,color=white];
    edge [style=dotted];
    "Mise à jour des informations sur le matériel" -> "processeur" -> "disque dur";
    color=lightgrey;
    style=filled;
  }
  subgraph cluster_4 {
    node [style=filled,color=white];
    edge [style=dotted];
    "une procédure d'enregistrement à déjà eu lieu pour ce serveur";
    color=lightgrey;
    style=filled;
  }
  subgraph cluster_5 {
    node [style=filled,color=white];
    edge [style=dotted];
    "Entrez le numéro de votre choix";
    color=lightgrey;
    style=filled;
  }
  subgraph cluster_6 {
    node [style=filled,color=white];
    edge [style=dotted];
    "Entrez l'adresse (nom DNS) du serveur Zéphir" -> "Entrez votre login pour l'application Zéphir" -> "Mot de passe pour l'application Zéphir";
    color=lightgrey;
    style=filled;
  }
  "Début de la procédure" -> "Voulez-vous établir une configuration réseau minimale";
  "Début de la procédure" -> "Serveur déjà enregistré";
  "Voulez-vous établir une configuration réseau minimale" -> "Entrez l'adresse (nom DNS) du serveur Zéphir" [label="non"];
  "adresse de la passerelle" -> "Entrez l'adresse (nom DNS) du serveur Zéphir";
  "Mot de passe pour l'application Zéphir" -> "créer le serveur dans la base du serveur Zéphir";
  "entrez le n° identifiant le serveur l'application Zéphir" -> "une procédure d'enregistrement à déjà eu lieu pour ce serveur";
  "entrez le n° identifiant le serveur l'application Zéphir" -> "Mise à jour des informations sur le matériel";
  "disque dur" -> "Entrez le numéro de votre choix";
  "Entrez le numéro de votre choix" -> "Fin de la procédure";

  "Début de la procédure" [shape=Mdiamond];
  "Fin de la procédure" [shape=Msquare];
}

Ce qui s’implémente de la façon suivante :

expectations = ExpectationCollection()

already_registered = Expectation("""1 -> Désinscrire ce serveur du serveur Zéphir
2 -> Relancer l'enregistrement
3 -> Ne rien faire

  Entrez le numéro de votre choix :""", response='3', name='already_registered')
network_configuration = Expectation("""  Procédure d'enregistrement sur le serveur Zéphir 


Voulez-vous établir une configuration réseau minimale (O/N) :""", response='N', name='network_configuration')
interface_name = Expectation("""interface connectée sur l'extérieur""", response='ens0', name='interface_name')
network_address = Expectation("""adresse_ip {variable} :""", response='192.168.1.2', name='')
network_netmask = Expectation("""masque de réseau pour {variable} :""", response='255.255.255.0', name='network_netmask')
gateway = Expectation("""adresse de la passerelle :""", response='192.168.1.1', name='gateway')
zephir_address = Expectation("""Entrez l'adresse (nom DNS) du serveur Zéphir :""", response='zephir', name='zephir_address')
zephir_admin = Expectation("""Entrez votre login pour l'application Zéphir (rien pour sortir) :""", response='admin_zephir', name='zephir_admin')
zephir_admin_password = Expectation("""Mot de passe pour l'application Zéphir pour {variable} :""", response='eole', name='zephir_admin_password')
new_server = Expectation("""créer le serveur dans la base du serveur Zéphir (O/N) :""", response='N', name='new_server')
rne = Expectation("""entrez le RNE de l'établissement correspondant au serveur,
(rien pour saisir directement un n° de serveur) :""", response='', name='rne')
server_id = Expectation("""entrez le n° identifiant le serveur l'application Zéphir :""", response='1', name='server_id')
hardware = Expectation("""Mise à jour des informations sur le matériel
matériel (Standard PC (Q35 + ICH9, 2009) par défaut) :""", response='', name='hardware')
processor = Expectation("""processeur ( Intel Core Processor (Skylake, IBRS) 3191 MHz par défaut) :""", response='', name='processor')
harddrive = Expectation("""disque dur (19 Go par défaut) :""", response='', name='harddrive')
key_available = Expectation("""(une procédure d'enregistrement à déjà eu lieu pour ce serveur)
continuer l'enregistrement (O/N) ?""", response='O', name='key_available')
final = Expectation("""1 -> Ne rien faire
2 -> Utiliser la configuration définie sur le serveur Zéphir
3 -> Non disponible
4 -> Modifier la variante du serveur

  Entrez le numéro de votre choix :""", response='2', name='final')


expectations.add_expectation(already_registered)
expectations.add_expectation(network_configuration)
expectations.add_expectation(new_server)
expectations.add_expectation(hardware)
expectations.add_expectation(key_available)
expectations.add_expectation(final)

network_configuration.set_next_expectation(interface_name, triggers=['O', 'Oui', 'OUI'])
interface_name.set_next_expectation(network_address)
network_address.set_next_expectation(network_netmask)
network_netmask.set_next_expectation(gateway)
network_configuration.set_next_expectation(zephir_address, triggers=['N', 'Non', 'NON'])

new_server.set_next_expectation(rne, triggers=['N', 'Non', 'NON'])
rne.set_next_expectation(server_id, triggers='')

zephir_address.set_next_expectation(zephir_admin)
zephir_admin.set_next_expectation(zephir_admin_password)

hardware.set_next_expectation(processor)
processor.set_next_expectation(harddrive)

Et s’utilise comme suit :

patterns = expectations.get_patterns()
nb_expectations = expectations.count_expectations()

instance_process = pexpect.spawn('/usr/bin/enregistrement_zephir', encoding='utf-8', timeout=60, logfile=sys.stdout)

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

La méthode get_patterns renvoie uniquement les premières question des blocs (les questions qu’on peut discriminer avec le contexte ou qui ne suivent pas obligatoirement une autre question). La méthode is_the_one est celle utilisant le contexte pour déterminer si la question est bien celle attendue. La méthode answer envoie la réponse avec pexpect et enchaîne éventuellement sur les autres questions du bloc, en tenant compte des déclencheurs.