モック、彼らは何ですか？

このテキストは、ここと私の個人的なブログで紹介するデータ処理アプリケーションのテストに関する一連のテキストの最初のものです。

私がソフトウェア エンジニアからデータ エンジニアにキャリアを移行したとき、ソフトウェア エンジニアリングの背景を持たないデータ分野の人々と会話をするようになりました。これらの会話の中で、次のような質問が繰り返し生じました: テストはどうやって書くのですか?

実際、テストを書くことは、コードの書き方を変える必要があるため、慣れていない人にとっては複雑な作業のように思えるかもしれません。真実は、神秘などはなく、むしろ練習と反復の問題であるということです。この記事の主な目的は、データを処理するアプリケーションのテストを作成し、コードの品質と信頼性を確保する方法を示すプロセスを、始めたばかりのユーザーに案内することです。

このテキストは、データ エンジニアリングを目的としたコードで自動テストを作成する方法を共有する今後数週間にわたるシリーズの一部です。今日の記事では、モックについて少し探ってみたいと思います。いくつかのコード シナリオでは、データ パイプラインが接続、API 呼び出し、クラウド サービスとの統合などを行うため、このアプリケーションをテストする方法に混乱が生じる可能性があります。今日は、モックの使用に焦点を当てて、テストを作成するための興味深いライブラリをいくつか探索します。

結局のところ、モックとは何ですか?

モックは、テストの焦点では​​ない外部の依存関係やコンポーネントの動作を模倣するためにテストで使用されるモック オブジェクトです。これにより、テスト対象のコード単位を分離できるため、テストの制御性と予測性が向上します。モックの使用は、単体テストと統合テストでは一般的な方法です。

次のような場合にはモックを使用する必要があります。

• 依存関係はテストとは無関係です;
• 依存関係は利用できません;
• 特別な動作、エラー シミュレーション、または特定の応答をテストしたいと考えています。

データ パイプラインでは、モッキングを使用すると、実際のインフラストラクチャに依存せずに、データベース、メッセージング サービス、API などの外部コンポーネントの表現を作成できます。これは、分散処理用の PySpark、メッセージング用の Kafka、さらに AWS や GCP などのクラウド サービスなど、複数のテクノロジーを統合するデータ処理環境で特に役立ちます。

データ パイプラインがあるこれらのシナリオでは、モッキングにより分離された迅速なテストの実行が容易になり、コストと実行時間を最小限に抑えることができます。これにより、実際の接続や外部インフラストラクチャによって引き起こされる断続的な障害がなく、各統合が期待どおりに動作することを確信しながら、パイプラインの各部分を正確に検証できます。

各プログラミング言語では、実装されるモック関数をすでに提供する内部モジュールを見つけることができます。 Python では、ネイティブの Unittest.mock ライブラリがモックを作成するための主要なツールであり、オブジェクトや関数を簡単に制御してシミュレートできるようになります。 Go では、言語にはネイティブのモック ライブラリがないため、モッキング プロセスは通常、モッカリーなどの外部パッケージによってサポートされます。 mockery は、Go のネイティブ機能であるインターフェイスからモックを生成する場合に特に役立ちます。Java では、Mockito はモックを作成するための人気のある強力なライブラリとして際立っており、堅牢な単体テストを容易にするために JUnit と統合されています。これらのライブラリは、特に外部データ ソースと API のシミュレーションが重要なデータ パイプラインや分散システムにおいて、分離されたコンポーネントをテストするための重要な基盤を提供します。

モックの実装

モックの使用方法の基本的な例から始めましょう。 API 呼び出しを行う関数があり、この関数の単体テストを記述する必要があるとします。

def get_data_from_api(url):
    import requests
    response = requests.get(url)
    if response.status_code == 200:
        return response.json()
    else:
        return None

テスト シナリオに正しく取り組むには、まずどの状況をカバーする必要があるかを理解する必要があります。関数が REST 呼び出しを行うため、テストでは少なくとも 2 つの主なシナリオを考慮する必要があります。1 つはリクエストが成功するシナリオで、もう 1 つは応答が期待どおりにならないシナリオです。実際の URL を使用してコードを実行して動作を観察することもできますが、このアプローチには欠点があります。さまざまな種類の応答を制御できないことに加えて、テストが URL 応答の変更や最終的に使用できなくなる可能性に対して脆弱になるためです。 。これらの不一致を避けるために、モックを使用します。

