🦫

GitLab CICD上でClaude Codeを実行する!

に公開

はじめに

GitLabのCICDにて生成AIを動作させたいとふと思いつき、調査と検証を行ってみました。
その結果を以下にまとめました。

今回のゴールのイメージ

今回AIに生成させるものとしてバックエンドのコードに対する単体テストを書いてもらい、それをマージクエストを介して人間がレビューできる状態にするということをゴールに進めてみました。

また生成AIにもいろいろサービスがありますが今回はClaudeCode GitLab CICDを利用することとしました。

今回の構成

以下のような構成にて実装することを決めました

マージリクエストを作成(またはコードをリモートへPush)
     │  
     ▼
Claude CLI (GitLab CICD上で動作)
     │  
     ▼
生成したテストコードをリモートへpush
     │
     ▼
CICD上でHTTPリクエストを介してGitLab (リポジトリ / MR / API)を作成

前準備

今回は簡易的なFlaskアプリを作成して、APIの改修を行なった際にその改修箇所に対してのテストコードを生成するような状況を作りました。

今回作成したディレクトリ構成

.
├── Dockerfile
├── Makefile
├── README.md
├── app
│   ├── data.db
│   ├── db.py...db接続の管理周り
│   ├── main.py...Flaskアプリのエントリポイント
│   ├── requirements.txt...依存関係の管理
│   ├── routes
│   │   └── items.py...APIのルーティング設定、ビジネスロジックの記述
│   └── seed.py...DBへのシードファイル
├── docker-compose.yml
└── tests

テストコード生成に関係のありそうなコードの詳細を以下に記載します

db.py
import psycopg2
import os

# 環境変数から設定を読み込む
DB_HOST = os.getenv("POSTGRES_HOST", "db")
DB_PORT = os.getenv("POSTGRES_PORT", "5432")
DB_NAME = os.getenv("POSTGRES_DB", "testgen")
DB_USER = os.getenv("POSTGRES_USER", "user")
DB_PASS = os.getenv("POSTGRES_PASSWORD", "pass")


def get_connection():
    """PostgreSQL へのコネクションを返す"""
    conn = psycopg2.connect(
        host=DB_HOST,
        port=DB_PORT,
        dbname=DB_NAME,
        user=DB_USER,
        password=DB_PASS
    )
    return conn


def init_db():
    """items テーブルを作成"""
    conn = get_connection()
    cur = conn.cursor()
    cur.execute('''
        CREATE TABLE IF NOT EXISTS items (
            id SERIAL PRIMARY KEY,
            name TEXT NOT NULL
        )
    ''')
    conn.commit()
    cur.close()
    conn.close()


def insert_item(name):
    """アイテムを挿入"""
    conn = get_connection()
    cur = conn.cursor()
    cur.execute('INSERT INTO items (name) VALUES (%s)', (name,))
    conn.commit()
    cur.close()
    conn.close()


def get_all_items():
    """全アイテムを取得"""
    conn = get_connection()
    cur = conn.cursor()
    cur.execute('SELECT id, name FROM items ORDER BY id')
    rows = cur.fetchall()
    cur.close()
    conn.close()

    return [{"id": row[0], "name": row[1]} for row in rows]

def update_item(item_id, name):
    """アイテム名を更新"""
    conn = get_connection()
    cur = conn.cursor()

    cur.execute('UPDATE items SET name = %s WHERE id = %s', (name, item_id))
    updated_rows = cur.rowcount  # 更新された件数を取得

    conn.commit()
    cur.close()
    conn.close()

    return updated_rows
items.py
from flask import Blueprint, request, jsonify
import db
import psycopg2

bp = Blueprint('items', __name__, url_prefix='/items')


# GET /items
@bp.route('/', methods=['GET'])
def get_items():
    try:
        items = db.get_all_items()
        return jsonify(items)
    except psycopg2.Error as e:
        return jsonify({"error": f"データ取得時にエラーが発生しました: {e.pgerror}"}), 500


# POST /items
@bp.route('/', methods=['POST'])
def add_item():
    data = request.get_json()
    name = data.get('name')

    if not name:
        return jsonify({"error": "nameフィールドが必要です"}), 400

    try:
        db.insert_item(name)
        return jsonify({"message": f"'{name}' を追加しました"}), 201
    except psycopg2.Error as e:
        return jsonify({"error": f"データ追加時にエラーが発生しました: {e.pgerror}"}), 500

