IntellijのMCP Serverプラグインを使ってAI駆動開発をおこなってみる

{
  "mcpServers": {
    "jetbrains": {
      "command": "npx",
      "args": ["-y", "@jetbrains/mcp-proxy"]
    }
  }
}

$ cat claude_desktop_config.json 
{
  "mcpServers": {
    "jetbrains": {
      "command": "npx",
      "args": ["-y", "@jetbrains/mcp-proxy"]
    }
  }
}

import boto3
import pandas as pd
import duckdb
from typing import List, Dict, Any

class DuckDBManager:
    def __init__(self, db_path: str, is_read_only: bool = True):
        """
        Args:
            db_path: データベースファイルのパス
            is_read_only: コネクションのタイプ
        """
        self.db_path = db_path
        self.is_read_only = is_read_only
        self.connection = None

    def __enter__(self):
        """コンテキストマネージャーの開始でデータベースに接続"""
        try:
            self.connection = duckdb.connect(self.db_path, read_only=self.is_read_only)
        except Exception as e:
            print(e)
            raise
        return self.connection

    def __exit__(self, exc_type, exc_val, exc_tb):
        """コンテキストマネージャーの終了で接続を閉じる"""
        if self.connection:
            try:
                self.connection.close()
            except Exception as e:
                print(e)

def _deserialize_dynamodb_item(item: Dict[str, Any]) -> Dict[str, Any]:
    """DynamoDBのデータ型をPythonのネイティブ型に変換"""
    result = {}
    for key, value in item.items():
        # DynamoDBの型に応じて変換
        if "S" in value:  # 文字列
            result[key] = value["S"]
        elif "N" in value:  # 数値
            result[key] = float(value["N"])
        elif "BOOL" in value:  # ブール値
            result[key] = value["BOOL"]
    return result

def fetch_from_dynamodb(table_name: str) -> List[Dict[str, Any]]:
    """DynamoDBからデータを取得"""
    dynamodb = boto3.client("dynamodb")

    items = []
    paginator = dynamodb.get_paginator("scan")

    try:
        # ページネーションを使用して全データを取得
        for page in paginator.paginate(TableName=table_name):
            for item in page["Items"]:
                deserialized_item = _deserialize_dynamodb_item(item)
                items.append(deserialized_item)

        return items

    except Exception as e:
        print(f"Error fetching data from DynamoDB: {e}")
        raise

def load_to_duckdb(db_path: str, table_name: str):
    """データをDuckDBに読み込む"""
    try:
        data = fetch_from_dynamodb(table_name)
        # DataFrameに変換
        df = pd.DataFrame(data)

        with DuckDBManager(db_path, False) as conn:
            # DuckDBにテーブルとして保存
            conn.register("temp_view", df)
            conn.execute(f"""
                CREATE TABLE IF NOT EXISTS {table_name} AS 
                SELECT * FROM temp_view
            """)

    except Exception as e:
        print(f"Error loading data to DuckDB: {e}")
        raise

def main():
    load_to_duckdb("dynamodb.db", "test_user")

if __name__ == "__main__":
    main()

JetBrainsで開いているファイル一覧を教えて

d2d.pyは何をするファイルですか？

d2d.pyではmain関数内でload_to_duckdbでtest_userテーブルだけなのでコードを変更したい
- DynamoDBの一覧を取得してテーブル名がtestから始まるテーブルをすべて取り込むようにしたい
- 作成したテーブル数とデータ件数を最後に出力してほしい

import boto3
import pandas as pd
import duckdb
from typing import List, Dict, Any

class DuckDBManager:
    def __init__(self, db_path: str, is_read_only: bool = True):
        """
        Args:
            db_path: データベースファイルのパス
            is_read_only: コネクションのタイプ
        """
        self.db_path = db_path
        self.is_read_only = is_read_only
        self.connection = None

    def __enter__(self):
        """コンテキストマネージャーの開始でデータベースに接続"""
        try:
            self.connection = duckdb.connect(self.db_path, read_only=self.is_read_only)
        except Exception as e:
            print(e)
            raise
        return self.connection

    def __exit__(self, exc_type, exc_val, exc_tb):
        """コンテキストマネージャーの終了で接続を閉じる"""
        if self.connection:
            try:
                self.connection.close()
            except Exception as e:
                print(e)

