RO/osm_ro/tests/db_helpers.py

   1 # -*- coding: utf-8 -*-
   2 ##
   3 # Copyright 2018 University of Bristol - High Performance Networks Research
   4 # Group
   5 # All Rights Reserved.
   6 #
   7 # Contributors: Anderson Bravalheri, Dimitrios Gkounis, Abubakar Siddique
   8 # Muqaddas, Navdeep Uniyal, Reza Nejabati and Dimitra Simeonidou
   9 #
  10 # Licensed under the Apache License, Version 2.0 (the "License"); you may
  11 # not use this file except in compliance with the License. You may obtain
  12 # a copy of the License at
  13 #
  14 #         http://www.apache.org/licenses/LICENSE-2.0
  15 #
  16 # Unless required by applicable law or agreed to in writing, software
  17 # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
  18 # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
  19 # License for the specific language governing permissions and limitations
  20 # under the License.
  21 #
  22 # For those usages not covered by the Apache License, Version 2.0 please
  23 # contact with: <highperformance-networks@bristol.ac.uk>
  24 #
  25 # Neither the name of the University of Bristol nor the names of its
  26 # contributors may be used to endorse or promote products derived from
  27 # this software without specific prior written permission.
  28 #
  29 # This work has been performed in the context of DCMS UK 5G Testbeds
  30 # & Trials Programme and in the framework of the Metro-Haul project -
  31 # funded by the European Commission under Grant number 761727 through the
  32 # Horizon 2020 and 5G-PPP programmes.
  33 ##
  34 import hashlib
  35 import shlex
  36 import unittest
  37 from contextlib import contextmanager
  38 from functools import wraps
  39 from hashlib import md5
  40 from os import environ, pathsep
  41 from subprocess import STDOUT, check_output
  42 from uuid import UUID
  43
  44 from MySQLdb import connect
  45
  46 from ..nfvo_db import nfvo_db
  47
  48 HOST = environ.get('TEST_DB_HOST', 'localhost')
  49 USER = environ.get('TEST_DB_USER', 'mano')
  50 PASSWORD = environ.get('TEST_DB_PASSWORD', 'manopw')
  51 DATABASE = environ.get('TEST_DB_DATABASE', 'mano_db')
  52
  53
  54 def uuid(seed):
  55     """Generates strings with a UUID format in a repeatable way"""
  56     return str(UUID(md5(str(seed)).hexdigest()))
  57
  58
  59 def sha1(text):
  60     """Generates SHA1 hash code from a text string"""
  61     return hashlib.sha1(text).hexdigest()
  62
  63
  64 def run(*args, **kwargs):
  65     """Run a command inside a subprocess, raising an exception when it fails
  66
  67     Arguments:
  68         *args: you can pass any number of arquments as separated words in the
  69             shell, or just a single string with the entire command
  70         **kwargs: proxied to subprocess.check_output (by default
  71             ``stderr=STDOUT`` and ``universal_newlines=True``
  72     """
  73     if len(args) == 1 and isinstance(args[0], str):
  74         args = shlex.split(args[0])
  75
  76     opts = dict(stderr=STDOUT, universal_newlines=True)
  77     opts.update(kwargs)
  78     return check_output(args, **opts)
  79
  80
  81 # In order to not mess around, enforce user to explicit set the
  82 # test database in a env variable
  83 @unittest.skipUnless(
  84     environ.get('TEST_DB_HOST'),
  85     'Test database not available. Please set TEST_DB_HOST env var')
  86 class TestCaseWithDatabase(unittest.TestCase):
  87     """Connect to the database and provide methods to facilitate isolating the
  88     database stored inside it between tests.
  89
  90     In order to avoid connecting, reconnecting, creating tables and destroying
  91     tables all the time, this class manage the database using class-level
  92     fixtures. This reduce the cost of performing these actions but not
  93     guarantees isolation in the DB state between the tests.
  94     To enforce isolation, please call the ``setup_tables`` and
  95     ``empty_database`` directly, or write one single test per class.
  96     """
  97
  98     host = HOST
  99     user = USER
 100     password = PASSWORD
 101     database = DATABASE
 102
 103     @classmethod
 104     def setup_tables(cls):
 105         """Make sure the database is set up and in the right version, with all the
 106         required tables.
 107         """
 108         dbutils = environ.get('DBUTILS')
 109
 110         if dbutils:
 111             environ["PATH"] += pathsep + dbutils
 112
 113         return run('init_mano_db.sh',
 114                    '-u', cls.user,
 115                    '-p', cls.password,
 116                    '-h', cls.host,
 117                    '-d', cls.database)
 118
 119     @classmethod
 120     def empty_database(cls):
 121         """Clear the database, so one test does not interfere with the other"""
 122         # Create a custom connection not attached to the database, so we can
 123         # destroy and recreate the database itself
 124         connection = connect(cls.host, cls.user, cls.password)
 125         cursor = connection.cursor()
 126         cursor.execute(
 127             "DROP DATABASE {};".format(
 128                 connection.escape_string(cls.database)))
 129         cursor.execute(
 130             "CREATE DATABASE {};".format(
 131                 connection.escape_string(cls.database)))
 132         cursor.close()
 133         connection.close()
 134
 135
 136 class TestCaseWithDatabasePerTest(TestCaseWithDatabase):
 137     """Ensure a connection to the database before and
 138     drop tables after each test runs
 139     """
 140
 141     def setUp(self):
 142         self.setup_tables()
 143         self.addCleanup(self.empty_database)
 144
 145         self.maxDiff = None
 146
 147         self.db = nfvo_db(self.host, self.user, self.password, self.database)
 148         self.db.connect()
 149
 150     def populate(self, seeds=None, **kwargs):
 151         """Seed the database with initial values"""
 152         if not seeds:
 153             seeds = []
 154         if not isinstance(seeds, (list, tuple)):
 155             seeds = [seeds]
 156         if kwargs:
 157             seeds.append(kwargs)
 158         self.db.new_rows(seeds)
 159
 160     def count(self, table):
 161         """Count number of rows in a table"""
 162         return self.db.get_rows(
 163             SELECT='COUNT(*) as count', FROM=table)[0]['count']
 164
 165     @contextmanager
 166     def disable_foreign_keys(self):
 167         """Do the test without checking foreign keys"""
 168         try:
 169             cursor = self.db.con.cursor()
 170             cursor.execute('SET FOREIGN_KEY_CHECKS=0;')
 171             yield
 172         finally:
 173             cursor.execute('SET FOREIGN_KEY_CHECKS=1;')
 174
 175
 176 def disable_foreign_keys(test):
 177     """Do the test without checking foreign keys.
 178     To be used together in subclasses of TestCaseWithDatabasePerTest
 179     """
 180     @wraps(test)
 181     def _no_check(self, *args, **kwargs):
 182         with self.disable_foreign_keys():
 183             result = test(self, *args, **kwargs)
 184
 185         return result
 186
 187     return _no_check