Cacher is the code snippet organizer for pro developers

# -*- coding: utf-8 -*-
import pytest
from fastdb import FastDB, Types, to_dict

MAX_ROWS = 2
MAX_PARTS = 3
SCHEMAS = dict(x=Types.TYPE_UINT, y=Types.TYPE_INT, z=Types.TYPE_FLOAT)
TABLE_NAME = 'default'

NON_FLUSHED_ROWS = [
    [dict(x=10, y=-1, z=-0.1)],
    [dict(x=10, y=-1, z=-0.1), dict(x=10, y=-2, z=-0.2)],
]

FLUSHED_ROWS = [
    [dict(x=10, y=-1, z=-0.1), dict(x=10, y=-2, z=-0.2), dict(x=15, y=-3, z=-0.5)],
    [dict(x=10, y=-1, z=-0.1), dict(x=10, y=-2, z=-0.2), dict(x=15, y=-3, z=-0.5),
     dict(x=30, y=-4, z=-0.3)],
]

CONDITION_SELECT_ROWS = [
    [dict(x=10, y=-1, z=-0.1), dict(x=10, y=-2, z=-0.2), dict(x=15, y=-3, z=-0.5),
     dict(x=30, y=-4, z=-0.3), dict(x=45, y=-5, z=-0.4)],
]

NO_TABLE_SPECIFIED_QUERY = [
    '',
    'select *',
    'select x,y,z',
]

TABLE_SPECIFIED_QUERY = [
    'select * from {}'.format(TABLE_NAME),
    'select x,y,z from {}'.format(TABLE_NAME),
]

CONDITION_QUERY = [
    'select x,y where z < -0.15 orderby y asc from {}'.format(TABLE_NAME),
    'select x,y where z < -0.15 from {} orderby y asc'.format(TABLE_NAME),
    'select x,y from {} where z < -0.15 orderby y asc'.format(TABLE_NAME),
    'select x,y from {} where z < -0.15 and 1 = 1 orderby y asc'.format(TABLE_NAME),
    'select x,y from {} where z < -0.15 or 1 = 0 orderby y asc'.format(TABLE_NAME),
]

CONDITION_COMPLEX_QUERY = [
    'select x, _y as y where _y == -2 from {select x, y as _y where z < -0.15 from %s}' % TABLE_NAME,
    'select x, _y as y from {select x, y as _y where z < -0.15 from %s} where _y == -2' % TABLE_NAME,
    'select x, y from {select x, y where z < -0.15 from %s} where y == -2' % TABLE_NAME,
    'select x, y from {select x, y where z < -0.15 and y == -2 from %s}' % TABLE_NAME,
]

def is_floats_almost_equal(list1, list2):
    return all([round(x-y, 5) == 0 for x,y in zip(list1, list2)])


@pytest.fixture
def empty_db():
    return FastDB()


@pytest.fixture
def empty_table_db():
    db = FastDB()
    db.create_table(TABLE_NAME, MAX_ROWS, MAX_PARTS, SCHEMAS)
    def fin():
        db.drop_table(TABLE_NAME)
    return db


@pytest.fixture(params=NON_FLUSHED_ROWS)
def non_flushed_db(request):
    db = FastDB()
    db.create_table(TABLE_NAME, MAX_ROWS, MAX_PARTS, SCHEMAS)
    
    rows = request.param
    for row in rows:
        db.append_row(TABLE_NAME, row)

    def fin():
        db.drop_table(TABLE_NAME)
    return db


@pytest.fixture(params=FLUSHED_ROWS)
def flushed_db(request):
    db = FastDB()
    db.create_table(TABLE_NAME, MAX_ROWS, MAX_PARTS, SCHEMAS)
    
    rows = request.param
    for row in rows:
        db.append_row(TABLE_NAME, row)

    def fin():
        db.drop_table(TABLE_NAME)
    return db


@pytest.fixture(params=CONDITION_SELECT_ROWS)
def condition_query_db(request):
    db = FastDB()
    db.create_table(TABLE_NAME, MAX_ROWS, MAX_PARTS, SCHEMAS)
    
    rows = request.param
    for row in rows:
        db.append_row(TABLE_NAME, row)

    def fin():
        db.drop_table(TABLE_NAME)
    return db


@pytest.fixture(params=NO_TABLE_SPECIFIED_QUERY)
def no_table_query_expr(request):
    return request.param


@pytest.fixture(params=TABLE_SPECIFIED_QUERY)
def has_table_query_expr(request):
    return request.param


@pytest.fixture(params=CONDITION_QUERY)
def condition_query_expr(request):
    return request.param


@pytest.fixture(params=CONDITION_COMPLEX_QUERY)
def condition_complex_query_expr(request):
    return request.param


