Daniel Allaire
8251022148
site.fiche_pour(locataire) reunit attribution, offre, racine et les trois lectures directes de l instancier ; verifier_fiche la confronte aux copies, a la racine et a l inventaire du locataire. P84 : aucune ecart sur les deux couples ; sept alterations vues en ecart. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
594 lines
26 KiB
Python
594 lines
26 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`.
|
|
|
|
ETAPES 1 ET 2 DU CHEMIN (§6) : le module existe, il est teste, et RIEN NE L'UTILISE ENCORE.
|
|
La fiche du site (`site.fiche_pour(locataire)`) est calculee A COTE de l'existant, et
|
|
`verifier_fiche` prouve qu'elle dit exactement ce que le locataire porte aujourd'hui. La face
|
|
reseau du locataire suit. Les consommateurs basculeront a l'etape 3, a resultat identique.
|
|
|
|
Usage :
|
|
python3 scripts/contexte.py # le contexte actif, et ce que chaque indice dit
|
|
python3 scripts/contexte.py --json
|
|
python3 scripts/contexte.py --fiche OPS-Technolibre # la fiche que son hebergeur lui destine
|
|
python3 scripts/contexte.py --verifier-fiches # chaque fiche dit-elle ce que le locataire porte ?
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import contextlib
|
|
import hashlib
|
|
import json
|
|
import os
|
|
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 {}
|
|
|
|
|
|
def _empreinte(chemin: Path) -> str:
|
|
return hashlib.sha256(chemin.read_bytes()).hexdigest() if chemin.is_file() else ""
|
|
|
|
|
|
@contextlib.contextmanager
|
|
def _monter(site: "Site | None" = None, locataire: "Locataire | None" = None):
|
|
"""Monte CE site et CE locataire le temps d'interroger les modules d'avant.
|
|
|
|
`underlay`, `site_intrants`, `instancier` lisent « le site monte » et « l'instance
|
|
courante » (variables `SETOPS_*`, sinon les liens). Plutot que de recopier leurs calculs,
|
|
on leur designe l'ecosysteme voulu, puis on remet l'environnement tel quel. C'est une
|
|
passerelle de transition : a l'etape 3, ces modules recevront leurs objets.
|
|
"""
|
|
clefs = ("SETOPS_UNDERLAY", "SETOPS_INSTANCE")
|
|
avant = {k: os.environ.get(k) for k in clefs}
|
|
try:
|
|
if site is not None:
|
|
os.environ["SETOPS_UNDERLAY"] = str(site.underlay)
|
|
if locataire is not None:
|
|
os.environ["SETOPS_INSTANCE"] = str(locataire.depot)
|
|
yield
|
|
finally:
|
|
for k, v in avant.items():
|
|
if v is None:
|
|
os.environ.pop(k, None)
|
|
else:
|
|
os.environ[k] = v
|
|
|
|
|
|
# --- 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]
|
|
|
|
# LE CONTRAT, SENS SITE -> LOCATAIRE (§2.4) ---------------------------------------------
|
|
def fiche_pour(self, locataire: "Locataire") -> dict:
|
|
"""Ce que CE site attribue et offre a CE locataire, en une fiche.
|
|
|
|
Trois canaux le portaient jusqu'ici, sans fiche (relevé du 2026-10-04) :
|
|
- des COPIES a la main : index, adresse publique, intrants de service, racine ;
|
|
- une LECTURE DIRECTE du site par `instancier.py` au moment de generer l'inventaire :
|
|
delegation DNS, serveurs de temps, mode de routage ;
|
|
- le genome (hors fiche : il passe par la forge).
|
|
La fiche les reunit, calcules par LES MEMES fonctions qu'aujourd'hui. Elle porte
|
|
l'empreinte de ses sources : une fiche qui ne correspond plus a son site se voit.
|
|
"""
|
|
import site_intrants as si
|
|
with _monter(self, locataire):
|
|
contrat = si.contrat()
|
|
carte = underlay_mod.charger() or {}
|
|
intrants = underlay_mod.lire_plan_site("10-intrants.yml")
|
|
# Les trois lectures d'`instancier.py`, a l'identique (`_zones_deleguees`, la patte
|
|
# de transit de la frontiere, `routage_tenants`).
|
|
zone = str(intrants.get("domaine_interne") or "").strip()
|
|
resolveurs = [a for a in underlay_mod.adresses_site_portant("serveur_resolveur") if a]
|
|
transit = underlay_mod.reseau_transit(carte) or {}
|
|
reseaux_transit = {str(transit.get("nom"))} if transit.get("nom") else set()
|
|
temps = sorted({str(h["ip"]) for h in underlay_mod.hotes(carte)
|
|
if h.get("role") == "frontiere" and h.get("ip")
|
|
and str(h.get("reseau")) in reseaux_transit
|
|
and str(h.get("etat", "actif")) != "reserve"})
|
|
routage = underlay_mod.routage_tenants(carte)
|
|
ip_publique = si.ip_publique_de(locataire.nom)
|
|
racine = self.depot / "ac-racine-site.crt"
|
|
offre = {k: v for k, v in contrat.items() if k not in ("ip_publique", "forge_amont_ac")}
|
|
return {
|
|
"site": self.nom,
|
|
"locataire": locataire.nom,
|
|
"sources": {f: _empreinte(self.depot / f) for f in
|
|
("underlay.yml", "plan/10-intrants.yml", "opnsense.yml", "ac-racine-site.crt")},
|
|
"attribue": {"index": self.allocations().get(locataire.nom), "ip_publique": ip_publique},
|
|
"offre": offre,
|
|
"racine_confiance": {"fichier": racine.name, "sha256": _empreinte(racine),
|
|
"pem": racine.read_text(encoding="utf-8") if racine.is_file() else ""},
|
|
"generation": {"zone_site": zone, "resolveurs_site": resolveurs,
|
|
"serveurs_temps": temps, "routage_tenants": routage},
|
|
}
|
|
|
|
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 domaine_interne(self) -> str:
|
|
"""Son domaine interne : le premier `domaine_interne` non vide de `group_vars/all`,
|
|
dans l'ordre des fichiers — la lecture meme d'`instancier._domaine_interne`."""
|
|
base = self.dossier_inventaire() / "group_vars" / "all"
|
|
for fichier in sorted(base.glob("*.yml")) if base.is_dir() else []:
|
|
try:
|
|
d = yaml.safe_load(fichier.read_text(encoding="utf-8")) or {}
|
|
except yaml.YAMLError:
|
|
continue
|
|
if isinstance(d, dict) and str(d.get("domaine_interne", "")).strip():
|
|
return str(d["domaine_interne"]).strip()
|
|
return ""
|
|
|
|
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
|
|
|
|
|
|
# --- LA FICHE DIT-ELLE CE QUE LE LOCATAIRE PORTE ? ----------------------------------------
|
|
|
|
def _hostvars(inventaire: Path) -> dict:
|
|
d = _yaml(inventaire)
|
|
hv: dict = {}
|
|
|
|
def parcourir(g: dict) -> None:
|
|
for h, v in (g.get("hosts") or {}).items():
|
|
hv.setdefault(h, {}).update(v or {})
|
|
for c in (g.get("children") or {}).values():
|
|
parcourir(c or {})
|
|
parcourir(d.get("all") or {})
|
|
return hv
|
|
|
|
|
|
def verifier_fiche(site: "Site", locataire: "Locataire", fiche: dict | None = None) -> list[str]:
|
|
"""Les ecarts entre la fiche du site et ce que le locataire porte AUJOURD'HUI.
|
|
|
|
Vide = la fiche dit exactement ce que les trois canaux d'avant ont depose chez le locataire :
|
|
ses copies (intrants, index, adresse publique, racine) et son inventaire genere (temps,
|
|
delegation DNS, rattachement SDN). C'est la condition pour que l'instancier puisse, a
|
|
l'etape 3, lire la fiche au lieu du site sans que l'inventaire change d'un octet.
|
|
"""
|
|
import site_intrants as si
|
|
f = fiche if fiche is not None else site.fiche_pour(locataire)
|
|
ecarts: list[str] = []
|
|
|
|
# 1. L'attribution.
|
|
if f["attribue"]["index"] != locataire.index:
|
|
ecarts.append(f"index : le site attribue {f['attribue']['index']}, "
|
|
f"le locataire porte {locataire.index}")
|
|
|
|
# 2. L'offre et l'adresse publique, contre les copies du locataire (la table de `site_intrants`).
|
|
with _monter(site, locataire):
|
|
vu = si._declare_par_le_locataire()
|
|
valeurs = dict(f["offre"], ip_publique=f["attribue"]["ip_publique"])
|
|
for clef, copie in vu.items():
|
|
attendu = valeurs.get(clef)
|
|
if not attendu or "{{" in str(copie):
|
|
continue
|
|
if si._normaliser(copie) != si._normaliser(attendu):
|
|
fichier, variable = si.OU_LE_LOCATAIRE_LE_DIT[clef]
|
|
ecarts.append(f"{variable} ({fichier}) : le locataire dit « {copie} », la fiche « {attendu} »")
|
|
|
|
# 3. La racine de confiance.
|
|
copie_ac = locataire.depot / "ac-racine-site.crt"
|
|
if copie_ac.is_file() and _empreinte(copie_ac) != f["racine_confiance"]["sha256"]:
|
|
ecarts.append("ac-racine-site.crt : la copie du locataire n'est pas la racine du site")
|
|
|
|
# 4. Ce que l'instancier a lu dans le site, tel qu'il l'a ecrit dans l'inventaire.
|
|
hv = _hostvars(locataire.inventaire())
|
|
if not hv:
|
|
ecarts.append(f"inventaire introuvable ou vide : {locataire.inventaire()}")
|
|
return ecarts
|
|
gen = f["generation"]
|
|
temps = gen["serveurs_temps"] or None
|
|
for h, v in sorted(hv.items()):
|
|
if v.get("chrony_serveurs") != temps:
|
|
ecarts.append(f"{h} : chrony_serveurs = {v.get('chrony_serveurs')}, la fiche dit {temps}")
|
|
break
|
|
domaine_loc = locataire.domaine_interne()
|
|
attendu_zones = ([{"adresses": gen["resolveurs_site"], "nom": gen["zone_site"]}]
|
|
if gen["zone_site"] and gen["resolveurs_site"] and gen["zone_site"] != domaine_loc
|
|
else [])
|
|
portees = [v["serveur_resolveur_zones_deleguees"] for v in hv.values()
|
|
if "serveur_resolveur_zones_deleguees" in v]
|
|
normal = [sorted(({"adresses": list(z.get("adresses") or []), "nom": z.get("nom")} for z in p),
|
|
key=lambda z: str(z["nom"])) for p in portees]
|
|
if attendu_zones and not normal:
|
|
ecarts.append("delegation DNS : la fiche en offre une, l'inventaire n'en porte aucune")
|
|
for p in normal:
|
|
if p != attendu_zones and (attendu_zones or p):
|
|
ecarts.append(f"delegation DNS : l'inventaire porte {p}, la fiche dit {attendu_zones}")
|
|
break
|
|
sdn = gen["routage_tenants"] == "sdn"
|
|
for h, v in sorted(hv.items()):
|
|
if sdn and (not v.get("proxmox_pont") or v.get("proxmox_etiquette_vlan")):
|
|
ecarts.append(f"{h} : routage SDN selon la fiche, mais pont « {v.get('proxmox_pont')} », "
|
|
f"etiquette « {v.get('proxmox_etiquette_vlan')} »")
|
|
break
|
|
if not sdn and v.get("proxmox_pont"):
|
|
ecarts.append(f"{h} : routage « {gen['routage_tenants']} » selon la fiche, mais un pont SDN")
|
|
break
|
|
return ecarts
|
|
|
|
|
|
def couples_hebergeur() -> list[tuple["Site", "Locataire"]]:
|
|
"""Chaque (site, locataire) dont ce site est l'hebergeur ACTIF — les seuls qui ont une fiche
|
|
a verifier : un site de reprise ne depose rien tant que la bascule n'a pas eu lieu."""
|
|
out = []
|
|
for underlay in sorted(DOSSIER_ECOSYSTEMES.glob("*/underlay.yml")):
|
|
site = Site(underlay.parent)
|
|
out += [(site, loc) for loc in site.locataires() if loc.nom_hebergeur() == site.nom]
|
|
return out
|
|
|
|
|
|
# --- 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")
|
|
ap.add_argument("--fiche", metavar="LOCATAIRE", help="la fiche que son hebergeur lui destine")
|
|
ap.add_argument("--verifier-fiches", action="store_true",
|
|
help="chaque fiche dit-elle ce que le locataire porte ?")
|
|
args = ap.parse_args(argv)
|
|
if args.fiche:
|
|
loc = Locataire.charger(args.fiche)
|
|
site = loc.site()
|
|
if site is None:
|
|
print(f"{loc.nom} ne nomme aucun hebergeur.")
|
|
return 1
|
|
print(yaml.safe_dump(site.fiche_pour(loc), allow_unicode=True, sort_keys=False), end="")
|
|
return 0
|
|
if args.verifier_fiches:
|
|
couples = couples_hebergeur()
|
|
if not couples:
|
|
print("Aucun site frere : aucune fiche a verifier.")
|
|
return 0
|
|
total = 0
|
|
for site, loc in couples:
|
|
e = verifier_fiche(site, loc)
|
|
total += len(e)
|
|
print(f"{'OK ' if not e else 'ECART'} {site.nom} -> {loc.nom}"
|
|
+ "".join(f"\n - {x}" for x in e))
|
|
return 1 if total else 0
|
|
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())
|