"""Tests for generic utils.""" from flexmock import flexmock from paramiko import rsakey import pytest import raven from datalytics import utils # Key contents taken from paramiko/tests/test_pkey.py TEST_RSA_PRIVATE_OUT = """\ -----BEGIN RSA PRIVATE KEY----- MIICWgIBAAKBgQDTj1bqB4WmayWNPB+8jVSYpZYk80Ujvj680pOTh2bORBjbIAyz oWGW+GUjzKxTiiPvVmxFgx5wdsFvF03v34lEVVhMpouqPAYQ15N37K/ir5XY+9m/ d8ufMCkjeXsQkKqFbAlQcnWMCRnOoPHS3I4vi6hmnDDeeYTSRvfLbW0fhwIBIwKB gBIiOqZYaoqbeD9OS9z2K9KR2atlTxGxOJPXiP4ESqP3NVScWNwyZ3NXHpyrJLa0 EbVtzsQhLn6rF+TzXnOlcipFvjsem3iYzCpuChfGQ6SovTcOjHV9z+hnpXvQ/fon soVRZY65wKnF7IAoUwTmJS9opqgrN6kRgCd3DASAMd1bAkEA96SBVWFt/fJBNJ9H tYnBKZGw0VeHOYmVYbvMSstssn8un+pQpUm9vlG/bp7Oxd/m+b9KWEh2xPfv6zqU avNwHwJBANqzGZa/EpzF4J8pGti7oIAPUIDGMtfIcmqNXVMckrmzQ2vTfqtkEZsA 4rE1IERRyiJQx6EJsz21wJmGV9WJQ5kCQQDwkS0uXqVdFzgHO6S++tjmjYcxwr3g H0CoFYSgbddOT6miqRskOQF3DZVkJT3kyuBgU2zKygz52ukQZMqxCb1fAkASvuTv qfpH87Qq5kQhNKdbbwbmd2NxlNabazPijWuphGTdW0VfJdWfklyS2Kr+iqrs/5wV HhathJt636Eg7oIjAkA8ht3MQ+XSl9yIJIS8gVpbPxSw5OMfw0PjVE7tBdQruiSc nvuQES5C9BMHjF39LZiGH1iLQy7FgdHyoP+eodI7 -----END RSA PRIVATE KEY----- """ def test_paramiko_rsa_pkey_from_str(): """Test creating Paramiko RSA private key object from string.""" key = utils.paramiko_rsa_pkey_from_str(TEST_RSA_PRIVATE_OUT) # If we were able to create a key, the key contents a most likely valid. assert isinstance(key, rsakey.RSAKey) @pytest.fixture def mock_raven_client(): """Fixture that returns mock raven client.""" client = flexmock() flexmock(raven.Client).new_instances(client) return client @pytest.mark.parametrize( 'exc, should_capture', ( (StopIteration(), False), (GeneratorExit(), False), (SystemExit(), False), (SystemExit(0), False), (SystemExit(1), True), (SystemExit(-1), True), (Exception(), True), (KeyboardInterrupt(), True), ), ) def test_exception_handler_raises(exc, should_capture, mock_raven_client): """Test exception_handler raises all exception types and captures some.""" mock_capture = flexmock(mock_raven_client).should_receive( 'captureException') if should_capture: # If the exception is interesting to us – make sure we capture it. mock_capture.once() else: mock_capture.never() expected_result = 4 # We are raising exceptions and all of them must be bubbled up. with pytest.raises(type(exc)): with utils.exception_handler(): # Do something useful. result = 2 + 2 # Explicitly raise expected exception. raise exc assert result == expected_result def test_exception_handler_with_no_exceptions(mock_raven_client): """Test exception_handler is a noop with no exceptions in the context.""" (flexmock(mock_raven_client) .should_receive('captureException') .never()) expected_result = 4 with utils.exception_handler(): # Do something useful. result = 2 + 2 assert result == expected_result @pytest.mark.parametrize( 'query_string, replacement_string, expected_result', ( ( 'FOO AWS_SECRET_KEY=\'SECRET\' BAR = 1', '***', 'FOO AWS_SECRET_KEY=*** BAR = 1'), ( 'FOO AWS_SECRET_KEY=SECRET BAR = 1', '***', 'FOO AWS_SECRET_KEY=*** BAR = 1'), ( 'FOO AWS_SECRET_KEY = SECRET BAR = 1', '***', 'FOO AWS_SECRET_KEY=*** BAR = 1'), ( 'FOO AWS_SECRET_KEY=SECRET BAR = 1 AWS_SECRET_KEY=SECRET', '***', 'FOO AWS_SECRET_KEY=*** BAR = 1 AWS_SECRET_KEY=***'), ( 'FOO FOO_AWS_SECRET_KEY=SECRET BAR = 1', '***', 'FOO FOO_AWS_SECRET_KEY=*** BAR = 1'), ) ) def test_strip_sql_query_secrets( query_string, replacement_string, expected_result): """Test secrets are stripped from an SQL query.""" result = utils.strip_sql_query_secrets(query_string, replacement_string) assert result == expected_result