genquery-limit-offset-totalrows-tests.py

class Fail(Exception): pass

# run as e.g.: test("/tempZone/home/rods/xyz");
def test(a, c, r):
    import genquery
    # First arg: name of an empty or non-existent collection where we can
    # create subcollections for testing.
    # Test queries will be on COLL_PARENT_NAME = test_coll
    test_coll = a[0]

    # Given a collection number, get the expected COLL_NAME value.
    coll_name = lambda n: '{}/coll{:04d}'.format(test_coll, n)

    # "interesting" row counts that are likely to trigger bugs.
    # used for the amount of DB entries, LIMIT, and OFFSET.
    interesting  = [0,1,2,100,254,255,256,257,258,510,511,512,513,514]

    # Sanity check (which assumes total_rows correctly returns non-zero).
    if genquery.Query(c, 'DATA_ID', "COLL_NAME = '{}'".format(test_coll)).total_rows()\
    or genquery.Query(c, 'DATA_ID', "COLL_NAME like '{}/%'".format(test_coll)).total_rows():
        raise Fail('collection <{}> not empty'.format(test_coll))

    # Remove test_coll recursively.
    def cleanup():
        try:    res = c.msiRmColl(test_coll, 'forceFlag=', 0)
        except: pass
    cleanup()

    # Run tests with all possible combinations of coll_count, limit and offset,
    # taken from the `interesting` list.

    # The amount of new colls to create in each step in `interesting`.
    increments = (lambda xs: [0] + [xs[i]-xs[i-1] for i in range(1, len(xs))])(interesting)

    # each amount of existing collections...
    for coll_count, new_colls in zip(interesting, increments):
        for i in range(new_colls):
            c.msiCollCreate('{}/coll{:04d}'.format(test_coll, coll_count-new_colls + i), 1, 0)

        # each limit...
        for limit in [None] + interesting: # Test with no LIMIT as well.
            print('EXIST/TOTALROWS  EXPECTED/GOT   OFFSET LIMIT')

            # each offset...
            for offset in interesting:
                # Test #1: Do we get the right indicated total row count,
                #          regardless of offset/limit?
                q = genquery.Query(c, 'COLL_ID', "COLL_PARENT_NAME = '{}'".format(test_coll),
                                    offset=offset, limit=limit)
                total_rows = q.total_rows()

                if total_rows != coll_count:
                    raise Fail('expected <{}> total_rows, got <{}>'.format(coll_count, total_rows))

                # Test #2: Do we get the right (and right amount of) results?
                q = genquery.Query(c, 'COLL_NAME', "COLL_PARENT_NAME = '{}'".format(test_coll),
                                   offset=offset, limit=limit)

                i = 0 # amount of fetched rows
                expected_results = max(0, coll_count - offset)
                if limit is not None:
                    expected_results = min(limit, expected_results)

                for x in q:
                    if x != coll_name(i + offset):
                        raise Fail('bad result, expected <{}>, got <{}>'.format(coll_name(i + offset), x))
                    i += 1
                    if i > expected_results:
                        break # should not happen.

                if i != expected_results:
                    raise Fail('expected <{}> results, got <{}>'.format(expected_results, i))

                print('{:5}/{:4}  {:13}/{:4}  {:6} {:5}'
                      .format(coll_count, total_rows, expected_results, i, offset, limit))

    cleanup()