<commit_before>""" Tests for table scans """
import six
from datetime import datetime, date, timedelta

import inspect
from decimal import Decimal

from .test_queries import User
from flywheel import Model, Field
from flywheel.tests import DynamoSystemTest


class TestScan(DynamoSystemTest):

    """ Tests for table scans """
    models = [User]

    def test_all(self):
        """ Scan returns all elements in a table """
        u = User(id='a', name='Adam')
        u2 = User(id='b', name='Billy')
        self.engine.save([u, u2])

        results = self.engine.scan(User).all()
        self.assertEquals(len(results), 2)
        self.assertTrue(u in results)
        self.assertTrue(u2 in results)

    def test_gen(self):
        """ Scan can get a generator for items """
        u = User(id='a', name='Adam')
        u2 = User(id='b', name='Billy')
        self.engine.save([u, u2])

        results = self.engine.scan(User).gen()
        self.assertTrue(inspect.isgenerator(results))
        results = list(results)
        self.assertEquals(len(results), 2)
        self.assertTrue(u in results)
        self.assertTrue(u2 in results)

    def test_count(self):
        """ Can return a count instead of the models """
        u = User(id='a', name='Adam')
        u2 = User(id='b', name='Billy')
        self.engine.save([u, u2])
        count = self.engine.scan(User).count()
        self.assertEqual(count, 2)
        count = self.engine.scan(User).filter(id='a').count()
        self.assertEqual(count, 1)

    def test_iter(self):
        """ Scan can iterate over items """
        u = User(id='a', name='Adam')
        u2 = User(id='b', name='Billy')
        self.engine.save([u, u2])

        users = [u, u2]
        for item in self.engine.scan(User):
            self.assertTrue(item in users)
            users.remove(item)

    def test_limit(self):
        """ Scan can have a limit """
        u = User(id='a', name='Adam')
        u2 = User(id='a', name='Aaron')
        self.engine.save([u, u2])

        results = self.engine.scan(User).limit(1).all()
        self.assertEquals(len(results), 1)

    def test_delete(self):
        """ Scan can selectively delete items """
        u = User(id='a', name='Adam')
        u2 = User(id='b', name='Billy')
        self.engine.save([u, u2])

        count = self.engine.scan(User).filter(User.name == 'Adam').delete()
        self.assertEquals(count, 1)
        results = self.engine.scan(User).all()
        self.assertEquals(results, [u2])

    def test_filter_eq(self):
        """ Scan can filter eq """
        u = User(id='a', name='Adam')
        u2 = User(id='b', name='Billy')
        self.engine.save([u, u2])

        results = self.engine.scan(User).filter(User.name == 'Adam').all()
        self.assertEquals(results, [u])

    def test_filter_lt(self):
        """ Scan can filter lt """
        u = User(id='a', name='Adam')
        u2 = User(id='a', name='Aaron')
        self.engine.save([u, u2])

        results = self.engine.scan(User).filter(User.name < 'Adam').all()
        self.assertEquals(results, [u2])

    def test_filter_lte(self):
        """ Scan can filter lte """
        u = User(id='a', name='Aaron')
        u2 = User(id='a', name='Adam')
        u3 = User(id='a', name='Alison')
        self.engine.save([u, u2, u3])

        results = self.engine.scan(User).filter(User.name <= u2.name).all()
        self.assertEquals(len(results), 2)
        self.assertTrue(u in results)
        self.assertTrue(u2 in results)

    def test_filter_gt(self):
        """ Scan can filter gt """
        u = User(id='a', name='Adam')
        u2 = User(id='a', name='Aaron')
        self.engine.save([u, u2])

        results = self.engine.scan(User).filter(User.name > 'Aaron').all()
        self.assertEquals(results, [u])

    def test_filter_gte(self):
        """ Scan can filter gte """
        u = User(id='a', name='Aaron')
        u2 = User(id='a', name='Adam')
        u3 = User(id='a', name='Alison')
        self.engine.save([u, u2, u3])

        results = self.engine.scan(User).filter(User.name >= u2.name).all()
        self.assertEquals(len(results), 2)
        self.assertTrue(u2 in results)
        self.assertTrue(u3 in results)

    def test_filter_beginswith(self):
        """ Scan can filter beginswith """
        u = User(id='a', name='Adam')
        u2 = User(id='a', name='Aaron')
        self.engine.save([u, u2])

        results = self.engine.scan(User)\
            .filter(User.name.beginswith_('Ad')).all()
        self.assertEquals(results, [u])

    def test_filter_ne(self):
        """ Scan can filter ne """
        u = User(id='a', name='Adam')
        u2 = User(id='a', name='Aaron')
        self.engine.save([u, u2])

        results = self.engine.scan(User).filter(User.name != 'Adam').all()
        self.assertEquals(results, [u2])

    def test_filter_in(self):
        """ Scan can filter in """
        u = User(id='a', name='Adam')
        u2 = User(id='a', name='Aaron')
        self.engine.save([u, u2])
        names = set([u.name])

        results = self.engine.scan(User).filter(User.name.in_(names)).all()
        self.assertEquals(results, [u])

    def test_filter_contains(self):
        """ Scan can filter contains """
        u = User(id='a', name='Adam', str_set=set(['hi']))
        u2 = User(id='a', name='Aaron')
        self.engine.save([u, u2])

        results = self.engine.scan(User)\
            .filter(User.str_set.contains_('hi')).all()
        self.assertEquals(results, [u])

    def test_filter_or(self):
        """ Scan can join filter constraints with OR """
        u = User(id='a', name='Adam', foo='bar')
        u2 = User(id='b', name='Billy', bar='baz')
        u3 = User(id='c', name='Celine', foo='not', bar='this')
        self.engine.save([u, u2, u3])

        results = self.engine.scan(User).filter(foo='bar', bar='baz') \
            .all(filter_or=True)
        self.assertItemsEqual(results, [u, u2])


