import time import random import json from .api_client import ApiClient from allure import step from assertpy import assert_that api_user_cleanup = "/api/dna/profile/reset" api_saved_discoveries = "/api/dna/user_searches" api_categories = "/api/dna/categories" api_category_entities = "/api/dna/categories/{}/entities" api_total_count = "/api/{}/discovery/total_count" api_discovery = "/api/{}/discovery" api_recent_searches = "/api/dna/recent_searches" api_external_links = "/api/v4/{}/{}/external_links" api_favorites = "/api/dna/favorites/{}/{}" api_discovery_result = "api/{}/{}/discovery_result" api_spotify_tracks = "api/v2/{}/{}/tracks/spotify" api_soundcloud_tracks = "api/v2/{}/{}/tracks/soundcloud" api_tiktok_videos = "api/v2/{}/{}/videos/tiktok" api_youtube_videos = "api/v2/{}/{}/videos/youtube" api_amplitude = "https://api2.amplitude.com/2/httpapi" class ApiHelper(ApiClient): error_rate = 0.05 def getting(self, api_url): for i in range(5): resp = self.get(api_url) body = resp.json() if len(body) > 0: return body time.sleep(1.0) return None @step("Cleanup user attributes") def user_cleanup(self): self.log.info('User Clean-Up') self.patch(api_user_cleanup) return @step("Return categories") def get_user_categories(self): return self.getting(api_categories) @step("Return category entities") def get_category_entities(self, category_id): return self.getting(api_category_entities.format(category_id)) @step("Delete all user categories") def delete_all_user_categories(self): categories = self.get_user_categories() print(categories) if categories is not None: for category in categories: print("Delete category id={}".format(category['id'])) self.delete(f"{api_categories}/{category['id']}") @step("Return {count} {entity_type}s") def get_entities(self, entity_type, count=100, discovery_data=None): if discovery_data is None: with open('src/data/discovery_payload_default.json') as default_file: discovery_data = json.load(default_file) discovery_data.update({"limit": count, "offset": 0, "error_rate": self.error_rate}) entities = self.post(api_discovery.format(entity_type), discovery_data).json() print('Returned number of entities: {}'.format(len(entities))) if len(entities) == 1: print(entities) return entities @step("Return random entity id") def get_random_entity(self, entity_type): return random.choice(self.get_entities(entity_type, count=50)) @step("Return random entity id with performance data") def get_random_entity_with_performance_data(self, entity_type): entities_bunch = self.get_entities(entity_type, count=50) chart_types = { 'artist': {'tiktok': 'followers', 'youtube': 'subscribers', 'spotify': 'followers', 'soundcloud': 'followers'}, 'track': { 'tiktok': 'views', 'youtube': 'likes', 'spotify': 'streams', 'soundcloud': 'plays'} } for entity in entities_bunch: for dsp, chart_type in chart_types[entity_type].items(): if not entity[dsp][chart_type]['chart_data']: entities_bunch.remove(entity) break return random.choice(entities_bunch) @step("Add entity to Recent Searches") def add_to_recent_searches(self, entity_id, entity_type): data = {"entity_id": entity_id, "entity_type": entity_type} self.post(api_recent_searches, data).json() @step("Insert Category") def insert_category(self, name, is_homepage=False): category = self.post(api_categories, {"name": name, "is_homepage": is_homepage}).json() self.log.info(f"Inserted category item: {category}") return category['id'] @step("Insert entity to category") def insert_entity_to_category(self, category_id, entity_id, entity_type): entity_data = { "category_id": category_id, "entity_id": entity_id, "entity_type": entity_type, "chart_type": "followers"} resp = self.post(api_category_entities.format(category_id), data=entity_data).json() self.log.info(f"Inserted entity to category: {resp}") @step("Prepare category with entity") def prepare_category_with_entity(self, entity_type='artist'): category_id = self.insert_category('test search category') entity_id = self.get_random_entity(entity_type)['id'] self.insert_entity_to_category(category_id, entity_id, entity_type) return category_id @step("Pre-conditions: Create Discovery via API") def insert_discovery(self, entity_type): with open(f'src/data/{entity_type}_discovery.json') as json_file: discovery_json = json.load(json_file) resp = self.post(api_saved_discoveries, discovery_json).json() self.log.info(f"Inserted discovery: {resp}") @step("Return saved discoveries") def get_saved_discoveries(self): return self.getting(api_saved_discoveries) @step("Delete all users Saved Discoveries") def delete_all_saved_discoveries(self): saved_discoveries = self.get_saved_discoveries() print(saved_discoveries) if saved_discoveries is not None: for discovery in saved_discoveries: print("Delete saved discovery id={}".format(discovery['id'])) self.delete(f"{api_saved_discoveries}/{discovery['id']}") @step("Add {filter_type}:{value} to Favorites") def add_to_favorite(self, filter_type, value): self.log.info(f'Add {filter_type}: {value} to Favorites') resp = self.put(api_favorites.format(filter_type, value)) self.log.info(f"RESPONSE 'status_code': {resp.status_code}") return @step("Check that entity's parameter is in range") def check_entity_in_range(self, entity, tested_value, min_val=None, max_val=None): print(f"Verify entity_id = {entity['id']}") print(f"{min_val} < {tested_value} < {max_val}") if min_val is not None: assert_that(tested_value).is_greater_than_or_equal_to(min_val * (1 - self.error_rate)) if max_val is not None: assert_that(tested_value).is_less_than_or_equal_to(max_val * (1 + self.error_rate))