lambda_function.py

# Copyright 2019 getcarrier.io

# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

import json
from os import environ
from email_client import EmailClient
from api_email_notification import ApiEmailNotification
from ui_email_notification import UIEmailNotification
from time import sleep
from typing import Union


def lambda_handler(event: Union[list, dict], context):
    try:
        args = parse_args(event)
        print(args)
        if not args['notification_type']:
            raise Exception('notification_type parameter is not passed')

        # Send notification
        if args['notification_type'] == 'api':
            # Check required params
            if not all([args['influx_host'], args['smtp_user'], args['test'], args['test_type'],
                        args['notification_type'], args['smtp_password'], args['user_list']]):
                raise Exception('Some required parameters not passed')

            email = ApiEmailNotification(args).api_email_notification()
        elif args['notification_type'] == 'ui':
            if not all([args['test_id'], args['report_id']]):
                raise Exception('test_id and report_id are required for UI reports')

            email = UIEmailNotification(args).ui_email_notification()
        else:
            raise Exception('Incorrect value for notification_type: {}. Must be api or ui'
                            .format(args['notification_type']))

        EmailClient(args).send_email(email)

    except Exception as e:
        from traceback import format_exc
        print(format_exc())
        return {
            'statusCode': 500,
            'body': json.dumps(str(e))
        }
    return {
        'statusCode': 200,
        'body': json.dumps('Email has been sent')
    }


def parse_args(event: Union[list, dict]):
    if isinstance(event, list):
        _event = event
    else:
        _event = [event]
    args = {}

    # Galloper or AWS Lambda service
    event = _event[0] if not _event[0].get('body') else json.loads(_event[0]['body'])
    # Galloper
    args['galloper_url'] = environ.get("galloper_url") if not event.get('galloper_url') else event.get('galloper_url')
    args['token'] = environ.get("token") if not event.get('token') else event.get('token')
    args['project_id'] = environ.get("project_id") if not event.get('project_id') else event.get('project_id')

    # Influx Config
    args['influx_host'] = environ.get("influx_host") if not event.get('influx_host') else event.get('influx_host')
    args['influx_port'] = event.get("influx_port", 8086)
    args['influx_user'] = event.get("influx_user", "")
    args['influx_password'] = event.get("influx_password", "")

    # Influx DBs
    args['comparison_db'] = event.get("comparison_db")
    args['influx_db'] = event.get("influx_db")

    # SMTP Config
    args['smtp_port'] = environ.get("smtp_port", 465) if not event.get('smtp_port') else event.get('smtp_port')
    args['smtp_host'] = environ.get("smtp_host", "smtp.gmail.com") if not event.get('smtp_host') else event.get(
        'smtp_host')
    args['smtp_user'] = environ.get('smtp_user') if not event.get('smtp_user') else event.get('smtp_user')
    args['smtp_sender'] = args['smtp_user'] if not event.get('smtp_sender') else event.get('smtp_sender')

    if not event.get('notification_type'):
        args['notification_type'] = environ.get('notification_type')
    else:
        args['notification_type'] = event.get('notification_type')
    if args['notification_type'] == 'ui':
        args['test_type'] = event.get('test_suite')
    if args['notification_type'] == 'api':
        args['test_type'] = event.get('test_type')

    args['type'] = args['test_type']

    if args['notification_type'] == "api":
        args['smtp_password'] = event.get("smtp_password")
    else:
        args['smtp_password'] = event.get('smtp_password')["value"]
    # Test Config
    args['users'] = event.get('users', 1)
    args['test'] = event.get('test')
    args['simulation'] = event.get('test')
    args['env'] = event.get('env')

    # summary data
    args["test_data"] = event.get('test_data', {})

    # Notification Config
    args['user_list'] = event.get('user_list')
    args['test_limit'] = event.get("test_limit", 5)
    # args['comparison_metric'] = event.get("comparison_metric", 'pct95')
    # Try to extract comparison_metric from multiple sources (in priority order)
    comparison_metric_from_event = None

    # 1. Check quality_gate_config for comparison_metric (try multiple locations)
    quality_gate_config = event.get('quality_gate_config', {})
    if isinstance(quality_gate_config, dict):
        # Try baseline.rt_baseline_comparison_metric (used for both baseline and SLA)
        baseline_config = quality_gate_config.get('baseline', {})
        if isinstance(baseline_config, dict):
            comparison_metric_from_event = baseline_config.get('rt_baseline_comparison_metric')

        # Try settings.comparison_metric
        if not comparison_metric_from_event:
            settings = quality_gate_config.get('settings', {})
            if isinstance(settings, dict):
                comparison_metric_from_event = settings.get('comparison_metric')

        # Try direct quality_gate_config.comparison_metric
        if not comparison_metric_from_event:
            comparison_metric_from_event = quality_gate_config.get('comparison_metric')

    # 2. Check direct comparison_metric field in event
    if not comparison_metric_from_event:
        comparison_metric_from_event = event.get("comparison_metric")

    # 3. Parse from reasons_to_fail_report as fallback
    if not comparison_metric_from_event and event.get('reasons_to_fail_report'):
        import re
        for reason in event.get('reasons_to_fail_report', []):
            match = re.search(r'by (pct\d+)', reason)
            if match:
                comparison_metric_from_event = match.group(1)
                break

    args['comparison_metric'] = comparison_metric_from_event or 'pct95'

    # ui data
    args['test_id'] = event.get('test_id')
    args['report_id'] = event.get('report_id')

    # SLA's
    args['performance_degradation_rate'] = event.get('performance_degradation_rate')
    args['missed_threshold_rate'] = event.get('missed_threshold_rate')
    args['reasons_to_fail_report'] = event.get('reasons_to_fail_report')

    # Quality gate thresholds (for UI notifications)
    if args['notification_type'] == "ui":
        args['deviation'] = event.get('deviation', 0)
        args['baseline_deviation'] = event.get('baseline_deviation', 0)
        args['missed_threshold_rate'] = event.get('missed_thresholds', 0)

    args['performance_degradation_rate_qg'] = event.get('performance_degradation_rate_qg')
    args['missed_thresholds_qg'] = event.get('missed_thresholds_qg')
    args['status'] = event.get('status')
    args['quality_gate_config'] = event.get('quality_gate_config')

    # AI Analysis Configuration (FR-001 through FR-007)
    # All parameters are optional with defaults for backward compatibility
    args['enable_ai_analysis'] = event.get('enable_ai_analysis', True)
    args['ai_provider'] = event.get('ai_provider', 'azure_openai')
    args['azure_openai_api_key'] = event.get('azure_openai_api_key', '')
    args['azure_openai_endpoint'] = event.get('azure_openai_endpoint', 'https://ai-proxy.lab.epam.com')
    args['azure_openai_api_version'] = event.get('azure_openai_api_version', '2024-02-15-preview')
    args['ai_model'] = event.get('ai_model', 'gpt-4o')
    args['ai_temperature'] = event.get('ai_temperature', 0.0)

    return args