#
# ============LICENSE_START==========================================
# org.onap.vvp/engagementmgr
# ===================================================================
# Copyright © 2017 AT&T Intellectual Property. All rights reserved.
# ===================================================================
#
# Unless otherwise specified, all software contained herein is licensed
# under the Apache License, Version 2.0 (the “License”);
# you may not use this software 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.
#
#
#
# Unless otherwise specified, all documentation contained herein is licensed
# under the Creative Commons License, Attribution 4.0 Intl. (the “License”);
# you may not use this documentation except in compliance with the License.
# You may obtain a copy of the License at
#
# https://creativecommons.org/licenses/by/4.0/
#
# Unless required by applicable law or agreed to in writing, documentation
# 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.
#
# ============LICENSE_END============================================
#
# ECOMP is a trademark and service mark of AT&T Intellectual Property.
import json
from django.utils import timezone
from engagementmanager.bus.messages.new_notification_message import NewNotificationMessage
from engagementmanager.models import IceUserProfile, Activity, Notification
from engagementmanager.utils import activities_data
from engagementmanager.utils.constants import Constants
from engagementmanager.service.logging_service import LoggingServiceFactory
logger = LoggingServiceFactory.get_logger()
class ActivitiesSvc:
def __init__(self):
pass
def generate_activity(self, activity_data):
if hasattr(activity_data, "users_list") and activity_data.users_list:
for user in activity_data.users_list:
activity_data.user = user
if hasattr(activity_data, "multiple_users_as_owners") and activity_data.multiple_users_as_owners:
activity_data.owner = activity_data.user
self.set_description(activity_data)
self.set_metadata(activity_data)
self.__create_activity(activity_data)
else:
self.set_description(activity_data)
self.set_metadata(activity_data)
self.__create_activity(activity_data)
def __create_activity(self, activity_data):
if activity_data.engagement is None:
msg = "Engagement provided is a None object, be careful not to generate " \
"description and metadata with engagement properties."
logger.warn(msg)
if activity_data.owner and not isinstance(activity_data.owner, IceUserProfile):
raise ValueError("owner should be IceUserProfile, was %r", activity_data.owner)
if activity_data.description is None:
logger.warn('createActivity called with description=None; setting to ""')
activity_data.description = ''
new_activity = Activity.objects.create(
activity_owner=activity_data.owner,
description=activity_data.description,
is_notification=activity_data.is_notification,
engagement=activity_data.engagement,
activity_type=activity_data.activity_type.name,
metadata=json.dumps(activity_data.metadata, ensure_ascii=False),)
if activity_data.engagement:
activity_data.engagement.activity_set.add(new_activity)
new_activity.save()
if activity_data.is_notification:
users_to_notify = []
if activity_data.owner is None:
if activity_data.engagement:
users_to_notify = activity_data.engagement.engagement_team.all()
else:
users_to_notify.append(activity_data.owner)
for user_to_notify in users_to_notify:
new_notification = Notification.objects.create(user=user_to_notify, activity=new_activity)
new_activity.notification_set.add(new_notification)
user_to_notify.notification_set.add(new_notification)
new_notification.save()
from engagementmanager.apps import bus_service
bus_service.send_message(NewNotificationMessage(new_notification))
def pull_recent_activities(self, engagement, recent_activities_limit):
""" expected: engagement object (Activity), recent_activities_limit (integer) - number of recent activities
result: Top-X Dict Activity objects (sort by create_time) """
logger.debug("Pulling top X activities from DB")
activities = Activity.objects.filter(
engagement=engagement, activity_owner=None).order_by('-create_time')[:recent_activities_limit]
return activities
def set_description(self, activity_data):
dt = timezone.now()
description = ''
if isinstance(activity_data, activities_data.UserJoinedEngagementActivityData):
description = "##user_name## joined ##vf_name## "
elif isinstance(activity_data, activities_data.UpdateNextStepsActivityData):
description = "##user_name## " + activity_data.update_type.lower() + " a next step"
elif isinstance(activity_data, activities_data.VFProvisioningActivityData):
description = "Failed Gitlab and/or Jenkins Provision ##vf_name##: " + activity_data.description
elif isinstance(activity_data, activities_data.TestFinishedActivityData):
description = "Failure in Jenkins Job: " + activity_data.description
elif isinstance(activity_data, activities_data.ChangeEngagementStageActivityData):
description = "Engagement stage is now " + activity_data.stage + " for the following VF: ##vf_name##"
elif isinstance(activity_data, activities_data.DeleteNextStepsActivityData):
description = activity_data.user.full_name + \
" has deleted a next step at " + dt.strftime("%Y-%m-%d %H:%M:%S")
elif isinstance(activity_data, activities_data.NoticeEmptyEngagementData):
description = "You have not added any parts of the VNF package to your engagement ##vf_name## since it was created " + activity_data.delta_days_from_creation +\
" days ago. Do note that if you have not added any parts of the VNF package by " + \
activity_data.max_empty_time + ", we will be automatically archive it."
elif isinstance(activity_data, activities_data.AddNextStepsActivityData):
description = activity_data.user.full_name + " has added a next step at " + dt.strftime("%Y-%m-%d %H:%M:%S")
elif isinstance(activity_data, activities_data.SSHKeyAddedActivityData):
if activity_data.action == 'add':
description = "You have added an SSH key to your profile"
elif activity_data.action == 'set':
description = "You have set a new SSH key in your profile"
else:
description = "SSH key activity"
activity_data.description = description
def set_metadata(self, activity_data):
dt = timezone.now()
metadata = {}
if isinstance(activity_data, activities_data.UserJoinedEngagementActivityData):
activity_data.is_notification = True
metadata['notification_subject'] = "Someone has joined the " + activity_data.vf.name + " team"
metadata['notification_message'] = activity_data.user.full_name + " joined the " + activity_data.vf.name + \
" team. You can reach the dashboard by going to this link: " + \
Constants.dashboard_href
metadata['macros'] = {
'##vf_name##': {
'type': 'select_engagement',
'short': activity_data.vf.name,
'eng_uuid': str(activity_data.engagement.uuid),
},
'##user_name##': {
'type': 'popover',
'short': activity_data.user.full_name,
'long': activity_data.user.email,
}
}
elif isinstance(activity_data, activities_data.UpdateNextStepsActivityData):
metadata['macros'] = {
'##user_name##': {
'type': 'popover',
'short': activity_data.user.full_name,
'long': activity_data.user.email,
}
}
elif isinstance(activity_data, activities_data.NoticeEmptyEngagementData):
activity_data.is_notification = True
metadata['notification_subject'] = "Inactive Engagement Alert - " + activity_data.vf_name
metadata['notification_message'] = "We have noticed that you have not added any parts of the VNF package to your engagement " +\
activity_data.engagement.engagement_manual_id + ": " + activity_data.vf_name + " since it was created " + activity_data.delta_days_from_creation +\
" days ago. If you have any questions around how you add your VNF package please check the relevant parts of the online documentation.
" +\
"Do note that if you have not added any parts of the VNF package by " + \
activity_data.max_empty_time + ", we will be automatically archive it."
metadata['macros'] = {
'##vf_name##': {
'type': 'select_engagement',
'short': activity_data.vf_name,
'eng_uuid': activity_data.engagement.uuid,
},
}
elif isinstance(activity_data, activities_data.VFProvisioningActivityData):
activity_data.is_notification = True
metadata['notification_subject'] = "Failed Gitlab and/or Jenkins Provision: " + activity_data.vf.name
metadata['notification_message'] = activity_data.description
metadata['macros'] = {
'##vf_name##': {
'type': 'select_engagement',
'short': activity_data.vf.name,
'eng_uuid': activity_data.vf.engagement.uuid,
},
}
elif isinstance(activity_data, activities_data.TestFinishedActivityData):
activity_data.is_notification = True
metadata['notification_subject'] = "Failed test_finished signal "
metadata['notification_message'] = activity_data.description
elif isinstance(activity_data, activities_data.ChangeEngagementStageActivityData):
activity_data.is_notification = True
metadata['notification_subject'] = "Engagement stage was changed for the following VF: " + \
activity_data.vf.name
metadata['notification_message'] = "Engagement stage is now " + activity_data.stage +\
" for the following VF: " + activity_data.vf.name
metadata['macros'] = {
'##vf_name##': {
'type': 'select_engagement',
'short': activity_data.vf.name,
'eng_uuid': activity_data.engagement.uuid,
}
}
elif isinstance(activity_data, activities_data.AddNextStepsActivityData):
activity_data.is_notification = True
metadata['notification_subject'] = "New next-step was added to the following VF: " + activity_data.vf.name
metadata['notification_message'] = activity_data.user.full_name + " has added a next step at " +\
dt.strftime("%Y-%m-%d %H:%M:%S") + \
", You can reach the dashboard by going to this link: " \
+ Constants.dashboard_href
elif isinstance(activity_data, activities_data.SSHKeyAddedActivityData):
activity_data.is_notification = True
metadata['notification_subject'] = "You have set an SSH key"
metadata['notification_message'] = "You have set an SSH key to your profile. Please allow some time for" \
" it to propagate across the system."
activity_data.metadata = metadata