def _deserialize_dynamodb_item(item: Dict[str, Any]) -> Dict[str, Any]:
    """DynamoDBのデータ型をPythonのネイティブ型に変換"""
    result = {}
    for key, value in item.items():
        # DynamoDBの型に応じて変換
        if "S" in value:  # 文字列
            result[key] = value["S"]
        elif "N" in value:  # 数値
            result[key] = float(value["N"])
        elif "BOOL" in value:  # ブール値
            result[key] = value["BOOL"]
    return result

def fetch_from_dynamodb(table_name: str) -> List[Dict[str, Any]]:
    """DynamoDBからデータを取得"""
    dynamodb = boto3.client("dynamodb")

    items = []
    paginator = dynamodb.get_paginator("scan")

    try:
        # ページネーションを使用して全データを取得
        for page in paginator.paginate(TableName=table_name):
            for item in page["Items"]:
                deserialized_item = _deserialize_dynamodb_item(item)
                items.append(deserialized_item)

        return items

    except Exception as e:
        print(f"Error fetching data from DynamoDB: {e}")
        raise

def load_to_duckdb(db_path: str, table_name: str):
    """データをDuckDBに読み込む"""
    try:
        data = fetch_from_dynamodb(table_name)
        # DataFrameに変換
        df = pd.DataFrame(data)

        with DuckDBManager(db_path, False) as conn:
            # DuckDBにテーブルとして保存
            conn.register("temp_view", df)
            conn.execute(f"""
                CREATE TABLE IF NOT EXISTS {table_name} AS 
                SELECT * FROM temp_view
            """)

    except Exception as e:
        print(f"Error loading data to DuckDB: {e}")
        raise

def list_dynamodb_tables():
    """DynamoDBのテーブル一覧を取得"""
    dynamodb = boto3.client("dynamodb")

    tables = []
    paginator = dynamodb.get_paginator("list_tables")

    try:
        for page in paginator.paginate():
            tables.extend(page["TableNames"])
        return tables
    except Exception as e:
        print(f"Error listing DynamoDB tables: {e}")
        raise

def get_table_count(conn, table_name):
    """テーブルのレコード数を取得"""
    try:
        result = conn.execute(f"SELECT COUNT(*) FROM {table_name}").fetchone()
        return result[0] if result else 0
    except Exception as e:
        print(f"Error getting count for table {table_name}: {e}")
        return 0

def main():
    db_path = "dynamodb.db"
    # テスト用テーブルを格納するリスト
    processed_tables = []
    total_records = 0

    # DynamoDBのテーブル一覧を取得
    tables = list_dynamodb_tables()

    # "test_"で始まるテーブルを処理
    for table_name in tables:
        if table_name.startswith("test_"):
            try:
                print(f"Processing table: {table_name}")
                load_to_duckdb(db_path, table_name)
                processed_tables.append(table_name)
            except Exception as e:
                print(f"Error processing table {table_name}: {e}")

    # 結果の出力
    with DuckDBManager(db_path, True) as conn:
        print("\n===== 処理結果 =====")
        print(f"取り込んだテーブル数: {len(processed_tables)}")

        if processed_tables:
            print("\nテーブル別レコード数:")
            for table in processed_tables:
                count = get_table_count(conn, table)
                total_records += count
                print(f"  - {table}: {count}件")

            print(f"\n合計レコード数: {total_records}件")

if __name__ == "__main__":
    main()

