return pgwui_core.core.DataLineProcessor(ue, self)
+def log_response(response):
+ if response['report_success']:
+ if pgwui_core.utils.is_checked(response['csv_checked']):
+ download_fmt = 'CSV'
+ else:
+ download_fmt = 'TAB'
+ if pgwui_core.utils.is_checked(response['one_file_checked']):
+ download_as = 'One file'
+ else:
+ download_as = 'A zip file of one file per result set'
+ log.info(': '.join(
+ [f'Successful sql: DB {response["db"]}',
+ # ' File ({filename}):'
+ f'Download Format {download_fmt}',
+ f'Download AS {download_as}',
+ f'By user {response["user"]}']))
+
+
@view_config(route_name='pgwui_sql',
renderer='pgwui_sql:templates/sql.mak')
@auth_base_view
response['result_rows'] = uh.sql_results
- if response['report_success']:
- # if pgwui_core.utils.is_checked(response['csv_checked']):
- # download_fmt = 'CSV'
- # else:
- # download_fmt = 'TAB'
- log.info('Successful sql: DB {db}:'
- # ' File ({filename}):'
- # ' Format {format}:'
- ' By user {user}'
- .format( # filename=response['filename'],
- # format=download_fmt,
- db=response['db'],
- user=response['user']))
+ log_response(response)
return response
# Karl O. Pinc <kop@karlpinc.com>
-import logging
import pytest
from pyramid.testing import DummyRequest
from pyramid.threadlocal import get_current_request, get_current_registry
@pytest.fixture
def isolate_sql_view(monkeypatch, pyramid_request_config):
- '''Keep upload_view() from calling anything
+ '''Keep sql_view() from calling anything
Also, have isolate_sql_view(response)
cause UploadEngine to return the supplied "response".
# sql_view()
@pytest.fixture
-def return_log_tuples(isolate_sql_view, caplog):
- '''Get result and the caplog.record_tuples from the upload_view() call'''
- caplog.set_level(logging.DEBUG)
-
+def cleanup_result(isolate_sql_view):
+ '''Get result from the upload_view() call, removing pgwui settings'''
def run(response):
isolate_sql_view(response)
result = sql.sql_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(cleanup_result):
'''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_report_success_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)
-
- assert result == response
- assert ([tup[:2] for tup in log_tuples]
- == [('pgwui_sql.views.sql', logging.INFO)])
-
-
-def test_upload_view_report_success_no_csv(return_log_tuples):
- '''When the db did change from not-CSV input something logs'''
- response = CHANGED_RESPONSE
- response['csv_checked'] = constants.UNCHECKED
- (result, log_tuples) = return_log_tuples(response)
-
+ result = cleanup_result(response)
assert result == response
- assert ([tup[:2] for tup in log_tuples]
- == [('pgwui_sql.views.sql', logging.INFO)])