from unittest import mock

@mock.patch('requests.get')
    def test_get_data_from_api_success(mock_get):
        # Configura o mock para retornar uma resposta simulada
        mock_get.return_value.status_code = 200
        mock_get.return_value.json.return_value = {"key": "value"}

        # Chama a função com o mock ativo
        result = get_data_from_api("http://fakeurl.com")

        # Verifica se o mock foi chamado corretamente e o resultado é o esperado
        mock_get.assert_called_once_with("http://fakeurl.com")
        self.assertEqual(result, {"key": "value"})

Python 単体テスト ライブラリの @mock.patch 装飾を使用すると、requests.get 呼び出しをモック (テスト コンテキスト内で get 関数の動作をシミュレートする「偽のオブジェクト」) に置き換え、外部依存関係を排除できます。 .

モックの return_value の値を定義することで、テストしている関数で呼び出されたときにオブジェクトが返すものを正確に指定できます。 return_value 構造が、置換する実際のオブジェクトと同じであることが重要です。たとえば、requests モジュールからの応答オブジェクトには、status_code などの属性と json() などのメソッドがあります。したがって、requests.get 関数からの応答をシミュレートするには、モック内でこれらの属性とメソッドに期待値を直接割り当てることができます。

def get_data_from_api(url):
    import requests
    response = requests.get(url)
    if response.status_code == 200:
        return response.json()
    else:
        return None

この特定のケースでは、リクエストの応答をシミュレートすること、つまり、外部 URL に依存せず、テスト環境に影響を与えることなく、期待されるさまざまな結果で関数の動作をテストすることに重点を置いています。

from unittest import mock

@mock.patch('requests.get')
    def test_get_data_from_api_success(mock_get):
        # Configura o mock para retornar uma resposta simulada
        mock_get.return_value.status_code = 200
        mock_get.return_value.json.return_value = {"key": "value"}

        # Chama a função com o mock ativo
        result = get_data_from_api("http://fakeurl.com")

        # Verifica se o mock foi chamado corretamente e o resultado é o esperado
        mock_get.assert_called_once_with("http://fakeurl.com")
        self.assertEqual(result, {"key": "value"})

テストで API エラー応答をシミュレートすることで、基本を超えて、404、401、500、503 などのさまざまな種類の HTTP ステータス コードに対してアプリケーションの動作をチェックできます。これにより、より広範囲の範囲が提供され、アプリケーションが適切に処理することが保証されます。それぞれの種類の障害について、呼び出しにおけるこれらの変化がアプリケーションやデータの処理にどのような影響を与えるかを理解しています。 POST メソッド呼び出しでは、追加の検証レイヤーを追加して、status_code と呼び出しの基本的な機能だけでなく、送受信された応答の スキーマ もチェックして、返されたデータが規則に従っていることを確認できます。予想される形式。このより詳細なテスト手法は、アプリケーションがさまざまなエラー シナリオを処理できるように準備されており、受信したデータが常に設計内容と一致していることを確認することで、将来の問題を防ぐのに役立ちます。

PySpark を使用したモックの実装

純粋な Python コードでモックを使用する簡単なケースを見てきました。次に、ケースを Pyspark を使用するコードのスニペットに拡張しましょう。

PySpark 関数、特に filter、groupBy、join などの DataFrame 操作をテストするには、モックを使用することが効果的なアプローチです。これにより、実際の Spark を実行する必要がなくなり、テスト時間が短縮され、開発環境が簡素化されます。 Python のunittest.mock ライブラリを使用すると、これらのメソッドの動作をシミュレートできるため、Spark インフラストラクチャに依存せずにコード フローとロジックを検証できます。

Spark のデータフレームに対してフィルター、groupBy、結合操作を実行する変換がある次の関数を見てみましょう。

def get_data_from_api(url):
    import requests
    response = requests.get(url)
    if response.status_code == 200:
        return response.json()
    else:
        return None