< def list_dynamodb_tables():
<     """DynamoDBのテーブル一覧を取得"""
<     dynamodb = boto3.client("dynamodb")
< 
<     tables = []
<     paginator = dynamodb.get_paginator("list_tables")
< 
<     try:
<         for page in paginator.paginate():
<             tables.extend(page["TableNames"])
<         return tables
<     except Exception as e:
<         print(f"Error listing DynamoDB tables: {e}")
<         raise
< 
< 
< def get_table_count(conn, table_name):
<     """テーブルのレコード数を取得"""
<     try:
<         result = conn.execute(f"SELECT COUNT(*) FROM {table_name}").fetchone()
<         return result[0] if result else 0
<     except Exception as e:
<         print(f"Error getting count for table {table_name}: {e}")
<         return 0
< 
< 
118,121c92
<     db_path = "dynamodb.db"
<     # テスト用テーブルを格納するリスト
<     processed_tables = []
<     total_records = 0
---
>     load_to_duckdb("dynamodb.db", "test_user")
123,124d93
<     # DynamoDBのテーブル一覧を取得
<     tables = list_dynamodb_tables()
126,150d94
<     # "test_"で始まるテーブルを処理
<     for table_name in tables:
<         if table_name.startswith("test_"):
<             try:
<                 print(f"Processing table: {table_name}")
<                 load_to_duckdb(db_path, table_name)
<                 processed_tables.append(table_name)
<             except Exception as e:
<                 print(f"Error processing table {table_name}: {e}")
< 
<     # 結果の出力
<     with DuckDBManager(db_path, True) as conn:
<         print("\n===== 処理結果 =====")
<         print(f"取り込んだテーブル数: {len(processed_tables)}")
< 
<         if processed_tables:
<             print("\nテーブル別レコード数:")
<             for table in processed_tables:
<                 count = get_table_count(conn, table)
<                 total_records += count
<                 print(f"  - {table}: {count}件")
< 
<             print(f"\n合計レコード数: {total_records}件")
< 
< 
(venv)$ diff d2d_org.py d2d.py                                                                                   
90a91,116
> def list_dynamodb_tables():
>     """DynamoDBのテーブル一覧を取得"""
>     dynamodb = boto3.client("dynamodb")
> 
>     tables = []
>     paginator = dynamodb.get_paginator("list_tables")
> 
>     try:
>         for page in paginator.paginate():
>             tables.extend(page["TableNames"])
>         return tables
>     except Exception as e:
>         print(f"Error listing DynamoDB tables: {e}")
>         raise
> 
> 
> def get_table_count(conn, table_name):
>     """テーブルのレコード数を取得"""
>     try:
>         result = conn.execute(f"SELECT COUNT(*) FROM {table_name}").fetchone()
>         return result[0] if result else 0
>     except Exception as e:
>         print(f"Error getting count for table {table_name}: {e}")
>         return 0
> 
> 
92c118,121
<     load_to_duckdb("dynamodb.db", "test_user")
---
>     db_path = "dynamodb.db"
>     # テスト用テーブルを格納するリスト
>     processed_tables = []
>     total_records = 0
93a123,124
>     # DynamoDBのテーブル一覧を取得
>     tables = list_dynamodb_tables()
94a126,150
>     # "test_"で始まるテーブルを処理
>     for table_name in tables:
>         if table_name.startswith("test_"):
>             try:
>                 print(f"Processing table: {table_name}")
>                 load_to_duckdb(db_path, table_name)
>                 processed_tables.append(table_name)
>             except Exception as e:
>                 print(f"Error processing table {table_name}: {e}")
> 
>     # 結果の出力
>     with DuckDBManager(db_path, True) as conn:
>         print("\n===== 処理結果 =====")
>         print(f"取り込んだテーブル数: {len(processed_tables)}")
> 
>         if processed_tables:
>             print("\nテーブル別レコード数:")
>             for table in processed_tables:
>                 count = get_table_count(conn, table)
>                 total_records += count
>                 print(f"  - {table}: {count}件")
> 
>             print(f"\n合計レコード数: {total_records}件")
> 
> 

$ python d2d.py 
Processing table: test_group
Processing table: test_sales
Processing table: test_user

===== 処理結果 =====
取り込んだテーブル数: 3

テーブル別レコード数:
  - test_group: 4件
  - test_sales: 6件
  - test_user: 800件

合計レコード数: 810件

d2d.pyのSmall testを作成したいです。
* テストはpytestを使って記述してください
* botoを使っている箇所はテストダブルでmotoを使ってください
* パラメータの異なるテストはpytest.mark.parametrizeを使ってください
* Small Testなので他のメソッドを使用している箇所はmock化してください
* Mock化した際は適切に使われていることを確認するためcall_countで確認してください
* テスト内ではGiven-When-Then構文を使ってテスト操作を分けてください
* カバレッジはC1で80%を達成してください

import pytest
import pandas as pd
import duckdb
import boto3
from moto import mock_dynamodb
from unittest.mock import patch, MagicMock, call
from d2d import (
    DuckDBManager,
    _deserialize_dynamodb_item,
    fetch_from_dynamodb,
    load_to_duckdb,
    list_dynamodb_tables,
    get_table_count,
    main
)