# PUT /items/<int:item_id>
@bp.route('/<int:item_id>', methods=['PUT'])
def update_item(item_id):
    data = request.get_json()
    name = data.get('name')

    if not name:
        return jsonify({"error": "nameフィールドが必要です"}), 400

    try:
        updated_rows = db.update_item(item_id, name)

        if updated_rows == 0:
            return jsonify({"error": f"ID {item_id} のデータが見つかりません"}), 404

        return jsonify({"message": f"ID {item_id} のアイテムを '{name}' に更新しました"}), 200

    except psycopg2.Error as e:
        return jsonify({"error": f"データ更新時にエラーが発生しました: {e.pgerror}"}), 500

Claude Code x GitLab CI/CDの設定

Claude CodeをGitLab CI/CD上で動作またMRをHTTPリクエスト介して実行させるために以下を行います

  • GitLab PATの設定
  • Anthropic API Keyの取得
  • Claude Codeのクレジットを有効化
  • GitLabの対象のリポジトリに対し環境変数を設定
  • .gitlab-ci.ymlの設定

GitLab PATの設定

https://qiita.com/turupon/items/17ca6f3c770fe82ead38

上記記事を参考にKeyを取得しました。
今回リポジトリの読み書きが必要であるため「read_repository」「write_repository」を許可してください。

Anthropic API Keyの取得

https://console.anthropic.com/dashboard

上記より「Get Key」を押下しKeyを取得してください。(Claude Codeのクレジットを有効化が必要な点に注意です)

GitLabの対象のリポジトリに対し環境変数を設定

GitLab -> 対象のリポジトリ -> サイドメニューSettings -> CICDより
Variablesの項目から以下をVisibilityをMaskedを選択した上で設定してください。

  • ANTHROPIC_API_KEY
  • GITLAB_TOKEN(PAT)
  • GITLAB_USERNAME

.gitlab-ci.ymlの設定

私が作成したciのymlファイルになります

.gitlab-ci.yml

stages:
  - generate_tests
  - merge_request

claude:
  stage: generate_tests
  image: node:24-alpine3.21

  rules:
  - if: '$CI_PIPELINE_SOURCE == "web"'
    when: always

  - if: '$CI_PIPELINE_SOURCE == "push" && $RUN_AI_PIPELINE == "true"'
    when: always

  - when: never

  only:
  variables:
    GIT_STRATEGY: fetch
    GIT_DEPTH: 0  # MRベースブランチの履歴が必要
    CLAUDE_LOG: claude_output.log

  before_script:
    - set -e
    - apk add --no-cache git curl bash python3 py3-pip
    - npm install -g @anthropic-ai/claude-code
    - export PATH="$PATH:$(npm bin -g)"

  cache:
    key: claude-global-cache
    paths:
      - /usr/local/lib/node_modules/

  script:
    - echo "Running Claude Code to generate Python tests for MR !${CI_MERGE_REQUEST_IID}"
    - |
      git fetch origin ${CI_MERGE_REQUEST_TARGET_BRANCH_NAME:-$CI_DEFAULT_BRANCH}
      CHANGED_FILES=$(git diff --name-only origin/${CI_MERGE_REQUEST_TARGET_BRANCH_NAME:-$CI_DEFAULT_BRANCH})
      echo "Changed files: $CHANGED_FILES"
      if [ -z "$CHANGED_FILES" ]; then
        echo "No file changes detected. Skipping Claude execution."
        exit 0
      fi
    - |
      claude \
        --model claude-sonnet-4-5-20250929 \
        -p "Generate Python unit tests for the new or modified functions in merge request !${CI_MERGE_REQUEST_IID} of project ${CI_PROJECT_PATH}. \
        Use pytest conventions, place the tests in the 'tests/unit/' directory, and ensure each public function or class has at least one test case. \
        Avoid changing production code unless strictly necessary for testability." \
        --permission-mode acceptEdits \
        --allowedTools 'Read(*) Write(*) Edit(*) mcp__gitlab' \
        --debug | tee $CLAUDE_LOG

  artifacts:
    paths:
      - tests/
      - $CLAUDE_LOG
    expire_in: 24 hour


