def test()

in asfpy/sqlite.py [0:0]

• 53 lines of code
• 8 McCabe index (conditional complexity)

def test(dbname=':memory:'):
    testdb = db(dbname)
    cstatement = '''CREATE TABLE test (
                      foo   varchar unique,
                      bar   varchar,
                      baz   real
                      )'''

    # Create if not already here
    try:
        testdb.runc(cstatement)
    except sqlite3.OperationalError as e:  # Table exists
        assert str(e) == "table test already exists"

    # Insert (may fail if already tested)
    try:
        testdb.insert('test', {'foo': 'foo1234', 'bar': 'blorgh', 'baz': 5})
    except sqlite3.IntegrityError as e:
        assert str(e) == "UNIQUE constraint failed: test.foo"

    # This must fail
    try:
        testdb.insert('test', {'foo': 'foo1234', 'bar': 'blorgh', 'baz': 2})
    except sqlite3.IntegrityError as e:
        assert str(e) == "UNIQUE constraint failed: test.foo"

    # This must pass
    testdb.upsert('test', {'foo': 'foo1234', 'bar': 'blorgssh', 'baz': 8}, foo='foo1234')

    # This should fail with no target specified
    try:
        testdb.upsert('test', {'foo': 'foo1234', 'bar': 'blorgssh', 'baz': 8})
    except AsfpyDBError as e:
        assert str(e) == "UPSERTs must have at least one defined target value for locating where to upsert"

    # This should all pass
    testdb.update('test', {'foo': 'foo4321'}, foo='foo1234')
    obj = testdb.fetchone('test', foo='foo4321')
    assert type(obj) is dict and obj.get('foo') == 'foo4321'
    obj = testdb.fetch('test', limit=5, foo = 'foo4321')
    assert str(type(obj)) == "<class 'generator'>"
    assert next(obj).get('foo') == 'foo4321'
    obj = testdb.fetchone('test', foo='foo9999')
    assert obj is None
    testdb.delete('test', foo='foo4321')
    assert testdb.table_exists('test')
    assert not testdb.table_exists('test2')

    # Let's insert 1000 rows, and perform a repeated fetch.
    for i in range(1000):
        testdb.insert('test', {'foo': str(i), 'bar': str(i), 'baz': i})
    count = 0
    for row in testdb.fetch('test', limit=None):
        assert int(row['foo']) == count
        count += 1
    assert count == 1000

    # Change the arraysize, and run it again.
    testdb.cursor.arraysize = 97  # ensure last fetch is short
    count = 0
    for row in testdb.fetch('test', limit=None):
        assert int(row['foo']) == count
        count += 1
    assert count == 1000

    # One more run, with a limit. Leave the arraysize.
    count = 0
    for row in testdb.fetch('test', limit=30):
        assert int(row['foo']) == count
        count += 1
    assert count == 30