class TestDuckDBManager:
    def test_context_manager(self):
        """
        DuckDBManagerのコンテキストマネージャー機能が正常に動作することをテストする。
        メモリ上に一時的なDBを作成し、テーブル作成と簡単なクエリが実行できることを確認する。
        """
        # Given
        db_path = ":memory:"

        # When
        with DuckDBManager(db_path, is_read_only=False) as conn:
            # Then
            assert conn is not None
            # データベースが機能することを確認
            conn.execute("CREATE TABLE test (id INTEGER, name VARCHAR)")
            conn.execute("INSERT INTO test VALUES (1, 'test')")
            result = conn.execute("SELECT * FROM test").fetchall()
            assert result == [(1, 'test')]

    def test_context_manager_exception(self):
        """
        DuckDBManagerの接続失敗時の例外処理を検証するテスト。
        データベース接続時に例外が発生した場合、適切に例外が伝播することを確認する。
        """
        # Given
        db_path = ":memory:"

        # When/Then
        with patch('duckdb.connect', side_effect=Exception("Connection error")):
            with pytest.raises(Exception) as exc_info:
                with DuckDBManager(db_path, is_read_only=False) as conn:
                    pass
            assert "Connection error" in str(exc_info.value)

    def test_context_manager_exit_exception(self):
        """
        DuckDBManagerのクローズ処理での例外をテスト。
        データベース接続のクローズ時に例外が発生しても、例外がコンテキストマネージャから
        外部に伝播せず、プログラムが続行できることを確認する。
        """
        # Given
        db_path = ":memory:"

        # When
        with patch.object(duckdb.DuckDBPyConnection, 'close', side_effect=Exception("Close error")):
            # Thenはassertがないが、例外が発生してもコンテキストマネージャが
            # プログラムを中断しないことを暗黙的にテスト
            with DuckDBManager(db_path, is_read_only=False) as conn:
                pass

class TestDeserializeItem:
    @pytest.mark.parametrize("item, expected", [
        ({"name": {"S": "test"}}, {"name": "test"}),
        ({"age": {"N": "25"}}, {"age": 25.0}),
        ({"active": {"BOOL": True}}, {"active": True}),
        ({"mixed": {"S": "text"}, "num": {"N": "10"}}, {"mixed": "text", "num": 10.0}),
        ({}, {})
    ])
    def test_deserialize_dynamodb_item(self, item, expected):
        """
        DynamoDBの項目データ形式をPythonのネイティブデータ型に変換する関数をテスト。
        文字列(S)、数値(N)、真偽値(BOOL)などの様々なDynamoDBデータ型の変換が
        正しく行われることを確認する。複数のテストケースをパラメータ化して検証する。
        """
        # Given - item and expected are provided by parametrize

        # When
        result = _deserialize_dynamodb_item(item)

        # Then
        assert result == expected

@mock_dynamodb
class TestDynamoDBOperations:
    def setup_method(self, method):
        """
        各テストメソッド実行前にDynamoDBのモックテーブルを作成する。
        テスト用のテーブルと2つのテストデータを挿入し、後続のテストの準備を行う。
        """
        self.dynamodb = boto3.resource('dynamodb')
        self.table_name = 'test_table'

        # テーブルの作成
        table = self.dynamodb.create_table(
            TableName=self.table_name,
            KeySchema=[{'AttributeName': 'id', 'KeyType': 'HASH'}],
            AttributeDefinitions=[{'AttributeName': 'id', 'AttributeType': 'S'}],
            ProvisionedThroughput={'ReadCapacityUnits': 5, 'WriteCapacityUnits': 5}
        )

        # テストデータの挿入
        table.put_item(Item={'id': 'item1', 'name': 'Item 1', 'value': 100})
        table.put_item(Item={'id': 'item2', 'name': 'Item 2', 'value': 200})

    def test_fetch_from_dynamodb(self):
        """
        DynamoDBからデータを取得する関数のテスト。
        モックDynamoDBテーブルから正しく全データを取得できることを確認する。
        取得したデータの数と内容が期待通りであることを検証する。
        """
        # Given
        # setup_methodでテーブルとデータが設定済み

        # When
        result = fetch_from_dynamodb(self.table_name)

        # Then
        assert len(result) == 2
        items = {item.get('id'): item for item in result}
        assert 'item1' in items
        assert 'item2' in items

    def test_fetch_from_dynamodb_error(self):
        """
        DynamoDBデータ取得時のエラー処理をテスト。
        存在しないテーブル名を指定した場合に適切に例外が発生することを確認する。
        """
        # Given
        # 存在しないテーブル名
        non_existent_table = 'non_existent_table'

        # When/Then
        with pytest.raises(Exception):
            fetch_from_dynamodb(non_existent_table)

    def test_list_dynamodb_tables(self):
        """
        DynamoDBのテーブル一覧を取得する関数のテスト。
        モックDynamoDBに作成したテーブルが一覧に含まれているかを確認する。
        """
        # Given
        # setup_methodでテーブルが設定済み

        # When
        tables = list_dynamodb_tables()

        # Then
        assert self.table_name in tables
        assert len(tables) >= 1

    @patch('d2d.fetch_from_dynamodb')
    def test_load_to_duckdb(self, mock_fetch):
        """
        DynamoDBからデータを取得してDuckDBに保存する関数をテスト。
        データ取得とDuckDBへの保存が正しい順序で呼び出されることを確認する。
        モックを使用してDuckDBへの実際の保存操作をシミュレートする。
        """
        # Given
        test_data = [
            {'id': 'item1', 'name': 'Item 1', 'value': 100},
            {'id': 'item2', 'name': 'Item 2', 'value': 200}
        ]
        mock_fetch.return_value = test_data
        db_path = ":memory:"
        table_name = "test_table"

        # When
        with patch('d2d.DuckDBManager') as mock_manager:
            mock_conn = MagicMock()
            mock_manager.return_value.__enter__.return_value = mock_conn
            load_to_duckdb(db_path, table_name)

        # Then
        mock_fetch.assert_called_once_with(table_name)
        mock_conn.register.assert_called_once()
        mock_conn.execute.assert_called_once()

    @patch('d2d.fetch_from_dynamodb')
    def test_load_to_duckdb_error(self, mock_fetch):
        """
        DynamoDBからDuckDBへのデータ読み込み時のエラー処理をテスト。
        DynamoDBからのデータ取得で例外が発生した場合に、適切なエラーメッセージとともに
        例外が再発生することを確認する。
        """
        # Given
        mock_fetch.side_effect = Exception("Fetch error")
        db_path = ":memory:"
        table_name = "test_table"

        # When/Then
        with pytest.raises(Exception) as exc_info:
            load_to_duckdb(db_path, table_name)

        # Then
        mock_fetch.assert_called_once_with(table_name)
        assert "Fetch error" in str(exc_info.value)

