テストとロガーを追加します

.vscode/settings.json

@@ -1,6 +1,7 @@
 {
     "python.testing.pytestArgs": [
-        "tests"
+        "tests",
+        "tests_integration"
     ],
     "python.testing.unittestEnabled": false,
     "python.testing.pytestEnabled": true

examples/examle_request.py

@@ -0,0 +1,104 @@
+"""
+Cloud Functionにリクエストを送信するサンプルコード
+"""
+
+import requests
+
+BASE_URL = "http://localhost:8080"
+
+
+def example_get_request():
+    """GETリクエストの例"""
+    print("=== GET Request ===")
+    # クエリパラメータを指定
+    params = {"name": "Python"}
+    response = requests.get(BASE_URL, params=params)
+
+    response.raise_for_status()
+    print(f"Status Code: {response.status_code}")
+    print(f"Response: {response.json()}")
+
+
+def example_get_request_no_params():
+    """GETリクエスト（パラメータなし）の例"""
+    print("=== GET Request (No Parameters) ===")
+    response = requests.get(BASE_URL)
+
+    response.raise_for_status()
+    print(f"Status Code: {response.status_code}")
+    print(f"Response: {response.json()}")
+
+
+def example_post_request():
+    """POSTリクエストの例"""
+    print("=== POST Request ===")
+    # JSONデータを送信
+    data = {"name": "Python"}
+    headers = {"Content-Type": "application/json"}
+
+    response = requests.post(BASE_URL, json=data, headers=headers)
+
+    response.raise_for_status()
+    print(f"Status Code: {response.status_code}")
+    print(f"Response: {response.json()}")
+
+
+def example_put_request():
+    """PUTリクエストの例（サポートされていないメソッド）"""
+    print("=== PUT Request (Unsupported Method) ===")
+
+    data = {"name": "Python"}
+    headers = {"Content-Type": "application/json"}
+
+    response = requests.put(BASE_URL, json=data, headers=headers)
+
+    response.raise_for_status()
+    print(f"Status Code: {response.status_code}")
+    print(f"Response: {response.json()}")
+
+
+def example_event_request():
+    """Cloud Eventリクエストの例"""
+    print("=== Cloud Event Request ===")
+    # Cloud Eventのペイロードを作成
+    event_payload = {
+        "message": {
+            "data": "aGVsbG8tbG9jYWw="  # "hello-local"をBase64エンコードしたもの
+        }
+    }
+    headers = {
+        "Content-Type": "application/json",
+        "Ce-Specversion": "1.0",
+        "Ce-Type": "google.cloud.pubsub.topic.v1.messagePublished",
+        "Ce-Source": "//pubsub.googleapis.com/projects/test-project/topics/test-topic",
+        "Ce-Id": "1234567890",
+        "Ce-Time": "2025-12-06T00:00:00Z",
+    }
+    response = requests.post(BASE_URL, json=event_payload, headers=headers)
+    response.raise_for_status()
+    print(f"Status Code: {response.status_code}")
+    print(f"Response: {response.json()}")
+
+
+if __name__ == "__main__":
+    print("Cloud Function Request Examples")
+    print(f"Target URL: {BASE_URL}")
+    print("=" * 50)
+    print()
+
+    try:
+        # 各種リクエストの実行例
+        example_get_request()
+        example_get_request_no_params()
+        example_post_request()
+        example_put_request()
+        print("All examples completed!")
+    except requests.exceptions.ConnectionError:
+        print("Error: Could not connect to the server.")
+        print("Please make sure the Cloud Function is running on port 8080.")
+        print("\nStart the server with:")
+        print(
+            "functions-framework --source=src/main.py --target=main --signature-type=http --port=8080"
+        )
+    except Exception as e:
+        print(f"Error: {e}")

readme/cloud_functions.md

@@ -102,3 +102,14 @@ curl -X POST \
   }' \
   http://localhost:8080
 ```
+
+## ログについて
+
+### `google-cloud-logging`を使う場合
+
+* カスタム logName を使い分けたい
+* OpenTelemetry
+* エラーレポーティングを細かく制御したい
+
+以外のものがなければ
+標準 logging + stdout/stderrで十分対応可能です。

ruff.toml

@@ -8,5 +8,5 @@ line-length = 79
 # BXX（バグの可能性）
 
 [lint]
-select = ["F", "E", "W", "D101", "D102", "D103", "B"] 
+select = ["F", "E", "W", "D101", "B"] 
 ignore = []

src/main.py

@@ -1,9 +1,14 @@
 from flask import Request
 import functions_framework
-from utils.custom_logger import get_logger
 
+import os
+os.environ["ENV"]="dev"  # For testing purposes
+
+
+from utils.custom_logger import get_logger
 logger = get_logger(__name__)
 
+
 @functions_framework.http
 def main(request: Request):
     """HTTPリクエストを処理するエンドポイント"""

src/utils/custom_logger.py

@@ -1,8 +1,59 @@
 import os
 import logging
+import json
 import functools
 from .singleton import Singleton
 
+class CoogelCustomLogger():
+    """Google Cloud Functions用のシンプルなカスタムロガー"""
+
+    def __init__(self, name="main"):
+        self.logger = logging.getLogger(name)
+        self.logger.setLevel(logging.INFO)
+
+        handler = logging.StreamHandler()
+        handler.setLevel(logging.INFO)
+        # メッセージのみ(フォーマットなし)
+        formatter = logging.Formatter("%(message)s")
+        handler.setFormatter(formatter)
+
+        if not self.logger.handlers:
+            self.logger.addHandler(handler)
+
+    def _log(self, message,level="INFO",**fields):
+        payload = {
+            "serverity": level,
+            "message": f"{message}",
+            **fields
+        }
+        self.logger.info(json.dumps(payload, ensure_ascii=False))
+
+    def info(self, message, **fields):
+        self._log(message, level="INFO", **fields)
+
+    def warning(self, message, **fields):
+        self._log(message, level="WARNING", **fields)
+
+    def error(self, message, **fields):
+        self._log(message, level="ERROR", **fields)
+
+    def exception(self, message, **fields):
+        payload = {
+            "serverity": "ERROR",
+            "message": f"{message}",
+            **fields
+        }
+        self.logger.info(
+            json.dumps(payload, ensure_ascii=False),
+            exc_info=True
+        )
+
+    def debug(self, message, **fields):
+        self._log(message, level="DEBUG", **fields)
+
+
+
+
 
 class CustomLogger(Singleton):
     """
@@ -13,9 +64,8 @@ class CustomLogger(Singleton):
     def __init__(self, name="main", log_file=None, level=logging.INFO):
         if hasattr(self, "_initialized") and self._initialized:
             return  # すでに初期化済みなら何もしない
-            # self.logger.setLevel(level)
 
-        if os.getenv("ENV", "local"):
+        if os.getenv("ENV", "local")=="local":
             self.logger = logging.getLogger(name)
             self.logger.setLevel(level)
             self.logger.propagate = False
@@ -35,8 +85,12 @@ class CustomLogger(Singleton):
                 fh = logging.FileHandler(log_file, encoding="utf-8")
                 fh.setFormatter(formatter)
                 self.logger.addHandler(fh)
-
                 self._initialized = True
+        elif os.getenv("ENV") in ["dev", "prd"]:
+            self.logger = CoogelCustomLogger(name)
+            self._initialized = True
+
+
 
     def get_logger(self):
         return self.logger

tests/test_mock_request.py

@@ -0,0 +1,71 @@
+import pytest
+from unittest.mock import Mock, patch
+from flask import Request
+import json
+
+
+class TestRequest:
+    """Request handling tests"""
+
+    def test_get_json_data_valid(self):
+        """Test getting valid JSON data from request"""
+        mock_request = Mock(spec=Request)
+        mock_request.get_json.return_value = {"key": "value"}
+
+        result = mock_request.get_json()
+        assert result == {"key": "value"}
+
+    def test_get_json_data_empty(self):
+        """Test getting empty JSON data from request"""
+        mock_request = Mock(spec=Request)
+        mock_request.get_json.return_value = {}
+
+        result = mock_request.get_json()
+        assert result == {}
+
+    def test_get_json_data_none(self):
+        """Test getting None when no JSON data"""
+        mock_request = Mock(spec=Request)
+        mock_request.get_json.return_value = None
+
+        result = mock_request.get_json()
+        assert result is None
+
+    def test_request_headers(self):
+        """Test accessing request headers"""
+        mock_request = Mock(spec=Request)
+        mock_request.headers = {"Content-Type": "application/json"}
+
+        assert mock_request.headers["Content-Type"] == "application/json"
+
+    def test_request_args(self):
+        """Test accessing URL query parameters"""
+        mock_request = Mock(spec=Request)
+        mock_request.args = {"param1": "value1", "param2": "value2"}
+
+        assert mock_request.args["param1"] == "value1"
+        assert mock_request.args["param2"] == "value2"
+
+    def test_request_method(self):
+        """Test request HTTP methods"""
+        mock_request = Mock(spec=Request)
+        mock_request.method = "POST"
+
+        assert mock_request.method == "POST"
+
+    def test_request_data_raw(self):
+        """Test accessing raw request data"""
+        mock_request = Mock(spec=Request)
+        mock_request.data = b'{"test": "data"}'
+
+        assert mock_request.data == b'{"test": "data"}'
+        data = json.loads(mock_request.data)
+        assert data["test"] == "data"
+
+    def test_request_form_data(self):
+        """Test accessing form data"""
+        mock_request = Mock(spec=Request)
+        mock_request.form = {"username": "testuser", "password": "testpass"}
+
+        assert mock_request.form["username"] == "testuser"
+        assert mock_request.form["password"] == "testpass"

tests_integration/test_int_request.py

@@ -0,0 +1,18 @@
+import pytest
+import requests
+
+BASEURL = "http://localhost:8080/"
+
+
+class TestIntegrationRequest:
+    """統合テスト: 実際のCloud Functionエンドポイントにリクエストを送信"""
+
+    def test_get_request_default(self):
+        """GETリクエスト: デフォルトパラメータ"""
+        response = requests.get(BASEURL)
+
+        assert response.status_code == 200
+        data = response.json()
+        assert "message" in data
+        assert data["message"] == "Hello, World!"
+