PySpark テストを実行するには、Spark 構成をローカルで行う必要があります。この構成は、クラスのすべてのテストで使用される Spark のインスタンスを作成する setUpClass メソッドで行われます。これにより、PySpark を分離して実行できるようになり、クラスター全体に依存せずに実際の変換操作を実行できるようになります。テストが完了すると、tearDownClass メソッドは Spark セッションを終了し、すべてのリソースが適切に解放され、テスト環境がクリーンになるようにします。

from unittest import mock

@mock.patch('requests.get')
    def test_get_data_from_api_success(mock_get):
        # Configura o mock para retornar uma resposta simulada
        mock_get.return_value.status_code = 200
        mock_get.return_value.json.return_value = {"key": "value"}

        # Chama a função com o mock ativo
        result = get_data_from_api("http://fakeurl.com")

        # Verifica se o mock foi chamado corretamente e o resultado é o esperado
        mock_get.assert_called_once_with("http://fakeurl.com")
        self.assertEqual(result, {"key": "value"})

test_transform_data テストでは、変換で使用されるデータを含む df および df_other のサンプル データフレームを作成することから始めます。次に、モックを適用せずにtransform_data関数を実行します。これにより、filter、groupBy、joinの操作が実際に実行され、新しいDataFrameが生成されます。実行後、collect() メソッドを使用して結果の DataFrame からデータを抽出します。これにより、このデータを期待値と比較できるようになり、実際かつ正確な方法で実行された変換を検証できます。

しかし、これらの pyspark 関数の 1 つの結果をテストしたいシナリオも考えられます。実行時にボトルネックとなっている可能性があり、プロセスに対するリスクを示さないコードの別の部分をモックする必要があります。したがって、リクエストを使用した前の例で見たように、関数/モジュールをモックする手法を使用できます。

response.status_code = mock_get.return_value.status_code
response.json() = mock_get.return_value.json.return_value

特定の操作のモック テストは test_transform_data_with_mocked_join メソッドで実行され、特にフィルター メソッドにモックを適用しました。このモックは、結合操作の結果をシミュレートされた DataFrame に置き換え、groupBy や join などの以前の操作を実際の方法で実行できるようにします。次にテストでは、結果の DataFrame を期待値と比較し、実行された他の変換を妨げることなく、結合モックが正しく使用されたことを確認します。

このハイブリッド アプローチにはいくつかの利点があります。 join や groupBy などの実際の PySpark 操作が確実に維持されるようにすることで、フィルターなどの特定の操作をモックに置き換える柔軟性を失うことなく、変換のロジックを検証できます。これにより、テストがより堅牢かつ高速になり、完全な Spark クラスターが不要になり、継続的なコード開発と検証が容易になります。

この戦略は慎重に使用し、結果に偏りが生じないシナリオでのみ使用する必要があることを強調することが重要です。テストの目的は、処理が正しく行われていることを確認することです。実際に関数をテストせずに、単に値を割り当てるべきではありません。単体テストのプロセスに影響を与えないことが保証できるセクションをモックすることは有効ですが、実際の動作を検証するには関数を実行する必要があることを覚えておくことが重要です。

したがって、この関数に他のタイプの処理を追加すると、ハイブリッド アプローチがより合理的になります。この戦略により、実際の操作とシミュレートされた操作を効果的に組み合わせることができ、より堅牢で信頼性の高いテストが保証されます

結論

モックは、特に PySpark やその他のクラウド サービスと連携する場合、効果的なテストを作成する際の貴重な味方です。 Python でのunittest を使用して検討した実装は、操作をシミュレートするだけでなく、データとプロセスの整合性を維持するのにも役立ちました。モックが提供する柔軟性により、実稼働環境に大混乱をもたらすことを恐れることなくパイプラインをテストできます。さて、次の挑戦の準備はできましたか?次のテキストでは、AWS および GCP サービスとの統合の世界に踏み込み、これらの呼び出しをモックしてパイプラインが完全に動作することを確認する方法を示します。次回まで！

以上がモック、彼らは何ですか？の詳細内容です。詳細については、PHP 中国語 Web サイトの他の関連記事を参照してください。