class TestTableOperations:
    def test_get_table_count(self):
        """
        DuckDBテーブルのレコード数を取得する関数をテスト。
        テスト用テーブルに3つのレコードを挿入し、カウント関数が正確に数を返すことを確認する。
        """
        # Given
        conn = duckdb.connect(":memory:")
        table_name = "test_table"
        conn.execute(f"CREATE TABLE {table_name} (id INTEGER, name VARCHAR)")
        conn.execute(f"INSERT INTO {table_name} VALUES (1, 'test1'), (2, 'test2'), (3, 'test3')")

        # When
        count = get_table_count(conn, table_name)

        # Then
        assert count == 3

        # クリーンアップ
        conn.close()

    def test_get_table_count_error(self):
        """
        存在しないテーブルのレコード数取得時の動作をテスト。
        存在しないテーブルに対して関数を実行した場合、エラーを発生させず0を返すことを確認する。
        """
        # Given
        conn = duckdb.connect(":memory:")
        table_name = "non_existent_table"

        # When
        count = get_table_count(conn, table_name)

        # Then
        assert count == 0

        # クリーンアップ
        conn.close()

@mock_dynamodb
class TestMain:
    def setup_method(self, method):
        """
        mainメソッドのテスト用にDynamoDBのモックテーブルを複数作成する。
        'test_'で始まるテーブルと'prod_'で始まるテーブルを作成し、
        選択的な処理の検証を可能にする。
        """
        self.dynamodb = boto3.resource('dynamodb')

        # テスト用テーブルの作成
        table_names = ['test_table1', 'test_table2', 'prod_table']

        for table_name in table_names:
            table = self.dynamodb.create_table(
                TableName=table_name,
                KeySchema=[{'AttributeName': 'id', 'KeyType': 'HASH'}],
                AttributeDefinitions=[{'AttributeName': 'id', 'AttributeType': 'S'}],
                ProvisionedThroughput={'ReadCapacityUnits': 5, 'WriteCapacityUnits': 5}
            )

            # テストデータの挿入
            table.put_item(Item={'id': 'item1', 'name': 'Item 1', 'value': 100})

    @patch('d2d.load_to_duckdb')
    @patch('d2d.get_table_count')
    @patch('d2d.DuckDBManager')
    def test_main(self, mock_manager, mock_get_count, mock_load):
        """
        メイン処理関数の正常系をテスト。
        'test_'で始まるテーブルのみが処理対象となり、それぞれに対して
        load_to_duckdbとget_table_countが呼ばれることを確認する。
        'prod_'で始まるテーブルは処理されないことも検証する。
        """
        # Given
        mock_get_count.return_value = 1
        mock_conn = MagicMock()
        mock_manager.return_value.__enter__.return_value = mock_conn

        # When
        main()

        # Then
        # load_to_duckdbが2回呼ばれる（test_で始まるテーブルが2つあるため）
        assert mock_load.call_count == 2
        # get_table_countが2回呼ばれる（処理されたテーブルが2つあるため）
        assert mock_get_count.call_count == 2

        # test_で始まるテーブルのみが処理されることを確認
        mock_load.assert_has_calls([
            call('dynamodb.db', 'test_table1'),
            call('dynamodb.db', 'test_table2')
        ], any_order=True)

    @patch('d2d.load_to_duckdb')
    @patch('d2d.get_table_count')
    @patch('d2d.DuckDBManager')
    def test_main_with_error(self, mock_manager, mock_get_count, mock_load):
        """
        メイン処理関数のエラー処理をテスト。
        テーブル処理中にエラーが発生しても処理が中断せず、残りのテーブルの処理を
        継続することを確認する。エラーが発生した場合はget_table_countが呼ばれないことも検証する。
        """
        # Given
        mock_load.side_effect = Exception("Load error")
        mock_conn = MagicMock()
        mock_manager.return_value.__enter__.return_value = mock_conn

        # When
        main()  # エラーが発生しても処理は続行

        # Then
        # エラーが発生してもすべてのtest_テーブルで処理が試行される
        assert mock_load.call_count == 2
        # エラーのためget_table_countは呼ばれない
        assert mock_get_count.call_count == 0

