· 5 years ago · Sep 28, 2020, 07:14 PM
1class EventParser:
2 task_keys = {
3 "title": "properties.title",
4 "description": "task_content.description.current",
5 "priority": "task_content.priority.current",
6 "status": "task_content.taskAction",
7 "due_date": "task_content.due_date.current",
8 "activity": "properties.activity_type_id",
9 "assignee": "task_content.assignee.current",
10 "assigner": "task_content.assigner.current",
11 "estimated_effort": "properties.estimated_effort",
12 "actual_effort": "properties.actual_effort",
13 "is_associated_user": "properties.is_associated_with_user",
14 "associated_users": "properties.associated_users"
15 }
16
17 touchpoint_event_keys = {
18 "subject": "properties.subject",
19 "content": "note_content.text",
20 "creator": "note_content.totango_user.user_name",
21 "participants": "properties.participants",
22 "updated": {
23 "key": "properties.last_edit_time",
24 "functor": lambda x: strftime("%Y-%m-%d", gmtime(x/1000.))
25 },
26 "updater": "properties.last_edit_user",
27 "created": {
28 "key": "timestamp",
29 "functor": lambda x: strftime("%Y-%m-%d", gmtime(x/1000.))
30 },
31 "activity": {
32 "key": "properties.activity_type_id",
33 "functor": lambda x: EventParser._get_activity_by_id(x)
34 },
35 "type": {
36 "key": "properties.meeting_type",
37 "functor": lambda x: EventParser._get_touchpoint_type_by_id(x)
38 },
39 "tags": {
40 "key": "properties.touchpoint_tags",
41 "functor": lambda x: EventParser._get_touchpoint_tag_by_id(x)
42 },
43 "assets": {
44 "key": "properties.assets",
45 "functor": lambda x: EventParser._get_asset_name(x)
46 }
47 }
48 touchpoint_db_keys = {
49 "subject": "subject",
50 "content": "content",
51 "creator": "created_by",
52 "participants": "participantsList",
53 "updated": {
54 "key": "last_update",
55 "functor": lambda x: str(parse(x).date())
56 },
57 "updater": "last_updater",
58 "created": {
59 "key": "create_date",
60 "functor": lambda x: str(parse(x).date())
61 },
62 "activity": {
63 "key": "activity_type_id",
64 "functor": lambda x: EventParser._get_activity_by_id(x)
65 },
66 "type": {
67 "key": "meeting_type",
68 "functor": lambda x: EventParser._get_touchpoint_type_by_id(x)
69 },
70 "tags": {
71 "key": "touchpoint_tags",
72 "functor": lambda x: EventParser._get_touchpoint_tag_by_id(x)
73 }
74 }
75
76 @classmethod
77 def _get_activity_by_id(cls, act_id):
78 api = SuccessFlowAPI()
79 return api.find_by_id(act_id)
80
81 @classmethod
82 def _get_touchpoint_type_by_id(cls, type_id):
83 api = OutcomeSuccessPlan()
84 return api.get_tp_type_by_id(type_id)
85
86 @classmethod
87 def _get_touchpoint_tag_by_id(cls, tags_id):
88 api = OutcomeSuccessPlan()
89 return api.get_tp_tags_by_id(tags_id)
90
91 @classmethod
92 def _get_asset_name(cls, assets):
93 data = json.loads(assets[0])
94 return data.get('name')
95
96 def __init__(self, csv):
97 self.csv = csv
98
99 def get_task_event_path_by_key(self, key):
100 path = self.task_keys.get(key)
101 if not path:
102 raise ValueError(f'unknown key for event assertion: {key}\n'
103 f'Possible on of : {self.task_keys.keys()}')
104 return path
105
106 def get_touchpoint_path_by_key(self, key, db=False, event=False):
107 sourse = None
108 if db:
109 sourse = self.touchpoint_db_keys
110 if event:
111 sourse = self.touchpoint_event_keys
112 path = sourse.get(key)
113 if not path:
114 raise ValueError(f'unknown key for touchpoint event assertion: {key}\n'
115 f'Possible on of : {self.task_keys.keys()}')
116 return path
117
118 def get_recursive_from_json(self, json, keys):
119 k = keys.split('.')
120 if isinstance(json, list):
121 v = next(x for x in json if x['name'] == k[0])
122 else:
123 v = json.get(k[0])
124 if len(k) == 1:
125 return v
126 return self.get_recursive_from_json(v, '.'.join(k[1:]))
127
128 @wait_true_v2(timeout=360, wait=20)
129 def get_by_ext_id(self, account_id, ext_id):
130 data = btest.steps.search.get_all_events_per_account(account_id).json()
131 event = next((x for x in data if x['properties']['external_id'] == str(ext_id)), None)
132 if event:
133 return event, True
134 return data, False
135
136 @wait_true_v2(timeout=360, wait=20)
137 def get_tp_id_by_subject_from_event(self, account_id, subject):
138 data = btest.steps.search.get_all_events_per_account(account_id).json()
139 event = next((x['note_content']['note_id'] for x in data if x['properties']['subject'] == str(subject)), None)
140 if event:
141 return event, True
142 return 0, False
143
144 @step('Validation Account Touchpoint DB by CSV')
145 def touchpoint_db_validate(self):
146 csv = btest.csv_helper(csv_name=self.csv, csv_type='assertion')
147 attachments.attach_csv(csv.raw_file, "Validation Touchpoint DB CSV") # attach input file to the Report
148 with Assertion('Validate Touchpoint DB Json response') as a:
149 for line in csv.lines_as_dict():
150 note_id = self.get_tp_id_by_subject_from_event(line.get('account_id'), line.get('subject'))
151 note = btest.steps.outcome_success_plan.get_touch_point_by_id(note_id)
152 expected_value = line.get('value')
153 path = self.get_touchpoint_path_by_key(line.get('key'), db=True)
154 if isinstance(path, dict):
155 key = path.get('key')
156 functor = path.get('functor')
157 actual_value = functor(self.get_recursive_from_json(note, key))
158 else:
159 actual_value = self.get_recursive_from_json(note, path)
160 key = path
161 a.contains_str(actual_value, expected_value,
162 f'Validate {key} value for {line.get("account_id")} account {line.get("external_id")} touchpoint')
163
164 @step('Validation Account TouchpointEvent by CSV')
165 def touchpoint_validate(self):
166 csv = btest.csv_helper(csv_name=self.csv, csv_type='assertion')
167 attachments.attach_csv(csv.raw_file, "Validation TouchpointEvent CSV") # attach input file to the Report
168 with Assertion('Validate TouchpointEvent Json response') as a:
169 for line in csv.lines_as_dict():
170 event = self.get_by_ext_id(line.get('account_id'), line.get('external_id'))
171 attachments.attach_json(event, "Current assertion TouchpointEvent JSON")
172 expected_value = line.get('value')
173 path = self.get_touchpoint_path_by_key(line.get('key'), event=True)
174 if isinstance(path, dict):
175 key = path.get('key')
176 functor = path.get('functor')
177 actual_value = functor(self.get_recursive_from_json(event, key))
178 else:
179 actual_value = self.get_recursive_from_json(event, path)
180 key = path
181 a.contains_str(actual_value, expected_value,
182 f'Validate {key} value for {line.get("account_id")} account {line.get("external_id")} touchpoint event')
183
184 @step('Validation Account TaskEvent by CSV')
185 def task_validate(self):
186 csv = btest.csv_helper(csv_name=self.csv, csv_type='assertion')
187 attachments.attach_csv(csv.raw_file, "Validation TaskEvent CSV") # attach input file to the Report
188 with Assertion('Validate TaskEvent Json response') as a:
189 for line in csv.lines_as_dict():
190 event = self.get_by_ext_id(line.get('account_id'), line.get('external_id'))
191 attachments.attach_json(event, "Current assertion TaskEvent JSON")
192 key = line.get('key')
193 expected_value = line.get('value')
194 actual_value = self.get_recursive_from_json(event, self.get_task_event_path_by_key(key))
195 a.contains_str(actual_value, expected_value,
196 f'Validate {key} value for {line.get("account_id")} account {line.get("external_id")} task')
197