Dagger: CI/CD Pipeline Guide

Table of Contents

• Dagger setup
• Quick Start
  • What Is Dagger?
  • Basic CLI Usage
  • Simple Go SDK Example
• Introduction
• Prerequisites
• Architecture Overview
• Python Implementation
  • Project Setup
  • Basic Build
  • Production Build with JCasC
• Go Implementation
  • Project Setup
  • Basic Build
  • Production Build with JCasC
• Pytest Integration
  • Test Dependencies
  • Test Fixtures
  • Test Suite
  • Running Tests
• Deployment
  • Loading and Running the Image
  • Docker Compose
  • Kubernetes Deployment
• Troubleshooting
  • Common Issues
  • Debugging Tips
• Summary
• References

Dagger setup

Quick Start

New to Dagger? Dagger is a CI/CD engine that lets you define pipelines using containers as code. Pipelines run identically locally and in CI, eliminating environment drift and YAML-heavy configurations.

What Is Dagger?

Key properties:

• Language-native pipelines (Python, Go, TypeScript, etc.)
• Deterministic, reproducible builds
• Local-first development
• CI portability

Basic CLI Usage

Build an application:

dagger call build --src https://github.com/example/app.git

Export build artifacts:

dagger call build --src . export --path ./build

Disable focus mode for CI:

dagger --focus=false call build --src ./ export --path ./build

Simple Go SDK Example

package main

import (
    "context"
    "dagger.io/dagger"
)

func main() {
    ctx := context.Background()
    client, _ := dagger.Connect(ctx)
    defer client.Close()

    container := client.Container().
        From("golang:1.22").
        WithDirectory("/app", client.Host().Directory(".")).
        WithWorkdir("/app").
        WithExec([]string{"go", "build", "-o", "app"})

    container.Export(ctx, "./build")
}

Why use the SDK?

• Conditional logic
• Reusable pipeline components
• Strong typing and IDE support
• Easier testing and refactoring

For a comprehensive example with Jenkins integration, continue reading below.

---

Building Production-Ready Jenkins Images with Dagger

A complete guide to containerizing Jenkins using Dagger’s programmable CI/CD engine, with implementations in Python and Go.

---

Introduction

Jenkins remains the backbone of CI/CD for thousands of organizations, but building and maintaining Jenkins Docker images presents unique challenges: plugin dependency hell, configuration drift between environments, slow feedback loops when testing changes, and the brittleness of shell-script-heavy Dockerfiles.

Dagger solves these problems by letting you define container builds as code in real programming languages. Your Jenkins image build becomes a testable, cacheable, portable function that runs identically on a developer laptop and in production CI.

This guide covers:

• Building Jenkins images with the Dagger Python and Go SDKs
• Pre-installing plugins using jenkins-plugin-cli
• Configuring authentication and security via JCasC (Jenkins Configuration as Code)
• Running Jenkins as a Dagger service for integration testing
• Writing pytest suites that validate your Jenkins setup

All examples use admin/admin credentials for simplicity—replace these with secrets management in production.

---

Prerequisites

| Tool | Version | Installation | | ---------- | ------- | ------------------------------------------------------ | --- | | Docker | 20.10+ | docs.docker.com | | Dagger CLI | 0.9+ | curl -fsSL https://dl.dagger.io/dagger/install.sh | sh | | Python | 3.11+ | With uv package manager | | Go | 1.21+ | Optional, for Go examples |

Verify your setup:

docker --version
dagger version

---

Architecture Overview

┌─────────────────────────────────────────────────────────────┐
│                     Dagger Pipeline                         │
├─────────────────────────────────────────────────────────────┤
│  ┌─────────────┐    ┌─────────────┐    ┌─────────────────┐  │
│  │ Base Image  │───▶│  Plugins    │───▶│  Configuration  │  │
│  │ jenkins/    │    │  jenkins-   │    │  JCasC + Groovy │  │
│  │ jenkins:lts │    │  plugin-cli │    │  init scripts   │  │
│  └─────────────┘    └─────────────┘    └─────────────────┘  │
│                                                │             │
│                                                ▼             │
│                                    ┌─────────────────────┐  │
│                                    │   Final Image       │  │
│                                    │   Ready to deploy   │  │
│                                    └─────────────────────┘  │
└─────────────────────────────────────────────────────────────┘
                              │
              ┌───────────────┼───────────────┐
              ▼               ▼               ▼
         Export to       Publish to      Run as
         tarball         registry        service

The build proceeds in layers:

1. Base image: Official Jenkins LTS with JDK 17
2. Plugin installation: jenkins-plugin-cli resolves and installs plugins with dependencies
3. Configuration: JCasC YAML and/or Groovy init scripts configure security, users, and settings
4. Output: Export as tarball, push to registry, or run as ephemeral service for testing

---

Python Implementation

Project Setup

mkdir jenkins-dagger && cd jenkins-dagger
uv init
uv add dagger-io anyio

Your pyproject.toml should include:

[project]
name = "jenkins-dagger"
version = "0.1.0"
requires-python = ">=3.11"
dependencies = [
    "dagger-io>=0.9.0",
    "anyio>=4.0.0",
]

Basic Build

Create build_jenkins.py:

