02.05.07 — `TestCycle[Driver]` + modes#

Fichier source : src/ocarina/dsl/testing/oc_test_cycle.py
Responsabilité unique : orchestrer les campagnes smoke puis main, et appliquer un mode de gestion d’échec des smoke.

Deux modes#

type Mode = Literal[
    "fail-fast-on-first-smoke-campaigns-sequence-fail",
    "wait-for-all-smoke-tests",
]

Mode	Comportement
`fail-fast-on-first-smoke-campaigns-sequence-fail` (par défaut)	Dès qu’une campagne de smoke fail, les suivantes sont skippées.
`wait-for-all-smoke-tests`	Toutes les campagnes de smoke s’exécutent. Si au moins une fail, les campagnes main sont skippées.

Cas	Mode
Smoke par dépendance (par exemple : “login” puis “dashboard accessible”)	`fail-fast` (si login échoue, dashboard est inutile)
Smoke parallèles (par exemple : “homepage” + “API ping”)	`wait-for-all` (on veut voir les deux)

Le default est fail-fast parce que c’est le cas le plus fréquent.

Code#

@final
class TestCycle[Driver]:
    def __init__(
        self,
        *,
        name: str,
        campaigns: Sequence[TestCampaign[Driver]],
        smoke_tests_campaigns: Sequence[TestCampaign[Driver]] | None = None,
        mode: Mode | None = None,
    ) -> None:
        if mode is None:
            mode = "fail-fast-on-first-smoke-campaigns-sequence-fail"

        smoke_campaigns = smoke_tests_campaigns or []
        all_campaigns = [*smoke_campaigns, *campaigns]

        for campaign in all_campaigns:
            for suite in campaign._suites:
                chain_validations(
                    validate_test_cycle_name(cycle_name=name, name="test_cycle"),
                    validate_campaigns_names(campaigns=all_campaigns, name="all campaigns (smoke + deep tests)"),
                    validate_test_suites_names(suites=campaign._suites, name="suites"),
                    validate_test_runners_names(tests=suite._tests, name="tests"),
                ).execute().raise_if_invalid()

        self.name = name
        self._campaigns = campaigns
        self._smoke_campaigns = smoke_campaigns
        self._results: TestCycleResults = {}
        self._mode = mode

        # Injection du nom de cycle dans toutes les suites
        for campaign in self._campaigns:
            for suite in campaign._suites:
                suite._cycle_name = name
        for smoke_campaign in self._smoke_campaigns:
            for suite in smoke_campaign._suites:
                suite._cycle_name = name


    def run_all(self, *, saturate_workers: bool = True) -> TestCycleResults:
        self._results.clear()

        def _run_smoke(*, fail_fast: bool) -> bool:
            failed = False
            for campaign in self._smoke_campaigns:
                results = campaign.run_all(
                    skip_all=failed if fail_fast else False,
                    saturate_workers=saturate_workers,
                )
                self._results[campaign.name] = results
                if campaign_has_failed(results):
                    failed = True
            return failed

        dispatch: dict[Mode, Thunk[bool]] = {
            "fail-fast-on-first-smoke-campaigns-sequence-fail":
                lambda: _run_smoke(fail_fast=True),
            "wait-for-all-smoke-tests":
                lambda: _run_smoke(fail_fast=False),
        }

        skip_all = dispatch[self._mode]()

        for campaign in self._campaigns:
            self._results[campaign.name] = campaign.run_all(
                skip_all=skip_all, saturate_workers=saturate_workers
            )

        return self._results


def has_test_cycle_failed(results: TestCycleResults) -> bool:
    return any(
        is_test_result_fail(outcome)
        for campaigns in results.values()
        for tests in campaigns.values()
        for outcome, _, _ in tests.values()
    )

Invariants de construction#

