Source code for test_feed
"""
Test Feeder module
Provides test methods for automated unit testing of Feeder module
"""
import os
import json
import string
import random
import tempfile
import unittest
from feed import Feeder
[docs]class TestFeeder(unittest.TestCase):
"""
Test Feeder class for unittesting Feeder module
"""
[docs] def setUp(self):
"""
set up method for test fixture
This method runs before each test method. Basically,
it setups the test environment
"""
self.feeder = Feeder()
[docs] def tearDown(self):
"""
set up tear down method for test fixture
This method runs after each test method and cleans
the test environment created in setUp method
"""
# clean everything stored with test_index in elastic search
if self.feeder.es.indices.exists('test_index'):
self.feeder.es.indices.delete('test_index')
del self.feeder
[docs] def test_readContent(self):
"""
tests readContent method of Feeder object
"""
data = {
'singer': 'Ellie Goulding',
'title': 'love me like you do',
'tags': ['cherrytree', 'interscope', 'republic'],
'release': '7 January 2015',
'genre': 'electropop',
'length': 4.12,
'awesomeness': 0.8
}
# use the temporary file and then clean it up after operation
fd, path = tempfile.mkstemp()
try:
with os.fdopen(fd, 'w') as tmp:
jsonDump = json.dumps(data)
tmp.write(jsonDump)
tmp.close()
content = self.feeder.readContent(path)
self.assertDictEqual(data, content)
finally:
os.remove(path)
[docs] def id_generator(self, size=10, chars=string.ascii_uppercase +
string.digits):
"""
generate random token of given size
"""
tok = ''.join(random.SystemRandom().choice(
string.ascii_uppercase + string.digits) for _ in range(size))
return tok
[docs] def test_feed(self):
"""
tests feed method of Feeder object
"""
iname = "test_index"
doc_type = "test_doc"
cont_id = self.id_generator()
data = {
'singer': 'Ellie Goulding',
'title': 'love me like you do',
'tags': ['cherrytree', 'interscope', 'republic'],
'release': '7 January 2015',
'genre': 'electropop',
'length': 4.12,
'awesomeness': 0.8
}
# use temporary file and clean it up after operation
fd, path = tempfile.mkstemp()
with os.fdopen(fd, 'w') as tmpfile:
jsonDump = json.dumps(data)
tmpfile.write(jsonDump)
tmpfile.close()
retVal = self.feeder.feed(path, cont_id, iname, doc_type)
self.assertTrue(retVal)
self.assertTrue(self.feeder.es.indices.exists('test_index'))
if __name__ == '__main__':
unittest.main()