"""
Build a production-ready Jenkins Docker image using Dagger.

This script demonstrates:
- Plugin installation via jenkins-plugin-cli
- Security configuration via Groovy init scripts
- Image export for deployment
"""

from __future__ import annotations

import anyio
import dagger

# Plugin selection: these cover most CI/CD workflows
# See https://plugins.jenkins.io/ for the full catalog
PLUGINS = [
    # Core pipeline functionality
    "workflow-aggregator",        # Pipeline: umbrella plugin for Pipeline suite
    "pipeline-stage-view",        # Visual pipeline stage status
    "pipeline-graph-view",        # Pipeline graph visualization

    # Source control
    "git",                        # Git integration
    "github",                     # GitHub integration
    "gitlab-plugin",              # GitLab integration

    # Docker & Kubernetes
    "docker-workflow",            # Docker Pipeline steps
    "docker-plugin",              # Docker cloud agents
    "kubernetes",                 # Kubernetes cloud agents

    # Credentials & security
    "credentials-binding",        # Inject credentials into builds
    "ssh-credentials",            # SSH key management
    "matrix-auth",                # Fine-grained permissions

    # Configuration & management
    "configuration-as-code",      # JCasC support
    "job-dsl",                    # Programmatic job creation

    # UI & usability
    "blueocean",                  # Modern UI
    "antisamy-markup-formatter",  # Safe HTML in descriptions
    "timestamper",                # Timestamps in console output
]


async def build_jenkins(
    plugins: list[str] = PLUGINS,
    username: str = "admin",
    password: str = "admin",
    export_path: str | None = "./jenkins-image.tar",
    publish_to: str | None = None,
) -> dagger.Container:
    """
    Build a fully configured Jenkins Docker image.

    Args:
        plugins: List of Jenkins plugin short names to install
        username: Admin username for initial setup
        password: Admin password for initial setup
        export_path: Local path to export tarball (None to skip)
        publish_to: Registry address to publish (None to skip)

    Returns:
        The configured Jenkins container
    """

    # Groovy script for initial security setup
    # This runs once on first boot before JCasC loads
    security_groovy = f"""
import jenkins.model.*
import hudson.security.*
import jenkins.security.s2m.AdminWhitelistRule

def instance = Jenkins.getInstance()

// Create the admin user
def hudsonRealm = new HudsonPrivateSecurityRealm(false)
hudsonRealm.createAccount("{username}", "{password}")
instance.setSecurityRealm(hudsonRealm)

// Allow logged-in users full control, deny anonymous access
def strategy = new FullControlOnceLoggedInAuthorizationStrategy()
strategy.setAllowAnonymousRead(false)
instance.setAuthorizationStrategy(strategy)

// Enable agent-to-controller security
instance.getInjector().getInstance(AdminWhitelistRule.class).setMasterKillSwitch(false)

instance.save()
println("Security configured: user '{username}' created")
"""

    config = dagger.Config(log_output=True)

    async with dagger.Connection(config) as client:
        jenkins = (
            client.container()
            .from_("jenkins/jenkins:lts-jdk17")
        )

        # Disable setup wizard - we're configuring programmatically
        # CSRF crumb exclusion helps with API calls in testing
        jenkins = jenkins.with_env_variable(
            "JAVA_OPTS",
            " ".join([
                "-Djenkins.install.runSetupWizard=false",
                "-Dhudson.security.csrf.DefaultCrumbIssuer.EXCLUDE_SESSION_ID=true",
                "-Dhudson.model.DirectoryBrowserSupport.CSP=",
            ])
        )

        # Install plugins using the official CLI tool
        # This resolves dependencies automatically and fails fast on conflicts
        jenkins = jenkins.with_exec(
            ["jenkins-plugin-cli", "--verbose", "--plugins", *plugins]
        )

        # Add the security initialization script
        # Scripts in init.groovy.d/ run on every startup in alphabetical order
        jenkins = (
            jenkins
            .with_user("root")
            .with_exec(["mkdir", "-p", "/usr/share/jenkins/ref/init.groovy.d"])
            .with_new_file(
                "/usr/share/jenkins/ref/init.groovy.d/00-security.groovy",
                security_groovy,
            )
            .with_user("jenkins")
        )

        # Export and/or publish
        if export_path:
            await jenkins.export(export_path)
            print(f"✓ Exported to {export_path}")

        if publish_to:
            ref = await jenkins.publish(publish_to)
            print(f"✓ Published to {ref}")

        return jenkins


if __name__ == "__main__":
    anyio.run(build_jenkins)

Run the build:

uv run python build_jenkins.py

Production Build with JCasC

Jenkins Configuration as Code (JCasC) provides declarative, version-controlled configuration. This is the recommended approach for production deployments.

Create build_jenkins_jcasc.py:

"""
Production Jenkins build using JCasC for declarative configuration.

JCasC advantages over Groovy init scripts:
- Declarative YAML instead of imperative code
- Easier to review in PRs
- Validates on startup
- Supports configuration reload without restart
"""

from __future__ import annotations

import anyio
import dagger
from textwrap import dedent

