SOURCE CODE PYTEST_HELM_CHARTS.FLUX.FIXTURES DOCS

from typing import Iterable, List

import pykube
import pytest

from pytest_helm_charts.k8s.deployment import wait_for_deployments_to_run
from pytest_helm_charts.k8s.fixtures import NamespaceFactoryFunc
from pytest_helm_charts.clusters import Cluster
from pytest_helm_charts.flux.git_repository import (
    GitRepositoryCR,
    GitRepositoryFactoryFunc,
    git_repository_factory_func,
)
from pytest_helm_charts.flux.helm_release import HelmReleaseCR, HelmReleaseFactoryFunc, helm_release_factory_func
from pytest_helm_charts.flux.helm_repository import (
    HelmRepositoryCR,
    HelmRepositoryFactoryFunc,
    helm_repository_factory_func,
)
from pytest_helm_charts.flux.kustomization import KustomizationCR, KustomizationFactoryFunc, kustomization_factory_func
from pytest_helm_charts.utils import delete_and_wait_for_objects

FLUX_NAMESPACE_NAME = "default"
FLUX_DEPLOYMENTS_READY_TIMEOUT: int = 180


@pytest.fixture(scope="module")DOCS
def flux_deployments(kube_cluster: Cluster) -> List[pykube.Deployment]:
    """
    Waits (blocks) until all the Deployments used to deploy Flux itself run
    (a.k.a. "wait for Flux to be ready for use").

    Args:
        kube_cluster: Cluster to use to connect to the k8s cluster

    Returns: A List of `pykube.Deployment` objects used by Flux itself.

    """
    deployments = wait_for_deployments_to_run(
        kube_cluster.kube_client,
        [
            "helm-controller",
            "image-automation-controller",
            "image-reflector-controller",
            "kustomize-controller",
            "notification-controller",
            "source-controller",
        ],
        FLUX_NAMESPACE_NAME,
        FLUX_DEPLOYMENTS_READY_TIMEOUT,
    )
    return deployments


@pytest.fixture(scope="function")DOCS
def kustomization_factory_function_scope(
    kube_cluster: Cluster, namespace_factory: NamespaceFactoryFunc
) -> Iterable[KustomizationFactoryFunc]:
    """Returns function-scoped [Kustomization](https://fluxcd.io/docs/components/kustomize/kustomization/)
    factory."""
    yield from _kustomization_factory_impl(kube_cluster, namespace_factory)


@pytest.fixture(scope="module")DOCS
def kustomization_factory(
    kube_cluster: Cluster, namespace_factory: NamespaceFactoryFunc
) -> Iterable[KustomizationFactoryFunc]:
    """Returns module-scoped [Kustomization](https://fluxcd.io/docs/components/kustomize/kustomization/)
    factory."""
    yield from _kustomization_factory_impl(kube_cluster, namespace_factory)


def _kustomization_factory_impl(
    kube_cluster: Cluster, namespace_factory: NamespaceFactoryFunc
) -> Iterable[KustomizationFactoryFunc]:
    created_objects: List[KustomizationCR] = []

    yield kustomization_factory_func(kube_cluster.kube_client, namespace_factory, created_objects)

    delete_and_wait_for_objects(kube_cluster.kube_client, KustomizationCR, created_objects)


@pytest.fixture(scope="function")DOCS
def git_repository_factory_function_scope(
    kube_cluster: Cluster, namespace_factory: NamespaceFactoryFunc
) -> Iterable[GitRepositoryFactoryFunc]:
    """Returns function-scoped [Git Repository](https://fluxcd.io/docs/components/source/gitrepositories/) factory."""
    yield from _git_repository_factory_impl(kube_cluster, namespace_factory)


@pytest.fixture(scope="module")DOCS
def git_repository_factory(
    kube_cluster: Cluster, namespace_factory: NamespaceFactoryFunc
) -> Iterable[GitRepositoryFactoryFunc]:
    """Returns module-scoped [Git Repository](https://fluxcd.io/docs/components/source/gitrepositories/) factory."""
    yield from _git_repository_factory_impl(kube_cluster, namespace_factory)


def _git_repository_factory_impl(
    kube_cluster: Cluster, namespace_factory: NamespaceFactoryFunc
) -> Iterable[GitRepositoryFactoryFunc]:
    created_objects: List[GitRepositoryCR] = []

    yield git_repository_factory_func(kube_cluster.kube_client, namespace_factory, created_objects)

    delete_and_wait_for_objects(kube_cluster.kube_client, GitRepositoryCR, created_objects)


@pytest.fixture(scope="function")DOCS
def helm_repository_factory_function_scope(
    kube_cluster: Cluster, namespace_factory: NamespaceFactoryFunc
) -> Iterable[HelmRepositoryFactoryFunc]:
    """Returns function-scoped [Helm Repository](https://fluxcd.io/docs/components/source/helmrepositories/) factory."""
    yield from _helm_repository_factory_impl(kube_cluster, namespace_factory)


@pytest.fixture(scope="module")DOCS
def helm_repository_factory(
    kube_cluster: Cluster, namespace_factory: NamespaceFactoryFunc
) -> Iterable[HelmRepositoryFactoryFunc]:
    """Returns module-scoped [Helm Repository](https://fluxcd.io/docs/components/source/helmrepositories/) factory."""
    yield from _helm_repository_factory_impl(kube_cluster, namespace_factory)


def _helm_repository_factory_impl(
    kube_cluster: Cluster, namespace_factory: NamespaceFactoryFunc
) -> Iterable[HelmRepositoryFactoryFunc]:
    created_objects: List[HelmRepositoryCR] = []

    yield helm_repository_factory_func(kube_cluster.kube_client, namespace_factory, created_objects)

    delete_and_wait_for_objects(kube_cluster.kube_client, HelmRepositoryCR, created_objects)


@pytest.fixture(scope="function")DOCS
def helm_release_factory_function_scope(
    kube_cluster: Cluster, namespace_factory: NamespaceFactoryFunc
) -> Iterable[HelmReleaseFactoryFunc]:
    """Returns function-scoped [Helm Release](https://fluxcd.io/docs/components/helm/helmreleases/) factory."""
    yield from _helm_release_factory_impl(kube_cluster, namespace_factory)


@pytest.fixture(scope="module")DOCS
def helm_release_factory(
    kube_cluster: Cluster, namespace_factory: NamespaceFactoryFunc
) -> Iterable[HelmReleaseFactoryFunc]:
    """Returns module-scoped [Helm Release](https://fluxcd.io/docs/components/helm/helmreleases/) factory."""
    yield from _helm_release_factory_impl(kube_cluster, namespace_factory)


def _helm_release_factory_impl(
    kube_cluster: Cluster, namespace_factory: NamespaceFactoryFunc
) -> Iterable[HelmReleaseFactoryFunc]:
    created_objects: List[HelmReleaseCR] = []

    yield helm_release_factory_func(kube_cluster.kube_client, namespace_factory, created_objects)

    delete_and_wait_for_objects(kube_cluster.kube_client, HelmReleaseCR, created_objects)