create_mr:
  stage: merge_request
  image: alpine:latest
  needs:
    - job: claude
      artifacts: true
  dependencies:
    - claude

  rules:
  - if: '$CI_PIPELINE_SOURCE == "web"'
    when: always

  - if: '$CI_PIPELINE_SOURCE == "push" && $RUN_AI_PIPELINE == "true"'
    when: always

  - when: never

  before_script:
    - set -e
    - apk add --no-cache git curl bash
    - git config --global user.email "ci-bot@creationline.com"
    - git config --global user.name "GitLab CI Bot"

  script:
    - export TARGET_BRANCH="${CI_MERGE_REQUEST_TARGET_BRANCH_NAME:-$CI_COMMIT_REF_NAME}"
    - export BRANCH_NAME="ai/generated-tests-${CI_MERGE_REQUEST_IID}-$(date +%Y%m%d-%H%M%S)"

    - git fetch origin $TARGET_BRANCH
    - git checkout -b ${BRANCH_NAME} origin/$TARGET_BRANCH

    - git add tests/ || true

    - if git diff --cached --quiet; then
        echo "No new test files to commit. Skipping MR creation.";
        exit 0;
      fi
  
    - git commit -m "Add AI-generated tests for MR !${CI_MERGE_REQUEST_IID}" || echo "No changes to commit"
    - git push -f https://gitlab-ci-token:${CI_JOB_TOKEN}@${CI_SERVER_HOST}/${CI_PROJECT_PATH}.git ${BRANCH_NAME}

    - |
      echo "Creating MR from ${BRANCH_NAME} -> ${TARGET_BRANCH}"
      curl --fail --request POST \
        --header "PRIVATE-TOKEN: ${GITLAB_TOKEN}" \
        "${CI_API_V4_URL}/projects/${CI_PROJECT_ID}/merge_requests" \
        --form "source_branch=${BRANCH_NAME}" \
        --form "target_branch=${TARGET_BRANCH}" \
        --form "title=Add AI-generated tests for MR !${CI_MERGE_REQUEST_IID}" \
        --form "description=This MR was automatically generated by Claude Code for !${CI_MERGE_REQUEST_IID}." \
        --form "labels=auto-generated,testing"

  artifacts:
    when: always
    expire_in: 24 hours

上記についての解説

rules
rules:
  - if: '$CI_PIPELINE_SOURCE == "web"'
    when: always

  - if: '$CI_PIPELINE_SOURCE == "push" && $RUN_AI_PIPELINE == "true"'
    when: always

  - when: never

GUIからの手動実行またはgit pushの際にオプションを渡した時のみ実行するように設定しています。

$ git push -o ci.variable="RUN_AI_PIPELINE=true" origin {ブランチ名}
generate_tests/scrpt
git fetch origin ${CI_MERGE_REQUEST_TARGET_BRANCH_NAME:-$CI_DEFAULT_BRANCH}
CHANGED_FILES=$(git diff --name-only origin/${CI_MERGE_REQUEST_TARGET_BRANCH_NAME:-$CI_DEFAULT_BRANCH})

カレントブランチとターゲットブランチの差分を取得し、変更されたファイルの情報を取得

claude \
  --model claude-sonnet-4-5-20250929 \
  -p " ... " \
  --permission-mode acceptEdits \
  --allowedTools 'Read(*) Write(*) Edit(*) mcp__gitlab' \
  --debug | tee $CLAUDE_LOG

--permission-mode...Claudeが自動でコードを変更できるよう許可
--allowedTools...ClaudeがGitLab操作などに使うツールの制限

実行してみる

今回items.py,db.pyに対し削除するメソッドを追加しテストコードを生成してみました。

追加したコード

items.py

@bp.route('/<int:item_id>', methods=['DELETE'])
def delete_item(item_id):
    try:
        deleted_rows = db.delete_item(item_id)

        if deleted_rows == 0:
            return jsonify({"error": f"ID {item_id} のデータが見つかりません"}), 404

        return jsonify({"message": f"ID {item_id} のアイテムを削除しました"}), 200

    except psycopg2.Error as e:
        return jsonify({"error": f"データ削除時にエラーが発生しました: {e.pgerror}"}), 500

db.py

def delete_item(item_id):
    conn = get_connection()
    cur = conn.cursor()

    cur.execute("DELETE FROM items WHERE id = %s", (item_id,))
    deleted_rows = cur.rowcount

    conn.commit()
    cur.close()
    conn.close()

    return deleted_rows

上記をリモートへpushしCIを走らせると以下コードがMRとして生成されました。

生成されたコード

test_db.py

