Daniel Allaire
c891d7fb26
scripts/contexte.py : Ecosysteme, Site, Locataire, contexte actif (fichier `contexte`, sinon les indices d avant via instance_courante et underlay.chemin). 40 controles dans make test. Corrige aussi P34 et P48, rouges depuis le commit de la page de conception. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
383 lines
15 KiB
Python
383 lines
15 KiB
Python
#!/usr/bin/env python3
|
|
"""Le contexte d'un ecosysteme : un tronc commun, deux classes (SITE et LOCATAIRE).
|
|
|
|
Conception : `docs/conception-contextes.md` (arretee avec l'exploitant le 2026-10-04).
|
|
|
|
POURQUOI CE MODULE. Le 2026-10-04, 33 scripts DEVINAIENT leur contexte, chacun a sa facon :
|
|
le lien `instance/`, le lien `underlay.yml`, les variables `SETOPS_*`, les dossiers freres,
|
|
les `SITE-*`. Les indices ne concordaient pas toujours, et chaque desaccord avait deja produit
|
|
un defaut silencieux (la frontiere d'un site recevant les regles de l'autre, la console d'un
|
|
site affichant zero machine, `make ci` melangeant le modele public et les ecosystemes reels).
|
|
|
|
Ce module est le SEUL endroit ou se lisent ces indices. Les scripts lui demandent « ou
|
|
suis-je ? » et recoivent un objet : un `Site` ou un `Locataire`, qui heritent d'`Ecosysteme`.
|
|
|
|
ETAPE 1 DU CHEMIN (§6) : le module existe, il est teste, et RIEN NE L'UTILISE ENCORE. Les
|
|
scripts l'adopteront un par un, chacun a resultat identique. Les fiches du contrat
|
|
(`site.fiche_pour`, `locataire.face_reseau`) sont l'etape 2.
|
|
|
|
Usage :
|
|
python3 scripts/contexte.py # le contexte actif, et ce que chaque indice dit
|
|
python3 scripts/contexte.py --json
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parent))
|
|
import yaml # noqa: E402
|
|
|
|
import underlay as underlay_mod # noqa: E402
|
|
import voutes as voutes_mod # noqa: E402
|
|
from inventory_rules import ORDRE_INVENTAIRE, dossier_inventaire, instance_courante # noqa: E402
|
|
|
|
RACINE = Path(__file__).resolve().parents[1]
|
|
# Les ecosystemes vivent a cote du moteur : des dossiers freres, pas de registre.
|
|
DOSSIER_ECOSYSTEMES = RACINE.parent
|
|
# Le contexte actif, nomme (decision du 2026-10-04) : une seule valeur, `site:<depot>` ou
|
|
# `locataire:<depot>`. Absent, on retombe sur les liens d'avant, le temps de la bascule.
|
|
NOM_FICHIER_CONTEXTE = "contexte"
|
|
NATURES = ("site", "locataire")
|
|
|
|
|
|
class ContexteInvalide(Exception):
|
|
"""Un contexte nomme qui ne designe rien, ou un depot qui ne dit pas ce qu'il est."""
|
|
|
|
|
|
def _yaml(chemin: Path) -> dict:
|
|
if not chemin.is_file():
|
|
return {}
|
|
return yaml.safe_load(chemin.read_text(encoding="utf-8")) or {}
|
|
|
|
|
|
# --- LE TRONC COMMUN ----------------------------------------------------------------------
|
|
|
|
class Ecosysteme:
|
|
"""Ce que tout ecosysteme possede, quel que soit son contexte (§2.1).
|
|
|
|
Les methodes qui dependent du contexte sont SURCHARGEES par `Site` et `Locataire` ; ici
|
|
elles refusent, pour qu'un oubli de surcharge se voie au premier appel.
|
|
"""
|
|
|
|
nature = "ecosysteme"
|
|
|
|
def __init__(self, depot: Path):
|
|
self.depot = Path(depot)
|
|
|
|
# Identite -------------------------------------------------------------------------
|
|
@property
|
|
def nom(self) -> str:
|
|
return self.depot.name
|
|
|
|
@property
|
|
def plan(self) -> Path:
|
|
return self.depot / "plan"
|
|
|
|
def lire_plan(self, fichier: str) -> dict:
|
|
"""Un fichier du plan, ou {} s'il n'existe pas."""
|
|
return _yaml(self.plan / fichier)
|
|
|
|
# Voute : un depot, une cle (2026-08-28) ---------------------------------------------
|
|
@property
|
|
def cle_voute(self) -> Path:
|
|
"""La cle qui ouvre SA voute, nommee d'apres le depot (`voutes.cle_de`)."""
|
|
return voutes_mod.cle_de(self.nom)
|
|
|
|
@property
|
|
def voute(self) -> Path:
|
|
raise NotImplementedError(f"{type(self).__name__} : `voute` non surchargee")
|
|
|
|
def a_sa_voute(self) -> bool:
|
|
return self.voute.is_file()
|
|
|
|
# Filiation ----------------------------------------------------------------------------
|
|
@property
|
|
def filiation(self) -> dict:
|
|
"""`parente.yml` : de quels depots (moteur, hebergeur, modeles) il descend."""
|
|
return (_yaml(self.depot / "parente.yml").get("parente")) or {}
|
|
|
|
# Ce que chaque contexte surcharge ------------------------------------------------------
|
|
@property
|
|
def index(self) -> int | None:
|
|
raise NotImplementedError(f"{type(self).__name__} : `index` non surcharge")
|
|
|
|
def inventaire(self) -> Path:
|
|
raise NotImplementedError(f"{type(self).__name__} : `inventaire` non surcharge")
|
|
|
|
# Commodites ----------------------------------------------------------------------------
|
|
def __repr__(self) -> str:
|
|
return f"{type(self).__name__}({self.nom!r})"
|
|
|
|
def __eq__(self, autre: object) -> bool:
|
|
return (isinstance(autre, Ecosysteme) and type(autre) is type(self)
|
|
and autre.depot.resolve() == self.depot.resolve())
|
|
|
|
def __hash__(self) -> int:
|
|
return hash((type(self).__name__, str(self.depot.resolve())))
|
|
|
|
def decrire(self) -> dict:
|
|
return {"nature": self.nature, "nom": self.nom, "depot": str(self.depot),
|
|
"index": self.index, "voute": str(self.voute),
|
|
"a_sa_voute": self.a_sa_voute(), "cle_voute": str(self.cle_voute)}
|
|
|
|
# Reconnaitre un depot ------------------------------------------------------------------
|
|
@staticmethod
|
|
def depuis(depot: Path) -> "Ecosysteme":
|
|
"""Le `Site` ou le `Locataire` que porte ce depot, d'apres CE QU'IL DECLARE.
|
|
|
|
Un site declare un `underlay.yml` ; un locataire, une nomenclature. Un depot qui
|
|
declare les deux (l'ancien modele `socle`) ne dit pas ce qu'il est : refuse, plutot
|
|
que de choisir pour lui.
|
|
"""
|
|
depot = Path(depot)
|
|
est_site = (depot / "underlay.yml").is_file()
|
|
est_locataire = (depot / "plan" / "nomenclature.yml").is_file()
|
|
if est_site and est_locataire:
|
|
raise ContexteInvalide(
|
|
f"{depot.name} porte un `underlay.yml` ET une nomenclature : un site ou un "
|
|
f"locataire, pas les deux.")
|
|
if est_site:
|
|
return Site(depot)
|
|
if est_locataire:
|
|
return Locataire(depot)
|
|
raise ContexteInvalide(
|
|
f"{depot} n'est ni un site (pas d'`underlay.yml`) ni un locataire (pas de "
|
|
f"`plan/nomenclature.yml`).")
|
|
|
|
|
|
# --- SITE ---------------------------------------------------------------------------------
|
|
|
|
class Site(Ecosysteme):
|
|
"""L'hebergeur : le materiel, la fabric, la frontiere, et ses locataires (§2.2)."""
|
|
|
|
nature = "site"
|
|
|
|
@classmethod
|
|
def charger(cls, nom: str, dossier: Path = DOSSIER_ECOSYSTEMES) -> "Site":
|
|
depot = Path(dossier) / nom
|
|
if not (depot / "underlay.yml").is_file():
|
|
raise ContexteInvalide(f"site « {nom} » introuvable : pas de {depot}/underlay.yml")
|
|
return cls(depot)
|
|
|
|
@property
|
|
def underlay(self) -> Path:
|
|
return self.depot / "underlay.yml"
|
|
|
|
def carte(self) -> dict:
|
|
"""Le contenu de `underlay:`."""
|
|
return _yaml(self.underlay).get("underlay") or {}
|
|
|
|
@property
|
|
def voute(self) -> Path:
|
|
return self.depot / "underlay.vault.yml"
|
|
|
|
@property
|
|
def index(self) -> int | None:
|
|
i = self.carte().get("index")
|
|
return i if isinstance(i, int) and not isinstance(i, bool) else None
|
|
|
|
def inventaire(self) -> Path:
|
|
"""Un site n'a pas de `hosts.yml` : son inventaire est un SCRIPT (`site_inventaire.py`)."""
|
|
return RACINE / "scripts" / "site_inventaire.py"
|
|
|
|
def allocations(self) -> dict[str, int]:
|
|
"""L'index que CE site attribue a chacun de ses locataires (`underlay.tenants`)."""
|
|
return underlay_mod.allocations({"underlay": self.carte()})
|
|
|
|
def noms_locataires(self) -> list[str]:
|
|
"""Les locataires que ce site declare, sous l'une ou l'autre forme de `tenants`."""
|
|
v = self.carte().get("tenants")
|
|
if isinstance(v, dict):
|
|
return sorted(str(n) for n in v)
|
|
if isinstance(v, list):
|
|
return sorted(str(n) for n in v)
|
|
return []
|
|
|
|
def locataires(self) -> list["Locataire"]:
|
|
"""Ses locataires, resolus en objets. Un nom sans depot frere n'est PAS rendu ici :
|
|
`locataires_absents()` le dit — un filtre qui oublie en silence est un defaut."""
|
|
out = []
|
|
for nom in self.noms_locataires():
|
|
depot = self.depot.parent / nom
|
|
if (depot / "plan" / "nomenclature.yml").is_file():
|
|
out.append(Locataire(depot))
|
|
return out
|
|
|
|
def locataires_absents(self) -> list[str]:
|
|
presents = {l.nom for l in self.locataires()}
|
|
return [n for n in self.noms_locataires() if n not in presents]
|
|
|
|
def decrire(self) -> dict:
|
|
d = super().decrire()
|
|
d.update({"locataires": self.noms_locataires(),
|
|
"locataires_absents": self.locataires_absents()})
|
|
return d
|
|
|
|
|
|
# --- LOCATAIRE ----------------------------------------------------------------------------
|
|
|
|
class Locataire(Ecosysteme):
|
|
"""Une organisation hebergee : son plan de services, son inventaire genere (§2.3).
|
|
|
|
UN LOCATAIRE PEUT FIGURER CHEZ DEUX SITES : son hebergeur actif, que nomme `parente.yml`,
|
|
et un site de reprise qui le declare aussi (SITE-Technolibre declare les deux locataires de
|
|
SITE-Chezlepro). `site()` rend l'hebergeur ACTIF ; les autres le connaissent par leur
|
|
propre `locataires()`.
|
|
"""
|
|
|
|
nature = "locataire"
|
|
|
|
@classmethod
|
|
def charger(cls, nom: str, dossier: Path = DOSSIER_ECOSYSTEMES) -> "Locataire":
|
|
depot = Path(dossier) / nom
|
|
if not (depot / "plan" / "nomenclature.yml").is_file():
|
|
raise ContexteInvalide(
|
|
f"locataire « {nom} » introuvable : pas de {depot}/plan/nomenclature.yml")
|
|
return cls(depot)
|
|
|
|
def nomenclature(self) -> dict:
|
|
return self.lire_plan("nomenclature.yml")
|
|
|
|
@property
|
|
def index(self) -> int | None:
|
|
i = self.nomenclature().get("index")
|
|
return i if isinstance(i, int) and not isinstance(i, bool) else None
|
|
|
|
def dossier_inventaire(self) -> Path:
|
|
return dossier_inventaire(self.depot, ORDRE_INVENTAIRE)
|
|
|
|
def inventaire(self) -> Path:
|
|
"""Son `hosts.yml` GENERE du plan. Le chemin est rendu meme s'il n'existe pas encore
|
|
(instance neuve) : c'est `instancier` qui l'ecrira."""
|
|
return self.dossier_inventaire() / "hosts.yml"
|
|
|
|
@property
|
|
def voute(self) -> Path:
|
|
return self.dossier_inventaire() / "group_vars" / "all" / "vault.yml"
|
|
|
|
def nom_hebergeur(self) -> str | None:
|
|
h = ((self.filiation.get("depots") or {}).get("hebergeur") or {}).get("nom")
|
|
return str(h) if h else None
|
|
|
|
def site(self) -> "Site | None":
|
|
"""Son hebergeur ACTIF. None s'il n'en nomme aucun ; refus s'il en nomme un absent."""
|
|
nom = self.nom_hebergeur()
|
|
if not nom:
|
|
return None
|
|
return Site.charger(nom, self.depot.parent)
|
|
|
|
def decrire(self) -> dict:
|
|
d = super().decrire()
|
|
d.update({"inventaire": str(self.inventaire()), "hebergeur": self.nom_hebergeur()})
|
|
return d
|
|
|
|
|
|
# --- LE CONTEXTE ACTIF --------------------------------------------------------------------
|
|
|
|
def _lire_fichier_contexte(fichier: Path, dossier: Path) -> Ecosysteme:
|
|
lignes = [l.split("#", 1)[0].strip()
|
|
for l in fichier.read_text(encoding="utf-8").splitlines()]
|
|
valeurs = [l for l in lignes if l]
|
|
if len(valeurs) != 1 or ":" not in valeurs[0]:
|
|
raise ContexteInvalide(
|
|
f"{fichier} doit porter UNE valeur, `site:<depot>` ou `locataire:<depot>` "
|
|
f"(lu : {valeurs or 'rien'}).")
|
|
nature, nom = (x.strip() for x in valeurs[0].split(":", 1))
|
|
if nature == "site":
|
|
return Site.charger(nom, dossier)
|
|
if nature == "locataire":
|
|
return Locataire.charger(nom, dossier)
|
|
raise ContexteInvalide(f"{fichier} : nature « {nature} » inconnue ({' ou '.join(NATURES)}).")
|
|
|
|
|
|
def montes(racine: Path = RACINE) -> list[Ecosysteme]:
|
|
"""Ce que les INDICES D'AVANT designent : variables `SETOPS_*`, puis liens a la racine.
|
|
|
|
Garde le temps de la bascule, et pour la dire : sur le poste d'aujourd'hui, cette liste
|
|
rend DEUX ecosystemes, ce qu'un selecteur ne doit plus faire.
|
|
"""
|
|
# LES DEUX RESOLUTIONS QUI FONT DEJA FOI, pas une troisieme copie (P41) :
|
|
# `instance_courante()` lit `SETOPS_INSTANCE`, puis le lien `instance/` ; `underlay.chemin()`
|
|
# lit `SETOPS_UNDERLAY`, puis le lien `underlay.yml`. Quand on interroge une AUTRE racine que
|
|
# celle du moteur (les tests), leurs liens par defaut sont remplaces par ceux de cette racine.
|
|
racine = Path(racine)
|
|
autre_racine = racine.resolve() != RACINE.resolve()
|
|
out: list[Ecosysteme] = []
|
|
inst_p = instance_courante()
|
|
if autre_racine and inst_p == RACINE.joinpath("instance"):
|
|
inst_p = racine.joinpath("instance")
|
|
if not inst_p.is_absolute():
|
|
inst_p = racine / inst_p
|
|
if (inst_p / "plan" / "nomenclature.yml").is_file():
|
|
out.append(Locataire(inst_p.resolve()))
|
|
und_p = underlay_mod.chemin()
|
|
if autre_racine and (und_p is None or und_p == RACINE.joinpath("underlay.yml")):
|
|
und_p = racine.joinpath("underlay.yml")
|
|
if und_p is not None and not und_p.is_absolute():
|
|
und_p = racine / und_p
|
|
if und_p is not None and und_p.is_file():
|
|
out.append(Site(und_p.resolve().parent))
|
|
return out
|
|
|
|
|
|
def actif(racine: Path = RACINE, dossier: Path | None = None) -> Ecosysteme | None:
|
|
"""LE contexte actif de cette machine.
|
|
|
|
1. Le fichier `contexte` a la racine du moteur, s'il existe : il fait foi.
|
|
2. Sinon, les indices d'avant (`montes`). S'ils designent un locataire ET un site — le
|
|
poste d'aujourd'hui —, le locataire l'emporte : c'est ce que fait deja la console
|
|
(`ConsolePoste` herite de `ConsoleLocataire`). Le fichier `contexte` leve l'ambiguite.
|
|
3. Rien : None. A l'appelant de le dire, jamais de le dessiner comme un plan vide.
|
|
"""
|
|
dossier = Path(dossier) if dossier is not None else Path(racine).parent
|
|
fichier = Path(racine) / NOM_FICHIER_CONTEXTE
|
|
if fichier.is_file():
|
|
return _lire_fichier_contexte(fichier, dossier)
|
|
m = montes(racine)
|
|
return m[0] if m else None
|
|
|
|
|
|
def main(argv: list[str] | None = None) -> int:
|
|
ap = argparse.ArgumentParser(description=__doc__.splitlines()[0])
|
|
ap.add_argument("--json", action="store_true")
|
|
args = ap.parse_args(argv)
|
|
try:
|
|
eco = actif()
|
|
erreur = None
|
|
except ContexteInvalide as e:
|
|
eco, erreur = None, str(e)
|
|
fichier = RACINE / NOM_FICHIER_CONTEXTE
|
|
rapport = {
|
|
"source": "fichier `contexte`" if fichier.is_file() else "indices d'avant (liens, variables)",
|
|
"actif": eco.decrire() if eco else None,
|
|
"erreur": erreur,
|
|
"indices": [e.decrire() for e in montes()],
|
|
}
|
|
if args.json:
|
|
print(json.dumps(rapport, indent=2, ensure_ascii=False))
|
|
return 0 if eco else 1
|
|
print(f"Source : {rapport['source']}")
|
|
if erreur:
|
|
print(f"REFUS : {erreur}")
|
|
elif eco:
|
|
d = eco.decrire()
|
|
print(f"Contexte actif : {d['nature']} {d['nom']} (index {d['index']})")
|
|
if isinstance(eco, Site):
|
|
print(f" locataires : {', '.join(d['locataires']) or 'aucun'}"
|
|
+ (f" ; ABSENTS : {', '.join(d['locataires_absents'])}" if d['locataires_absents'] else ""))
|
|
else:
|
|
print(f" hebergeur actif : {d['hebergeur'] or 'aucun nomme'}")
|
|
else:
|
|
print("Aucun contexte : ni fichier `contexte`, ni locataire, ni site monte.")
|
|
if len(rapport["indices"]) > 1:
|
|
print(" NOTE : les indices d'avant designent "
|
|
+ " ET ".join(f"{i['nature']} {i['nom']}" for i in rapport["indices"])
|
|
+ " — un selecteur n'en garde qu'un (fichier `contexte`).")
|
|
return 0 if eco else 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|