local-test-server.py

#!/usr/bin/env python3
"""
BeanRatio Local Test Server
Lambda関数をローカルでテストするためのシンプルなHTTPサーバー
"""

import sys
import json
import os
from http.server import HTTPServer, BaseHTTPRequestHandler
from urllib.parse import urlparse, parse_qs, unquote

# Lambda関数をインポート
sys.path.insert(0, '/home/ec2-user/BeanRatio/lambda')
os.environ['S3_BUCKET_NAME'] = 'beanratio-data'
os.environ['AUTH_TOKEN'] = 'test-token-12345'
os.environ['WEB_DIR'] = '/home/ec2-user/BeanRatio/web'
os.environ['FUNCTION_URL'] = 'http://localhost:3000'

# LocalStack S3エンドポイント設定
os.environ['AWS_ENDPOINT_URL_S3'] = 'http://localhost:4566'
os.environ['AWS_ACCESS_KEY_ID'] = 'test'
os.environ['AWS_SECRET_ACCESS_KEY'] = 'test'
os.environ['AWS_DEFAULT_REGION'] = 'ap-northeast-1'

print(f"DEBUG: AUTH_TOKEN環境変数 = {os.environ.get('AUTH_TOKEN')}")

from functions.api_handler import lambda_handler


class LambdaHTTPHandler(BaseHTTPRequestHandler):
    """Lambda関数をHTTPハンドラーでラップ"""

    def do_GET(self):
        self.handle_request()

    def do_POST(self):
        self.handle_request()

    def do_PUT(self):
        self.handle_request()

    def do_DELETE(self):
        self.handle_request()

    def do_OPTIONS(self):
        self.send_response(200)
        self.send_header('Access-Control-Allow-Origin', '*')
        self.send_header('Access-Control-Allow-Methods', 'GET, POST, PUT, DELETE, OPTIONS')
        self.send_header('Access-Control-Allow-Headers', 'Content-Type')
        self.end_headers()

    def handle_request(self):
        """HTTPリクエストをLambda形式に変換して実行"""
        try:
            # URLをパース
            parsed_url = urlparse(self.path)
            path = parsed_url.path
            query_string = parsed_url.query
            query_params = parse_qs(query_string)

            # クエリパラメータを単一値に変換
            query_string_parameters = {k: v[0] for k, v in query_params.items()}

            # ボディを読み込み
            body = None
            if self.command in ['POST', 'PUT']:
                content_length = int(self.headers.get('Content-Length', 0))
                if content_length > 0:
                    body = self.rfile.read(content_length).decode('utf-8')

            # Lambda Event形式に変換
            event = {
                'requestContext': {
                    'http': {
                        'method': self.command,
                        'path': path,
                        'sourceIp': self.client_address[0]
                    }
                },
                'rawPath': path,
                'body': body,
                'queryStringParameters': query_string_parameters if query_string_parameters else None,
                'headers': dict(self.headers)
            }

            print(f"\n{'='*60}")
            print(f"REQUEST: {self.command} {path}")
            print(f"Query: {query_string_parameters}")
            if body:
                print(f"Body: {body[:200]}")
            print(f"{'='*60}")

            # Lambda関数を実行
            response = lambda_handler(event, None)

            # レスポンスを送信
            status_code = response.get('statusCode', 200)
            response_headers = response.get('headers', {})
            response_body = response.get('body', '{}')

            self.send_response(status_code)

            # Lambda関数からのヘッダーを優先して設定
            content_type = response_headers.get('Content-Type', 'application/json')
            self.send_header('Content-Type', content_type)
            self.send_header('Access-Control-Allow-Origin', '*')
            self.send_header('Access-Control-Allow-Methods', 'GET, POST, PUT, DELETE, OPTIONS')
            self.send_header('Access-Control-Allow-Headers', 'Content-Type')

            for header, value in response_headers.items():
                if header.lower() not in ['content-type', 'access-control-allow-origin', 'access-control-allow-methods', 'access-control-allow-headers']:
                    self.send_header(header, value)

            self.end_headers()
            self.wfile.write(response_body.encode('utf-8'))

            print(f"RESPONSE: {status_code}")
            print(f"Body: {response_body[:200]}")

        except Exception as e:
            print(f"ERROR: {e}")
            import traceback
            traceback.print_exc()

            self.send_response(500)
            self.send_header('Content-Type', 'application/json')
            self.end_headers()
            error_response = json.dumps({'error': str(e)})
            self.wfile.write(error_response.encode('utf-8'))

    def log_message(self, format, *args):
        """ログ出力を制御"""
        # BaseHTTPRequestHandlerのログは出力しない（自前でやる）
        pass


def main():
    """HTTPサーバーを起動"""
    host = 'localhost'
    port = 3000

    server = HTTPServer((host, port), LambdaHTTPHandler)

    print(f"\n{'='*60}")
    print(f"BeanRatio Local Test Server")
    print(f"{'='*60}")
    print(f"Server running at: http://{host}:{port}")
    print(f"Token: test-token-12345")
    print(f"\nExample requests:")
    print(f"  GET  http://{host}:{port}/beans?token=test-token-12345")
    print(f"  POST http://{host}:{port}/beans?token=test-token-12345")
    print(f"        Body: {{'name':'Test Bean','notes':'Test notes'}}")
    print(f"\nPress Ctrl+C to stop")
    print(f"{'='*60}\n")

    try:
        server.serve_forever()
    except KeyboardInterrupt:
        print("\n\nServer stopped.")
        server.server_close()


if __name__ == '__main__':
    main()