use temp directory for all code files in tests

This commit is contained in:
Alan Hamlett 2016-06-08 20:33:36 +02:00
parent e918bb0997
commit 429c61967a
3 changed files with 890 additions and 749 deletions

File diff suppressed because it is too large Load diff

View file

@ -7,6 +7,7 @@ from wakatime.packages import requests
import os import os
import time import time
import re import re
import shutil
import sys import sys
from wakatime.compat import u, is_py3 from wakatime.compat import u, is_py3
from wakatime.constants import ( from wakatime.constants import (
@ -57,320 +58,372 @@ class BaseTestCase(utils.TestCase):
response.status_code = 201 response.status_code = 201
self.patched['wakatime.packages.requests.adapters.HTTPAdapter.send'].return_value = response self.patched['wakatime.packages.requests.adapters.HTTPAdapter.send'].return_value = response
entity = 'tests/samples/codefiles/twolinefile.txt' with utils.TemporaryDirectory() as tempdir:
config = 'tests/samples/configs/good_config.cfg' entity = 'tests/samples/codefiles/twolinefile.txt'
args = ['--file', entity, '--key', '123', '--config', config] shutil.copy(entity, os.path.join(tempdir, 'twolinefile.txt'))
entity = os.path.realpath(os.path.join(tempdir, 'twolinefile.txt'))
retval = execute(args) config = 'tests/samples/configs/good_config.cfg'
self.assertEquals(retval, SUCCESS) args = ['--file', entity, '--key', '123', '--config', config]
self.assertEquals(sys.stdout.getvalue(), '')
self.assertEquals(sys.stderr.getvalue(), '')
self.patched['wakatime.session_cache.SessionCache.get'].assert_called_once_with() retval = execute(args)
self.patched['wakatime.session_cache.SessionCache.delete'].assert_not_called() self.assertEquals(retval, SUCCESS)
self.patched['wakatime.session_cache.SessionCache.save'].assert_called_once_with(ANY) self.assertEquals(sys.stdout.getvalue(), '')
self.assertEquals(sys.stderr.getvalue(), '')
self.patched['wakatime.offlinequeue.Queue.push'].assert_not_called() self.patched['wakatime.session_cache.SessionCache.get'].assert_called_once_with()
self.patched['wakatime.offlinequeue.Queue.pop'].assert_called_once_with() self.patched['wakatime.session_cache.SessionCache.delete'].assert_not_called()
self.patched['wakatime.session_cache.SessionCache.save'].assert_called_once_with(ANY)
self.patched['wakatime.offlinequeue.Queue.push'].assert_not_called()
self.patched['wakatime.offlinequeue.Queue.pop'].assert_called_once_with()
def test_config_file_not_passed_in_command_line_args(self): def test_config_file_not_passed_in_command_line_args(self):
response = Response() response = Response()
response.status_code = 201 response.status_code = 201
self.patched['wakatime.packages.requests.adapters.HTTPAdapter.send'].return_value = response self.patched['wakatime.packages.requests.adapters.HTTPAdapter.send'].return_value = response
with utils.mock.patch('wakatime.main.open') as mock_open: with utils.TemporaryDirectory() as tempdir:
mock_open.side_effect = IOError('')
config = os.path.join(os.path.expanduser('~'), '.wakatime.cfg')
entity = 'tests/samples/codefiles/emptyfile.txt' entity = 'tests/samples/codefiles/emptyfile.txt'
args = ['--file', entity] shutil.copy(entity, os.path.join(tempdir, 'emptyfile.txt'))
entity = os.path.realpath(os.path.join(tempdir, 'emptyfile.txt'))
with self.assertRaises(SystemExit): with utils.mock.patch('wakatime.main.open') as mock_open:
execute(args) mock_open.side_effect = IOError('')
expected_stdout = u("Error: Could not read from config file {0}\n").format(u(config))
expected_stderr = open('tests/samples/output/test_missing_config_file').read() config = os.path.join(os.path.expanduser('~'), '.wakatime.cfg')
self.assertEquals(sys.stdout.getvalue(), expected_stdout) args = ['--file', entity]
self.assertEquals(sys.stderr.getvalue(), expected_stderr)
self.patched['wakatime.session_cache.SessionCache.get'].assert_not_called() with self.assertRaises(SystemExit):
execute(args)
expected_stdout = u("Error: Could not read from config file {0}\n").format(u(config))
expected_stderr = open('tests/samples/output/test_missing_config_file').read()
self.assertEquals(sys.stdout.getvalue(), expected_stdout)
self.assertEquals(sys.stderr.getvalue(), expected_stderr)
self.patched['wakatime.session_cache.SessionCache.get'].assert_not_called()
def test_missing_config_file(self): def test_missing_config_file(self):
config = 'foo' config = 'foo'
entity = 'tests/samples/codefiles/emptyfile.txt'
args = ['--file', entity, '--config', config]
with self.assertRaises(SystemExit):
execute(args)
expected_stdout = u("Error: Could not read from config file foo\n")
expected_stderr = open('tests/samples/output/test_missing_config_file').read()
self.assertEquals(sys.stdout.getvalue(), expected_stdout)
self.assertEquals(sys.stderr.getvalue(), expected_stderr)
self.patched['wakatime.session_cache.SessionCache.get'].assert_not_called() with utils.TemporaryDirectory() as tempdir:
entity = 'tests/samples/codefiles/emptyfile.txt'
shutil.copy(entity, os.path.join(tempdir, 'emptyfile.txt'))
entity = os.path.realpath(os.path.join(tempdir, 'emptyfile.txt'))
args = ['--file', entity, '--config', config]
with self.assertRaises(SystemExit):
execute(args)
expected_stdout = u("Error: Could not read from config file foo\n")
expected_stderr = open('tests/samples/output/test_missing_config_file').read()
self.assertEquals(sys.stdout.getvalue(), expected_stdout)
self.assertEquals(sys.stderr.getvalue(), expected_stderr)
self.patched['wakatime.session_cache.SessionCache.get'].assert_not_called()
def test_good_config_file(self): def test_good_config_file(self):
response = Response() response = Response()
response.status_code = 201 response.status_code = 201
self.patched['wakatime.packages.requests.adapters.HTTPAdapter.send'].return_value = response self.patched['wakatime.packages.requests.adapters.HTTPAdapter.send'].return_value = response
entity = 'tests/samples/codefiles/emptyfile.txt' with utils.TemporaryDirectory() as tempdir:
config = 'tests/samples/configs/has_everything.cfg' entity = 'tests/samples/codefiles/emptyfile.txt'
args = ['--file', entity, '--config', config] shutil.copy(entity, os.path.join(tempdir, 'emptyfile.txt'))
retval = execute(args) entity = os.path.realpath(os.path.join(tempdir, 'emptyfile.txt'))
self.assertEquals(retval, SUCCESS)
expected_stdout = open('tests/samples/output/main_test_good_config_file').read()
traceback_file = os.path.realpath('wakatime/main.py')
lineno = int(re.search(r' line (\d+),', sys.stdout.getvalue()).group(1))
self.assertEquals(sys.stdout.getvalue(), expected_stdout.format(file=traceback_file, lineno=lineno))
self.assertEquals(sys.stderr.getvalue(), '')
self.patched['wakatime.session_cache.SessionCache.get'].assert_called_once_with() config = 'tests/samples/configs/has_everything.cfg'
self.patched['wakatime.session_cache.SessionCache.delete'].assert_not_called() args = ['--file', entity, '--config', config]
self.patched['wakatime.session_cache.SessionCache.save'].assert_called_once_with(ANY) retval = execute(args)
self.assertEquals(retval, SUCCESS)
expected_stdout = open('tests/samples/output/main_test_good_config_file').read()
traceback_file = os.path.realpath('wakatime/main.py')
lineno = int(re.search(r' line (\d+),', sys.stdout.getvalue()).group(1))
self.assertEquals(sys.stdout.getvalue(), expected_stdout.format(file=traceback_file, lineno=lineno))
self.assertEquals(sys.stderr.getvalue(), '')
self.patched['wakatime.offlinequeue.Queue.push'].assert_not_called() self.patched['wakatime.session_cache.SessionCache.get'].assert_called_once_with()
self.patched['wakatime.offlinequeue.Queue.pop'].assert_called_once_with() self.patched['wakatime.session_cache.SessionCache.delete'].assert_not_called()
self.patched['wakatime.session_cache.SessionCache.save'].assert_called_once_with(ANY)
self.patched['wakatime.offlinequeue.Queue.push'].assert_not_called()
self.patched['wakatime.offlinequeue.Queue.pop'].assert_called_once_with()
def test_api_key_without_underscore_accepted(self): def test_api_key_without_underscore_accepted(self):
response = Response() response = Response()
response.status_code = 201 response.status_code = 201
self.patched['wakatime.packages.requests.adapters.HTTPAdapter.send'].return_value = response self.patched['wakatime.packages.requests.adapters.HTTPAdapter.send'].return_value = response
entity = 'tests/samples/codefiles/emptyfile.txt' with utils.TemporaryDirectory() as tempdir:
config = 'tests/samples/configs/sample_alternate_apikey.cfg' entity = 'tests/samples/codefiles/emptyfile.txt'
args = ['--file', entity, '--config', config] shutil.copy(entity, os.path.join(tempdir, 'emptyfile.txt'))
retval = execute(args) entity = os.path.realpath(os.path.join(tempdir, 'emptyfile.txt'))
self.assertEquals(retval, SUCCESS)
self.assertEquals(sys.stdout.getvalue(), '')
self.assertEquals(sys.stderr.getvalue(), '')
self.patched['wakatime.session_cache.SessionCache.get'].assert_called_once_with() config = 'tests/samples/configs/sample_alternate_apikey.cfg'
self.patched['wakatime.session_cache.SessionCache.delete'].assert_not_called() args = ['--file', entity, '--config', config]
self.patched['wakatime.session_cache.SessionCache.save'].assert_called_once_with(ANY) retval = execute(args)
self.assertEquals(retval, SUCCESS)
self.assertEquals(sys.stdout.getvalue(), '')
self.assertEquals(sys.stderr.getvalue(), '')
self.patched['wakatime.offlinequeue.Queue.push'].assert_not_called() self.patched['wakatime.session_cache.SessionCache.get'].assert_called_once_with()
self.patched['wakatime.offlinequeue.Queue.pop'].assert_called_once_with() self.patched['wakatime.session_cache.SessionCache.delete'].assert_not_called()
self.patched['wakatime.session_cache.SessionCache.save'].assert_called_once_with(ANY)
self.patched['wakatime.offlinequeue.Queue.push'].assert_not_called()
self.patched['wakatime.offlinequeue.Queue.pop'].assert_called_once_with()
def test_bad_config_file(self): def test_bad_config_file(self):
entity = 'tests/samples/codefiles/emptyfile.txt' with utils.TemporaryDirectory() as tempdir:
config = 'tests/samples/configs/bad_config.cfg' entity = 'tests/samples/codefiles/emptyfile.txt'
args = ['--file', entity, '--config', config] shutil.copy(entity, os.path.join(tempdir, 'emptyfile.txt'))
retval = execute(args) entity = os.path.realpath(os.path.join(tempdir, 'emptyfile.txt'))
self.assertEquals(retval, CONFIG_FILE_PARSE_ERROR)
self.assertIn('ParsingError', sys.stdout.getvalue()) config = 'tests/samples/configs/bad_config.cfg'
self.assertEquals(sys.stderr.getvalue(), '') args = ['--file', entity, '--config', config]
self.patched['wakatime.offlinequeue.Queue.push'].assert_not_called() retval = execute(args)
self.patched['wakatime.session_cache.SessionCache.get'].assert_not_called() self.assertEquals(retval, CONFIG_FILE_PARSE_ERROR)
self.patched['wakatime.session_cache.SessionCache.delete'].assert_not_called() self.assertIn('ParsingError', sys.stdout.getvalue())
self.patched['wakatime.session_cache.SessionCache.save'].assert_not_called() self.assertEquals(sys.stderr.getvalue(), '')
self.patched['wakatime.offlinequeue.Queue.push'].assert_not_called()
self.patched['wakatime.session_cache.SessionCache.get'].assert_not_called()
self.patched['wakatime.session_cache.SessionCache.delete'].assert_not_called()
self.patched['wakatime.session_cache.SessionCache.save'].assert_not_called()
def test_non_hidden_filename(self): def test_non_hidden_filename(self):
response = Response() response = Response()
response.status_code = 0 response.status_code = 0
self.patched['wakatime.packages.requests.adapters.HTTPAdapter.send'].return_value = response self.patched['wakatime.packages.requests.adapters.HTTPAdapter.send'].return_value = response
now = u(int(time.time())) with utils.TemporaryDirectory() as tempdir:
entity = 'tests/samples/codefiles/twolinefile.txt' entity = 'tests/samples/codefiles/twolinefile.txt'
config = 'tests/samples/configs/good_config.cfg' shutil.copy(entity, os.path.join(tempdir, 'twolinefile.txt'))
entity = os.path.realpath(os.path.join(tempdir, 'twolinefile.txt'))
args = ['--file', entity, '--key', '123', '--config', config, '--time', now] now = u(int(time.time()))
config = 'tests/samples/configs/good_config.cfg'
retval = execute(args) args = ['--file', entity, '--key', '123', '--config', config, '--time', now]
self.assertEquals(retval, API_ERROR)
self.assertEquals(sys.stdout.getvalue(), '')
self.assertEquals(sys.stderr.getvalue(), '')
self.patched['wakatime.session_cache.SessionCache.get'].assert_called_once_with() retval = execute(args)
self.patched['wakatime.session_cache.SessionCache.delete'].assert_called_once_with() self.assertEquals(retval, API_ERROR)
self.patched['wakatime.session_cache.SessionCache.save'].assert_not_called() self.assertEquals(sys.stdout.getvalue(), '')
self.assertEquals(sys.stderr.getvalue(), '')
heartbeat = { self.patched['wakatime.session_cache.SessionCache.get'].assert_called_once_with()
'language': 'Text only', self.patched['wakatime.session_cache.SessionCache.delete'].assert_called_once_with()
'lines': 2, self.patched['wakatime.session_cache.SessionCache.save'].assert_not_called()
'entity': os.path.abspath(entity),
'project': os.path.basename(os.path.abspath('.')),
'time': float(now),
'type': 'file',
}
stats = {
u('cursorpos'): None,
u('dependencies'): [],
u('language'): u('Text only'),
u('lineno'): None,
u('lines'): 2,
}
self.patched['wakatime.offlinequeue.Queue.push'].assert_called_once_with(ANY, ANY, None) heartbeat = {
for key, val in self.patched['wakatime.offlinequeue.Queue.push'].call_args[0][0].items(): 'language': 'Text only',
self.assertEquals(heartbeat[key], val) 'lines': 2,
self.assertEquals(stats, json.loads(self.patched['wakatime.offlinequeue.Queue.push'].call_args[0][1])) 'entity': os.path.realpath(entity),
self.patched['wakatime.offlinequeue.Queue.pop'].assert_not_called() 'project': os.path.basename(os.path.abspath('.')),
'time': float(now),
'type': 'file',
}
stats = {
u('cursorpos'): None,
u('dependencies'): [],
u('language'): u('Text only'),
u('lineno'): None,
u('lines'): 2,
}
self.patched['wakatime.offlinequeue.Queue.push'].assert_called_once_with(ANY, ANY, None)
for key, val in self.patched['wakatime.offlinequeue.Queue.push'].call_args[0][0].items():
self.assertEquals(heartbeat[key], val)
self.assertEquals(stats, json.loads(self.patched['wakatime.offlinequeue.Queue.push'].call_args[0][1]))
self.patched['wakatime.offlinequeue.Queue.pop'].assert_not_called()
def test_hidden_filename(self): def test_hidden_filename(self):
response = Response() response = Response()
response.status_code = 0 response.status_code = 0
self.patched['wakatime.packages.requests.adapters.HTTPAdapter.send'].return_value = response self.patched['wakatime.packages.requests.adapters.HTTPAdapter.send'].return_value = response
now = u(int(time.time())) with utils.TemporaryDirectory() as tempdir:
entity = 'tests/samples/codefiles/twolinefile.txt' entity = 'tests/samples/codefiles/twolinefile.txt'
config = 'tests/samples/configs/paranoid.cfg' shutil.copy(entity, os.path.join(tempdir, 'twolinefile.txt'))
entity = os.path.realpath(os.path.join(tempdir, 'twolinefile.txt'))
args = ['--file', entity, '--key', '123', '--config', config, '--time', now] now = u(int(time.time()))
config = 'tests/samples/configs/paranoid.cfg'
retval = execute(args) args = ['--file', entity, '--key', '123', '--config', config, '--time', now]
self.assertEquals(retval, API_ERROR)
self.assertEquals(sys.stdout.getvalue(), '')
self.assertEquals(sys.stderr.getvalue(), '')
self.patched['wakatime.session_cache.SessionCache.get'].assert_called_once_with() retval = execute(args)
self.patched['wakatime.session_cache.SessionCache.delete'].assert_called_once_with() self.assertEquals(retval, API_ERROR)
self.patched['wakatime.session_cache.SessionCache.save'].assert_not_called() self.assertEquals(sys.stdout.getvalue(), '')
self.assertEquals(sys.stderr.getvalue(), '')
heartbeat = { self.patched['wakatime.session_cache.SessionCache.get'].assert_called_once_with()
'language': 'Text only', self.patched['wakatime.session_cache.SessionCache.delete'].assert_called_once_with()
'lines': 2, self.patched['wakatime.session_cache.SessionCache.save'].assert_not_called()
'entity': 'HIDDEN.txt',
'project': os.path.basename(os.path.abspath('.')),
'time': float(now),
'type': 'file',
}
stats = {
u('cursorpos'): None,
u('dependencies'): [],
u('language'): u('Text only'),
u('lineno'): None,
u('lines'): 2,
}
self.patched['wakatime.offlinequeue.Queue.push'].assert_called_once_with(ANY, ANY, None) heartbeat = {
for key, val in self.patched['wakatime.offlinequeue.Queue.push'].call_args[0][0].items(): 'language': 'Text only',
self.assertEquals(heartbeat[key], val) 'lines': 2,
self.assertEquals(stats, json.loads(self.patched['wakatime.offlinequeue.Queue.push'].call_args[0][1])) 'entity': 'HIDDEN.txt',
self.patched['wakatime.offlinequeue.Queue.pop'].assert_not_called() 'project': os.path.basename(os.path.abspath('.')),
'time': float(now),
'type': 'file',
}
stats = {
u('cursorpos'): None,
u('dependencies'): [],
u('language'): u('Text only'),
u('lineno'): None,
u('lines'): 2,
}
self.patched['wakatime.offlinequeue.Queue.push'].assert_called_once_with(ANY, ANY, None)
for key, val in self.patched['wakatime.offlinequeue.Queue.push'].call_args[0][0].items():
self.assertEquals(heartbeat[key], val)
self.assertEquals(stats, json.loads(self.patched['wakatime.offlinequeue.Queue.push'].call_args[0][1]))
self.patched['wakatime.offlinequeue.Queue.pop'].assert_not_called()
def test_timeout_passed_via_command_line(self): def test_timeout_passed_via_command_line(self):
response = Response() response = Response()
response.status_code = 201 response.status_code = 201
self.patched['wakatime.packages.requests.adapters.HTTPAdapter.send'].return_value = response self.patched['wakatime.packages.requests.adapters.HTTPAdapter.send'].return_value = response
entity = 'tests/samples/codefiles/twolinefile.txt' with utils.TemporaryDirectory() as tempdir:
config = 'tests/samples/configs/good_config.cfg' entity = 'tests/samples/codefiles/twolinefile.txt'
args = ['--file', entity, '--key', '123', '--config', config, '--timeout', 'abc'] shutil.copy(entity, os.path.join(tempdir, 'twolinefile.txt'))
entity = os.path.realpath(os.path.join(tempdir, 'twolinefile.txt'))
with self.assertRaises(SystemExit): config = 'tests/samples/configs/good_config.cfg'
execute(args) args = ['--file', entity, '--key', '123', '--config', config, '--timeout', 'abc']
self.assertEquals(sys.stdout.getvalue(), '')
expected_stderr = open('tests/samples/output/main_test_timeout_passed_via_command_line').read()
self.assertEquals(sys.stderr.getvalue(), expected_stderr)
self.patched['wakatime.offlinequeue.Queue.push'].assert_not_called() with self.assertRaises(SystemExit):
self.patched['wakatime.offlinequeue.Queue.pop'].assert_not_called() execute(args)
self.patched['wakatime.session_cache.SessionCache.get'].assert_not_called() self.assertEquals(sys.stdout.getvalue(), '')
expected_stderr = open('tests/samples/output/main_test_timeout_passed_via_command_line').read()
self.assertEquals(sys.stderr.getvalue(), expected_stderr)
self.patched['wakatime.offlinequeue.Queue.push'].assert_not_called()
self.patched['wakatime.offlinequeue.Queue.pop'].assert_not_called()
self.patched['wakatime.session_cache.SessionCache.get'].assert_not_called()
def test_500_response(self): def test_500_response(self):
response = Response() response = Response()
response.status_code = 500 response.status_code = 500
self.patched['wakatime.packages.requests.adapters.HTTPAdapter.send'].return_value = response self.patched['wakatime.packages.requests.adapters.HTTPAdapter.send'].return_value = response
now = u(int(time.time())) 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'))
args = ['--file', 'tests/samples/codefiles/twolinefile.txt', '--key', '123', now = u(int(time.time()))
'--config', 'tests/samples/configs/paranoid.cfg', '--time', now]
args = ['--file', entity, '--key', '123',
'--config', 'tests/samples/configs/paranoid.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(), '')
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()
heartbeat = { heartbeat = {
'language': 'Text only', 'language': 'Text only',
'lines': 2, 'lines': 2,
'entity': 'HIDDEN.txt', 'entity': 'HIDDEN.txt',
'project': os.path.basename(os.path.abspath('.')), 'project': os.path.basename(os.path.abspath('.')),
'time': float(now), 'time': float(now),
'type': 'file', 'type': 'file',
} }
stats = { stats = {
u('cursorpos'): None, u('cursorpos'): None,
u('dependencies'): [], u('dependencies'): [],
u('language'): u('Text only'), u('language'): u('Text only'),
u('lineno'): None, u('lineno'): None,
u('lines'): 2, u('lines'): 2,
} }
self.patched['wakatime.offlinequeue.Queue.push'].assert_called_once_with(ANY, ANY, None) self.patched['wakatime.offlinequeue.Queue.push'].assert_called_once_with(ANY, ANY, None)
for key, val in self.patched['wakatime.offlinequeue.Queue.push'].call_args[0][0].items(): for key, val in self.patched['wakatime.offlinequeue.Queue.push'].call_args[0][0].items():
self.assertEquals(heartbeat[key], val) self.assertEquals(heartbeat[key], val)
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_400_response(self): def test_400_response(self):
response = Response() response = Response()
response.status_code = 400 response.status_code = 400
self.patched['wakatime.packages.requests.adapters.HTTPAdapter.send'].return_value = response self.patched['wakatime.packages.requests.adapters.HTTPAdapter.send'].return_value = response
now = u(int(time.time())) 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'))
args = ['--file', 'tests/samples/codefiles/twolinefile.txt', '--key', '123', now = u(int(time.time()))
'--config', 'tests/samples/configs/paranoid.cfg', '--time', now]
args = ['--file', entity, '--key', '123',
'--config', 'tests/samples/configs/paranoid.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(), '')
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()
self.patched['wakatime.offlinequeue.Queue.push'].assert_not_called() self.patched['wakatime.offlinequeue.Queue.push'].assert_not_called()
self.patched['wakatime.offlinequeue.Queue.pop'].assert_not_called() self.patched['wakatime.offlinequeue.Queue.pop'].assert_not_called()
def test_401_response(self): def test_401_response(self):
response = Response() response = Response()
response.status_code = 401 response.status_code = 401
self.patched['wakatime.packages.requests.adapters.HTTPAdapter.send'].return_value = response self.patched['wakatime.packages.requests.adapters.HTTPAdapter.send'].return_value = response
now = u(int(time.time())) 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'))
args = ['--file', 'tests/samples/codefiles/twolinefile.txt', '--key', '123', now = u(int(time.time()))
'--config', 'tests/samples/configs/paranoid.cfg', '--time', now]
args = ['--file', entity, '--key', '123',
'--config', 'tests/samples/configs/paranoid.cfg', '--time', now]
retval = execute(args) retval = execute(args)
self.assertEquals(retval, AUTH_ERROR) self.assertEquals(retval, AUTH_ERROR)
self.assertEquals(sys.stdout.getvalue(), '') self.assertEquals(sys.stdout.getvalue(), '')
self.assertEquals(sys.stderr.getvalue(), '') self.assertEquals(sys.stderr.getvalue(), '')
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()
heartbeat = { heartbeat = {
'language': 'Text only', 'language': 'Text only',
'lines': 2, 'lines': 2,
'entity': 'HIDDEN.txt', 'entity': 'HIDDEN.txt',
'project': os.path.basename(os.path.abspath('.')), 'project': os.path.basename(os.path.abspath('.')),
'time': float(now), 'time': float(now),
'type': 'file', 'type': 'file',
} }
stats = { stats = {
u('cursorpos'): None, u('cursorpos'): None,
u('dependencies'): [], u('dependencies'): [],
u('language'): u('Text only'), u('language'): u('Text only'),
u('lineno'): None, u('lineno'): None,
u('lines'): 2, u('lines'): 2,
} }
self.patched['wakatime.offlinequeue.Queue.push'].assert_called_once_with(ANY, ANY, None) self.patched['wakatime.offlinequeue.Queue.push'].assert_called_once_with(ANY, ANY, None)
for key, val in self.patched['wakatime.offlinequeue.Queue.push'].call_args[0][0].items(): for key, val in self.patched['wakatime.offlinequeue.Queue.push'].call_args[0][0].items():
self.assertEquals(heartbeat[key], val) self.assertEquals(heartbeat[key], val)
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_missing_entity_file(self): def test_missing_entity_file(self):
response = Response() response = Response()
@ -378,6 +431,7 @@ class BaseTestCase(utils.TestCase):
self.patched['wakatime.packages.requests.adapters.HTTPAdapter.send'].return_value = response self.patched['wakatime.packages.requests.adapters.HTTPAdapter.send'].return_value = response
entity = 'tests/samples/codefiles/missingfile.txt' entity = 'tests/samples/codefiles/missingfile.txt'
config = 'tests/samples/configs/good_config.cfg' config = 'tests/samples/configs/good_config.cfg'
args = ['--file', entity, '--config', config] args = ['--file', entity, '--config', config]
retval = execute(args) retval = execute(args)
@ -397,22 +451,26 @@ class BaseTestCase(utils.TestCase):
response.status_code = 201 response.status_code = 201
self.patched['wakatime.packages.requests.adapters.HTTPAdapter.send'].return_value = response self.patched['wakatime.packages.requests.adapters.HTTPAdapter.send'].return_value = response
entity = 'tests/samples/codefiles/emptyfile.txt' with utils.TemporaryDirectory() as tempdir:
config = 'tests/samples/configs/good_config.cfg' entity = 'tests/samples/codefiles/emptyfile.txt'
args = ['--file', entity, '--config', config, '--proxy', 'localhost:1234'] shutil.copy(entity, os.path.join(tempdir, 'emptyfile.txt'))
retval = execute(args) entity = os.path.realpath(os.path.join(tempdir, 'emptyfile.txt'))
self.assertEquals(retval, SUCCESS)
self.assertEquals(sys.stdout.getvalue(), '')
self.assertEquals(sys.stderr.getvalue(), '')
self.patched['wakatime.session_cache.SessionCache.get'].assert_called_once_with() config = 'tests/samples/configs/good_config.cfg'
self.patched['wakatime.session_cache.SessionCache.delete'].assert_not_called() args = ['--file', entity, '--config', config, '--proxy', 'localhost:1234']
self.patched['wakatime.session_cache.SessionCache.save'].assert_called_once_with(ANY) retval = execute(args)
self.assertEquals(retval, SUCCESS)
self.assertEquals(sys.stdout.getvalue(), '')
self.assertEquals(sys.stderr.getvalue(), '')
self.patched['wakatime.offlinequeue.Queue.push'].assert_not_called() self.patched['wakatime.session_cache.SessionCache.get'].assert_called_once_with()
self.patched['wakatime.offlinequeue.Queue.pop'].assert_called_once_with() self.patched['wakatime.session_cache.SessionCache.delete'].assert_not_called()
self.patched['wakatime.session_cache.SessionCache.save'].assert_called_once_with(ANY)
self.patched['wakatime.packages.requests.adapters.HTTPAdapter.send'].assert_called_once_with(ANY, cert=None, proxies={'https': 'localhost:1234'}, stream=False, timeout=60, verify=True) self.patched['wakatime.offlinequeue.Queue.push'].assert_not_called()
self.patched['wakatime.offlinequeue.Queue.pop'].assert_called_once_with()
self.patched['wakatime.packages.requests.adapters.HTTPAdapter.send'].assert_called_once_with(ANY, cert=None, proxies={'https': 'localhost:1234'}, stream=False, timeout=60, verify=True)
def test_entity_type_domain(self): def test_entity_type_domain(self):
response = Response() response = Response()
@ -493,48 +551,83 @@ class BaseTestCase(utils.TestCase):
response.status_code = 201 response.status_code = 201
self.patched['wakatime.packages.requests.adapters.HTTPAdapter.send'].return_value = response self.patched['wakatime.packages.requests.adapters.HTTPAdapter.send'].return_value = response
hostname = 'test汉语' if is_py3 else 'test\xe6\xb1\x89\xe8\xaf\xad' with utils.TemporaryDirectory() as tempdir:
with utils.mock.patch('socket.gethostname') as mock_gethostname:
mock_gethostname.return_value = hostname
self.assertEquals(type(hostname).__name__, 'str')
entity = 'tests/samples/codefiles/emptyfile.txt' entity = 'tests/samples/codefiles/emptyfile.txt'
config = 'tests/samples/configs/has_everything.cfg' shutil.copy(entity, os.path.join(tempdir, 'emptyfile.txt'))
args = ['--file', entity, '--config', config] entity = os.path.realpath(os.path.join(tempdir, 'emptyfile.txt'))
retval = execute(args)
self.assertEquals(retval, SUCCESS)
expected_stdout = open('tests/samples/output/main_test_good_config_file').read()
traceback_file = os.path.realpath('wakatime/main.py')
lineno = int(re.search(r' line (\d+),', sys.stdout.getvalue()).group(1))
self.assertEquals(sys.stdout.getvalue(), expected_stdout.format(file=traceback_file, lineno=lineno))
self.assertEquals(sys.stderr.getvalue(), '')
self.patched['wakatime.session_cache.SessionCache.get'].assert_called_once_with() hostname = 'test汉语' if is_py3 else 'test\xe6\xb1\x89\xe8\xaf\xad'
self.patched['wakatime.session_cache.SessionCache.delete'].assert_not_called() with utils.mock.patch('socket.gethostname') as mock_gethostname:
self.patched['wakatime.session_cache.SessionCache.save'].assert_called_once_with(ANY) mock_gethostname.return_value = hostname
self.assertEquals(type(hostname).__name__, 'str')
self.patched['wakatime.offlinequeue.Queue.push'].assert_not_called() config = 'tests/samples/configs/has_everything.cfg'
self.patched['wakatime.offlinequeue.Queue.pop'].assert_called_once_with() args = ['--file', entity, '--config', config]
retval = execute(args)
self.assertEquals(retval, SUCCESS)
expected_stdout = open('tests/samples/output/main_test_good_config_file').read()
traceback_file = os.path.realpath('wakatime/main.py')
lineno = int(re.search(r' line (\d+),', sys.stdout.getvalue()).group(1))
self.assertEquals(sys.stdout.getvalue(), expected_stdout.format(file=traceback_file, lineno=lineno))
self.assertEquals(sys.stderr.getvalue(), '')
headers = self.patched['wakatime.packages.requests.adapters.HTTPAdapter.send'].call_args[0][0].headers self.patched['wakatime.session_cache.SessionCache.get'].assert_called_once_with()
self.assertEquals(headers.get('X-Machine-Name'), hostname.encode('utf-8') if is_py3 else hostname) self.patched['wakatime.session_cache.SessionCache.delete'].assert_not_called()
self.patched['wakatime.session_cache.SessionCache.save'].assert_called_once_with(ANY)
self.patched['wakatime.offlinequeue.Queue.push'].assert_not_called()
self.patched['wakatime.offlinequeue.Queue.pop'].assert_called_once_with()
headers = self.patched['wakatime.packages.requests.adapters.HTTPAdapter.send'].call_args[0][0].headers
self.assertEquals(headers.get('X-Machine-Name'), hostname.encode('utf-8') if is_py3 else hostname)
def test_nonascii_timezone(self): def test_nonascii_timezone(self):
response = Response() response = Response()
response.status_code = 201 response.status_code = 201
self.patched['wakatime.packages.requests.adapters.HTTPAdapter.send'].return_value = response self.patched['wakatime.packages.requests.adapters.HTTPAdapter.send'].return_value = response
package_path = 'wakatime.packages.py3.tzlocal.get_localzone' if is_py3 else 'wakatime.packages.py2.tzlocal.get_localzone' with utils.TemporaryDirectory() as tempdir:
timezone = tzlocal.get_localzone()
timezone.zone = 'tz汉语' if is_py3 else 'tz\xe6\xb1\x89\xe8\xaf\xad'
with utils.mock.patch(package_path) as mock_getlocalzone:
mock_getlocalzone.return_value = timezone
entity = 'tests/samples/codefiles/emptyfile.txt' entity = 'tests/samples/codefiles/emptyfile.txt'
config = 'tests/samples/configs/has_everything.cfg' shutil.copy(entity, os.path.join(tempdir, 'emptyfile.txt'))
args = ['--file', entity, '--config', config, '--timeout', '15'] entity = os.path.realpath(os.path.join(tempdir, 'emptyfile.txt'))
package_path = 'wakatime.packages.py3.tzlocal.get_localzone' if is_py3 else 'wakatime.packages.py2.tzlocal.get_localzone'
timezone = tzlocal.get_localzone()
timezone.zone = 'tz汉语' if is_py3 else 'tz\xe6\xb1\x89\xe8\xaf\xad'
with utils.mock.patch(package_path) as mock_getlocalzone:
mock_getlocalzone.return_value = timezone
config = 'tests/samples/configs/has_everything.cfg'
args = ['--file', entity, '--config', config, '--timeout', '15']
retval = execute(args)
self.assertEquals(retval, SUCCESS)
self.patched['wakatime.session_cache.SessionCache.get'].assert_called_once_with()
self.patched['wakatime.session_cache.SessionCache.delete'].assert_not_called()
self.patched['wakatime.session_cache.SessionCache.save'].assert_called_once_with(ANY)
self.patched['wakatime.offlinequeue.Queue.push'].assert_not_called()
self.patched['wakatime.offlinequeue.Queue.pop'].assert_called_once_with()
headers = self.patched['wakatime.packages.requests.adapters.HTTPAdapter.send'].call_args[0][0].headers
self.assertEquals(headers.get('TimeZone'), u(timezone.zone).encode('utf-8') if is_py3 else timezone.zone)
def test_timezone_header(self):
response = Response()
response.status_code = 201
self.patched['wakatime.packages.requests.adapters.HTTPAdapter.send'].return_value = response
with utils.TemporaryDirectory() as tempdir:
entity = 'tests/samples/codefiles/emptyfile.txt'
shutil.copy(entity, os.path.join(tempdir, 'emptyfile.txt'))
entity = os.path.realpath(os.path.join(tempdir, 'emptyfile.txt'))
config = 'tests/samples/configs/good_config.cfg'
args = ['--file', entity, '--config', config]
retval = execute(args) retval = execute(args)
self.assertEquals(retval, SUCCESS) self.assertEquals(retval, SUCCESS)
self.assertEquals(sys.stdout.getvalue(), '')
self.assertEquals(sys.stderr.getvalue(), '')
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.delete'].assert_not_called() self.patched['wakatime.session_cache.SessionCache.delete'].assert_not_called()
@ -543,77 +636,59 @@ class BaseTestCase(utils.TestCase):
self.patched['wakatime.offlinequeue.Queue.push'].assert_not_called() self.patched['wakatime.offlinequeue.Queue.push'].assert_not_called()
self.patched['wakatime.offlinequeue.Queue.pop'].assert_called_once_with() self.patched['wakatime.offlinequeue.Queue.pop'].assert_called_once_with()
timezone = tzlocal.get_localzone()
headers = self.patched['wakatime.packages.requests.adapters.HTTPAdapter.send'].call_args[0][0].headers headers = self.patched['wakatime.packages.requests.adapters.HTTPAdapter.send'].call_args[0][0].headers
self.assertEquals(headers.get('TimeZone'), u(timezone.zone).encode('utf-8') if is_py3 else timezone.zone) self.assertEquals(headers.get('TimeZone'), u(timezone.zone).encode('utf-8') if is_py3 else timezone.zone)
def test_timezone_header(self):
response = Response()
response.status_code = 201
self.patched['wakatime.packages.requests.adapters.HTTPAdapter.send'].return_value = response
entity = 'tests/samples/codefiles/emptyfile.txt'
config = 'tests/samples/configs/good_config.cfg'
args = ['--file', entity, '--config', config]
retval = execute(args)
self.assertEquals(retval, SUCCESS)
self.assertEquals(sys.stdout.getvalue(), '')
self.assertEquals(sys.stderr.getvalue(), '')
self.patched['wakatime.session_cache.SessionCache.get'].assert_called_once_with()
self.patched['wakatime.session_cache.SessionCache.delete'].assert_not_called()
self.patched['wakatime.session_cache.SessionCache.save'].assert_called_once_with(ANY)
self.patched['wakatime.offlinequeue.Queue.push'].assert_not_called()
self.patched['wakatime.offlinequeue.Queue.pop'].assert_called_once_with()
timezone = tzlocal.get_localzone()
headers = self.patched['wakatime.packages.requests.adapters.HTTPAdapter.send'].call_args[0][0].headers
self.assertEquals(headers.get('TimeZone'), u(timezone.zone).encode('utf-8') if is_py3 else timezone.zone)
def test_extra_heartbeats_argument(self): def test_extra_heartbeats_argument(self):
response = Response() response = Response()
response.status_code = 201 response.status_code = 201
self.patched['wakatime.packages.requests.adapters.HTTPAdapter.send'].return_value = response self.patched['wakatime.packages.requests.adapters.HTTPAdapter.send'].return_value = response
project1 = os.path.basename(os.path.abspath('.')) with utils.TemporaryDirectory() as tempdir:
project2 = 'xyz' entity = 'tests/samples/codefiles/twolinefile.txt'
entity1 = os.path.abspath('tests/samples/codefiles/emptyfile.txt') shutil.copy(entity, os.path.join(tempdir, 'twolinefile.txt'))
entity2 = os.path.abspath('tests/samples/codefiles/twolinefile.txt') entity = os.path.realpath(os.path.join(tempdir, 'twolinefile.txt'))
config = 'tests/samples/configs/good_config.cfg'
args = ['--file', entity1, '--config', config, '--extra-heartbeats']
with utils.mock.patch('wakatime.main.sys.stdin') as mock_stdin: project1 = os.path.basename(os.path.abspath('.'))
now = int(time.time()) project2 = 'xyz'
heartbeats = json.dumps([{ entity1 = os.path.abspath('tests/samples/codefiles/emptyfile.txt')
'timestamp': now, entity2 = os.path.abspath('tests/samples/codefiles/twolinefile.txt')
'entity': entity2, config = 'tests/samples/configs/good_config.cfg'
'entity_type': 'file', args = ['--file', entity1, '--config', config, '--extra-heartbeats']
'project': project2,
'is_write': True,
}])
mock_stdin.readline.return_value = heartbeats
retval = execute(args) with utils.mock.patch('wakatime.main.sys.stdin') as mock_stdin:
now = int(time.time())
heartbeats = json.dumps([{
'timestamp': now,
'entity': entity2,
'entity_type': 'file',
'project': project2,
'is_write': True,
}])
mock_stdin.readline.return_value = heartbeats
self.assertEquals(retval, SUCCESS) retval = execute(args)
self.assertEquals(sys.stdout.getvalue(), '')
self.assertEquals(sys.stderr.getvalue(), '')
self.patched['wakatime.session_cache.SessionCache.get'].assert_has_calls([call(), call()]) self.assertEquals(retval, SUCCESS)
self.patched['wakatime.session_cache.SessionCache.delete'].assert_not_called() self.assertEquals(sys.stdout.getvalue(), '')
self.patched['wakatime.session_cache.SessionCache.save'].assert_has_calls([call(ANY), call(ANY)]) self.assertEquals(sys.stderr.getvalue(), '')
self.patched['wakatime.offlinequeue.Queue.push'].assert_not_called() self.patched['wakatime.session_cache.SessionCache.get'].assert_has_calls([call(), call()])
self.patched['wakatime.offlinequeue.Queue.pop'].assert_called_once_with() self.patched['wakatime.session_cache.SessionCache.delete'].assert_not_called()
self.patched['wakatime.session_cache.SessionCache.save'].assert_has_calls([call(ANY), call(ANY)])
calls = self.patched['wakatime.packages.requests.adapters.HTTPAdapter.send'].call_args_list self.patched['wakatime.offlinequeue.Queue.push'].assert_not_called()
self.patched['wakatime.offlinequeue.Queue.pop'].assert_called_once_with()
body = calls[0][0][0].body calls = self.patched['wakatime.packages.requests.adapters.HTTPAdapter.send'].call_args_list
data = json.loads(body)
self.assertEquals(data.get('entity'), entity1)
self.assertEquals(data.get('project'), project1)
body = calls[1][0][0].body body = calls[0][0][0].body
data = json.loads(body) data = json.loads(body)
self.assertEquals(data.get('entity'), entity2) self.assertEquals(data.get('entity'), entity1)
self.assertEquals(data.get('project'), project2) self.assertEquals(data.get('project'), project1)
body = calls[1][0][0].body
data = json.loads(body)
self.assertEquals(data.get('entity'), entity2)
self.assertEquals(data.get('project'), project2)

View file

@ -52,3 +52,24 @@ class TestCase(unittest.TestCase):
@property @property
def isPy35(self): def isPy35(self):
return (sys.version_info[0] == 3 and sys.version_info[1] == 5) return (sys.version_info[0] == 3 and sys.version_info[1] == 5)
try:
# Python >= 3
from tempfile import TemporaryDirectory
except ImportError:
# Python < 3
import shutil
import tempfile
class TemporaryDirectory(object):
"""Context manager for tempfile.mkdtemp().
Adds the ability to use with a `with` statement.
"""
def __enter__(self):
self.name = tempfile.mkdtemp()
return self.name
def __exit__(self, exc_type, exc_value, traceback):
shutil.rmtree(self.name)