Test covar_samp(), covar_pip() and corr() functions with Firebird3

test_corr_fb.py

# coding:utf-8

import unittest
import tempfile
import firebirdsql

values_with_null = [
    (1, 1),
    (2, 2),
    (3, 4),
    (4, None),
    (5, 16),
    (None, 32),
    (7, 64),
    (8, 128),
    (9, 256),
    (10, 512),
]

values_without_null = [
    (1, 1),
    (2, 2),
    (3, 4),
    (4, 8),
    (5, 16),
    (6, 32),
    (7, 64),
    (8, 128),
    (9, 256),
    (10, 512),
]

class TestAggregate(unittest.TestCase):
    host='localhost'
    port=3050
    user='sysdba'
    password='masterkey'
    def setUp(self):
        self.database=tempfile.mktemp()
        self.connection = firebirdsql.create_database(
                host=self.host,
                port=self.port,
                database=self.database,
                user=self.user,
                password=self.password)

        cur = self.connection.cursor()
        cur.execute("""
            create table test_int_with_null (
                x integer,
                y integer
            )
        """)
        cur.execute("""
            create table test_double_with_null (
                x double precision,
                y double precision
            )
        """)
        cur.execute("""
            create table test_int_without_null (
                x integer,
                y integer
            )
        """)
        cur.execute("""
            create table test_double_without_null (
                x double precision,
                y double precision
            )
        """)
        self.connection.commit()

        self.connection.begin()
        cur = self.connection.cursor()
        for x, y in values_with_null:
            cur.execute("insert into test_int_with_null (X, Y) values (?, ?)", (x, y))
            cur.execute("insert into test_double_with_null (X, Y) values (?, ?)", (x, y))

        for x, y in values_without_null:
            cur.execute("insert into test_int_without_null (X, Y) values (?, ?)", (x, y))
            cur.execute("insert into test_double_without_null (X, Y) values (?, ?)", (x, y))

        self.connection.commit()

    def tearDown(self):
        self.connection.close()

    def test_corr(self):
        cur = self.connection.cursor()
        cur.execute("select corr(y, x) from test_double_with_null")
        self.assertEqual(cur.fetchone()[0], 0.8122519602602442)
        cur.execute("select corr(y, x) from test_double_without_null")
        self.assertEqual(cur.fetchone()[0], 0.7988372426181232)
        cur.execute("select corr(y, x) from test_int_with_null")
        self.assertEqual(cur.fetchone()[0], 0.8122519602602442)
        cur.execute("select corr(y, x) from test_int_without_null")
        self.assertEqual(cur.fetchone()[0], 0.7988372426181232)


    def test_covar_pop(self):
        cur = self.connection.cursor()
        cur.execute("select covar_pop(y, x) from test_double_with_null")
        self.assertEqual(cur.fetchone()[0], 432.953125)
        cur.execute("select covar_pop(y, x) from test_double_without_null")
        self.assertEqual(cur.fetchone()[0], 359.05)
        cur.execute("select covar_pop(y, x) from test_int_with_null")
        self.assertEqual(cur.fetchone()[0], 432.953125)
        cur.execute("select covar_pop(y, x) from test_int_without_null")
        self.assertEqual(cur.fetchone()[0], 359.05)

    def test_covar_samp(self):
        cur = self.connection.cursor()
        cur.execute("select covar_samp(y, x) from test_double_with_null")
        self.assertEqual(cur.fetchone()[0], 494.80357142857144)
        cur.execute("select covar_samp(y, x) from test_double_without_null")
        self.assertEqual(cur.fetchone()[0], 398.94444444444446)
        cur.execute("select covar_samp(y, x) from test_int_with_null")
        self.assertEqual(cur.fetchone()[0], 494.80357142857144)
        cur.execute("select covar_samp(y, x) from test_int_without_null")
        self.assertEqual(cur.fetchone()[0], 398.94444444444446)

if __name__ == "__main__":
    unittest.main()