return SaveLine(ue, self, insert_stmt)
+def log_success(response):
+ if is_checked(response['csv_checked']):
+ upload_fmt = 'CSV'
+ else:
+ upload_fmt = 'TAB'
+ log.info('Successful upload: DB {db}: Table ({table}):'
+ ' File ({filename}): Lines {lines}:'
+ ' Format {format}: Upload Null {null}: Null Rep ({null_rep}):'
+ ' Trim {trim}: By user {user}'
+ .format(filename=response['filename'],
+ lines=response['lines'],
+ format=upload_fmt,
+ null=is_checked(response['upload_null']),
+ null_rep=escape_eol(response['null_rep']),
+ trim=is_checked(response['trim_upload']),
+ db=response['db'],
+ table=response['table'],
+ user=response['user']))
+
+
@view_config(route_name='pgwui_upload',
renderer='pgwui_upload:templates/upload.mak')
@auth_base_view
response['pgwui']['pgwui_upload'] = settings['pgwui']['pgwui_upload']
if response['db_changed']:
- if is_checked(response['csv_checked']):
- upload_fmt = 'CSV'
- else:
- upload_fmt = 'TAB'
- log.info('Successful upload: DB {db}: Table ({table}):'
- ' File ({filename}): Lines {lines}:'
- ' Format {format}: Upload Null {null}: Null Rep ({null_rep}):'
- ' Trim {trim}: By user {user}'
- .format(filename=response['filename'],
- lines=response['lines'],
- format=upload_fmt,
- null=is_checked(response['upload_null']),
- null_rep=escape_eol(response['null_rep']),
- trim=is_checked(response['trim_upload']),
- db=response['db'],
- table=response['table'],
- user=response['user']))
+ log_success(response)
return response
import logging
import pytest
+from pgwui_testing import testing
from pyramid.testing import DummyRequest
from pyramid.threadlocal import get_current_request, get_current_registry
from pgwui_common.__init__ import includeme as pgwui_common_includeme
assert result is False
+# log_success()
+@pytest.mark.parametrize(
+ ('checked',), [
+ (constants.CHECKED,),
+ (constants.UNCHECKED,)])
+def test_log_success(caplog, checked):
+ '''A "info" level log message is logged
+ '''
+ caplog.set_level(logging.DEBUG)
+
+ response = CHANGED_RESPONSE.copy()
+ response['csv_checked'] = checked
+ upload.log_success(response)
+ logs = caplog.record_tuples
+ assert len(logs) == 1
+ assert logs[0][1] == logging.INFO
+
+
+mock_log_success = testing.make_mock_fixture(
+ upload, 'log_success')
+
+
# upload_view()
@pytest.fixture
-def return_log_tuples(isolate_upload_view, caplog):
- '''Get result and the caplog.record_tuples from the upload_view() call'''
- caplog.set_level(logging.DEBUG)
-
+def undecorate(isolate_upload_view):
+ '''Get result and the caplog.record_tuples from the upload_view(),
+ removing stuff added by the view decorators
+ '''
def run(response):
isolate_upload_view(response)
result = upload.upload_view(get_current_request())
del result['pgwui'] # Remove variables added by pgwui view decorators
- return (result, caplog.record_tuples)
+ return result
return run
-def test_upload_view_db_not_changed(return_log_tuples):
+def test_upload_view_db_not_changed(undecorate, mock_log_success):
'''When the db did not change nothing logs'''
response = UNCHANGED_RESPONSE
- (result, log_tuples) = return_log_tuples(response)
- assert result == response
- assert log_tuples == []
-
-
-def test_upload_view_db_changed_csv(return_log_tuples):
- '''When the db did change from CSV input something logs'''
- response = CHANGED_RESPONSE
- response['csv_checked'] = constants.CHECKED
- (result, log_tuples) = return_log_tuples(response)
+ result = undecorate(response)
assert result == response
- assert ([tup[:2] for tup in log_tuples]
- == [('pgwui_upload.views.upload', logging.INFO)])
+ assert mock_log_success.call_count == 0
-def test_upload_view_db_changed_no_csv(return_log_tuples):
- '''When the db did change from not-CSV input something logs'''
+def test_upload_view_db_changed(undecorate, mock_log_success):
+ '''When the db did change from CSV input log_success() is called'''
response = CHANGED_RESPONSE
- response['csv_checked'] = constants.UNCHECKED
- (result, log_tuples) = return_log_tuples(response)
+ result = undecorate(response)
assert result == response
- assert ([tup[:2] for tup in log_tuples]
- == [('pgwui_upload.views.upload', logging.INFO)])
+ assert mock_log_success.call_count == 1
-def test_upload_view_literal_cols_ask(isolate_upload_view):
+def test_upload_view_literal_cols_ask(undecorate, mock_log_success):
'''When literal_column_headings == ask the respose should reflect this'''
response = UNCHANGED_RESPONSE
- isolate_upload_view(response)
+ undecorate(response)
settings = get_current_request().registry.settings
settings['pgwui'].update({'literal_column_headings': 'ask'})
assert result['ask_about_literal_cols']
-def test_upload_view_literal_cols_noask(isolate_upload_view):
+def test_upload_view_literal_cols_noask(undecorate, mock_log_success):
'''When literal_column_headings != ask the respose should reflect this'''
response = UNCHANGED_RESPONSE
- isolate_upload_view(response)
+ undecorate(response)
settings = get_current_registry().settings
settings['pgwui'].update({'literal_column_headings': 'no'})