lavatech/rana-cli
2
0
Fork 0
Code Issues Pull requests Releases Wiki Activity

improve test coverage for main module

Browse source
This commit is contained in:
Alan Hamlett 2016-08-31 22:37:02 +02:00
parent 7bf637a42e
commit 9c5b31ce04
2 changed files with 138 additions and 4 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

88
tests/test_main.py
View file

@ -504,7 +504,45 @@ class MainTestCase(utils.TestCase):
self.assertEquals(stats, json.loads(self.patched['wakatime.offlinequeue.Queue.push'].call_args[0][1])) self.assertEquals(stats, json.loads(self.patched['wakatime.offlinequeue.Queue.push'].call_args[0][1]))
self.patched['wakatime.offlinequeue.Queue.pop'].assert_not_called() self.patched['wakatime.offlinequeue.Queue.pop'].assert_not_called()
def test_requests_exception(self): @log_capture()
def test_500_response_without_offline_logging(self, logs):
logging.disable(logging.NOTSET)
response = Response()
response.status_code = 500
response._content = 'fake content'
self.patched['wakatime.packages.requests.adapters.HTTPAdapter.send'].return_value = response
with utils.TemporaryDirectory() as tempdir:
entity = 'tests/samples/codefiles/twolinefile.txt'
shutil.copy(entity, os.path.join(tempdir, 'twolinefile.txt'))
entity = os.path.realpath(os.path.join(tempdir, 'twolinefile.txt'))
now = u(int(time.time()))
args = ['--file', entity, '--key', '123', '--disableoffline',
'--config', 'tests/samples/configs/good_config.cfg', '--time', now]
retval = execute(args)
self.assertEquals(retval, API_ERROR)
self.assertEquals(sys.stdout.getvalue(), '')
self.assertEquals(sys.stderr.getvalue(), '')
log_output = u("\n").join([u(' ').join(x) for x in logs.actual()])
expected = "WakaTime ERROR {'response_code': 500, 'response_content': u'fake content'}"
self.assertEquals(expected, log_output)
self.patched['wakatime.session_cache.SessionCache.delete'].assert_called_once_with()
self.patched['wakatime.session_cache.SessionCache.get'].assert_called_once_with()
self.patched['wakatime.session_cache.SessionCache.save'].assert_not_called()
self.patched['wakatime.offlinequeue.Queue.push'].assert_not_called()
self.patched['wakatime.offlinequeue.Queue.pop'].assert_not_called()
@log_capture()
def test_requests_exception(self, logs):
logging.disable(logging.NOTSET)
self.patched['wakatime.packages.requests.adapters.HTTPAdapter.send'].side_effect = RequestException('requests exception') self.patched['wakatime.packages.requests.adapters.HTTPAdapter.send'].side_effect = RequestException('requests exception')
with utils.TemporaryDirectory() as tempdir: with utils.TemporaryDirectory() as tempdir:
@ -514,14 +552,24 @@ class MainTestCase(utils.TestCase):
now = u(int(time.time())) now = u(int(time.time()))
args = ['--file', entity, '--key', '123', args = ['--file', entity, '--key', '123', '--verbose',
'--config', 'tests/samples/configs/paranoid.cfg', '--time', now] '--config', 'tests/samples/configs/good_config.cfg', '--time', now]
retval = execute(args) retval = execute(args)
self.assertEquals(retval, API_ERROR) self.assertEquals(retval, API_ERROR)
self.assertEquals(sys.stdout.getvalue(), '') self.assertEquals(sys.stdout.getvalue(), '')
self.assertEquals(sys.stderr.getvalue(), '') self.assertEquals(sys.stderr.getvalue(), '')
log_output = u("\n").join([u(' ').join(x) for x in logs.actual()])
expected = 'ImportError: No module named special'
self.assertIn(expected, log_output)
expected = 'WakaTime DEBUG Sending heartbeat to api at https://api.wakatime.com/api/v1/heartbeats'
self.assertIn(expected, log_output)
expected = 'WakaTime DEBUG Traceback'
self.assertIn(expected, log_output)
expected = "RequestException': u'requests exception'"
self.assertIn(expected, log_output)
self.patched['wakatime.session_cache.SessionCache.delete'].assert_called_once_with() self.patched['wakatime.session_cache.SessionCache.delete'].assert_called_once_with()
self.patched['wakatime.session_cache.SessionCache.get'].assert_called_once_with() self.patched['wakatime.session_cache.SessionCache.get'].assert_called_once_with()
self.patched['wakatime.session_cache.SessionCache.save'].assert_not_called() self.patched['wakatime.session_cache.SessionCache.save'].assert_not_called()
@ -529,7 +577,7 @@ class MainTestCase(utils.TestCase):
heartbeat = { heartbeat = {
'language': 'Text only', 'language': 'Text only',
'lines': 2, 'lines': 2,
'entity': 'HIDDEN.txt', 'entity': entity,
'project': os.path.basename(os.path.abspath('.')), 'project': os.path.basename(os.path.abspath('.')),
'time': float(now), 'time': float(now),
'type': 'file', 'type': 'file',
@ -548,6 +596,38 @@ class MainTestCase(utils.TestCase):
self.assertEquals(stats, json.loads(self.patched['wakatime.offlinequeue.Queue.push'].call_args[0][1])) self.assertEquals(stats, json.loads(self.patched['wakatime.offlinequeue.Queue.push'].call_args[0][1]))
self.patched['wakatime.offlinequeue.Queue.pop'].assert_not_called() self.patched['wakatime.offlinequeue.Queue.pop'].assert_not_called()
@log_capture()
def test_requests_exception_without_offline_logging(self, logs):
logging.disable(logging.NOTSET)
self.patched['wakatime.packages.requests.adapters.HTTPAdapter.send'].side_effect = RequestException('requests exception')
with utils.TemporaryDirectory() as tempdir:
entity = 'tests/samples/codefiles/twolinefile.txt'
shutil.copy(entity, os.path.join(tempdir, 'twolinefile.txt'))
entity = os.path.realpath(os.path.join(tempdir, 'twolinefile.txt'))
now = u(int(time.time()))
args = ['--file', entity, '--key', '123', '--disableoffline',
'--config', 'tests/samples/configs/good_config.cfg', '--time', now]
retval = execute(args)
self.assertEquals(retval, API_ERROR)
self.assertEquals(sys.stdout.getvalue(), '')
self.assertEquals(sys.stderr.getvalue(), '')
log_output = u("\n").join([u(' ').join(x) for x in logs.actual()])
expected = "WakaTime ERROR {'RequestException': u'requests exception'}"
self.assertEquals(expected, log_output)
self.patched['wakatime.session_cache.SessionCache.delete'].assert_called_once_with()
self.patched['wakatime.session_cache.SessionCache.get'].assert_called_once_with()
self.patched['wakatime.session_cache.SessionCache.save'].assert_not_called()
self.patched['wakatime.offlinequeue.Queue.push'].assert_not_called()
self.patched['wakatime.offlinequeue.Queue.pop'].assert_not_called()
@log_capture() @log_capture()
def test_missing_entity_file(self, logs): def test_missing_entity_file(self, logs):
logging.disable(logging.NOTSET) logging.disable(logging.NOTSET)

54
tests/test_offlinequeue.py
View file

@ -167,6 +167,60 @@ class OfflineQueueTestCase(utils.TestCase):
self.assertEquals(data.get('project'), project2) self.assertEquals(data.get('project'), project2)
self.assertEquals(u(int(data.get('time'))), now2) self.assertEquals(u(int(data.get('time'))), now2)
def test_auth_error_when_sending_offline_heartbeats(self):
with tempfile.NamedTemporaryFile() as fh:
with utils.mock.patch('wakatime.offlinequeue.Queue.get_db_file') as mock_db_file:
mock_db_file.return_value = fh.name
response = Response()
response.status_code = 500
self.patched['wakatime.packages.requests.adapters.HTTPAdapter.send'].return_value = response
config = 'tests/samples/configs/good_config.cfg'
now1 = u(int(time.time()))
entity1 = 'tests/samples/codefiles/emptyfile.txt'
project1 = 'proj1'
args = ['--file', entity1, '--config', config, '--time', now1, '--project', project1]
execute(args)
now2 = u(int(time.time()))
entity2 = 'tests/samples/codefiles/twolinefile.txt'
project2 = 'proj2'
args = ['--file', entity2, '--config', config, '--time', now2, '--project', project2]
execute(args)
# send heartbeats from offline queue after 201 response
now3 = u(int(time.time()))
entity3 = 'tests/samples/codefiles/python.py'
project3 = 'proj3'
args = ['--file', entity3, '--config', config, '--time', now3, '--project', project3]
class CustomResponse(Response):
count = 0
@property
def status_code(self):
if self.count > 0:
return 401
self.count += 1
return 201
@status_code.setter
def status_code(self, value):
pass
response = CustomResponse()
self.patched['wakatime.packages.requests.adapters.HTTPAdapter.send'].return_value = response
execute(args)
# offline queue should be empty
queue = Queue()
saved_heartbeat = queue.pop()
self.assertEquals(sys.stdout.getvalue(), '')
self.assertEquals(sys.stderr.getvalue(), '')
self.assertEquals(os.path.realpath(entity1), saved_heartbeat['entity'])
def test_empty_project_can_be_saved(self): def test_empty_project_can_be_saved(self):
with tempfile.NamedTemporaryFile() as fh: with tempfile.NamedTemporaryFile() as fh:
with utils.mock.patch('wakatime.offlinequeue.Queue.get_db_file') as mock_db_file: with utils.mock.patch('wakatime.offlinequeue.Queue.get_db_file') as mock_db_file: