【Python】発電所登録処理1

Python
2025.01.06

こんにちは、かつコーチです。

Pythonで開発しているコードを備忘録として記載します。

file_upload_power_plant.py

def file_upload_power_plant(arguments: list[str]) -> int:
    """
    発電所を登録する
    """

    # 開始ログを出力する
    logger.info(f"発電所登録 開始 {arguments=}")

    # 引数チェック
    args = ucom.check_args(arguments, "power-plant")
    if isinstance(args, str):
        logger.error(f"発電所登録 引数異常:{args}")
        return ucom.RETURN_VALUE_INVALID_ARGUMENT

    # ステータスファイル作成
    status_file_result = ucom.create_status_file(args.file_time)
    if isinstance(status_file_result, int):  # 数値が返された場合はエラー
        logger.error(f"発電所登録 ステータスファイル作成失敗: {status_file_result=}")
        return status_file_result

    # メイン処理実施
    errors: Optional[Exception] | list[ucom.CsvError]
    main_result = file_upload_power_plant_main(args.file_path)
    if isinstance(main_result, tuple):
        code = main_result[0]
        errors = main_result[1]
    else:
        code = ucom.RETURN_VALUE_INVALID_FILE_CONTENT
        errors = main_result

    # ステータスファイル更新
    status_file_result = ucom.update_status_file(args.file_time, code, errors)
    if isinstance(status_file_result, int):  # 数値が返された場合はエラー
        logger.error(f"発電所登録 ステータスファイル更新失敗: {status_file_result=}")

    return code

解説

1. 開始ログを出力

logger.info(f"発電所登録 開始 {arguments=}")
  • 目的: 処理の開始をログに記録します。
  • arguments は外部から渡される引数であり、ログとして出力して、どの引数でこの関数が呼び出されたかを記録します。

2. 引数チェック

args = ucom.check_args(arguments, "power-plant")
if isinstance(args, str):
    logger.error(f"発電所登録 引数異常:{args}")
    return ucom.RETURN_VALUE_INVALID_ARGUMENT
  • ucom.check_args(arguments, "power-plant")
    • ucom モジュールの check_args 関数で、arguments の形式や内容を検証します。
    • 引数が適切でない場合（例: 必須引数が不足、形式が異常）、str 型のエラー理由が返されます。
  • 異常時の処理
    • エラー内容をログに出力し、ucom.RETURN_VALUE_INVALID_ARGUMENT（定数として定義されているエラーコード）を返して終了します。

3. ステータスファイル作成

status_file_result = ucom.create_status_file(args.file_time)
if isinstance(status_file_result, int):
    logger.error(f"発電所登録 ステータスファイル作成失敗: {status_file_result=}")
    return status_file_result
  • 目的: 処理状態を管理するステータスファイルを作成します。
  • ucom.create_status_file(args.file_time):
    • ステータスファイルを作成し、その結果を返します。
    • エラーが発生した場合、数値（エラーコード）が返されます。
  • 異常時の処理
    • 作成失敗時はログにエラーを出力し、エラーコードを返して終了します。

4. メイン処理の実施

main_result = file_upload_power_plant_main(args.file_path)
if isinstance(main_result, tuple):
    code = main_result[0]
    errors = main_result[1]
else:
    code = ucom.RETURN_VALUE_INVALID_FILE_CONTENT
    errors = main_result
  • file_upload_power_plant_main(args.file_path)
    • 発電所登録のメイン処理を呼び出します。
    • この関数が正常に動作した場合、結果がタプル (code, errors) として返されます。
      • code: 処理結果のステータスコード。
      • errors: エラー情報（通常はリスト）。
  • 異常時の処理
    • メイン処理がエラーを返した場合、適切なエラーコード (ucom.RETURN_VALUE_INVALID_FILE_CONTENT) とエラー内容を設定します。

5. ステータスファイル更新

status_file_result = ucom.update_status_file(args.file_time, code, errors)
if isinstance(status_file_result, int):
    logger.error(f"発電所登録 ステータスファイル更新失敗: {status_file_result=}")
  • 目的: ステータスファイルを更新して、処理の結果を記録します。
  • ucom.update_status_file:
    • ステータスファイルを更新し、その結果を返します。
    • エラーが発生した場合、数値（エラーコード）が返されます。
  • 異常時の処理
    • 更新失敗時はログにエラーを出力します。
    • ただし、この時点では処理を中断せずに次に進みます。

6. 結果の返却

return code

この関数をテストする際に考慮すべきケースは以下の通りです。

  • メイン処理の結果コード（code）を返します。

ユニットテスト作成時に必要なポイント

  1. 正常系
    • 引数が正しい場合に、正常終了することを確認します。
    • 期待される code が返されることを検証します。
  2. 引数チェックの異常系
    • 引数が不足または異常な形式の場合に、ucom.RETURN_VALUE_INVALID_ARGUMENT が返されることを確認します。
  3. ステータスファイル作成の異常系
    • ucom.create_status_file がエラーを返した場合に、適切なエラーコードが返されることを確認します。
  4. メイン処理の異常系
    • file_upload_power_plant_main がエラーを返した場合に、適切なエラーコードとエラー内容が処理されることを確認します。
  5. ステータスファイル更新の異常系
    • ucom.update_status_file がエラーを返した場合に、エラーがログに記録されることを確認します。

ユニットテスト

@patch('file_upload_power_plant.ucom.check_args')
@patch('file_upload_power_plant.ucom.create_status_file')
@patch('file_upload_power_plant.ucom.update_status_file')
@patch('file_upload_power_plant.file_upload_power_plant_main')
@patch('lib.logger.info')
@patch('lib.logger.error')
def test_file_upload_power_plant_success(
    mock_log_error,
    mock_log_info,
    mock_main,
    mock_update_status_file,
    mock_create_status_file,
    mock_check_args,
):
    # モックの設定
    mock_check_args.return_value = MagicMock(file_path="dummy_path.csv", file_time="20250101")
    mock_create_status_file.return_value = None
    mock_main.return_value = (0, None)  # 正常コード
    mock_update_status_file.return_value = None

    # 関数実行
    result = file_upload_power_plant(["dummy", "arguments"])

    # 検証
    assert result == 0
    mock_log_info.assert_called()
    mock_log_error.assert_not_called()

まとめ

コメント

タイトルとURLをコピーしました