"""Test user for integration API testing.""" from os import getenv from auth.auth import GrassAuth, UserType import requests from tests.consts.api import ( DEMOGRAPHIC_INSIGHTS, GEOGRAPHIC_INSIGHTS_BY_REGION, GEOGRAPHIC_INSIGHTS_BY_TERRITORY, GEOGRAPHIC_ORCHARD_LATEST_DATE, GEOGRAPHIC_ORCHARD_REGIONS, PLAYLIST_METADATA, PLAYLIST_PLACEMENTS, STREAM_CHARTS) class User(object): """Test user for integration API testing.""" def __init__(self, user_id): """Create a test user of ows-analytics.""" self.auth = {} self.user_id = user_id def get_demographic_insights(self, **body): """Get demographic insights response. Args: body (dict): dict should include: from_date to_date stores isrcs """ headers = { 'session': self.auth.get('session_token', ''), 'Grass-Account-Type': 'vendor', 'Grass-Account-Id': '6971', 'Orchard-User-Id': 'alw:68791' } url = self.workstation_url + DEMOGRAPHIC_INSIGHTS response = requests.get(url, params=body, headers=headers) return response def get_stream_charts(self, **body): """Get stream charts response. Args: body (dict): dict should include: from_date to_date stores isrcs """ headers = { 'session': self.auth.get('session_token', ''), 'Grass-Account-Type': 'vendor', 'Grass-Account-Id': '6971', 'Orchard-User-Id': 'alw:68791' } url = self.workstation_url + STREAM_CHARTS response = requests.get(url, headers=headers, params=body) return response def get_playlist_placements(self, **body): """Get playlist placements response. Args: body (dict): dict should include: from_date to_date stores isrcs """ headers = { 'session': self.auth.get('session_token', ''), 'Grass-Account-Type': 'vendor', 'Grass-Account-Id': '6971', 'Orchard-User-Id': 'alw:68791' } url = self.workstation_url + PLAYLIST_PLACEMENTS response = requests.get(url, headers=headers, params=body) return response def get_playlist_metadata(self, **body): """Get playlist metadata response. Args: body (dict): dict should include: playlist_links """ headers = { 'session': self.auth.get('session_token', ''), 'Grass-Account-Type': 'vendor', 'Grass-Account-Id': '6971', 'Orchard-User-Id': 'alw:68791' } url = self.workstation_url + PLAYLIST_METADATA response = requests.get(url, headers=headers, params=body) return response def get_geographic_by_territory(self, **body): """Get geographic insights by territory response. Args: body (dict): dict should include: from_date to_date """ headers = { 'session': self.auth.get('session_token', ''), 'Grass-Account-Type': 'vendor', 'Grass-Account-Id': '6971', 'Orchard-User-Id': 'alw:68791' } url = self.workstation_url + GEOGRAPHIC_INSIGHTS_BY_TERRITORY response = requests.get(url, headers=headers, params=body) return response def get_geographic_by_region(self, **body): """Get geographic insights by region response. Args: body (dict): dict should include: from_date to_date """ headers = { 'session': self.auth.get('session_token', ''), 'Grass-Account-Type': 'vendor', 'Grass-Account-Id': '6971', 'Orchard-User-Id': 'alw:68791' } url = self.workstation_url + GEOGRAPHIC_INSIGHTS_BY_REGION response = requests.get(url, headers=headers, params=body) return response def get_geographic_orchard_regions(self): """Get geographic orchard regions response.""" body = {} headers = { 'session': self.auth.get('session_token', ''), 'Grass-Account-Type': 'vendor', 'Grass-Account-Id': '6971', 'Orchard-User-Id': 'alw:68791' } url = self.workstation_url + GEOGRAPHIC_ORCHARD_REGIONS response = requests.get(url, headers=headers, params=body) return response def get_geographics_latest_date(self): """Get geographic orchard latest available date response.""" body = {} headers = { 'session': self.auth.get('session_token', ''), 'Grass-Account-Type': 'vendor', 'Grass-Account-Id': '6971', 'Orchard-User-Id': 'alw:68791' } url = self.workstation_url + GEOGRAPHIC_ORCHARD_LATEST_DATE response = requests.get(url, headers=headers, params=body) return response class QAUser(User): """Test user for integration API testing in QA environment.""" def __init__(self, user_id): """Create a test user of ows-analytics.""" super().__init__(user_id) self.auth['session_token'] = GrassAuth(UserType.WORKSTATION, user_id).grass_token self.workstation_url = getenv('QA_BASE_URL')