class TestDeleteItem:
    """Test cases for delete_item function."""

    @patch('db.get_connection')
    def test_delete_item_success(self, mock_get_connection):
        """Test successful item deletion."""
        import db

        mock_conn = MagicMock()
        mock_cursor = MagicMock()
        mock_conn.cursor.return_value = mock_cursor
        mock_cursor.rowcount = 1
        mock_get_connection.return_value = mock_conn

        result = db.delete_item(1)

        assert result == 1
        mock_get_connection.assert_called_once()
        mock_conn.cursor.assert_called_once()
        mock_cursor.execute.assert_called_once_with(
            'DELETE FROM items WHERE id = %s',
            (1,)
        )
        mock_conn.commit.assert_called_once()
        mock_cursor.close.assert_called_once()
        mock_conn.close.assert_called_once()

    @patch('db.get_connection')
    def test_delete_item_not_found(self, mock_get_connection):
        """Test delete_item returns 0 when item doesn't exist."""
        import db

        mock_conn = MagicMock()
        mock_cursor = MagicMock()
        mock_conn.cursor.return_value = mock_cursor
        mock_cursor.rowcount = 0
        mock_get_connection.return_value = mock_conn

        result = db.delete_item(999)

        assert result == 0
        mock_cursor.execute.assert_called_once_with(
            'DELETE FROM items WHERE id = %s',
            (999,)
        )

    @patch('db.get_connection')
    def test_delete_item_database_error(self, mock_get_connection):
        """Test delete_item handles database errors."""
        import db

        mock_conn = MagicMock()
        mock_cursor = MagicMock()
        mock_conn.cursor.return_value = mock_cursor
        mock_cursor.execute.side_effect = psycopg2.Error("Delete failed")
        mock_get_connection.return_value = mock_conn

        with pytest.raises(psycopg2.Error, match="Delete failed"):
            db.delete_item(1)

    @patch('db.get_connection')
    def test_delete_item_with_large_id(self, mock_get_connection):
        """Test delete_item handles large ID numbers."""
        import db

        mock_conn = MagicMock()
        mock_cursor = MagicMock()
        mock_conn.cursor.return_value = mock_cursor
        mock_cursor.rowcount = 1
        mock_get_connection.return_value = mock_conn

        result = db.delete_item(999999999)

        assert result == 1
        mock_cursor.execute.assert_called_once_with(
            'DELETE FROM items WHERE id = %s',
            (999999999,)
        )

    @patch('db.get_connection')
    def test_delete_item_with_zero_id(self, mock_get_connection):
        """Test delete_item with zero ID."""
        import db

        mock_conn = MagicMock()
        mock_cursor = MagicMock()
        mock_conn.cursor.return_value = mock_cursor
        mock_cursor.rowcount = 0
        mock_get_connection.return_value = mock_conn

        result = db.delete_item(0)

        assert result == 0
        mock_cursor.execute.assert_called_once_with(
            'DELETE FROM items WHERE id = %s',
            (0,)
        )

    @patch('db.get_connection')
    def test_delete_item_connection_cleanup_on_error(self, mock_get_connection):
        """Test that connections are cleaned up even when delete fails."""
        import db

        mock_conn = MagicMock()
        mock_cursor = MagicMock()
        mock_conn.cursor.return_value = mock_cursor
        mock_cursor.execute.side_effect = psycopg2.Error("Delete failed")
        mock_get_connection.return_value = mock_conn

        with pytest.raises(psycopg2.Error):
            db.delete_item(1)

        # Connection should still be attempted to be created
        mock_get_connection.assert_called_once()

    @patch('db.get_connection')
    def test_delete_item_commit_is_called(self, mock_get_connection):
        """Test that commit is called after successful deletion."""
        import db

        mock_conn = MagicMock()
        mock_cursor = MagicMock()
        mock_conn.cursor.return_value = mock_cursor
        mock_cursor.rowcount = 1
        mock_get_connection.return_value = mock_conn

        db.delete_item(5)

        # Verify commit is called before closing
        assert mock_conn.commit.called
        assert mock_cursor.close.called
        assert mock_conn.close.called

    @patch('db.get_connection')
    def test_delete_item_multiple_calls(self, mock_get_connection):
        """Test multiple sequential delete operations."""
        import db

        mock_conn = MagicMock()
        mock_cursor = MagicMock()
        mock_conn.cursor.return_value = mock_cursor
        mock_cursor.rowcount = 1
        mock_get_connection.return_value = mock_conn

        result1 = db.delete_item(1)
        result2 = db.delete_item(2)

        assert result1 == 1
        assert result2 == 1
        assert mock_get_connection.call_count == 2