if __name__ == "__main__":
    pytest.main(["-xvs", "test_d2d.py"])

$ pip install pytest pytest-cov moto
$ pytest test_d2d.py -v    
======================================================================================================= test session starts =======================================================================================================
platform darwin -- Python 3.11.4, pytest-7.4.3, pluggy-1.3.0 -- python
cachedir: .pytest_cache
rootdir: jetbrains_mcp
plugins: xdist-3.5.0, base-url-2.1.0, mock-3.14.0, playwright-0.5.2, anyio-3.7.0
collected 17 items                                                                                                                                                                                                                

test_d2d.py::TestDuckDBManager::test_context_manager PASSED                                                                                                                                                                 [  5%]
test_d2d.py::TestDuckDBManager::test_context_manager_exception PASSED                                                                                                                                                       [ 11%]
test_d2d.py::TestDuckDBManager::test_context_manager_exit_exception PASSED                                                                                                                                                  [ 17%]
test_d2d.py::TestDeserializeItem::test_deserialize_dynamodb_item[item0-expected0] PASSED                                                                                                                                    [ 23%]
test_d2d.py::TestDeserializeItem::test_deserialize_dynamodb_item[item1-expected1] PASSED                                                                                                                                    [ 29%]
test_d2d.py::TestDeserializeItem::test_deserialize_dynamodb_item[item2-expected2] PASSED                                                                                                                                    [ 35%]
test_d2d.py::TestDeserializeItem::test_deserialize_dynamodb_item[item3-expected3] PASSED                                                                                                                                    [ 41%]
test_d2d.py::TestDeserializeItem::test_deserialize_dynamodb_item[item4-expected4] PASSED                                                                                                                                    [ 47%]
test_d2d.py::TestDynamoDBOperations::test_fetch_from_dynamodb PASSED                                                                                                                                                        [ 52%]
test_d2d.py::TestDynamoDBOperations::test_fetch_from_dynamodb_error PASSED                                                                                                                                                  [ 58%]
test_d2d.py::TestDynamoDBOperations::test_list_dynamodb_tables PASSED                                                                                                                                                       [ 64%]
test_d2d.py::TestDynamoDBOperations::test_load_to_duckdb PASSED                                                                                                                                                             [ 70%]
test_d2d.py::TestDynamoDBOperations::test_load_to_duckdb_error PASSED                                                                                                                                                       [ 76%]
test_d2d.py::TestTableOperations::test_get_table_count PASSED                                                                                                                                                               [ 82%]
test_d2d.py::TestTableOperations::test_get_table_count_error PASSED                                                                                                                                                         [ 88%]
test_d2d.py::TestMain::test_main PASSED                                                                                                                                                                                     [ 94%]
test_d2d.py::TestMain::test_main_with_error PASSED                                                                                                                                                                          [100%]

======================================================================================================= 17 passed in 1.80s ========================================================================================================