"""Updates features for a particular user.""" from enum import Enum import json from math import isnan from os import getenv import requests class NewVariant(Enum): """Enum to define various states of feature flags.""" CONTROL = 1 ENABLED = 2 class UserType(Enum): """Enum to define various user types.""" WORKSTATION = 1 OA = 2 class OwsFeatures: """Provides utility method to update a feature for a user. Args: new_variant (NewVariant): marks feature as 'control' or 'enabled'. """ def __init__(self, feature_name, user_id, new_variant, user_type): """Initializes class with required fields to update a feature flag.""" self.feature_name = feature_name self.user_id = user_id self.new_variant = new_variant self.base_qa_url = getenv('BASE_QA_URL') self.user_type = user_type if not self.feature_name: raise ValueError('Feature Name is required.') if not self.user_id: raise ValueError('User ID is required.') if not self.new_variant: raise ValueError('New Variant is required.') if not self.base_qa_url: raise ValueError('BASE_QA_URL is required.') if not self.user_type: raise ValueError('User Type is required.') isnan(user_id) if not isinstance(new_variant, NewVariant): raise TypeError('New Variant must be a NewVariant Enum.') if not isinstance(user_type, UserType): raise TypeError('User Type must be a UserType Enum.') def update_feature(self): """Updates a feature flag for the provided user.""" body = json.dumps({'variant_name': self.new_variant.name.lower()}) response = requests.put( self._path(), data=body, headers=OwsFeatures._headers()) if not self._verify_response(response): raise UnexpectedResponse( 'Response came back in an unexpected format. ' 'See response: {}'.format(vars(response))) def _path(self): """Returns a path to ows_features with params filled in.""" return '{base_url}/features/{feature_name}/user/{user_string}'.format( base_url=self.base_qa_url, feature_name=self.feature_name, user_string=self._user_to_string()) @staticmethod def _headers(): """Returns request headers.""" return {'Content-Type': 'application/json'} def _verify_response(self, response): """ Returns false if provided response is not matching expectations. This could be a non-200 response or not matching expected enabled/control state. """ if response.status_code == 200: response_content = json.loads(response.content)[self.feature_name] if self.new_variant == NewVariant.ENABLED \ and response_content == 'enabled': return True elif self.new_variant == NewVariant.CONTROL \ and response_content == 'control': return True else: return False else: return False def _user_to_string(self): user_type_string = '' if self.user_type == UserType.WORKSTATION: user_type_string = 'alw' elif self.user_type == UserType.OA: user_type_string = 'oa' return '{}:{}'.format(user_type_string, self.user_id) class UnexpectedResponse(Exception): """Raised if response from OWS Features did not meet expectation.""" pass