PLUGINS = [
    # Core
    "workflow-aggregator",
    "pipeline-stage-view",
    "git",
    "github",

    # Docker support
    "docker-workflow",
    "docker-plugin",

    # Security & credentials
    "credentials-binding",
    "ssh-credentials",
    "matrix-auth",

    # Configuration
    "configuration-as-code",
    "job-dsl",

    # UI
    "blueocean",
    "dark-theme",
    "timestamper",
]


def generate_jcasc_config(
    admin_user: str = "admin",
    admin_password: str = "admin",
    jenkins_url: str = "http://localhost:8080",
    executor_count: int = 2,
) -> str:
    """Generate JCasC YAML configuration."""

    return dedent(f"""
        # Jenkins Configuration as Code
        # https://github.com/jenkinsci/configuration-as-code-plugin

        jenkins:
          systemMessage: |
            Jenkins configured via Dagger + JCasC
            Managed configuration - manual changes will be overwritten

          numExecutors: {executor_count}
          mode: NORMAL

          # Security realm: local user database
          securityRealm:
            local:
              allowsSignup: false
              enableCaptcha: false
              users:
                - id: "{admin_user}"
                  name: "Administrator"
                  password: "{admin_password}"

          # Authorization: logged-in users have full control
          authorizationStrategy:
            loggedInUsersCanDoAnything:
              allowAnonymousRead: false

          # Agent security
          remotingSecurity:
            enabled: true

          # Global credentials available to all jobs
          # In production, use external secrets management
          # credentials:
          #   system:
          #     domainCredentials:
          #       - credentials:
          #           - string:
          #               scope: GLOBAL
          #               id: "github-token"
          #               secret: "${{GITHUB_TOKEN}}"
          #               description: "GitHub API Token"

        # Unclassified configuration
        unclassified:
          location:
            url: "{jenkins_url}"
            adminAddress: "admin@example.com"

          # Timestamper plugin configuration
          timestamper:
            allPipelines: true

          # Theme
          themeManager:
            disableUserThemes: false
            theme: "dark"

        # Tool installations
        tool:
          git:
            installations:
              - name: "Default"
                home: "git"

        # Seed job example using Job DSL
        # Uncomment to auto-create jobs on startup
        # jobs:
        #   - script: |
        #       pipelineJob('example-pipeline') {{
        #         definition {{
        #           cps {{
        #             script('''
        #               pipeline {{
        #                 agent any
        #                 stages {{
        #                   stage('Hello') {{
        #                     steps {{ echo 'Hello from JCasC!' }}
        #                   }}
        #                 }}
        #               }}
        #             ''')
        #             sandbox()
        #           }}
        #         }}
        #       }}
    """).strip()


async def build_jenkins_production(
    admin_user: str = "admin",
    admin_password: str = "admin",
    jenkins_url: str = "http://localhost:8080",
    export_path: str | None = "./jenkins-production.tar",
    publish_to: str | None = None,
) -> dagger.Container:
    """Build production Jenkins with JCasC configuration."""

    jcasc_yaml = generate_jcasc_config(
        admin_user=admin_user,
        admin_password=admin_password,
        jenkins_url=jenkins_url,
    )

    async with dagger.Connection(dagger.Config(log_output=True)) as client:
        jenkins = (
            client.container()
            .from_("jenkins/jenkins:lts-jdk17")
            # Disable wizard, point to JCasC config
            .with_env_variable(
                "JAVA_OPTS",
                " ".join([
                    "-Djenkins.install.runSetupWizard=false",
                    "-Dhudson.security.csrf.DefaultCrumbIssuer.EXCLUDE_SESSION_ID=true",
                ])
            )
            .with_env_variable("CASC_JENKINS_CONFIG", "/var/jenkins_home/casc.yaml")
        )

        # Install plugins
        jenkins = jenkins.with_exec(
            ["jenkins-plugin-cli", "--verbose", "--plugins", *PLUGINS]
        )

        # Write JCasC configuration
        jenkins = (
            jenkins
            .with_user("root")
            .with_new_file("/var/jenkins_home/casc.yaml", jcasc_yaml)
            .with_exec(["chown", "jenkins:jenkins", "/var/jenkins_home/casc.yaml"])
            .with_user("jenkins")
        )

        if export_path:
            await jenkins.export(export_path)
            print(f"✓ Exported to {export_path}")

        if publish_to:
            ref = await jenkins.publish(publish_to)
            print(f"✓ Published to {ref}")

        return jenkins


if __name__ == "__main__":
    anyio.run(build_jenkins_production)

---

Go Implementation

Project Setup

mkdir jenkins-dagger-go && cd jenkins-dagger-go
go mod init jenkins-dagger
go get dagger.io/dagger@latest

Basic Build

Create main.go:

// Package main builds a production-ready Jenkins Docker image using Dagger.
package main

import (
  "context"
  "flag"
  "fmt"
  "os"
  "strings"

  "dagger.io/dagger"
)

// Default plugins covering common CI/CD workflows
var defaultPlugins = []string{
  // Core pipeline
  "workflow-aggregator",
  "pipeline-stage-view",
  "pipeline-graph-view",

  // SCM
  "git",
  "github",
  "gitlab-plugin",

  // Docker & Kubernetes
  "docker-workflow",
  "docker-plugin",
  "kubernetes",

  // Credentials
  "credentials-binding",
  "ssh-credentials",
  "matrix-auth",

  // Configuration
  "configuration-as-code",
  "job-dsl",

  // UI
  "blueocean",
  "antisamy-markup-formatter",
  "timestamper",
}

