エピソード紹介
- Ep.1 – クリーンアーキテクチャとは
- Ep.2 – 認証方式の実践的な紹介
- Ep.3 – ER設計と監査ログ
- Ep.4 – RepoScanner の実装とテスト ← 今回はこちら
- Ep.5 – Copilot プロンプトを効率化
こんな方へ特におすすめ
- クリーンアーキテクチャの理屈は分かったけど、どこから書き始めるの？と疑問な方
- クリーンアーキテクチャで小さな MVP を実装するワークフローに興味がある方
- TDD（テスト駆動開発）を実務に取り入れて、壊れにくいコードを書きたい方
概要
こんにちは。サイオステクノロジーのはらちゃんです！
シリーズ4本目となる今回は、いよいよ待望の実装編に突入します。
ここで大きな役割を果たすのが TDD（テスト駆動開発） です。テストファーストで進めることで、依存性の切り離しや境界の明確化が自然と行われます。
—
本シリーズでは、Copilotを活用しつつ、クリーンアーキテクチャに沿って小規模なプロダクト「RepoScanner」を設計・実装した経緯をまとめます。
このエピソードは、アプリのコア機能である「リポジトリ内のスナップショット一覧を取得する機能」に焦点を当てました。
さらに、「AIへの指示の出し方」や「テストの質の変化」についてもお伝えします。
実装前の準備
プロジェクトの全体像
実装作業を開始する前に、まずは「RepoScanner」の構成を整理しておきます。
- 目的
リポジトリのメタデータや集計結果を効率よく取得し、分析しやすくすること。 - 要件
スナップショット一覧のページング（limit / offset）、最大取得数の制限。
クリーンなユースケースを保つ「3つの設計ルール」
RepoScannerにおけるUse Cases層では、以下のルールを徹底しました。
- 入出力はDTO（Data Transfer Object）
HTTPリクエストなどのオブジェクトは、そのまま使わず単純なデータクラスに変換する。
- 依存はインターフェースを介して注入（DI）
DB処理などは、具体的な実装ではなく「Repository」などの抽象的な型に依存させる。
- 副作用の入り口を明示
「どこで外部APIを呼ぶ」など処理の流れがユースケースから分かる状態を保つ。
このように責務を分離することで、ユースケースが純粋なビジネスロジックだけに集中できる環境が整います。
クリーンアーキテクチャによる設計
コードの保守性を高めるため、クリーンアーキテクチャに従って責務を明確に分離しました。
- Domain層
SnapshotSummaryなどのエンティティ - Use Cases層
ListSnapshotSummariesUseCase - Interface Adapters層
HTTP エンドポイントの制御 - Infrastructure層
SnapshotQueryRepository
ここで重要なのは、DBアクセスへの依存を必ず「インターフェース」経由にすることです。
その結果、内側のロジックが外側の技術的な詳細を知らなくて済む「依存性の逆転」が成立します。
設計上の要点
永続化（DBアクセス）への依存は、必ずリポジトリの「インターフェース」を経由させます。
これにより、内側（Use Cases層）が外側（Infrastructure層）の技術詳細を知らない状態となり、依存の逆転を保ちます。
→ 詳細はエピソード1へ
階層ごとのテスト戦略
「どこからテストを書けばいいか分からない」という悩みは、クリーンアーキテクチャで解決します。
なぜなら、各階層の役割がはっきりしているため、テストの目的も自ずと定まるからです。
最優先: Use Cases層のユニットテスト
ここでは「仕様としての正しさ」を検証します。limitの上限クリッピングや、offsetの負値チェックなどが対象です。
このテストはフレームワーク更新や DB ドライバ変更に影響されない高速なフィードバックを得ることが可能です。
例えば、以下のような正常系のテストを用意してください。
/tests/use_cases/test_list_snapshot_summaries.py
# 正常範囲の limit がそのままリポジトリへ渡されることを確認
class ListSnapshotSummariesUseCaseTest(unittest.TestCase):
def test_execute_passes_limit_when_within_bounds(self) -> None:
repository = FakeSnapshotQueryRepository()
use_case = ListSnapshotSummariesUseCase(repository, max_read_limit=25)
use_case.execute(limit=1, offset=0)
self.assertEqual(1, repository.received_limit)
use_case.execute(limit=25, offset=0)
self.assertEqual(25, repository.received_limit)- 外部依存はすべてモック化する
- ドメインモデルに正しく仕事を任せているか確認
優先: Infrastructure層のテスト
次に、「技術的な正しさ」を検証します。
SQLの組み立てが正しいか、LIMIT / OFFSETのパラメータ順序が間違っていないかといったパラメータ順序や DBから取得したデータをエンティティへ正しくマッピングできているかを確認できます。
/tests/frameworks&drivers/persistence/test_postgres_snapshot_query_repository.py
# フィルタなしで LIMIT/OFFSET がパラメータに含まれ、`ORDER BY s.observed_at desc` が含まれる確認
class PostgresSnapshotQueryRepositoryTest(unittest.TestCase):
def test_list_snapshot_summaries_without_filters(self) -> None:
now = datetime(2026, 2, 24, tzinfo=UTC)
cursor = _FakeCursor(
rows=[
{
"snapshot_id": UUID("aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa"),
"fetch_run_id": UUID("bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb"),
"target_repo_id": UUID("cccccccc-cccc-cccc-cccc-cccccccccccc"),
"owner": "owner",
"name": "repo",
"default_branch": "main",
"requested_at": now,
"observed_at": now,
"head_sha": "deadbeef",
"scan_scope": "full",
"status": "succeeded",
"trigger_type": "manual",
}
]
)
with patch(
"application.backend.src.infrastructure.persistence.postgres_snapshot_query_repository.get_cursor",
return_value=_CursorContext(cursor),
):
repository = PostgresSnapshotQueryRepository()
items = repository.list_snapshot_summaries(
limit=10, offset=5, user_id=None, target_repo_id=None
)
self.assertEqual([10, 5], cursor.executed_params)- どのようなSQL（またはリクエスト）を組み立てたか検証
- インターフェースの「型」と「変換」を確認
最終: 統合テスト／E2E
最後に、GitHub Actionsでのデータ抽出からDB保存、そしてアプリでの読み取りまで、システム全体のフローが本番に近い環境で動作するかを確かめます。
/tests/http/test_main_api.py
# ユースケースのクリッピングを利用して 200 を返すことを確認
class MainApiTest(unittest.TestCase):
def test_list_snapshots_clamps_limit_and_returns_200(self) -> None:
repository = _RecordingSnapshotRepository()
cast(Any, main).list_snapshot_summaries_use_case = ListSnapshotSummariesUseCase(
repository, max_read_limit=5
)
response = self.client.get("/snapshots?limit=999&offset=0")
self.assertEqual(200, response.status_code)
self.assertEqual(5, repository.received_limit)- 本物のデータベースを使用
- セットアップとクリーンアップの仕組みが必須
実装
TDD（テスト駆動開発）
進め方
TDDは、単にバグを防ぐためだけでなく、「設計を洗練させるためのツール」です。
以下の3サイクルで進めます。
- Red: 失敗
仕様の最小ケースを満たす「テストコード」を書く。まだ実装がないので当然エラー。 - Green: 成功
そのテストが通るように、最小限の「実装コード」を書く。 - Refactor: リファクタリング
テストが通る状態を保ったまま「設計ルール」に合わせてコードをきれいに整理。
例えば、「1回の取得上限は100件まで」というルールはAPIの仕様ではなくドメインの規則です。これをユースケース層で確実に担保することで、フレームワークに依存しない堅牢なロジックが完成します。
【実践】スナップショット取得ユースケース
前回設計したsnapshotテーブルに対して、「スナップショットを取得する」というユースケースをTDDで作ってみましょう。
Use Cases層のユニットテスト
tests/use_cases/test_register_snapshot.py
import unittest
from uuid import UUID
from datetime import UTC, datetime
from application.backend.src.domain.entities.snapshot_summary import SnapshotSummary
from application.backend.src.use_cases.list_snapshot_summaries import ListSnapshotSummariesUseCase
class _FakeRepo:
def __init__(self):
self.called = False
self.last_params = {}
def list_snapshot_summaries(self, *, limit, offset, user_id, target_repo_id):
self.called = True
self.last_params = dict(limit=limit, offset=offset, user_id=user_id, target_repo_id=target_repo_id)
now = datetime.now(UTC)
return [SnapshotSummary(
snapshot_id=UUID("aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa"),
fetch_run_id=UUID("bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb"),
target_repo_id=UUID("cccccccc-cccc-cccc-cccc-cccccccccccc"),
owner="owner", name="repo", default_branch="main",
requested_at=now, observed_at=now, head_sha="deadbeef",
scan_scope="full", status="succeeded", trigger_type="manual"
)]
class ListSnapshotSummariesUseCaseTDD(unittest.TestCase):
def test_offset_negative_raises_and_repo_not_called(self):
repo = _FakeRepo()
uc = ListSnapshotSummariesUseCase(repo, max_read_limit=100)
with self.assertRaises(ValueError):
uc.execute(limit=10, offset=-1)
self.assertFalse(repo.called)
def test_limit_zero_becomes_one(self):
repo = _FakeRepo()
uc = ListSnapshotSummariesUseCase(repo, max_read_limit=100)
uc.execute(limit=0, offset=0)
self.assertEqual(1, repo.last_params["limit"])
def test_limit_clamped_to_max(self):
repo = _FakeRepo()
uc = ListSnapshotSummariesUseCase(repo, max_read_limit=25)
uc.execute(limit=999, offset=0)
self.assertEqual(25, repo.last_params["limit"])
def test_passes_filters_and_offset(self):
repo = _FakeRepo()
uc = ListSnapshotSummariesUseCase(repo, max_read_limit=100)
user_id = UUID("11111111-2222-3333-4444-555555555555")
target_repo_id = UUID("66666666-7777-8888-9999-aaaaaaaaaaaa")
uc.execute(limit=10, offset=5, user_id=user_id, target_repo_id=target_repo_id)
self.assertEqual(5, repo.last_params["offset"])
self.assertEqual(user_id, repo.last_params["user_id"])
self.assertEqual(target_repo_id, repo.last_params["target_repo_id"])
if __name__ == "__main__":
unittest.main()仕上がったら想定するエラーかどうか、テストを走らせてみることをお勧めします。
Use Cases層
テストを書いた後で、初めて ListSnapshotSummariesUseCase クラスを実装します。
python
class ListSnapshotSummariesUseCase:
def __init__(self, snapshot_query_repository: SnapshotQueryRepository, max_read_limit: int) -> None:
self.snapshot_query_repository = snapshot_query_repository
self.max_read_limit = max_read_limit
def execute(self, *, limit: int, offset: int, user_id: UUID | None = None, target_repo_id: UUID | None = None) -> list[SnapshotSummary]:
if offset < 0:
raise ValueError("offset must be greater than or equal to 0")
safe_limit = max(1, min(limit, self.max_read_limit))
return self.snapshot_query_repository.list_snapshot_summaries(
limit=safe_limit, offset=offset, user_id=user_id, target_repo_id=target_repo_id
)- ペイロード検証をUse Cases層で記述
Interface Adapters層のバリデーションだけに依存せず、ドメインルールとしてテスト可能に
このように1つテストを作成したら1つ実装するサイクルを作ると、意図したコード実装になりやすいです。
プロジェクトによって、1ファイルごとにするか全体のテストを先に作ってしまうかは調整すべきだと感じました。
Interface Adapters層
ユースケースの実装後、モック化していたDBへの具体的な接続処理は、後からインターフェースの中身として差し替えます。
python
class PostgresSnapshotQueryRepository(SnapshotQueryRepository):
def list_snapshot_summaries(self, *, limit: int, offset: int, user_id: UUID | None, target_repo_id: UUID | None) -> list[SnapshotSummary]:
conditions: list[sql.Composable] = []
params: list[object] = []
if user_id is not None:
conditions.append(sql.SQL("fr.user_id = %s"))
params.append(user_id)
if target_repo_id is not None:
conditions.append(sql.SQL("fr.target_repo_id = %s"))
params.append(target_repo_id)
# where_clause 組み立て、limit/offset を params に追加して実行Infrastructure層
さいごに、マイグレーションは以下のように追加します。
create table history_event (
history_event_id uuid primary key default gen_random_uuid(),
user_id uuid not null references app_user(user_id) on delete restrict,
fetch_run_id uuid references fetch_run(fetch_run_id) on delete set null,
event_type text not null,
happened_at timestamptz not null default now(),
summary text not null
);
create table operation_log (
operation_log_id uuid primary key default gen_random_uuid(),
actor_user_id uuid references app_user(user_id) on delete set null,
action text not null,
target_type text not null,
target_id text,
happened_at timestamptz not null default now(),
result text not null check (result in ('success', 'failure')),
error_code text,
trace_id text not null
);- 監査・運用観点で
operation_logを用意
CI 実行の失敗や権限エラーをすぐ追跡できる
Copilotを活用するプロンプト術
クリーンアーキテクチャの骨格を作る際、AIにゼロからコードを書かせると、層の境界が曖昧になりがちです。
そこで、「型」と「依存関係」を明示することで、一発で実用的なコードを得ることができます。
実際のプロンプト例
以下のように指示を出すと、Copilotはビジネスロジックを純粋なPythonコードとして抽出し、テストしやすい構造で出力してくれます。
RepoScanner のスナップショット一覧を返す `ListSnapshotSummariesUseCase` を実装してください。
入力: limit, offset, optional filter
出力: SnapshotSummary のリストと total_count
既存エンティティ: SnapshotSummary(id, repo_name, observed_at, summary)
依存: SnapshotQueryRepository.list_snapshots(limit, offset, filter) -> (items, total)
テスト（unittest スタイル）を先に示してください。生成されたコード（抜粋）
プロンプトの指示通り、まずはUse Cases層のコードが生成されます。
APIの検証とは別に、ビジネスロジックが純粋なPythonコードとして抽出され、テストしやすい構造になりました。
application/backend/src/use_cases/list_snapshot_summaries.py
class ListSnapshotSummariesUseCase:
def __init__(self, repo: SnapshotQueryRepository, max_read_limit: int = 100):
self.repo = repo
self.max_read_limit = max_read_limit
def execute(self, limit: int, offset: int = 0):
if limit <= 0:
raise ValueError("limit must be > 0")
if offset < 0:
raise ValueError("offset must be >= 0")
limit = min(limit, self.max_read_limit)
return self.repo.list_snapshots(limit=limit, offset=offset)Interface Adapters層はリクエストを受け取り、ユースケースを呼び出すだけの「薄い」層になります。
Python
@app.get("/snapshots")
def get_snapshots(limit: int = 20, offset: int = 0, repo=Depends(get_snapshot_repo)):
items, total = ListSnapshotSummariesUseCase(repo).execute(limit=limit, offset=offset)
return {"items": [i.to_dict() for i in items], "total": total}おまけ: GitHub Actions
エピソード2では、実行環境の分離や運用コストから、GitHub Actions による認証と実行をメインに据える決断をしました。
認証まわりのワークフロー（抜粋）
permissions:
contents: read
pull-requests: read
issues: read
jobs:
collect:
steps:
- name: Collect PR snapshot and persist
env:
GITHUB_TOKEN: ${{ github.token }}
DATABASE_URL: ${{ secrets.DATABASE_URL }}-
permissionsを明示し、github.token（実行時トークン）を活用 - 外部 DB への書き込みには
DATABASE_URLといった修飾済みのシークレットを利用
まとめ
今回は、「テストを先に書くことで、自然と依存を切り離す設計になる」という、TDDとクリーンアーキテクチャの相性の良さが伝わるご紹介をしました。
- 「どうテストするか」を考えるTDDは設計を導く最強のツール
- ユースケースは小さく、純粋に作り、外部の仕組みに依存しない純粋なビジネスロジックを保つ
- AI には型と依存関係を伝えることで、開発効率が劇的に向上する
エピソード5では、さらに一歩踏み込んだCopilot運用術についてお話しします。お楽しみに！
参考
