Sunday, August 28, 2016

Data containers for REST calls and responses.

I would say most people using python probably wouldn't do this, but coming from a c# and java background, it's ingrained in my head to make container classes for data.  I think it helps keep track of the data.  Admittedly, I've done some tricky things to do to make generic methods to translate the fields into json (or some other payload), and back.  But once they are written you don't have to deal with it anymore.


First up the container for posting a status.  As you saw earlier the status gets turned into a query string, and that query string is sent as the payload to the post.  I have decided to leave out several fields in the data class.  They can be easily added in, if we decided to test those features (location, picture assets, etc),

 from DataObjects.DataObjectBase import DataObjectBase  
   
   
 class PostStatusRequest(DataObjectBase):  
   status = None  
   in_reply_to_status_id = None  
   possibly_sensitive = None  
   trim_user = None  
   
   def __init__(self, status, trim_user=False, possibly_sensitive=None, in_reply_to_status_id=None):  
     self.status = status  
     self.trim_user = trim_user  
     self.possibly_sensitive = possibly_sensitive  
     self.in_reply_to_status_id = in_reply_to_status_id  
 

The only field we currently use in our tests is "status", but a class with just 1 field in it would be kinda boring, so I added the other fields.  I would write tests that use them, but my test suite is already pushing the boundaries of how many requests I make to Twitter in a 15 minute period.  Perhaps I'll disable some tests so I can write some tests that utilize these fields.

Not many methods here?  The "magic" happens in the base class:

 import urllib  
   
   
 class DataObjectBase:  
   
   def create_query_string(self):  
     payload = ""  
     items = self.__dict__  
     for key in items:  
       if items[key]:  
         payload += "{}={}&".format(key, urllib.parse.quote(str(items[key])))  
     if payload:  
       # removing trailing "&"  
       payload = payload[:-1]  
     return payload  
   
   @classmethod  
   def from_dict(cls, data_dict):  
     cls = cls()  
     for key in data_dict:  
       cls.__dict__[key] = data_dict[key]  
     return cls  
   

Here we have methods to create a query string, and load up the class from the dictionary response we get from our service calls.

So we have our request, now how about our response.  We have 2 classes for responses, StatusResponse, and StatusResponses.  As you can guess one of the classes is for the data of a single record, and the other handles data of a list of statuses returned.  For the record when I build classes like this I generally just use 1 class, creating a dummy field to hold the list data, but I thought having 2 classes would make it more clear what the intent is for this example.

from DataObjects.DataObjectBase import DataObjectBase  
   
   
class StatusResponse(DataObjectBase):  
     created_at = None  
     id = None  
     id_str = None  
     text = None  
     truncated = False  
     entities = {}  
     source = None  
     in_reply_to_status_id = None  
     in_reply_to_status_id_str = None  
     in_reply_to_user_id = None  
     in_reply_to_user_id_str = None  
     in_reply_to_screen_name = None  
     user = {}  
     geo = None  
   coordinates = None  
   place = None  
   contributors = None  
   is_quote_status = None  
   retweet_count = None  
   favorite_count = None  
   favorited = None  
   retweeted = None  
   lang = None  
   
   def __init__(self):  
     # Nothing here, as everything gets loaded via base class routines.  
     pass  

You will notice above some empty dictionaries.  This is data that comes from the json payload that I don't really need to put into data classes, as I'm not (yet) verifying the data.  If I did need to put them into data classes I would have to override the from_dict() method to create whatever data classes were there, and put the data into these fields.

And StatusResponses:

from DataObjects.DataObjectBase import DataObjectBase  
from DataObjects.StatusResponse import StatusResponse  
   
   
class StatusResponses(DataObjectBase):  
   status_list = []  
   
def __init__(self):  
     # Nothing here, as everything gets loaded below  
     pass  
   
@classmethod  
def list_from_dict(cls, data_list):  
     cls = cls()  
     cls.status_list = []  
     for item in data_list:  
         status = StatusResponse.from_dict(item)  
         cls.status_list.append(status)  
     return cls  
   
def find_record_with_status_text(self, status_text):  
     return next((r for r in self.status_list if r.text == status_text), None)  
   
def find_record_by_id(self, id):  
     return next((r for r in self.status_list if r.id == id), None)  

You can see StatusResponses is a little more interesting.  The method list_from_dict is used to assimilate the list of objects into the class.  I tried quite a few ways of just making this a base class method, but the problem is you need to know the field name of the list.  I got some very odd results when trying to do this in the base class (I'd almost say they were bugs in Python, but sometimes doing things from pytest causes odd things to happen)

There are also 2 find record methods.  clearly because we are looking at a particular field in the response class neither of them can be base classed either.

How do we use these classes?  Lets look at post_to_timeline, and get_timeline in our test suite:


 def get_timeline(self):  
     response = self.twitter_service.get_home_timeline()  
     TestHelpers.verify_http_response(response, 200, "Get Twitter Account Timeline")  
     response_record = StatusResponses.list_from_dict(response.response_json)  
     return response_record  
   
   def post_to_timeline(self, tweet_text=None):  
     # This will add a status to the timeline  
     if tweet_text is None:  
       tweet_text = "Test Status {}".format(TestHelpers.make_random_string(6))  
     payload = PostStatusRequest(tweet_text)  
     self.logger.debug("Post status payload = {}".format(payload.create_query_string()))  
     response = self.twitter_service.post_tweet(payload.create_query_string())  
     TestHelpers.verify_http_response(response, 200, "Post to Timeline")  
     response_record = StatusResponses.from_dict(response.response_json)  
     return response_record  

You can see in both get_timeline and post_to_timeline we load the response into our StatusResponses and StatusResponse records, using the appropriate loader.  And in post_to_timeline we create a payload using the PostStatusRequest class, and calling the create_query_string method to create a query string when we call post_tweet.

We can use the response record to verify our data.  Let's look at the get_timeline test.


 def test_get_timeline(self):  
     self.logger.info("")  
     self.logger.info("Get timeline test")  
     # Post something to the timeline to find later  
     post_response_record_text = self.post_to_timeline().text  
   
     response_record = self.get_timeline()  
     assert len(response_record.status_list) > 0  
     assert response_record.find_record_with_status_text(post_response_record_text), \  
       "Could not find record with posted text in timeline"  

You can that we call post_to_timeline to add a post to the timeline so we can find it when we retrieve out timeline.  We then call get_timeline, and search the response records for the status text we posted earlier.

No comments:

Post a Comment