Example: gitlab-ci-scaffold — generate a complete CI project

"""GitLab CI scaffold — generates a full project layout from a fluid contract."""

from __future__ import annotations

import json
from typing import Any, List, Mapping

from fluid_sdk import (
    ContractHelper,
    CustomScaffold,
    PluginMetadata,
    write_file_action,
)


class GitLabCIScaffold(CustomScaffold):
    """Generates README.md, .gitlab-ci.yml, and per-env config files."""

    name = "gitlab-ci"

    @classmethod
    def get_plugin_info(cls) -> PluginMetadata:
        return PluginMetadata(
            name=cls.name,
            role=cls.role,
            display_name="GitLab CI Scaffold",
            description="Generates a complete GitLab CI scaffold from a fluid contract.",
            version="0.1.0",
            author="FLUID SDK Examples",
            tags=["ci", "gitlab", "scaffold"],
        )

    def plan(self, contract: Mapping[str, Any]) -> List[dict]:
        c = ContractHelper(contract)
        actions: List[dict] = []

        # 1. The project README.
        actions.append(
            write_file_action(
                path="README.md",
                content=self._render_readme(c).encode("utf-8"),
                description="Project README",
            ).to_dict()
        )

        # 2. The CI definition.
        actions.append(
            write_file_action(
                path=".gitlab-ci.yml",
                content=self._render_ci(c).encode("utf-8"),
                description="GitLab CI definition",
            ).to_dict()
        )

        # 3. One config file per environment.
        for env_name in c.environment_names():
            actions.append(
                write_file_action(
                    path=f"config/{env_name}.json",
                    content=self._render_env_config(c, env_name).encode("utf-8"),
                    description=f"Config for environment {env_name!r}",
                ).to_dict()
            )

        return actions

    # The three private renderers below are mechanical f-strings.
    # Click to expand if you want the full source.
    # ↓

# tests/test_scaffold.py

MULTI_ENV_CONTRACT = {
    "fluidVersion": "0.7.4",
    "kind": "DataProduct",
    "id": "my-data-product",
    "name": "My Data Product",
    "description": "A nightly aggregation of yesterday's events.",
    "domain": "platform",
    "metadata": {
        "owner": {"team": "platform", "email": "platform@example.com"},
    },
    "environments": {
        "dev":     {"metadata": {"labels": {"cloud.accountId": "111111111111", "cloud.region": "eu-west-1"}}},
        "staging": {"metadata": {"labels": {"cloud.accountId": "222222222222", "cloud.region": "eu-west-1"}}},
        "prod":    {"metadata": {"labels": {"cloud.accountId": "333333333333", "cloud.region": "eu-west-1"}}},
    },
    "exposes": [], "consumes": [], "builds": [],
}


class TestGitLabCIScaffold(CustomScaffoldTestHarness):
    plugin_class = GitLabCIScaffold
    sample_contracts = [MULTI_ENV_CONTRACT]

    def _action_content(self, actions, path: str) -> str:
        action = next(a for a in actions if a["params"]["path"] == path)
        return base64.b64decode(action["params"]["content_b64"]).decode("utf-8")

    def test_readme_includes_owner_and_envs(self):
        readme = self._action_content(self.get_plugin().plan(MULTI_ENV_CONTRACT), "README.md")
        assert "platform@example.com" in readme
        assert "dev, prod, staging" in readme   # environment_names() returns sorted

    def test_ci_has_one_deploy_per_env(self):
        ci = self._action_content(self.get_plugin().plan(MULTI_ENV_CONTRACT), ".gitlab-ci.yml")
        assert all(f"deploy:{e}:" in ci for e in ("dev", "staging", "prod"))

    def test_prod_deploy_is_manual(self):
        """Prod deploys must be gated `when: manual` so they don't auto-run."""
        ci = self._action_content(self.get_plugin().plan(MULTI_ENV_CONTRACT), ".gitlab-ci.yml")
        prod_idx = ci.index("deploy:prod:")
        rest = ci[prod_idx + len("deploy:prod:"):]
        next_deploy = rest.find("\ndeploy:")
        end = prod_idx + len("deploy:prod:") + next_deploy if next_deploy != -1 else len(ci)
        assert "when: manual" in ci[prod_idx:end]

    def test_env_config_carries_account_id(self):
        dev_config = json.loads(self._action_content(self.get_plugin().plan(MULTI_ENV_CONTRACT), "config/dev.json"))
        assert dev_config["cloud"]["accountId"] == "111111111111"
        assert dev_config["cloud"]["region"] == "eu-west-1"

    def test_emits_correct_file_count(self):
        # 1 README + 1 CI + 3 env configs = 5
        assert len(self.get_plugin().plan(MULTI_ENV_CONTRACT)) == 5