mirror of
https://github.com/scylladb/scylladb.git
synced 2026-04-28 20:27:03 +00:00
5bc7bb00e0
Paginator is an implementation detail and does not belong in the name, and thus the test is renamed to test_query_basic_restrictions. Message-Id: <849bc9d210d0faee4bb8479306654f2a59e18517.1558524028.git.sarna@scylladb.com>
86 lines
3.7 KiB
Python
86 lines
3.7 KiB
Python
# Tests for the Query operation
|
|
|
|
import random
|
|
import string
|
|
|
|
import pytest
|
|
from botocore.exceptions import ClientError
|
|
|
|
def set_of_frozen_elements(list_of_dicts):
|
|
return {frozenset(item.items()) for item in list_of_dicts}
|
|
|
|
# Test that scanning works fine with in-stock paginator
|
|
def test_query_basic_restrictions(dynamodb, filled_test_table):
|
|
test_table, items = filled_test_table
|
|
paginator = dynamodb.meta.client.get_paginator('query')
|
|
|
|
# EQ
|
|
got_items = []
|
|
for page in paginator.paginate(TableName=test_table.name, KeyConditions={
|
|
'p' : {'AttributeValueList': ['long'], 'ComparisonOperator': 'EQ'}
|
|
}):
|
|
got_items += page['Items']
|
|
print(got_items)
|
|
assert set_of_frozen_elements([item for item in items if item['p'] == 'long']) == set_of_frozen_elements(got_items)
|
|
|
|
# LT
|
|
got_items = []
|
|
for page in paginator.paginate(TableName=test_table.name, KeyConditions={
|
|
'p' : {'AttributeValueList': ['long'], 'ComparisonOperator': 'EQ'},
|
|
'c' : {'AttributeValueList': ['12'], 'ComparisonOperator': 'LT'}
|
|
}):
|
|
got_items += page['Items']
|
|
print(got_items)
|
|
assert set_of_frozen_elements([item for item in items if item['p'] == 'long' and item['c'] < '12']) == set_of_frozen_elements(got_items)
|
|
|
|
# LE
|
|
got_items = []
|
|
for page in paginator.paginate(TableName=test_table.name, KeyConditions={
|
|
'p' : {'AttributeValueList': ['long'], 'ComparisonOperator': 'EQ'},
|
|
'c' : {'AttributeValueList': ['14'], 'ComparisonOperator': 'LE'}
|
|
}):
|
|
got_items += page['Items']
|
|
print(got_items)
|
|
assert set_of_frozen_elements([item for item in items if item['p'] == 'long' and item['c'] <= '14']) == set_of_frozen_elements(got_items)
|
|
|
|
# GT
|
|
got_items = []
|
|
for page in paginator.paginate(TableName=test_table.name, KeyConditions={
|
|
'p' : {'AttributeValueList': ['long'], 'ComparisonOperator': 'EQ'},
|
|
'c' : {'AttributeValueList': ['15'], 'ComparisonOperator': 'GT'}
|
|
}):
|
|
got_items += page['Items']
|
|
print(got_items)
|
|
assert set_of_frozen_elements([item for item in items if item['p'] == 'long' and item['c'] > '15']) == set_of_frozen_elements(got_items)
|
|
|
|
# GE
|
|
got_items = []
|
|
for page in paginator.paginate(TableName=test_table.name, KeyConditions={
|
|
'p' : {'AttributeValueList': ['long'], 'ComparisonOperator': 'EQ'},
|
|
'c' : {'AttributeValueList': ['14'], 'ComparisonOperator': 'GE'}
|
|
}):
|
|
got_items += page['Items']
|
|
print(got_items)
|
|
assert set_of_frozen_elements([item for item in items if item['p'] == 'long' and item['c'] >= '14']) == set_of_frozen_elements(got_items)
|
|
|
|
# BETWEEN
|
|
got_items = []
|
|
for page in paginator.paginate(TableName=test_table.name, KeyConditions={
|
|
'p' : {'AttributeValueList': ['long'], 'ComparisonOperator': 'EQ'},
|
|
'c' : {'AttributeValueList': ['155', '164'], 'ComparisonOperator': 'BETWEEN'}
|
|
}):
|
|
got_items += page['Items']
|
|
print(got_items)
|
|
assert set_of_frozen_elements([item for item in items if item['p'] == 'long' and item['c'] >= '155' and item['c'] <= '164']) == set_of_frozen_elements(got_items)
|
|
|
|
# BEGINS_WITH
|
|
got_items = []
|
|
for page in paginator.paginate(TableName=test_table.name, KeyConditions={
|
|
'p' : {'AttributeValueList': ['long'], 'ComparisonOperator': 'EQ'},
|
|
'c' : {'AttributeValueList': ['11'], 'ComparisonOperator': 'BEGINS_WITH'}
|
|
}):
|
|
print([item for item in items if item['p'] == 'long' and item['c'].startswith('11')])
|
|
got_items += page['Items']
|
|
print(got_items)
|
|
assert set_of_frozen_elements([item for item in items if item['p'] == 'long' and item['c'].startswith('11')]) == set_of_frozen_elements(got_items)
|