Testing Our Backend – Python For Engineers

Build a Twitter Analytics App

6 Writing the Backend Twitter Server

Writing the Code in Small Parts: Part 1, The Basic App

Part 2: Adding a Counter to Exit

Part 3: Adding Language and Retweet Count

Part 4: Organising Our Code

7 Adding the Data to a Database

8 Testing: What and How to Test

8.1 Testing Our Frontend

8.2 Testing Our Backend

9 Displaying our Data using the Flask Webserver

9.1 Introduction to Flask

9.2 Adding templates to our Flask app

9.3 Displaying our Tweets in the Flask Web Server

10 Future Work and Improvements

Let’s remind ourselves. What does our user expect on the page?

There will be some graphs with the languages and top languages statistics
They will see the top tweets
They will see the trends on Twitter

Now, all of these are webserver tests. However, they also affect the backend, as the backend produces these statistics. So these tests will need to be repeated for the backend.

But this won’t be hard; if you remember, we refactored our code, so that all the above features have their own functions now. All we need to do is write tests for them.

Let’s have a look. The code is here.

We are using the Python Unittest library.

class TestTwit(unittest.TestCase):
    def setUp(self):
        self.conn = sqlite3.connect(":memory:")
        c = self.conn.cursor()

class TestTwit(unittest.TestCase):

def setUp(self):

self.conn = sqlite3.connect(":memory:")

c = self.conn.cursor()

The setUp() function is called before running any tests.

The first thing we do is create a Sqlite database, this time in memory only. This will test the actual database code, without messing up our live database.

        cmd = "CREATE TABLE trend_data (trend TEXT, trend_id1 TEXT, trend_id2 TEXT, trend_id3 TEXT, datetime TEXT)"
        c.execute(cmd)

        cmd = "CREATE TABLE twit_data (top_tweet TEXT, datetime TEXT)"
        c.execute(cmd)

        cmd = "CREATE TABLE lang_data (language TEXT, top_language TEXT, datetime TEXT)"
        c.execute(cmd)


        self.conn.commit()

cmd = "CREATE TABLE trend_data (trend TEXT, trend_id1 TEXT, trend_id2 TEXT, trend_id3 TEXT, datetime TEXT)"

c.execute(cmd)

cmd = "CREATE TABLE twit_data (top_tweet TEXT, datetime TEXT)"

c.execute(cmd)

cmd = "CREATE TABLE lang_data (language TEXT, top_language TEXT, datetime TEXT)"

c.execute(cmd)

self.conn.commit()

We create the tables we need.

        num_tweets_to_grab = 5
        retweet_count = 0
        self.twit = TwitterMain(num_tweets_to_grab, retweet_count, self.conn)

num_tweets_to_grab = 5

retweet_count = 0

self.twit = TwitterMain(num_tweets_to_grab, retweet_count, self.conn)

And we create our Twitter class. We are only grabbing 5 tweets, as it’s good enough to prove our tests work. And the tests will be fast!

There are two helper functions to read data from our database:

    def read_data(self):
        '''
        Shamelssly copied from app.py. Could be common, but I want to keep the tests independent of the code.
        '''
        
        self.conn.row_factory = sqlite3.Row
        c = self.conn.cursor()
        c.execute("SELECT * from lang_data ORDER BY datetime DESC LIMIT 1")

        result = c.fetchone()
        lang = ast.literal_eval(result['language'])
        top_lang = ast.literal_eval(result['top_language'])
      
        return lang, top_lang

    def get_trends(self):
        self.conn.row_factory = sqlite3.Row
        c = self.conn.cursor()

        trend = []
        trend_tweet = []

        c.execute("SELECT * from trend_data ORDER BY datetime DESC LIMIT 5") 
        result = c.fetchall()

        for r in result:
            trend.append(r['trend'])
            trend_tweet.append(r['trend_id1'])
            trend_tweet.append(r['trend_id2'])
            trend_tweet.append(r['trend_id3'])



        return trend, trend_tweet

def read_data(self):

'''

Shamelssly copied from app.py. Could be common, but I want to keep the tests independent of the code.

'''

self.conn.row_factory = sqlite3.Row

c = self.conn.cursor()

c.execute("SELECT * from lang_data ORDER BY datetime DESC LIMIT 1")

result = c.fetchone()

lang = ast.literal_eval(result['language'])

top_lang = ast.literal_eval(result['top_language'])

return lang, top_lang

def get_trends(self):

self.conn.row_factory = sqlite3.Row

c = self.conn.cursor()

trend = []

trend_tweet = []

c.execute("SELECT * from trend_data ORDER BY datetime DESC LIMIT 5")

result = c.fetchall()

for r in result:

trend.append(r['trend'])

trend_tweet.append(r['trend_id1'])

trend_tweet.append(r['trend_id2'])

trend_tweet.append(r['trend_id3'])

return trend, trend_tweet

Okay, to test for our features:

    def test_streaming_data(self):
        self.twit.get_streaming_data()
        try:
            lang, top_lang = self.read_data()
        except:
            self.fail("test_streaming_data: Nothing written to database. Test Failed.")

    def test_trends(self):
        self.twit.get_trends()      
        try:
            trend, trend_tweet = self.get_trends()
        except:
            self.fail("test_trends: Nothing written to database. Test Failed.")

def test_streaming_data(self):

self.twit.get_streaming_data()

try:

lang, top_lang = self.read_data()

except:

self.fail("test_streaming_data: Nothing written to database. Test Failed.")

def test_trends(self):

self.twit.get_trends()

try:

trend, trend_tweet = self.get_trends()

except:

self.fail("test_trends: Nothing written to database. Test Failed.")

The tests are quite simple. I call the functions to gather the tweets and trends, and then check they have been written to the database. We don’t check what the actual content is, because that is not predictable. As long as something was written to our database, our tests will pass. This could be improved, of course.

We have one more test. It is not directly needed, but we do need to test it to ensure our code has no bugs. This is the stats class. The test for it is quite simple.

We just write some data, and read it back:

    def test_stats(self):
        s = stats()
        s.add_lang("English")
        s.add_top_lang("Wooki")

        s.add_top_tweets("Tweet")


        lang, top_lang, top_tweets = s.get_stats()

        self.assertEqual(lang[0], "English")
        self.assertEqual(top_lang[0], "Wooki")
        self.assertEqual(top_tweets[0], "Tweet")

def test_stats(self):

s = stats()

s.add_lang("English")

s.add_top_lang("Wooki")

s.add_top_tweets("Tweet")

lang, top_lang, top_tweets = s.get_stats()

self.assertEqual(lang[0], "English")

self.assertEqual(top_lang[0], "Wooki")

self.assertEqual(top_tweets[0], "Tweet")

Finally, at the end of our tests, we tear down our database.

    def tearDown(self):
        self.conn.close()

1 2	def tearDown(self): self.conn.close()

Next: We start writing our Flask webserver