// Config holds the build configuration
type Config struct {
  Plugins    []string
  Username   string
  Password   string
  ExportPath string
  PublishTo  string
}

func main() {
  cfg := Config{
    Plugins:  defaultPlugins,
    Username: "admin",
    Password: "admin",
  }

  flag.StringVar(&cfg.ExportPath, "export", "./jenkins-image.tar", "Path to export tarball")
  flag.StringVar(&cfg.PublishTo, "publish", "", "Registry to publish (e.g., ttl.sh/my-jenkins:1h)")
  flag.StringVar(&cfg.Username, "user", "admin", "Admin username")
  flag.StringVar(&cfg.Password, "pass", "admin", "Admin password")
  flag.Parse()

  if err := build(context.Background(), cfg); err != nil {
    fmt.Fprintf(os.Stderr, "Error: %v\n", err)
    os.Exit(1)
  }
}

func build(ctx context.Context, cfg Config) error {
  // Groovy init script for security setup
  securityGroovy := fmt.Sprintf(`
import jenkins.model.*
import hudson.security.*
import jenkins.security.s2m.AdminWhitelistRule

def instance = Jenkins.getInstance()

def hudsonRealm = new HudsonPrivateSecurityRealm(false)
hudsonRealm.createAccount("%s", "%s")
instance.setSecurityRealm(hudsonRealm)

def strategy = new FullControlOnceLoggedInAuthorizationStrategy()
strategy.setAllowAnonymousRead(false)
instance.setAuthorizationStrategy(strategy)

instance.getInjector().getInstance(AdminWhitelistRule.class).setMasterKillSwitch(false)
instance.save()
println("Security configured: user '%s' created")
`, cfg.Username, cfg.Password, cfg.Username)

  client, err := dagger.Connect(ctx, dagger.WithLogOutput(os.Stdout))
  if err != nil {
    return fmt.Errorf("connecting to dagger: %w", err)
  }
  defer client.Close()

  // Start from Jenkins LTS
  jenkins := client.Container().From("jenkins/jenkins:lts-jdk17")

  // Configure JVM options
  javaOpts := strings.Join([]string{
    "-Djenkins.install.runSetupWizard=false",
    "-Dhudson.security.csrf.DefaultCrumbIssuer.EXCLUDE_SESSION_ID=true",
    "-Dhudson.model.DirectoryBrowserSupport.CSP=",
  }, " ")
  jenkins = jenkins.WithEnvVariable("JAVA_OPTS", javaOpts)

  // Install plugins
  pluginArgs := append([]string{"jenkins-plugin-cli", "--verbose", "--plugins"}, cfg.Plugins...)
  jenkins = jenkins.WithExec(pluginArgs)

  // Add security init script
  jenkins = jenkins.
    WithUser("root").
    WithExec([]string{"mkdir", "-p", "/usr/share/jenkins/ref/init.groovy.d"}).
    WithNewFile("/usr/share/jenkins/ref/init.groovy.d/00-security.groovy", securityGroovy).
    WithUser("jenkins")

  // Export if path provided
  if cfg.ExportPath != "" {
    if _, err := jenkins.Export(ctx, cfg.ExportPath); err != nil {
      return fmt.Errorf("exporting image: %w", err)
    }
    fmt.Printf("✓ Exported to %s\n", cfg.ExportPath)
  }

  // Publish if registry provided
  if cfg.PublishTo != "" {
    ref, err := jenkins.Publish(ctx, cfg.PublishTo)
    if err != nil {
      return fmt.Errorf("publishing image: %w", err)
    }
    fmt.Printf("✓ Published to %s\n", ref)
  }

  return nil
}

Production Build with JCasC

Create main_jcasc.go:

// Package main builds production Jenkins with JCasC configuration.
package main

import (
  "context"
  "flag"
  "fmt"
  "os"
  "strings"

  "dagger.io/dagger"
)

var plugins = []string{
  "workflow-aggregator",
  "pipeline-stage-view",
  "git",
  "github",
  "docker-workflow",
  "docker-plugin",
  "credentials-binding",
  "ssh-credentials",
  "matrix-auth",
  "configuration-as-code",
  "job-dsl",
  "blueocean",
  "dark-theme",
  "timestamper",
}

// JCasC configuration template
const jcascTemplate = `
jenkins:
  systemMessage: |
    Jenkins configured via Dagger + JCasC
    Managed configuration - manual changes will be overwritten

  numExecutors: %d
  mode: NORMAL

  securityRealm:
    local:
      allowsSignup: false
      enableCaptcha: false
      users:
        - id: "%s"
          name: "Administrator"
          password: "%s"

  authorizationStrategy:
    loggedInUsersCanDoAnything:
      allowAnonymousRead: false

  remotingSecurity:
    enabled: true

unclassified:
  location:
    url: "%s"
    adminAddress: "admin@example.com"

  timestamper:
    allPipelines: true

  themeManager:
    disableUserThemes: false
    theme: "dark"

tool:
  git:
    installations:
      - name: "Default"
        home: "git"
`

