What we want to do make a web service client framework for our twitter requests. This will consist of 3 classes:
- A service request base class, that will be used for all our api web service tests for all basic calls (get, post, put, delete). Note that base class will try to catch all http errors, and put the error information in the Response class (see below). This is so the test can report the failure. Also we don't want to assume what the test's intentions are. It's possible it's trying to create some horrible error scenerio to see how the service response.
- An api service class that will define our interactions with the "statuses" api of twitter. It is assumed that later there may be more twitter api services we want to access, so "statuses" seems like the best level to code up a single service class
- A response class. This response class will be a container for our return data, or error code and messages. This way we have a unified way of returning data back to the calling test of what happened when we made the api service call. This data will include (though it could contain more)
- Response code. Did we get a 200, 201, 204, 400, 404, maybe a 500 level error. If something horrible happens and an error is thrown we'll just return a 0.
- Response json dictionary. Converting it to a dict before return will make it easier to process.
- Any messages or errors returned. Say we do catch an error. Put the error message in here so we can report it back to the test.
- I have found it useful to include the original request information, The action (get, put, post, delete), url, and payload so the test can report it when asserts fail. We'll see if we have time to get to this today.
What this part of our test framework doesn't deal with the data structure of the payload to the service call, or return json (except to serialize the json into a dictionary). We don't know at all what the user is trying to send to the service. We should support anything and everything. To that extent we just pass on the data that's given to us, and also return the json dictionary.
We'll begin with our WebServiceResponse class. It will contain the response from the web service. It's quite simple
# This is just a container class for response information from a web_service call.
class WebServiceResponse:
response_code = None
response_json = None
response_message = None
http_request_url = None
http_method = None
http_payload = None
def __init__(self, response_code, response_json, response_message):
self.response_code = response_code
self.response_json = response_json
self.response_message = response_message
Next we'll create WebServiceBase. This is the bulk of what we are doing today:
import json
from enum import Enum
import httplib2
import oauth2
from WebServicesClients.WebServiceResponse import WebServiceResponse
class HttpMethod(Enum):
GET = 1
PUT = 2
POST = 3
DELETE = 4
class WebServiceBase:
consumer_key = None
consumer_secret = None
access_token = None
access_token_secret = None
def __init__(self, base_url, consumer_key, consumer_secret, access_token, access_token_secret):
self.consumer_key = consumer_key
self.consumer_secret = consumer_secret
self.access_token = access_token
self.access_token_secret = access_token_secret
# lets do some funky python stuff to add the base url to all our urls
url_vars = [attr for attr in dir(self) if not callable(attr) and attr.endswith("_URL")]
for var in url_vars:
self.__dict__[var] = base_url + eval('self.'+var)
def _get(self, url, headers=None):
return self._service_call_impl(url, HttpMethod.GET, "", headers)
def _put(self, url, payload, headers=None):
return self._service_call_impl(url, HttpMethod.PUT, payload, headers)
def _post(self, url, payload, headers=None):
return self._service_call_impl(url, HttpMethod.POST, payload, headers)
def _delete(self, url, headers=None):
return self._service_call_impl(url, HttpMethod.DELETE, "", headers)
def _service_call_impl(self, url, http_method, payload, headers=None):
try:
consumer = oauth2.Consumer(key=self.consumer_key, secret=self.consumer_secret)
token = oauth2.Token(key=self.access_token, secret=self.access_token_secret)
client = oauth2.Client(consumer, token)
resp, content = client.request( url, method=http_method.name, body=payload, headers=headers)
except httplib2.HttpLib2Error as e:
service_response = WebServiceResponse(0, "", str(e))
return service_response
service_response = WebServiceResponse(resp['status'], self.loads_json(content), content)
return service_response
def loads_json(self, myjson):
try:
return json.loads(myjson.decode("utf-8"))
except ValueError:
return None
Some Notes about above:
- I think most of this code is self explanatory, except that business at the end of the constructor. It's a nifty python trick to add the base url to all the service call urls.
- Also you may be wondering about the numerous constructor arguments. I have too. They'll be refactored during the next check in.
Now that we have a base class, implementing our service class is a piece of cake!
Now let's update our tests to utilize the new service call library!
You can now see the following is changed:
from WebServicesClients.WebServiceBase import WebServiceBase
class TwitterStatusesService(WebServiceBase):
# All URLS variable names need to end in '_URL". We do some python magic to add the base url to the front of the
# variables.
HOME_TIMELINE_URL = "/1.1/statuses/home_timeline.json"
UPDATE_URL = "/1.1/statuses/update.json"
def __init__(self, base_url, consumer_key, consumer_secret, access_token, access_token_secret):
super().__init__(base_url, consumer_key, consumer_secret, access_token, access_token_secret)
def get_home_timeline(self):
return self._get(self.HOME_TIMELINE_URL)
def post_tweet(self, payload):
return self._post(self.UPDATE_URL, payload)
Now let's update our tests to utilize the new service call library!
import random
import string
import urllib
from urllib.parse import urlencode
import pytest
from Config import Config
from WebServicesClients.TwitterStatusesService import TwitterStatusesService
class TestTwitterCRUD:
config_file_name = None
config = None
twitter_service = None
def setup_class(self):
self.config_file_name = pytest.config.getoption('config')
print("\nconfig file is {}".format(self.config_file_name))
self.config = Config(self.config_file_name)
self.config.read_config()
self.twitter_service = TwitterStatusesService(self.config.twitter_api_base_url,
self.config.consumer_key,
self.config.consumer_secret,
self.config.access_token,
self.config.access_token_secret)
def test_get_timeline(self):
response = self.twitter_service.get_home_timeline()
assert response.response_code == '200'
assert response.response_json is not None
print(response.response_json)
def test_post_timeline(self):
status = "Test Status {}".format(self.make_random_string(6))
payload = "status={}".format(urllib.parse.quote(status))
response = self.twitter_service.post_tweet(payload)
assert response.response_code == '200'
assert response.response_json is not None
print(response.response_json)
assert status in str(response.response_json)
def make_random_string(self, num_chars):
return ''.join(random.choice(string.ascii_uppercase + string.digits) for _ in range(num_chars))
You can now see the following is changed:
- We're no longer hard-coding the service urls in our tests!
- Our api calls are simple method calls.
- We're still creating our payloads directly within the tests. We'll add some data classes later to deal with that.
- Also all these asserts to ensure the response code is correct and the json wasn't null is a little messy. We'll deal with that too.
This code can be found in https://github.com/lightmanca/TheTestFrameworkBlog/tree/twitter_with_api_service_library
No comments:
Post a Comment