infi.clickhouse_orm/tests/test_decimal_fields.py

# -*- coding: utf-8 -*-
from __future__ import unicode_literals
import unittest
from decimal import Decimal

from infi.clickhouse_orm.database import Database, ServerError
from infi.clickhouse_orm.models import Model
from infi.clickhouse_orm.fields import *
from infi.clickhouse_orm.engines import *


class DecimalFieldsTest(unittest.TestCase):

    def setUp(self):
        self.database = Database('test-db', log_statements=True)
        self.database.add_setting('allow_experimental_decimal_type', 1)
        try:
            self.database.create_table(DecimalModel)
        except ServerError as e:
            if 'Unknown setting' in e.message:
                # This ClickHouse version does not support decimals yet
                raise unittest.SkipTest(e.message)
            else:
                raise

    def tearDown(self):
        self.database.drop_database()

    def _insert_sample_data(self):
        self.database.insert([
            DecimalModel(date_field='2016-08-20'),
            DecimalModel(date_field='2016-08-21', dec=Decimal('1.234')),
            DecimalModel(date_field='2016-08-22', dec32=Decimal('12342.2345')),
            DecimalModel(date_field='2016-08-23', dec64=Decimal('12342.23456')),
            DecimalModel(date_field='2016-08-24', dec128=Decimal('-4545456612342.234567')),
        ])

    def _assert_sample_data(self, results):
        self.assertEqual(len(results), 5)
        self.assertEqual(results[0].dec, Decimal(0))
        self.assertEqual(results[0].dec32, Decimal(17))
        self.assertEqual(results[1].dec, Decimal('1.234'))
        self.assertEqual(results[2].dec32, Decimal('12342.2345'))
        self.assertEqual(results[3].dec64, Decimal('12342.23456'))
        self.assertEqual(results[4].dec128, Decimal('-4545456612342.234567'))

    def test_insert_and_select(self):
        self._insert_sample_data()
        query = 'SELECT * from $table ORDER BY date_field'
        results = list(self.database.select(query, DecimalModel))
        self._assert_sample_data(results)

    def test_ad_hoc_model(self):
        self._insert_sample_data()
        query = 'SELECT * from decimalmodel ORDER BY date_field'
        results = list(self.database.select(query))
        self._assert_sample_data(results)

    def test_rounding(self):
        d = Decimal('11111.2340000000000000001')
        self.database.insert([DecimalModel(date_field='2016-08-20', dec=d, dec32=d, dec64=d, dec128=d)])
        m = DecimalModel.objects_in(self.database)[0]
        for val in (m.dec, m.dec32, m.dec64, m.dec128):
            self.assertEqual(val, Decimal('11111.234'))

    def test_assignment_ok(self):
        for value in (True, False, 17, 3.14, '20.5', Decimal('20.5')):
            DecimalModel(dec=value)

    def test_assignment_error(self):
        for value in ('abc', u'זה ארוך', None, float('NaN'), Decimal('-Infinity')):
            with self.assertRaises(ValueError):
                DecimalModel(dec=value)

    def test_aggregation(self):
        self._insert_sample_data()
        result = DecimalModel.objects_in(self.database).aggregate(m='min(dec)', n='max(dec)')
        self.assertEqual(result[0].m, Decimal(0))
        self.assertEqual(result[0].n, Decimal('1.234'))

    def test_precision_and_scale(self):
        # Go over all valid combinations
        for precision in range(1, 39):
            for scale in range(0, precision + 1):
                f = DecimalField(precision, scale)
        # Some invalid combinations
        for precision, scale in [(0, 0), (-1, 7), (7, -1), (39, 5), (20, 21)]:
            with self.assertRaises(AssertionError):
                f = DecimalField(precision, scale)

    def test_min_max(self):
        # In range
        f = DecimalField(3, 1)
        f.validate(f.to_python('99.9', None))
        f.validate(f.to_python('-99.9', None))
        # In range after rounding
        f.validate(f.to_python('99.94', None))
        f.validate(f.to_python('-99.94', None))
        # Out of range
        with self.assertRaises(ValueError):
            f.validate(f.to_python('99.99', None))
        with self.assertRaises(ValueError):
            f.validate(f.to_python('-99.99', None))
        # In range
        f = Decimal32Field(4)
        f.validate(f.to_python('99999.9999', None))
        f.validate(f.to_python('-99999.9999', None))
        # In range after rounding
        f.validate(f.to_python('99999.99994', None))
        f.validate(f.to_python('-99999.99994', None))
        # Out of range
        with self.assertRaises(ValueError):
            f.validate(f.to_python('100000', None))
        with self.assertRaises(ValueError):
            f.validate(f.to_python('-100000', None))


class DecimalModel(Model):

    date_field  = DateField()
    dec         = DecimalField(15, 3)
    dec32       = Decimal32Field(4, default=17)
    dec64       = Decimal64Field(5)
    dec128      = Decimal128Field(6)

    engine = Memory()