chain_validations(
    validate_test_cycle_name(cycle_name=name, name="test_cycle"),
    validate_campaigns_names(campaigns=all_campaigns, name="..."),
    validate_test_suites_names(suites=campaign._suites, name="suites"),
    validate_test_runners_names(tests=suite._tests, name="tests"),
).execute().raise_if_invalid()

name est un nom de cycle valide (cross-OS filename).
Noms uniques + valides parmi toutes les campagnes (smoke + main).
Noms uniques + valides parmi les suites de chaque campagne.
Noms uniques + valides parmi les tests de chaque suite.

Depuis 1.1.10, l’unicité des noms aux trois niveaux est insensible à la casse (NFC + casefold).

Levé à la construction.
Si l’utilisateur a une collision ou un nommage invalide, il l’apprend immédiatement.

Sélection de mode via dispatch table#

dispatch: dict[Mode, Thunk[bool]] = {
    "fail-fast-on-first-smoke-campaigns-sequence-fail":
        lambda: _run_smoke(fail_fast=True),
    "wait-for-all-smoke-tests":
        lambda: _run_smoke(fail_fast=False),
}

skip_all = dispatch[self._mode]()

Pattern fonctionnel : on indexe une table de thunks par la valeur du Literal. Le checker peut vérifier l’exhaustivité de la table par rapport au Literal (Mode = Literal["a", "b"] → table doit avoir les 2 clés).

⚠️ Limite Python actuelle : à l’heure où l’on écrit ces lignes, le type checker Python ne vérifie pas réellement l’exhaustivité de ce pattern (ce qu’on aurait en TypeScript avec un Record<Mode, ...> exhaustif). C’est une limitation acceptée, à documenter, pas une feature à corriger. Si on ajoute un membre au Literal et qu’on oublie de l’ajouter au dispatch, mypy ne le criera pas (c’est ballot). De plus, l’idiome match de Python a le même comportement non exhaustif. :'(

Flux complet d’un cycle#

TestCycle.run_all(saturate_workers=True)
       │
       ▼
   dispatch[mode]()                           ▶︎  _run_smoke(fail_fast=<bool>)
       │
       │  for campaign in self._smoke_campaigns :
       │     results = campaign.run_all(skip_all=failed if fail_fast else False, ...)
       │           │
       │           ▼
       │     pour chaque suite :
       │       suite.run(max_workers=...) → dict[test_name -> TestSuiteResult]
       │
       │     self._results[campaign.name] = results
       │     if campaign_has_failed(results) :
       │         failed = True
       │
       └─ return failed   →  skip_all
       │
       ▼
   for campaign in self._campaigns :
       results = campaign.run_all(skip_all=skip_all, ...)
       self._results[campaign.name] = results
       │
       ▼
   return self._results

`has_test_cycle_failed`#

def has_test_cycle_failed(results: TestCycleResults) -> bool:
    return any(
        is_test_result_fail(outcome)
        for campaigns in results.values()
        for tests in campaigns.values()
        for outcome, _, _ in tests.values()
    )

results.values() : itère sur les TestCampaignResults (un par campagne).
campaigns.values() : itère sur les TestSuiteResults (un par suite).
tests.values() : itère sur les TestSuiteResult (un par test).

Court-circuit sur le premier Fail trouvé.

C’est ce que le main.py de l’utilisateur appelle dans son post_exec :

def _post_exec(results: TestCycleResults) -> None:
    print()
    pretty_print_results(results, with_colors=True)
    if has_test_cycle_failed(results):
        sys.exit(1)

Pourquoi `mode` est facultatif#

Le default "fail-fast-..." couvre 80% des cas.

L’exemple ocarina-example utilise wait-for-all-smoke-tests parce qu’il a deux campagnes smoke (global_smoke_tests + corsicamon_smoke) qui sont indépendantes, on veut voir les deux fail si elles fail.

L’exemple ocarina-with-ai-example reste sur le default (fail-fast) parce qu’il n’a qu’une campagne smoke (prerequisites).

02.05.07 — TestCycle[Driver] + modes#