type Config struct {
  Username   string
  Password   string
  JenkinsURL string
  Executors  int
  ExportPath string
  PublishTo  string
}

func main() {
  cfg := Config{
    Username:   "admin",
    Password:   "admin",
    JenkinsURL: "http://localhost:8080",
    Executors:  2,
    ExportPath: "./jenkins-production.tar",
  }

  flag.StringVar(&cfg.Username, "user", cfg.Username, "Admin username")
  flag.StringVar(&cfg.Password, "pass", cfg.Password, "Admin password")
  flag.StringVar(&cfg.JenkinsURL, "url", cfg.JenkinsURL, "Jenkins URL")
  flag.IntVar(&cfg.Executors, "executors", cfg.Executors, "Number of executors")
  flag.StringVar(&cfg.ExportPath, "export", cfg.ExportPath, "Export path")
  flag.StringVar(&cfg.PublishTo, "publish", "", "Registry to publish")
  flag.Parse()

  if err := buildJCasC(context.Background(), cfg); err != nil {
    fmt.Fprintf(os.Stderr, "Error: %v\n", err)
    os.Exit(1)
  }
}

func buildJCasC(ctx context.Context, cfg Config) error {
  jcascYAML := fmt.Sprintf(jcascTemplate,
    cfg.Executors,
    cfg.Username,
    cfg.Password,
    cfg.JenkinsURL,
  )

  client, err := dagger.Connect(ctx, dagger.WithLogOutput(os.Stdout))
  if err != nil {
    return err
  }
  defer client.Close()

  javaOpts := strings.Join([]string{
    "-Djenkins.install.runSetupWizard=false",
    "-Dhudson.security.csrf.DefaultCrumbIssuer.EXCLUDE_SESSION_ID=true",
  }, " ")

  jenkins := client.Container().
    From("jenkins/jenkins:lts-jdk17").
    WithEnvVariable("JAVA_OPTS", javaOpts).
    WithEnvVariable("CASC_JENKINS_CONFIG", "/var/jenkins_home/casc.yaml")

  // Install plugins
  pluginArgs := append([]string{"jenkins-plugin-cli", "--verbose", "--plugins"}, plugins...)
  jenkins = jenkins.WithExec(pluginArgs)

  // Write JCasC config
  jenkins = jenkins.
    WithUser("root").
    WithNewFile("/var/jenkins_home/casc.yaml", jcascYAML).
    WithExec([]string{"chown", "jenkins:jenkins", "/var/jenkins_home/casc.yaml"}).
    WithUser("jenkins")

  if cfg.ExportPath != "" {
    if _, err := jenkins.Export(ctx, cfg.ExportPath); err != nil {
      return err
    }
    fmt.Printf("✓ Exported to %s\n", cfg.ExportPath)
  }

  if cfg.PublishTo != "" {
    ref, err := jenkins.Publish(ctx, cfg.PublishTo)
    if err != nil {
      return err
    }
    fmt.Printf("✓ Published to %s\n", ref)
  }

  return nil
}

Run the build:

# Basic build
go run main.go

# With custom settings
go run main_jcasc.go -user admin -pass admin -export ./jenkins.tar

---

Pytest Integration

Testing your Jenkins image ensures plugins work correctly and configuration applies as expected. Dagger’s service binding lets you spin up Jenkins containers during test runs.

Test Dependencies

uv add --dev pytest pytest-asyncio httpx tenacity

Update pyproject.toml:

[tool.pytest.ini_options]
asyncio_mode = "auto"
asyncio_default_fixture_loop_scope = "session"
testpaths = ["tests"]

Test Fixtures

Create tests/conftest.py:

"""
Pytest fixtures for Jenkins integration testing.

These fixtures build and run Jenkins using Dagger, wait for full
initialization, and provide an authenticated HTTP client for API calls.
"""

from __future__ import annotations

import pytest
import httpx
import dagger
from tenacity import (
    retry,
    stop_after_delay,
    wait_exponential,
    retry_if_exception_type,
    before_sleep_log,
)
import logging

logger = logging.getLogger(__name__)

# Test-specific plugin set (minimal for faster startup)
TEST_PLUGINS = [
    "workflow-aggregator",
    "git",
    "configuration-as-code",
    "matrix-auth",
    "timestamper",
]

TEST_JCASC = """
jenkins:
  systemMessage: "Jenkins Test Instance"
  numExecutors: 2
  securityRealm:
    local:
      allowsSignup: false
      users:
        - id: "admin"
          password: "admin"
  authorizationStrategy:
    loggedInUsersCanDoAnything:
      allowAnonymousRead: false
  remotingSecurity:
    enabled: true

unclassified:
  location:
    url: http://localhost:8080/
"""


class JenkinsNotReadyError(Exception):
    """Raised when Jenkins hasn't fully initialized."""
    pass


