return SaveBulkLine(self.ue, self, self.insert_map())
def stats(self):
- return self.data.stats()
+ if self.data:
+ return self.data.stats()
+ return []
def statmap(stats):
def inserted_rows(stats, response):
'''Return the number of rows inserted
'''
- return response['lines'] - len(stats)
+ if stats:
+ return response['lines'] - len(stats)
+ return 0
def relation_stats(stats, quote_columns):
def analyze_results(uh, response):
- if response['db_changed']:
- stats = list(uh.stats())
- response['stats'] = stats
- response['statmap'] = statmap(stats)
- i_rows = inserted_rows(stats, response)
- response['inserted_rows'] = i_rows
- r_stats = relation_stats(stats, uh.quote_columns())
- response['relation_stats'] = r_stats
+ stats = list(uh.stats())
+ response['stats'] = stats
+ response['statmap'] = statmap(stats)
+ i_rows = inserted_rows(stats, response)
+ response['inserted_rows'] = i_rows
+ r_stats = relation_stats(stats, uh.quote_columns())
+ response['relation_stats'] = r_stats
+ if response['db_changed']:
log_success(response, stats, i_rows, r_stats)
return response
# analyze_results()
-def test_analyze_results_unchanged(
- mock_log_success, mock_relation_stats, mock_inserted_rows,
- mock_statmap, mock_stats, mock_quote_columns):
- '''Response is returned unchanged, nothing is logged
- '''
- response = UNCHANGED_RESPONSE
-
- uh = bulk_upload.BulkTableUploadHandler(response)
- mocked_stats = mock_stats(uh)
- mocked_quote_columns = mock_quote_columns(uh)
-
- result = bulk_upload.analyze_results(uh, response)
-
- assert mock_log_success.call_count == 0
- assert mock_relation_stats.call_count == 0
- assert mock_inserted_rows.call_count == 0
- assert mocked_stats.call_count == 0
- assert mocked_quote_columns.call_count == 0
- assert result == response
-
-
-def test_analyze_results_changed(
+@pytest.mark.parametrize(
+ ('response', 'log_calls'), [
+ (UNCHANGED_RESPONSE, 0),
+ (CHANGED_RESPONSE, 1)])
+def test_analyze_results(
+ response, log_calls,
mock_log_success, mock_relation_stats, mock_inserted_rows,
mock_statmap, mock_stats, mock_quote_columns):
'''Response is returned unchanged, nothing is logged
'''
- response = copy.deepcopy(CHANGED_RESPONSE)
+ response = copy.deepcopy(response)
uh = bulk_upload.BulkTableUploadHandler(response)
mocked_stats = mock_stats(uh)
result = bulk_upload.analyze_results(uh, response)
- assert mock_log_success.call_count == 1
+ assert mock_log_success.call_count == log_calls
assert mock_relation_stats.call_count == 1
assert mock_inserted_rows.call_count == 1
assert mocked_stats.call_count == 1
- assert mock_statmap.call_count == 1
assert mocked_quote_columns.call_count == 1
assert 'stats' in result
+ assert 'statmap' in result
assert 'inserted_rows' in result
assert 'relation_stats' in result
+ del result['stats']
+ del result['statmap']
+ del result['inserted_rows']
+ del result['relation_stats']
+ assert result == response
+
mock_analyze_results = testing.make_mock_fixture(
bulk_upload, 'analyze_results')