860 lines
31 KiB
Python
860 lines
31 KiB
Python
import asyncio
|
|
import collections
|
|
import contextlib
|
|
import json
|
|
import os
|
|
import typing
|
|
import unittest
|
|
import uuid
|
|
from urllib import parse
|
|
|
|
import aiopg
|
|
import asynctest
|
|
import problemdetails
|
|
import psycopg2
|
|
import pycares
|
|
from asynctest import mock
|
|
from psycopg2 import errors
|
|
from sprockets.http import app, testing
|
|
from tornado import ioloop, testing as ttesting, web
|
|
|
|
import sprockets_postgres
|
|
|
|
|
|
test_postgres_cursor_oer_invocation = 0
|
|
|
|
|
|
class RequestHandler(sprockets_postgres.RequestHandlerMixin,
|
|
web.RequestHandler):
|
|
"""Base RequestHandler for test endpoints"""
|
|
|
|
def cast_data(self, data: typing.Union[dict, list, None]) \
|
|
-> typing.Union[dict, list, None]:
|
|
if data is None:
|
|
return None
|
|
elif isinstance(data, list):
|
|
return [self.cast_data(row) for row in data]
|
|
return {k: str(v) for k, v in data.items()}
|
|
|
|
|
|
class CallprocRequestHandler(RequestHandler):
|
|
|
|
async def get(self):
|
|
result = await self.postgres_callproc(
|
|
'uuid_generate_v4', metric_name='uuid')
|
|
await self.finish({'value': str(result.row['uuid_generate_v4'])})
|
|
|
|
|
|
class CountRequestHandler(RequestHandler):
|
|
|
|
GET_SQL = """\
|
|
SELECT last_updated_at, count
|
|
FROM public.query_count
|
|
WHERE key = 'test';"""
|
|
|
|
async def get(self):
|
|
result = await self.postgres_execute(self.GET_SQL)
|
|
assert '<QueryResult row_count=1>' == repr(result)
|
|
assert result.rows[0] == result.row
|
|
await self.finish(self.cast_data(result.row))
|
|
|
|
|
|
class ErrorRequestHandler(RequestHandler):
|
|
|
|
GET_SQL = """\
|
|
SELECT last_updated_at, count
|
|
FROM public.query_count
|
|
WHERE key = 'test';"""
|
|
|
|
async def get(self):
|
|
await self.postgres_execute(self.GET_SQL)
|
|
self.set_status(204)
|
|
|
|
def on_postgres_error(self,
|
|
metric_name: str,
|
|
exc: Exception) -> typing.Optional[Exception]:
|
|
return RuntimeError()
|
|
|
|
|
|
class ErrorPassthroughRequestHandler(RequestHandler):
|
|
|
|
async def get(self):
|
|
exc = self.on_postgres_error('test', RuntimeError())
|
|
if isinstance(exc, RuntimeError):
|
|
self.set_status(204)
|
|
else:
|
|
raise web.HTTPError(500, 'Did not pass through')
|
|
|
|
|
|
class ExecuteRequestHandler(RequestHandler):
|
|
|
|
GET_SQL = 'SELECT %s::TEXT AS value;'
|
|
|
|
async def get(self):
|
|
timeout = self.get_argument('timeout', None)
|
|
if timeout is not None:
|
|
timeout = int(timeout)
|
|
result = await self.postgres_execute(
|
|
self.GET_SQL, [self.get_argument('value')], timeout=timeout)
|
|
await self.finish({
|
|
'value': result.row['value'] if result.row else None})
|
|
|
|
|
|
class ProblemDetailsExecuteRequestHandler(
|
|
problemdetails.ErrorWriter, ExecuteRequestHandler):
|
|
pass
|
|
|
|
|
|
class InfluxDBRequestHandler(ExecuteRequestHandler):
|
|
|
|
def __init__(self, *args, **kwargs):
|
|
super().__init__(*args, **kwargs)
|
|
self.influxdb = self.application.influxdb
|
|
self.influxdb.add_field = mock.Mock()
|
|
|
|
|
|
class RowCountNoRowsRequestHandler(RequestHandler):
|
|
|
|
GET_SQL = 'INSERT INTO public.row_count_no_rows (value) VALUES (%(value)s)'
|
|
|
|
async def get(self):
|
|
count = 0
|
|
for iteration in range(0, 5):
|
|
result = await self.postgres_execute(
|
|
self.GET_SQL, {'value': uuid.uuid4()})
|
|
count += len(result)
|
|
await self.finish({'count': count})
|
|
|
|
|
|
class MetricsMixinRequestHandler(ExecuteRequestHandler):
|
|
|
|
def __init__(self, *args, **kwargs):
|
|
super().__init__(*args, **kwargs)
|
|
self.record_timing = self.application.record_timing
|
|
|
|
|
|
class MultiRowRequestHandler(RequestHandler):
|
|
|
|
GET_SQL = 'SELECT * FROM public.test_rows;'
|
|
|
|
UPDATE_SQL = """\
|
|
UPDATE public.test_rows
|
|
SET toggle = %(to_value)s,
|
|
last_modified_at = CURRENT_TIMESTAMP
|
|
WHERE toggle IS %(from_value)s"""
|
|
|
|
async def get(self):
|
|
result = await self.postgres_execute(self.GET_SQL)
|
|
await self.finish({
|
|
'count': result.row_count,
|
|
'rows': self.cast_data(result.rows),
|
|
'iterator_rows': [self.cast_data(r) for r in result]})
|
|
|
|
async def post(self):
|
|
body = json.loads(self.request.body.decode('utf-8'))
|
|
result = await self.postgres_execute(
|
|
self.UPDATE_SQL, {
|
|
'to_value': body['value'], 'from_value': not body['value']})
|
|
await self.finish({
|
|
'count': result.row_count,
|
|
'rows': self.cast_data(result.rows)})
|
|
|
|
|
|
class NoErrorRequestHandler(ErrorRequestHandler):
|
|
|
|
def on_postgres_error(self,
|
|
metric_name: str,
|
|
exc: Exception) -> typing.Optional[Exception]:
|
|
return None
|
|
|
|
|
|
class NoMixinRequestHandler(web.RequestHandler):
|
|
|
|
async def get(self):
|
|
try:
|
|
async with self.application.postgres_connector() as connector:
|
|
await connector.execute('SELECT 1', {})
|
|
except sprockets_postgres.ConnectionException:
|
|
raise web.HTTPError(503)
|
|
else:
|
|
raise web.HTTPError(418)
|
|
|
|
|
|
class NoRowRequestHandler(RequestHandler):
|
|
|
|
GET_SQL = """\
|
|
SELECT * FROM information_schema.tables WHERE table_schema = 'foo';"""
|
|
|
|
async def get(self):
|
|
result = await self.postgres_execute(self.GET_SQL)
|
|
assert len(result) == result.row_count
|
|
await self.finish({
|
|
'count': result.row_count,
|
|
'rows': self.cast_data(result.rows)})
|
|
|
|
|
|
class TransactionRequestHandler(RequestHandler):
|
|
|
|
GET_SQL = """\
|
|
SELECT id, created_at, last_modified_at, value
|
|
FROM public.test
|
|
WHERE id = %(id)s;"""
|
|
|
|
POST_SQL = """\
|
|
INSERT INTO public.test (id, created_at, value)
|
|
VALUES (%(id)s, CURRENT_TIMESTAMP, %(value)s)
|
|
RETURNING id, created_at, value;"""
|
|
|
|
UPDATE_COUNT_SQL = """\
|
|
UPDATE public.query_count
|
|
SET count = count + 1,
|
|
last_updated_at = CURRENT_TIMESTAMP
|
|
WHERE key = 'test'
|
|
RETURNING last_updated_at, count;"""
|
|
|
|
async def get(self, test_id):
|
|
result = await self.postgres_execute(self.GET_SQL, {'id': test_id})
|
|
if not result.row_count:
|
|
raise web.HTTPError(404, 'Not Found')
|
|
await self.finish(self.cast_data(result.row))
|
|
|
|
async def post(self):
|
|
body = json.loads(self.request.body.decode('utf-8'))
|
|
async with self.postgres_transaction() as postgres:
|
|
|
|
# This should roll back on the second call to this endopoint
|
|
self.application.first_txn = await postgres.execute(
|
|
self.POST_SQL, {'id': str(uuid.uuid4()),
|
|
'value': str(uuid.uuid4())})
|
|
|
|
# This should roll back on the second call to this endopoint
|
|
count = await postgres.execute(self.UPDATE_COUNT_SQL)
|
|
|
|
# This will trigger an error on the second call to this endpoint
|
|
user = await postgres.execute(self.POST_SQL, body)
|
|
|
|
await self.finish({
|
|
'count': self.cast_data(count.row),
|
|
'user': self.cast_data(user.row)})
|
|
|
|
|
|
class TimeoutErrorRequestHandler(RequestHandler):
|
|
|
|
GET_SQL = 'SELECT 1;'
|
|
|
|
async def get(self):
|
|
await self.postgres_execute(self.GET_SQL)
|
|
raise web.HTTPError(500, 'This should have failed')
|
|
|
|
def on_postgres_error(self,
|
|
metric_name: str,
|
|
exc: Exception) -> typing.Optional[Exception]:
|
|
"""Override for different error handling behaviors
|
|
|
|
Return an exception if you would like for it to be raised, or swallow
|
|
it here.
|
|
|
|
"""
|
|
if isinstance(exc, asyncio.TimeoutError):
|
|
raise web.HTTPError(418)
|
|
return exc
|
|
|
|
|
|
class UnhandledExceptionRequestHandler(RequestHandler):
|
|
|
|
GET_SQL = 'SELECT 100 / 0;'
|
|
|
|
async def get(self):
|
|
try:
|
|
await self.postgres_execute(self.GET_SQL)
|
|
except psycopg2.DataError:
|
|
raise web.HTTPError(422)
|
|
raise web.HTTPError(418, 'This should have failed')
|
|
|
|
def on_postgres_error(self,
|
|
metric_name: str,
|
|
exc: Exception) -> typing.Optional[Exception]:
|
|
"""Override for different error handling behaviors
|
|
|
|
Return an exception if you would like for it to be raised, or swallow
|
|
it here.
|
|
|
|
"""
|
|
return exc
|
|
|
|
|
|
class DontRaiseRequestHandler(UnhandledExceptionRequestHandler):
|
|
|
|
def on_postgres_error(self,
|
|
metric_name: str,
|
|
exc: Exception) -> typing.Optional[Exception]:
|
|
return None
|
|
|
|
|
|
class Application(sprockets_postgres.ApplicationMixin,
|
|
app.Application):
|
|
|
|
def __init__(self, *args, **kwargs):
|
|
super().__init__(*args, **kwargs)
|
|
self.influxdb = mock.Mock()
|
|
self.record_timing = mock.Mock()
|
|
self.first_txn: typing.Optional[sprockets_postgres.QueryResult] = None
|
|
|
|
|
|
class TestCase(testing.SprocketsHttpTestCase):
|
|
|
|
def setUp(self):
|
|
super().setUp()
|
|
loop = asyncio.get_event_loop()
|
|
while self.app._postgres_connected is None:
|
|
loop.run_until_complete(asyncio.sleep(0.1))
|
|
loop.run_until_complete(self.app._postgres_connected.wait())
|
|
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
with open('build/test-environment') as f:
|
|
for line in f:
|
|
if line.startswith('export '):
|
|
line = line[7:]
|
|
name, _, value = line.strip().partition('=')
|
|
os.environ[name] = value
|
|
|
|
def tearDown(self):
|
|
if self.app._postgres_pool is not None:
|
|
self.app._postgres_pool.terminate()
|
|
super().tearDown()
|
|
|
|
def get_app(self):
|
|
self.app = Application(handlers=[
|
|
web.url('/callproc', CallprocRequestHandler),
|
|
web.url('/count', CountRequestHandler),
|
|
web.url('/dont-raise', DontRaiseRequestHandler),
|
|
web.url('/error', ErrorRequestHandler),
|
|
web.url('/error-passthrough', ErrorPassthroughRequestHandler),
|
|
web.url('/execute', ExecuteRequestHandler),
|
|
web.url('/influxdb', InfluxDBRequestHandler),
|
|
web.url('/metrics-mixin', MetricsMixinRequestHandler),
|
|
web.url('/multi-row', MultiRowRequestHandler),
|
|
web.url('/no-error', NoErrorRequestHandler),
|
|
web.url('/no-mixin', NoMixinRequestHandler),
|
|
web.url('/no-row', NoRowRequestHandler),
|
|
web.url('/pdexecute', ProblemDetailsExecuteRequestHandler),
|
|
web.url('/row-count-no-rows', RowCountNoRowsRequestHandler),
|
|
web.url('/status', sprockets_postgres.StatusRequestHandler),
|
|
web.url('/timeout-error', TimeoutErrorRequestHandler),
|
|
web.url('/transaction', TransactionRequestHandler),
|
|
web.url('/transaction/(?P<test_id>.*)', TransactionRequestHandler),
|
|
web.url('/unhandled-exception', UnhandledExceptionRequestHandler)
|
|
])
|
|
return self.app
|
|
|
|
|
|
class PostgresStatusTestCase(asynctest.TestCase):
|
|
|
|
async def test_postgres_status_before_first_connection(self):
|
|
app = Application()
|
|
status = await app.postgres_status()
|
|
self.assertEqual(
|
|
status,
|
|
{'available': False,
|
|
'pool_size': 0,
|
|
'pool_free': 0})
|
|
|
|
|
|
class ReconnectionTestCast(TestCase):
|
|
|
|
@ttesting.gen_test
|
|
async def test_postgres_reconnect(self):
|
|
response = await self.http_client.fetch(self.get_url('/callproc'))
|
|
self.assertEqual(response.code, 200)
|
|
self.assertIsInstance(
|
|
uuid.UUID(json.loads(response.body)['value']), uuid.UUID)
|
|
|
|
# Force close all open connections for tests
|
|
conn = await aiopg.connect(os.environ['POSTGRES_URL'].split('?')[0])
|
|
cursor = await conn.cursor()
|
|
await cursor.execute(
|
|
'SELECT pg_terminate_backend(pid)'
|
|
' FROM pg_stat_activity'
|
|
" WHERE application_name = 'sprockets_postgres'")
|
|
await cursor.fetchall()
|
|
await asyncio.sleep(1)
|
|
response = await self.http_client.fetch(
|
|
self.get_url('/callproc'), raise_error=False)
|
|
self.assertEqual(response.code, 200)
|
|
conn.close()
|
|
|
|
|
|
class RequestHandlerMixinTestCase(TestCase):
|
|
|
|
def test_postgres_connected(self):
|
|
response = self.fetch('/status')
|
|
data = json.loads(response.body)
|
|
self.assertEqual(data['status'], 'ok')
|
|
self.assertTrue(self.app.postgres_is_connected)
|
|
|
|
def test_postgres_status(self):
|
|
response = self.fetch('/status')
|
|
data = json.loads(response.body)
|
|
self.assertEqual(data['status'], 'ok')
|
|
self.assertGreaterEqual(data['postgres']['pool_size'], 1)
|
|
self.assertGreaterEqual(data['postgres']['pool_free'], 1)
|
|
|
|
@mock.patch('aiopg.pool.Pool.acquire')
|
|
def test_postgres_status_connect_error(self, acquire):
|
|
acquire.side_effect = asyncio.TimeoutError()
|
|
response = self.fetch('/status')
|
|
self.assertEqual(response.code, 503)
|
|
data = json.loads(response.body)
|
|
self.assertEqual(data['status'], 'unavailable')
|
|
|
|
def test_postgres_status_not_connected(self):
|
|
self.app._postgres_connected.clear()
|
|
response = self.fetch('/status')
|
|
self.assertEqual(response.code, 503)
|
|
data = json.loads(response.body)
|
|
self.assertEqual(data['status'], 'unavailable')
|
|
|
|
@mock.patch('aiopg.Cursor.execute')
|
|
def test_postgres_status_error(self, execute):
|
|
execute.side_effect = asyncio.TimeoutError()
|
|
response = self.fetch('/status')
|
|
data = json.loads(response.body)
|
|
self.assertEqual(data['status'], 'unavailable')
|
|
|
|
def test_postgres_callproc(self):
|
|
response = self.fetch('/callproc')
|
|
self.assertEqual(response.code, 200)
|
|
self.assertIsInstance(
|
|
uuid.UUID(json.loads(response.body)['value']), uuid.UUID)
|
|
|
|
@mock.patch('aiopg.Cursor.execute')
|
|
def test_postgres_error(self, execute):
|
|
execute.side_effect = asyncio.TimeoutError
|
|
response = self.fetch('/error')
|
|
self.assertEqual(response.code, 500)
|
|
self.assertIn(b'Internal Server Error', response.body)
|
|
|
|
@mock.patch('aiopg.pool.Pool.acquire')
|
|
def test_postgres_error_on_connect(self, acquire):
|
|
acquire.side_effect = asyncio.TimeoutError
|
|
response = self.fetch('/error')
|
|
self.assertEqual(response.code, 500)
|
|
self.assertIn(b'Internal Server Error', response.body)
|
|
|
|
def test_postgres_error_passthrough(self):
|
|
response = self.fetch('/error-passthrough')
|
|
self.assertEqual(response.code, 204)
|
|
|
|
def test_postgres_execute(self):
|
|
expectation = str(uuid.uuid4())
|
|
response = self.fetch('/execute?value={}'.format(expectation))
|
|
self.assertEqual(response.code, 200)
|
|
self.assertEqual(json.loads(response.body)['value'], expectation)
|
|
|
|
def test_postgres_execute_with_timeout(self):
|
|
expectation = str(uuid.uuid4())
|
|
response = self.fetch(
|
|
'/execute?value={}&timeout=5'.format(expectation))
|
|
self.assertEqual(response.code, 200)
|
|
self.assertEqual(json.loads(response.body)['value'], expectation)
|
|
|
|
def test_postgres_influxdb(self):
|
|
expectation = str(uuid.uuid4())
|
|
response = self.fetch(
|
|
'/influxdb?value={}'.format(expectation))
|
|
self.assertEqual(response.code, 200)
|
|
self.assertEqual(json.loads(response.body)['value'], expectation)
|
|
self.app.influxdb.set_field.assert_called_once()
|
|
|
|
def test_postgres_metrics_mixin(self):
|
|
expectation = str(uuid.uuid4())
|
|
response = self.fetch(
|
|
'/metrics-mixin?value={}'.format(expectation))
|
|
self.assertEqual(response.code, 200)
|
|
self.assertEqual(json.loads(response.body)['value'], expectation)
|
|
self.app.record_timing.assert_called_once()
|
|
duration, metric_name = self.app.record_timing.call_args[0]
|
|
self.assertGreater(duration, 0.0)
|
|
self.assertEqual('', metric_name)
|
|
|
|
def test_postgres_multirow_get(self):
|
|
response = self.fetch('/multi-row')
|
|
self.assertEqual(response.code, 200)
|
|
body = json.loads(response.body)
|
|
self.assertEqual(body['count'], 5)
|
|
self.assertIsInstance(body['rows'], list)
|
|
|
|
def test_postgres_multirow_no_data(self):
|
|
for value in [True, False]:
|
|
response = self.fetch(
|
|
'/multi-row', method='POST', body=json.dumps({'value': value}))
|
|
self.assertEqual(response.code, 200)
|
|
body = json.loads(response.body)
|
|
self.assertEqual(body['count'], 5)
|
|
self.assertListEqual(body['rows'], [])
|
|
|
|
def test_postgres_norow(self):
|
|
response = self.fetch('/no-row')
|
|
self.assertEqual(response.code, 200)
|
|
body = json.loads(response.body)
|
|
self.assertEqual(body['count'], 0)
|
|
self.assertListEqual(body['rows'], [])
|
|
|
|
def test_postgres_cursor_operational_error_reconnects(self):
|
|
original = aiopg.connection.Connection.cursor
|
|
|
|
@contextlib.asynccontextmanager
|
|
async def mock_cursor(self, name=None, cursor_factory=None,
|
|
scrollable=None, withhold=False, timeout=None):
|
|
global test_postgres_cursor_oer_invocation
|
|
|
|
test_postgres_cursor_oer_invocation += 1
|
|
if test_postgres_cursor_oer_invocation == 1:
|
|
raise psycopg2.OperationalError()
|
|
async with original(self, name, cursor_factory, scrollable,
|
|
withhold, timeout) as value:
|
|
yield value
|
|
|
|
aiopg.connection.Connection.cursor = mock_cursor
|
|
|
|
with mock.patch.object(self.app, '_postgres_connect') as connect:
|
|
response = self.fetch('/execute?value=1')
|
|
self.assertEqual(response.code, 200)
|
|
self.assertEqual(json.loads(response.body)['value'], '1')
|
|
connect.assert_called_once()
|
|
|
|
aiopg.connection.Connection.cursor = original
|
|
|
|
@mock.patch('aiopg.connection.Connection.cursor')
|
|
def test_postgres_cursor_raises_on_failed_reconnect(self, cursor):
|
|
cursor.side_effect = psycopg2.OperationalError()
|
|
with mock.patch.object(self.app, '_postgres_connect') as connect:
|
|
connect.return_value = False
|
|
response = self.fetch('/execute?value=1')
|
|
self.assertEqual(response.code, 503)
|
|
connect.assert_called_once()
|
|
|
|
@ttesting.gen_test()
|
|
@mock.patch('aiopg.connection.Connection.cursor')
|
|
async def test_postgres_cursor_failure_concurrency(self, cursor):
|
|
cursor.side_effect = psycopg2.OperationalError()
|
|
|
|
def on_error(*args):
|
|
return RuntimeError
|
|
|
|
async def invoke_cursor():
|
|
async with self.app.postgres_connector(on_error) as connector:
|
|
await connector.execute('SELECT 1')
|
|
|
|
with self.assertRaises(RuntimeError):
|
|
await asyncio.gather(invoke_cursor(), invoke_cursor())
|
|
|
|
def test_row_count_no_rows(self):
|
|
response = self.fetch('/row-count-no-rows')
|
|
self.assertEqual(response.code, 200)
|
|
data = json.loads(response.body)
|
|
self.assertEqual(data['count'], 5)
|
|
|
|
@mock.patch('aiopg.Cursor.execute')
|
|
def test_timeout_error_when_overriding_on_postgres_error(self, execute):
|
|
execute.side_effect = asyncio.TimeoutError
|
|
response = self.fetch('/timeout-error')
|
|
self.assertEqual(response.code, 418)
|
|
|
|
def test_unhandled_exception_in_on_postgres_error(self):
|
|
response = self.fetch('/unhandled-exception')
|
|
self.assertEqual(response.code, 422)
|
|
|
|
@mock.patch('aiopg.Cursor.execute')
|
|
def test_postgres_execute_timeout_error(self, execute):
|
|
execute.side_effect = asyncio.TimeoutError()
|
|
response = self.fetch('/pdexecute?value=1')
|
|
self.assertEqual(response.code, 500)
|
|
problem = json.loads(response.body)
|
|
self.assertEqual(problem['title'], 'Query Timeout')
|
|
|
|
@mock.patch('aiopg.Cursor.execute')
|
|
def test_postgres_execute_unique_violation(self, execute):
|
|
execute.side_effect = errors.UniqueViolation()
|
|
response = self.fetch('/pdexecute?value=1')
|
|
self.assertEqual(response.code, 409)
|
|
problem = json.loads(response.body)
|
|
self.assertEqual(problem['title'], 'Unique Violation')
|
|
|
|
@mock.patch('aiopg.Cursor.execute')
|
|
def test_postgres_execute_error(self, execute):
|
|
execute.side_effect = psycopg2.Error()
|
|
response = self.fetch('/pdexecute?value=1')
|
|
self.assertEqual(response.code, 500)
|
|
problem = json.loads(response.body)
|
|
self.assertEqual(problem['title'], 'Database Error')
|
|
|
|
@mock.patch('aiopg.Cursor.fetchone')
|
|
def test_postgres_programming_error(self, fetchone):
|
|
fetchone.side_effect = psycopg2.ProgrammingError()
|
|
response = self.fetch('/pdexecute?value=1')
|
|
self.assertEqual(response.code, 200)
|
|
self.assertIsNone(json.loads(response.body)['value'])
|
|
|
|
|
|
class HTTPErrorTestCase(TestCase):
|
|
|
|
def setUp(self):
|
|
super().setUp()
|
|
self._problemdetails = sprockets_postgres.problemdetails
|
|
sprockets_postgres.problemdetails = None
|
|
|
|
def tearDown(self):
|
|
sprockets_postgres.problemdetails = self._problemdetails
|
|
super().tearDown()
|
|
|
|
@mock.patch('aiopg.Cursor.execute')
|
|
def test_postgres_execute_timeout_error(self, execute):
|
|
execute.side_effect = asyncio.TimeoutError()
|
|
response = self.fetch('/execute?value=1')
|
|
self.assertEqual(response.code, 500)
|
|
self.assertIn(b'Query Timeout', response.body)
|
|
|
|
@mock.patch('aiopg.Cursor.execute')
|
|
def test_postgres_execute_unique_violation(self, execute):
|
|
execute.side_effect = errors.UniqueViolation()
|
|
response = self.fetch('/execute?value=1')
|
|
self.assertEqual(response.code, 409)
|
|
self.assertIn(b'Unique Violation', response.body)
|
|
|
|
@mock.patch('aiopg.Cursor.execute')
|
|
def test_postgres_execute_error(self, execute):
|
|
execute.side_effect = psycopg2.Error()
|
|
response = self.fetch('/execute?value=1')
|
|
self.assertEqual(response.code, 500)
|
|
self.assertIn(b'Database Error', response.body)
|
|
|
|
@mock.patch('aiopg.Cursor.fetchone')
|
|
def test_postgres_programming_error(self, fetchone):
|
|
fetchone.side_effect = psycopg2.ProgrammingError()
|
|
response = self.fetch('/execute?value=1')
|
|
self.assertEqual(response.code, 200)
|
|
self.assertIsNone(json.loads(response.body)['value'])
|
|
|
|
@mock.patch('aiopg.connection.Connection.cursor')
|
|
def test_postgres_cursor_raises(self, cursor):
|
|
cursor.side_effect = asyncio.TimeoutError()
|
|
response = self.fetch('/execute?value=1')
|
|
self.assertEqual(response.code, 503)
|
|
|
|
def test_on_error_no_exception_branch(self):
|
|
response = self.fetch('/dont-raise')
|
|
self.assertEqual(response.code, 418)
|
|
|
|
|
|
class NoMixinTestCase(TestCase):
|
|
|
|
@mock.patch('aiopg.Cursor.execute')
|
|
def test_postgres_cursor_raises(self, execute):
|
|
execute.side_effect = psycopg2.ProgrammingError()
|
|
response = self.fetch('/no-mixin')
|
|
self.assertEqual(response.code, 500)
|
|
|
|
@mock.patch('aiopg.pool.Pool.acquire')
|
|
def test_postgres_status_connect_error(self, acquire):
|
|
acquire.side_effect = asyncio.TimeoutError()
|
|
response = self.fetch('/no-mixin')
|
|
self.assertEqual(response.code, 503)
|
|
|
|
|
|
class TransactionTestCase(TestCase):
|
|
|
|
def test_transactions(self):
|
|
test_body = {
|
|
'id': str(uuid.uuid4()),
|
|
'value': str(uuid.uuid4())
|
|
}
|
|
response = self.fetch(
|
|
'/transaction', method='POST', body=json.dumps(test_body))
|
|
self.assertEqual(response.code, 200)
|
|
record = json.loads(response.body.decode('utf-8'))
|
|
self.assertEqual(record['user']['id'], test_body['id'])
|
|
self.assertEqual(record['user']['value'], test_body['value'])
|
|
count = record['count']['count']
|
|
last_updated = record['count']['last_updated_at']
|
|
|
|
response = self.fetch(
|
|
'/transaction', method='POST', body=json.dumps(test_body))
|
|
self.assertEqual(response.code, 409)
|
|
|
|
response = self.fetch(
|
|
'/transaction/{}'.format(self.app.first_txn.row['id']))
|
|
self.assertEqual(response.code, 404)
|
|
|
|
response = self.fetch('/count')
|
|
self.assertEqual(response.code, 200)
|
|
record = json.loads(response.body.decode('utf-8'))
|
|
self.assertEqual(record['count'], count)
|
|
self.assertEqual(record['last_updated_at'], last_updated)
|
|
|
|
|
|
class MissingURLTestCase(unittest.TestCase):
|
|
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
with open('build/test-environment') as f:
|
|
for line in f:
|
|
if line.startswith('export '):
|
|
line = line[7:]
|
|
name, _, value = line.strip().partition('=')
|
|
if name != 'POSTGRES_URL':
|
|
os.environ[name] = value
|
|
if 'POSTGRES_URL' in os.environ:
|
|
del os.environ['POSTGRES_URL']
|
|
|
|
def test_that_stop_is_invoked(self):
|
|
io_loop = ioloop.IOLoop.current()
|
|
obj = Application()
|
|
obj.stop = mock.Mock(wraps=obj.stop)
|
|
obj.start(io_loop)
|
|
io_loop.start()
|
|
obj.stop.assert_called_once()
|
|
|
|
|
|
class ObscurePasswordUrlTestCase(unittest.TestCase):
|
|
|
|
def test_passwords_obscured(self):
|
|
for url, expected in {
|
|
'postgresql://server:5432/database':
|
|
'postgresql://server:5432/database',
|
|
'postgresql://username:password@server:5432/database':
|
|
'postgresql://username:*****@server:5432/database',
|
|
'postgresql://username@server/database':
|
|
'postgresql://username@server/database'
|
|
}.items():
|
|
result = \
|
|
sprockets_postgres.ApplicationMixin._obscure_url_password(url)
|
|
self.assertEqual(result, expected)
|
|
|
|
|
|
SRV = collections.namedtuple(
|
|
'SRV', ['host', 'port', 'priority', 'weight', 'ttl'])
|
|
|
|
|
|
class SRVTestCase(asynctest.TestCase):
|
|
|
|
async def test_srv_result(self):
|
|
obj = Application()
|
|
result = await obj._resolve_srv('_xmpp-server._tcp.google.com')
|
|
self.assertIsInstance(result[0], pycares.ares_query_srv_result)
|
|
|
|
async def test_srv_error(self):
|
|
obj = Application()
|
|
with self.assertRaises(RuntimeError):
|
|
await obj._resolve_srv('_postgresql._tcp.foo')
|
|
|
|
@mock.patch('sprockets.http.app.Application.stop')
|
|
@mock.patch('sprockets_postgres.LOGGER.critical')
|
|
async def test_aws_srv_parsing(self, critical, stop):
|
|
obj = Application()
|
|
loop = ioloop.IOLoop.current()
|
|
with mock.patch.object(obj, '_resolve_srv') as resolve_srv:
|
|
resolve_srv.return_value = []
|
|
obj._postgres_settings['url'] = 'aws+srv://foo@bar/baz'
|
|
await obj._postgres_on_start(obj, loop)
|
|
stop.assert_called_once_with(loop)
|
|
critical.assert_any_call('No SRV records found')
|
|
|
|
@mock.patch('sprockets.http.app.Application.stop')
|
|
@mock.patch('sprockets_postgres.LOGGER.critical')
|
|
async def test_postgres_srv_parsing(self, critical, stop):
|
|
obj = Application()
|
|
loop = ioloop.IOLoop.current()
|
|
with mock.patch.object(obj, '_resolve_srv') as resolve_srv:
|
|
resolve_srv.return_value = []
|
|
obj._postgres_settings['url'] = 'postgresql+srv://foo@bar/baz'
|
|
await obj._postgres_on_start(obj, loop)
|
|
stop.assert_called_once_with(loop)
|
|
critical.assert_any_call('No SRV records found')
|
|
|
|
@mock.patch('sprockets.http.app.Application.stop')
|
|
@mock.patch('sprockets_postgres.LOGGER.critical')
|
|
async def test_unsupported_srv_uri(self, critical, stop):
|
|
obj = Application()
|
|
loop = ioloop.IOLoop.current()
|
|
obj._postgres_settings['url'] = 'postgres+srv://foo@bar/baz'
|
|
await obj._postgres_on_start(obj, loop)
|
|
stop.assert_called_once_with(loop)
|
|
critical.assert_any_call(
|
|
'Unsupported URI Scheme: postgres+srv')
|
|
|
|
@mock.patch('aiodns.DNSResolver.query')
|
|
async def test_aws_url_from_srv_variation_1(self, query):
|
|
obj = Application()
|
|
parsed = parse.urlparse('aws+srv://foo@bar.baz/qux')
|
|
future = asyncio.Future()
|
|
future.set_result([
|
|
SRV('foo2', 5432, 2, 0, 32),
|
|
SRV('foo1', 5432, 1, 1, 32),
|
|
SRV('foo3', 6432, 1, 0, 32)
|
|
])
|
|
query.return_value = future
|
|
url = await obj._postgres_url_from_srv(parsed)
|
|
query.assert_called_once_with('bar.baz', 'SRV')
|
|
self.assertEqual(
|
|
url, 'postgresql://foo@foo1:5432,foo3:6432,foo2:5432/qux')
|
|
|
|
@mock.patch('aiodns.DNSResolver.query')
|
|
async def test_postgresql_url_from_srv_variation_1(self, query):
|
|
obj = Application()
|
|
parsed = parse.urlparse('postgresql+srv://foo@bar.baz/qux')
|
|
future = asyncio.Future()
|
|
future.set_result([
|
|
SRV('foo2', 5432, 2, 0, 32),
|
|
SRV('foo1', 5432, 1, 0, 32)
|
|
])
|
|
query.return_value = future
|
|
url = await obj._postgres_url_from_srv(parsed)
|
|
query.assert_called_once_with('_bar._postgresql.baz', 'SRV')
|
|
self.assertEqual(url, 'postgresql://foo@foo1:5432,foo2:5432/qux')
|
|
|
|
@mock.patch('aiodns.DNSResolver.query')
|
|
async def test_postgresql_url_from_srv_variation_2(self, query):
|
|
obj = Application()
|
|
parsed = parse.urlparse('postgresql+srv://foo:bar@baz.qux/corgie')
|
|
future = asyncio.Future()
|
|
future.set_result([
|
|
SRV('foo2', 5432, 1, 0, 32),
|
|
SRV('foo1', 5432, 2, 0, 32)
|
|
])
|
|
query.return_value = future
|
|
url = await obj._postgres_url_from_srv(parsed)
|
|
query.assert_called_once_with('_baz._postgresql.qux', 'SRV')
|
|
self.assertEqual(
|
|
url, 'postgresql://foo:bar@foo2:5432,foo1:5432/corgie')
|
|
|
|
@mock.patch('aiodns.DNSResolver.query')
|
|
async def test_postgresql_url_from_srv_variation_3(self, query):
|
|
obj = Application()
|
|
parsed = parse.urlparse('postgresql+srv://foo.bar/baz')
|
|
future = asyncio.Future()
|
|
future.set_result([
|
|
SRV('foo2', 5432, 2, 0, 32),
|
|
SRV('foo1', 5432, 1, 0, 32),
|
|
SRV('foo3', 5432, 1, 10, 32),
|
|
])
|
|
query.return_value = future
|
|
url = await obj._postgres_url_from_srv(parsed)
|
|
query.assert_called_once_with('_foo._postgresql.bar', 'SRV')
|
|
self.assertEqual(url, 'postgresql://foo3:5432,foo1:5432,foo2:5432/baz')
|
|
|
|
@mock.patch('aiodns.DNSResolver.query')
|
|
async def test_resolve_srv_sorted(self, query):
|
|
obj = Application()
|
|
result = [
|
|
SRV('foo2', 5432, 2, 0, 32),
|
|
SRV('foo1', 5432, 1, 1, 32),
|
|
SRV('foo3', 6432, 1, 0, 32)
|
|
]
|
|
future = asyncio.Future()
|
|
future.set_result(result)
|
|
query.return_value = future
|
|
records = await obj._resolve_srv('foo')
|
|
self.assertListEqual(records, [result[1], result[2], result[0]])
|