@retry(
    stop=stop_after_delay(180),  # Jenkins can take a while with plugins
    wait=wait_exponential(multiplier=1, min=2, max=15),
    retry=retry_if_exception_type((httpx.RequestError, httpx.HTTPStatusError, JenkinsNotReadyError)),
    before_sleep=before_sleep_log(logger, logging.DEBUG),
    reraise=True,
)
async def wait_for_jenkins_ready(
    base_url: str,
    username: str,
    password: str,
) -> dict:
    """
    Wait for Jenkins to be fully operational.

    Checks:
    1. HTTP connectivity
    2. Authentication works
    3. Mode is NORMAL (not INITIALIZING)
    4. No pending plugin installations

    Returns:
        Jenkins API response data

    Raises:
        JenkinsNotReadyError: If Jenkins is still initializing
        httpx.RequestError: If connection fails
    """
    async with httpx.AsyncClient(timeout=15.0) as client:
        response = await client.get(
            f"{base_url}/api/json",
            auth=(username, password),
        )
        response.raise_for_status()

        data = response.json()

        # Check mode - NORMAL means fully initialized
        mode = data.get("mode", "UNKNOWN")
        if mode != "NORMAL":
            raise JenkinsNotReadyError(f"Jenkins mode is {mode}, waiting for NORMAL")

        # Verify we can access plugin manager (ensures plugins are loaded)
        plugin_response = await client.get(
            f"{base_url}/pluginManager/api/json?depth=1",
            auth=(username, password),
        )
        plugin_response.raise_for_status()

        return data


@pytest.fixture(scope="session")
async def dagger_client():
    """Session-scoped Dagger client."""
    async with dagger.Connection(dagger.Config(log_output=False)) as client:
        yield client


@pytest.fixture(scope="session")
async def jenkins_container(dagger_client: dagger.Client) -> dagger.Container:
    """Build the Jenkins container (cached for the session)."""
    jenkins = (
        dagger_client.container()
        .from_("jenkins/jenkins:lts-jdk17")
        .with_env_variable("JAVA_OPTS", "-Djenkins.install.runSetupWizard=false")
        .with_env_variable("CASC_JENKINS_CONFIG", "/var/jenkins_home/casc.yaml")
        .with_exec(["jenkins-plugin-cli", "--plugins", *TEST_PLUGINS])
        .with_user("root")
        .with_new_file("/var/jenkins_home/casc.yaml", TEST_JCASC)
        .with_exec(["chown", "jenkins:jenkins", "/var/jenkins_home/casc.yaml"])
        .with_user("jenkins")
        .with_exposed_port(8080)
    )
    return jenkins


@pytest.fixture(scope="session")
async def jenkins_service(jenkins_container: dagger.Container):
    """
    Run Jenkins as a Dagger service and wait for it to be ready.

    Yields a dict with:
        - service: The Dagger service object
        - endpoint: HTTP endpoint URL
        - username: Admin username
        - password: Admin password
    """
    service = jenkins_container.as_service()
    endpoint = await service.endpoint(port=8080, scheme="http")

    logger.info(f"Jenkins starting at {endpoint}")

    # Wait for full initialization
    await wait_for_jenkins_ready(endpoint, "admin", "admin")

    logger.info("Jenkins is ready")

    yield {
        "service": service,
        "endpoint": endpoint,
        "username": "admin",
        "password": "admin",
    }


@pytest.fixture
async def jenkins_client(jenkins_service: dict):
    """Authenticated HTTP client for Jenkins API calls."""
    async with httpx.AsyncClient(
        base_url=jenkins_service["endpoint"],
        auth=(jenkins_service["username"], jenkins_service["password"]),
        timeout=30.0,
    ) as client:
        yield client


@pytest.fixture
async def crumb(jenkins_client: httpx.AsyncClient) -> dict:
    """Get CSRF crumb for POST requests."""
    response = await jenkins_client.get("/crumbIssuer/api/json")
    response.raise_for_status()
    data = response.json()
    return {data["crumbRequestField"]: data["crumb"]}

Test Suite

Create tests/test_jenkins.py:

"""
Integration tests for Jenkins Docker image.

These tests validate:
- Jenkins starts correctly with our configuration
- Required plugins are installed
- API is accessible and authenticated
- Jobs can be created and executed
"""

from __future__ import annotations

import asyncio
import pytest
import httpx


class TestJenkinsHealth:
    """Basic health and configuration tests."""

    async def test_api_accessible(self, jenkins_client: httpx.AsyncClient):
        """Verify Jenkins API responds to authenticated requests."""
        response = await jenkins_client.get("/api/json")
        assert response.status_code == 200

        data = response.json()
        assert data["mode"] == "NORMAL"
        assert "_class" in data

    async def test_anonymous_access_denied(self, jenkins_service: dict):
        """Verify anonymous users cannot access the API."""
        async with httpx.AsyncClient() as client:
            response = await client.get(
                f"{jenkins_service['endpoint']}/api/json",
                follow_redirects=False,
            )
            # Should redirect to login or return 401/403
            assert response.status_code in (401, 403, 302)

    async def test_system_message_configured(self, jenkins_client: httpx.AsyncClient):
        """Verify JCasC applied the system message."""
        response = await jenkins_client.get("/api/json")
        data = response.json()

        # System message from JCasC config
        assert "Test Instance" in data.get("description", "")


