t_tmss_test_database.py

#!/usr/bin/env python3

# Copyright (C) 2018    ASTRON (Netherlands Institute for Radio Astronomy)
# P.O. Box 2, 7990 AA Dwingeloo, The Netherlands
#
# This file is part of the LOFAR software suite.
# The LOFAR software suite is free software: you can redistribute it and/or
# modify it under the terms of the GNU General Public License as published
# by the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# The LOFAR software suite is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.    See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License along
# with the LOFAR software suite. If not, see <http://www.gnu.org/licenses/>.

# $Id:  $

import os
import unittest
import logging
from datetime import datetime

logging.basicConfig(format='%(asctime)s %(levelname)s %(message)s', level=logging.INFO)

from lofar.common.test_utils import exit_with_skipped_code_if_skip_integration_tests
exit_with_skipped_code_if_skip_integration_tests()

from lofar.common.postgres import PostgresDatabaseConnection, FETCH_ONE
from lofar.sas.tmss.test.test_utils import TMSSPostgresTestMixin


class TMSSPostgresTestMixinTestCase(TMSSPostgresTestMixin, unittest.TestCase):
    '''
    Test the setup/teardown of a TMSS postgress test database environment.
    Thanks to the TMSSPostgresTestMixin we get a isolated postgres database with the django migrations applied to it.
    Test if we can connect to it and if we can execute some "basic" (non-django) queries on it.
    '''
    def test_db_connection(self):
        '''Can we connect/disconnect?'''
        with PostgresDatabaseConnection(self.dbcreds) as db:
            self.assertTrue(db.is_connected)

        self.assertFalse(db.is_connected)

    def test_db_basics(self):
        '''Can we do some simple plain sql queries?'''
        with PostgresDatabaseConnection(self.dbcreds) as db:
            cycle_count = db.executeQuery("SELECT COUNT(*) FROM tmssapp_cycle;", fetch=FETCH_ONE)['count']
            self.assertGreaterEqual(cycle_count, 0)

            now = datetime.utcnow()

            db.executeQuery('''INSERT INTO tmssapp_cycle VALUES (%s, %s, %s, %s, %s, %s, %s);''',
                            qargs=([], now, now, "my_description", "my_name", now, now))

            self.assertEqual(cycle_count+1, db.executeQuery("SELECT COUNT(*) FROM tmssapp_cycle;", fetch=FETCH_ONE)['count'])


if __name__ == "__main__":
    os.environ['TZ'] = 'UTC'
    unittest.main()