class TestFastDB(object):
    def test_not_table_provided(self, empty_db, has_table_query_expr):
        with pytest.raises(RuntimeError):
            empty_db.query(has_table_query_expr)

    def test_not_table_specified(self, empty_table_db, no_table_query_expr):
        with pytest.raises(KeyError):
            empty_table_db.query(no_table_query_expr)

    def test_not_flushed_part(self, non_flushed_db, has_table_query_expr):
        with pytest.raises(RuntimeError):
            non_flushed_db.query(has_table_query_expr)

    def test_flushed_part(self, flushed_db, has_table_query_expr):
        rs = flushed_db.query(has_table_query_expr)
        r = to_dict(rs)
        assert sorted(['x', 'y', 'z']) == sorted(r['columns'])
        assert [10, 10] == r['data']['x']
        assert [-1, -2] == r['data']['y']
        assert is_floats_almost_equal([-0.1, -0.2], r['data']['z'])

    def test_to_pandas(self, flushed_db, has_table_query_expr):
        from fastdb import to_pandas
        rs = flushed_db.query(has_table_query_expr)
        r = to_pandas(flushed_db.query(has_table_query_expr))
        assert sorted(['x', 'y', 'z']) == sorted(r.keys())
        assert [10, 10] == r['x'].tolist()
        assert [-1, -2] == r['y'].tolist()
        assert is_floats_almost_equal([-0.1, -0.2], r['z'].tolist())

    def test_condition_select(self, condition_query_db, condition_query_expr):
        rs = condition_query_db.query(condition_query_expr)
        r = to_dict(rs)
        assert sorted(['x', 'y']) == sorted(r['columns'])
        assert [30, 15, 10] == r['data']['x']
        assert [-4, -3, -2] == r['data']['y']

    def test_condition_complex_select(self, condition_query_db, condition_complex_query_expr):
        rs = condition_query_db.query(condition_complex_query_expr)
        r = to_dict(rs)
        assert sorted(['x', 'y']) == sorted(r['columns'])
        assert [10] == r['data']['x']
        assert [-2] == r['data']['y']

import os
import pkg_resources
import sys
from setuptools import setup
from setuptools.command.test import test as TestCommand


os.environ['OPT'] = '-DNDEBUG -g -fwrapv -O2 -Wall'

install_requires = [
    'Cython',
    'pandas'
]

tests_require = [
    'pytest',
    'pytest-cov',
] + install_requires

ext_modules = []
try:
    from Cython.Build import cythonize
    ext_modules += cythonize('fastdb/*.pyx')
except ImportError:
    sys.exit("""Cython-generated files not found.
                Cython is required to compile FastDB from a development branch.
                Please install Cython.
                """)

try:
    numpy_incl = pkg_resources.resource_filename('numpy', 'core/include')
except ImportError:
    sys.exit("""install requires: 'numpy'. use pip or easy_install.
                    $ pip install numpy""")

class PyTest(TestCommand):
    user_options = [('pytest-args=', 'a', "Arguments to pass to py.test")]

    def initialize_options(self):
        TestCommand.initialize_options(self)
        self.pytest_args = []

    def finalize_options(self):
        TestCommand.finalize_options(self)
        self.test_args = []
        self.test_suite = True

    def run_tests(self):
        #import here, cause outside the eggs aren't loaded
        import pytest
        errno = pytest.main(self.pytest_args)
        sys.exit(errno)


setup(
    name="fastdb",
    packages=['fastdb'],
    include_dirs=[numpy_incl, 'fastbit'],
    ext_modules=ext_modules,
    install_requires=install_requires,
    test_suite='pytest',
    tests_require=tests_require,
    cmdclass={'test': PyTest},
)

VENV=./venv
PYTHON=$(VENV)/bin/python
PIP=$(VENV)/bin/pip
 
PACKAGE=rtlogclient
SETUP=./setup.py
REQUIREMENTS=./requirements.txt
TEST_DIR=./tests/
 
.PHONY: all update_tools install_requirements install develop test
 
usage:
	@echo use `make [target]` to build project.
 
update_tools:
	$(PIP) install --upgrade setuptools pip
 
install_requirements: update_tools $(REQUIREMENTS)
	$(PIP) install -r $(REQUIREMENTS)
 
install: install_requirements $(SETUP)
	$(PYTHON) $(SETUP) install
 
develop: install_requirements $(SETUP)
	$(PYTHON) $(SETUP) develop
 
test: $(SETUP) $(TEST_DIR)
	$(PYTHON) $(SETUP) test --pytest-args='--cov $(PACKAGE) $(TEST_DIR)'