class TestPlugins:
    """Plugin installation verification tests."""

    async def test_required_plugins_installed(self, jenkins_client: httpx.AsyncClient):
        """Verify all required plugins are installed and active."""
        response = await jenkins_client.get("/pluginManager/api/json?depth=1")
        assert response.status_code == 200

        plugins = response.json()["plugins"]
        installed = {p["shortName"]: p for p in plugins}

        required = ["workflow-aggregator", "git", "configuration-as-code"]
        for plugin_name in required:
            assert plugin_name in installed, f"Plugin {plugin_name} not installed"
            plugin = installed[plugin_name]
            assert plugin["active"], f"Plugin {plugin_name} not active"
            assert plugin["enabled"], f"Plugin {plugin_name} not enabled"

    async def test_no_plugin_update_failures(self, jenkins_client: httpx.AsyncClient):
        """Verify no plugins failed to install."""
        response = await jenkins_client.get("/pluginManager/api/json?depth=1")
        plugins = response.json()["plugins"]

        failed = [p["shortName"] for p in plugins if p.get("hasUpdate") and not p.get("active")]
        assert not failed, f"Plugins with issues: {failed}"


class TestJobManagement:
    """Job creation and execution tests."""

    async def test_create_freestyle_job(
        self,
        jenkins_client: httpx.AsyncClient,
        crumb: dict,
    ):
        """Test creating a freestyle job via API."""
        job_name = "test-freestyle-job"
        job_config = """<?xml version='1.1' encoding='UTF-8'?>
<project>
  <description>Test freestyle job</description>
  <builders>
    <hudson.tasks.Shell>
      <command>echo "Hello from freestyle job"</command>
    </hudson.tasks.Shell>
  </builders>
</project>
"""
        # Create job
        response = await jenkins_client.post(
            "/createItem",
            params={"name": job_name},
            content=job_config,
            headers={**crumb, "Content-Type": "application/xml"},
        )
        assert response.status_code in (200, 201), response.text

        # Verify job exists
        response = await jenkins_client.get(f"/job/{job_name}/api/json")
        assert response.status_code == 200
        assert response.json()["name"] == job_name

        # Cleanup
        await jenkins_client.post(
            f"/job/{job_name}/doDelete",
            headers=crumb,
        )

    async def test_create_and_run_pipeline(
        self,
        jenkins_client: httpx.AsyncClient,
        crumb: dict,
    ):
        """Test creating and running a pipeline job."""
        job_name = "test-pipeline-job"
        job_config = """<?xml version='1.1' encoding='UTF-8'?>
<flow-definition plugin="workflow-job">
  <description>Test pipeline job</description>
  <definition class="org.jenkinsci.plugins.workflow.cps.CpsFlowDefinition" plugin="workflow-cps">
    <script>
pipeline {
    agent any
    stages {
        stage('Build') {
            steps {
                echo 'Building...'
            }
        }
        stage('Test') {
            steps {
                echo 'Testing...'
            }
        }
        stage('Deploy') {
            steps {
                echo 'Deploying...'
            }
        }
    }
}
    </script>
    <sandbox>true</sandbox>
  </definition>
</flow-definition>
"""
        # Create job
        response = await jenkins_client.post(
            "/createItem",
            params={"name": job_name},
            content=job_config,
            headers={**crumb, "Content-Type": "application/xml"},
        )
        assert response.status_code in (200, 201), response.text

        # Trigger build
        response = await jenkins_client.post(
            f"/job/{job_name}/build",
            headers=crumb,
        )
        assert response.status_code in (200, 201), response.text

        # Wait for build to complete
        build_result = await self._wait_for_build(
            jenkins_client, job_name, timeout=60
        )

        assert build_result["result"] == "SUCCESS"
        assert build_result["building"] is False

        # Cleanup
        await jenkins_client.post(
            f"/job/{job_name}/doDelete",
            headers=crumb,
        )

    async def _wait_for_build(
        self,
        client: httpx.AsyncClient,
        job_name: str,
        timeout: int = 60,
    ) -> dict:
        """Poll until build completes or timeout."""
        deadline = asyncio.get_event_loop().time() + timeout

        while asyncio.get_event_loop().time() < deadline:
            await asyncio.sleep(2)

            response = await client.get(f"/job/{job_name}/lastBuild/api/json")
            if response.status_code == 404:
                # Build hasn't started yet
                continue

            if response.status_code == 200:
                build = response.json()
                if not build.get("building", True):
                    return build

        pytest.fail(f"Build for {job_name} did not complete within {timeout}s")


class TestCredentials:
    """Credential management tests."""

    async def test_credential_domains_accessible(
        self,
        jenkins_client: httpx.AsyncClient,
    ):
        """Verify credential store is accessible."""
        response = await jenkins_client.get(
            "/credentials/store/system/api/json?depth=2"
        )
        assert response.status_code == 200

        data = response.json()
        assert "domains" in data


class TestSecurity:
    """Security configuration tests."""

    async def test_csrf_protection_enabled(
        self,
        jenkins_client: httpx.AsyncClient,
    ):
        """Verify CSRF protection is active."""
        response = await jenkins_client.get("/crumbIssuer/api/json")
        assert response.status_code == 200

        data = response.json()
        assert "crumb" in data
        assert "crumbRequestField" in data

    async def test_remoting_security_enabled(
        self,
        jenkins_client: httpx.AsyncClient,
    ):
        """Verify agent-to-controller security is enabled."""
        response = await jenkins_client.get(
            "/computer/(built-in)/api/json"
        )
        assert response.status_code == 200