class TestScanFilterOverflow(DynamoSystemTest):

    """ Filter tests on overflow fields """
    models = [User]

    def test_filter_eq(self):
        """ Scan overflow field can filter eq """<commit_msg>Fixing another python 2.6 test<commit_after>""" Tests for table scans """
import six
from datetime import datetime, date, timedelta

import inspect
from decimal import Decimal

from .test_queries import User
from flywheel import Model, Field
from flywheel.tests import DynamoSystemTest


class TestScan(DynamoSystemTest):

    """ Tests for table scans """
    models = [User]

    def test_all(self):
        """ Scan returns all elements in a table """
        u = User(id='a', name='Adam')
        u2 = User(id='b', name='Billy')
        self.engine.save([u, u2])

        results = self.engine.scan(User).all()
        self.assertEquals(len(results), 2)
        self.assertTrue(u in results)
        self.assertTrue(u2 in results)

    def test_gen(self):
        """ Scan can get a generator for items """
        u = User(id='a', name='Adam')
        u2 = User(id='b', name='Billy')
        self.engine.save([u, u2])

        results = self.engine.scan(User).gen()
        self.assertTrue(inspect.isgenerator(results))
        results = list(results)
        self.assertEquals(len(results), 2)
        self.assertTrue(u in results)
        self.assertTrue(u2 in results)

    def test_count(self):
        """ Can return a count instead of the models """
        u = User(id='a', name='Adam')
        u2 = User(id='b', name='Billy')
        self.engine.save([u, u2])
        count = self.engine.scan(User).count()
        self.assertEqual(count, 2)
        count = self.engine.scan(User).filter(id='a').count()
        self.assertEqual(count, 1)

    def test_iter(self):
        """ Scan can iterate over items """
        u = User(id='a', name='Adam')
        u2 = User(id='b', name='Billy')
        self.engine.save([u, u2])

        users = [u, u2]
        for item in self.engine.scan(User):
            self.assertTrue(item in users)
            users.remove(item)

    def test_limit(self):
        """ Scan can have a limit """
        u = User(id='a', name='Adam')
        u2 = User(id='a', name='Aaron')
        self.engine.save([u, u2])

        results = self.engine.scan(User).limit(1).all()
        self.assertEquals(len(results), 1)

    def test_delete(self):
        """ Scan can selectively delete items """
        u = User(id='a', name='Adam')
        u2 = User(id='b', name='Billy')
        self.engine.save([u, u2])

        count = self.engine.scan(User).filter(User.name == 'Adam').delete()
        self.assertEquals(count, 1)
        results = self.engine.scan(User).all()
        self.assertEquals(results, [u2])

    def test_filter_eq(self):
        """ Scan can filter eq """
        u = User(id='a', name='Adam')
        u2 = User(id='b', name='Billy')
        self.engine.save([u, u2])

        results = self.engine.scan(User).filter(User.name == 'Adam').all()
        self.assertEquals(results, [u])

    def test_filter_lt(self):
        """ Scan can filter lt """
        u = User(id='a', name='Adam')
        u2 = User(id='a', name='Aaron')
        self.engine.save([u, u2])

        results = self.engine.scan(User).filter(User.name < 'Adam').all()
        self.assertEquals(results, [u2])

    def test_filter_lte(self):
        """ Scan can filter lte """
        u = User(id='a', name='Aaron')
        u2 = User(id='a', name='Adam')
        u3 = User(id='a', name='Alison')
        self.engine.save([u, u2, u3])

        results = self.engine.scan(User).filter(User.name <= u2.name).all()
        self.assertEquals(len(results), 2)
        self.assertTrue(u in results)
        self.assertTrue(u2 in results)

    def test_filter_gt(self):
        """ Scan can filter gt """
        u = User(id='a', name='Adam')
        u2 = User(id='a', name='Aaron')
        self.engine.save([u, u2])

        results = self.engine.scan(User).filter(User.name > 'Aaron').all()
        self.assertEquals(results, [u])

    def test_filter_gte(self):
        """ Scan can filter gte """
        u = User(id='a', name='Aaron')
        u2 = User(id='a', name='Adam')
        u3 = User(id='a', name='Alison')
        self.engine.save([u, u2, u3])

        results = self.engine.scan(User).filter(User.name >= u2.name).all()
        self.assertEquals(len(results), 2)
        self.assertTrue(u2 in results)
        self.assertTrue(u3 in results)

    def test_filter_beginswith(self):
        """ Scan can filter beginswith """
        u = User(id='a', name='Adam')
        u2 = User(id='a', name='Aaron')
        self.engine.save([u, u2])

        results = self.engine.scan(User)\
            .filter(User.name.beginswith_('Ad')).all()
        self.assertEquals(results, [u])

    def test_filter_ne(self):
        """ Scan can filter ne """
        u = User(id='a', name='Adam')
        u2 = User(id='a', name='Aaron')
        self.engine.save([u, u2])

        results = self.engine.scan(User).filter(User.name != 'Adam').all()
        self.assertEquals(results, [u2])

    def test_filter_in(self):
        """ Scan can filter in """
        u = User(id='a', name='Adam')
        u2 = User(id='a', name='Aaron')
        self.engine.save([u, u2])
        names = set([u.name])

        results = self.engine.scan(User).filter(User.name.in_(names)).all()
        self.assertEquals(results, [u])

    def test_filter_contains(self):
        """ Scan can filter contains """
        u = User(id='a', name='Adam', str_set=set(['hi']))
        u2 = User(id='a', name='Aaron')
        self.engine.save([u, u2])

        results = self.engine.scan(User)\
            .filter(User.str_set.contains_('hi')).all()
        self.assertEquals(results, [u])

    def test_filter_or(self):
        """ Scan can join filter constraints with OR """
        u = User(id='a', name='Adam', foo='bar')
        u2 = User(id='b', name='Billy', bar='baz')
        u3 = User(id='c', name='Celine', foo='not', bar='this')
        self.engine.save([u, u2, u3])

        results = self.engine.scan(User).filter(foo='bar', bar='baz') \
            .all(filter_or=True)
        self.assertEqual(len(results), 2)
        self.assertTrue(u in results)
        self.assertTrue(u2 in results)


class TestScanFilterOverflow(DynamoSystemTest):

    """ Filter tests on overflow fields """
    models = [User]

    def test_filter_eq(self):
        """ Scan overflow field can filter eq """
{'fraction_non_alphanumeric': 0.17147435897435898, 'fraction_numerical': 0.00969551282051282, 'mean_word_length': 2.4602162461879677, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7417675', 'n_tokens_mistral': 4131, 'n_tokens_neox': 3974, 'n_words': 877}