test_items.py

class TestDeleteItemRoute:
    """Test cases for DELETE /items/<id> endpoint."""

    @patch('routes.items.db.delete_item')
    def test_delete_item_success(self, mock_delete_item, client):
        """Test DELETE /items/<id> successfully deletes an item."""
        mock_delete_item.return_value = 1

        response = client.delete('/items/1')

        assert response.status_code == 200
        assert "message" in response.json
        assert "ID 1" in response.json["message"]
        assert "削除しました" in response.json["message"]
        mock_delete_item.assert_called_once_with(1)

    @patch('routes.items.db.delete_item')
    def test_delete_item_not_found(self, mock_delete_item, client):
        """Test DELETE /items/<id> returns 404 when item doesn't exist."""
        mock_delete_item.return_value = 0

        response = client.delete('/items/999')

        assert response.status_code == 404
        assert "error" in response.json
        assert "ID 999" in response.json["error"]
        assert "見つかりません" in response.json["error"]

    @patch('routes.items.db.delete_item')
    def test_delete_item_database_error(self, mock_delete_item, client):
        """Test DELETE /items/<id> handles database errors gracefully."""
        db_error = psycopg2.Error("Delete failed")
        db_error.pgerror = "Foreign key constraint violation"
        mock_delete_item.side_effect = db_error

        response = client.delete('/items/1')

        assert response.status_code == 500
        assert "error" in response.json
        assert "データ削除時にエラーが発生しました" in response.json["error"]

    @patch('routes.items.db.delete_item')
    def test_delete_item_with_large_id(self, mock_delete_item, client):
        """Test DELETE /items/<id> handles large ID numbers."""
        mock_delete_item.return_value = 1

        response = client.delete('/items/999999999')

        assert response.status_code == 200
        mock_delete_item.assert_called_once_with(999999999)

    def test_delete_item_with_invalid_id_type(self, client):
        """Test DELETE /items/<id> handles non-integer ID."""
        response = client.delete('/items/abc')

        assert response.status_code == 404

    @patch('routes.items.db.delete_item')
    def test_delete_item_with_zero_id(self, mock_delete_item, client):
        """Test DELETE /items/0 is handled correctly."""
        mock_delete_item.return_value = 0

        response = client.delete('/items/0')

        assert response.status_code == 404
        mock_delete_item.assert_called_once_with(0)

    @patch('routes.items.db.delete_item')
    def test_delete_item_returns_json_content_type(self, mock_delete_item, client):
        """Test DELETE /items/<id> returns JSON content type."""
        mock_delete_item.return_value = 1

        response = client.delete('/items/1')

        assert 'application/json' in response.content_type

    @patch('routes.items.db.delete_item')
    def test_delete_item_multiple_sequential_deletes(self, mock_delete_item, client):
        """Test multiple sequential delete operations."""
        mock_delete_item.return_value = 1

        response1 = client.delete('/items/1')
        response2 = client.delete('/items/2')
        response3 = client.delete('/items/3')

        assert response1.status_code == 200
        assert response2.status_code == 200
        assert response3.status_code == 200
        assert mock_delete_item.call_count == 3

    @patch('routes.items.db.delete_item')
    def test_delete_item_with_negative_id(self, mock_delete_item, client):
        """Test DELETE /items/<id> with negative ID."""
        mock_delete_item.return_value = 0

        response = client.delete('/items/-1')

        assert response.status_code == 404

    @patch('routes.items.db.delete_item')
    def test_delete_item_error_message_includes_pgerror(self, mock_delete_item, client):
        """Test that database error messages include PostgreSQL error details."""
        db_error = psycopg2.Error("Delete failed")
        db_error.pgerror = "Detailed PostgreSQL error message"
        mock_delete_item.side_effect = db_error

        response = client.delete('/items/1')

        assert response.status_code == 500
        assert "Detailed PostgreSQL error message" in response.json["error"]

終わりに

思いつきで着手してみましたがパパッと実装することができました。
今回は単にテストコードを生成させてみましたが、テストケースの厳格な指定やカバレッジに言及することなどコンテキストを持たせることでより実践に近づけれると感じたので試してみたいと思います。

Discussion