Running Tests

# Run all tests
uv run pytest -v

# Run with detailed output
uv run pytest -v -s --log-cli-level=DEBUG

# Run specific test class
uv run pytest tests/test_jenkins.py::TestPlugins -v

# Run with coverage
uv add --dev pytest-cov
uv run pytest --cov=. --cov-report=html

---

Deployment

Loading and Running the Image

# Load the exported tarball
docker load -i jenkins-production.tar

# Find the image ID
IMAGE_ID=$(docker images --format "{{.ID}}" | head -1)

# Run with persistent storage
docker run -d \
  --name jenkins \
  -p 8080:8080 \
  -p 50000:50000 \
  -v jenkins_home:/var/jenkins_home \
  --restart unless-stopped \
  $IMAGE_ID

# View logs
docker logs -f jenkins

# Access at http://localhost:8080
# Login: admin / admin

Docker Compose

Create docker-compose.yml:

version: "3.8"

services:
  jenkins:
    image: ${JENKINS_IMAGE:-jenkins-production:latest}
    container_name: jenkins
    restart: unless-stopped
    ports:
      - "8080:8080"
      - "50000:50000"
    volumes:
      - jenkins_home:/var/jenkins_home
      - /var/run/docker.sock:/var/run/docker.sock # For Docker-in-Docker
    environment:
      - JAVA_OPTS=-Xmx2g -Xms512m
    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:8080/api/json"]
      interval: 30s
      timeout: 10s
      retries: 5
      start_period: 60s

volumes:
  jenkins_home:

Kubernetes Deployment

Create k8s/jenkins.yaml:

apiVersion: apps/v1
kind: Deployment
metadata:
  name: jenkins
  labels:
    app: jenkins
spec:
  replicas: 1
  selector:
    matchLabels:
      app: jenkins
  template:
    metadata:
      labels:
        app: jenkins
    spec:
      securityContext:
        fsGroup: 1000
      containers:
        - name: jenkins
          image: your-registry/jenkins-production:latest
          ports:
            - containerPort: 8080
            - containerPort: 50000
          resources:
            requests:
              memory: "1Gi"
              cpu: "500m"
            limits:
              memory: "4Gi"
              cpu: "2"
          volumeMounts:
            - name: jenkins-home
              mountPath: /var/jenkins_home
          livenessProbe:
            httpGet:
              path: /login
              port: 8080
            initialDelaySeconds: 60
            periodSeconds: 10
          readinessProbe:
            httpGet:
              path: /api/json
              port: 8080
            initialDelaySeconds: 60
            periodSeconds: 5
      volumes:
        - name: jenkins-home
          persistentVolumeClaim:
            claimName: jenkins-pvc
---
apiVersion: v1
kind: Service
metadata:
  name: jenkins
spec:
  selector:
    app: jenkins
  ports:
    - name: http
      port: 8080
      targetPort: 8080
    - name: agent
      port: 50000
      targetPort: 50000
  type: LoadBalancer

---

Troubleshooting

Common Issues

Issue	Cause	Solution
Plugins fail to install	Network issues or version conflicts	Check jenkins-plugin-cli output, pin plugin versions
Jenkins never reaches NORMAL mode	JCasC syntax error	Check /var/jenkins_home/casc.yaml syntax
Authentication fails	Groovy script error	Check init script logs in startup output
Slow startup	Too many plugins	Use minimal plugin set for tests
OOM errors	Insufficient heap	Increase -Xmx in JAVA_OPTS

Debugging Tips

# Check container logs
docker logs jenkins 2>&1 | grep -i error

# Verify JCasC loaded
docker exec jenkins cat /var/jenkins_home/casc.yaml

# Check plugin installation
docker exec jenkins ls /var/jenkins_home/plugins/

# Test API manually
curl -u admin:admin http://localhost:8080/api/json | jq .

# Reload JCasC without restart
curl -X POST -u admin:admin http://localhost:8080/reload-configuration-as-code/

---

Summary

This guide demonstrated building production Jenkins images with Dagger:

Approach	Best For
Groovy init scripts	Simple setups, quick prototypes
JCasC	Production, version-controlled config
Dagger services	Integration testing, CI pipelines

Key benefits of the Dagger approach:

• Reproducible: Same build on laptops, CI, and production
• Testable: Spin up real Jenkins instances in pytest
• Cached: Dagger caches layers intelligently across runs
• Portable: No vendor lock-in; runs anywhere Docker runs
• Type-safe: Catch configuration errors at build time

For production deployments, always:

1. Use secrets management instead of hardcoded passwords
2. Pin plugin versions for reproducibility
3. Enable persistent storage for jenkins_home
4. Configure proper resource limits
5. Set up health checks and monitoring

---

References

• Dagger Documentation
• Dagger Python SDK
• Dagger Go SDK
• Jenkins Configuration as Code
• Jenkins Plugin CLI
• Jenkins Docker Images