X-Git-Url: https://osm.etsi.org/gitweb/?a=blobdiff_plain;f=RO%2Fosm_ro%2Ftests%2Fdb_helpers.py;fp=RO%2Fosm_ro%2Ftests%2Fdb_helpers.py;h=bedf9a5c65cf04ac1765a56d5fe8de28ccaaab7a;hb=7d782eff123e5b44d41437377ccca66ad1e8b21b;hp=0000000000000000000000000000000000000000;hpb=5db670b68349fd1f00a5efc8c0ccd0ef9d073dca;p=osm%2FRO.git diff --git a/RO/osm_ro/tests/db_helpers.py b/RO/osm_ro/tests/db_helpers.py new file mode 100644 index 00000000..bedf9a5c --- /dev/null +++ b/RO/osm_ro/tests/db_helpers.py @@ -0,0 +1,187 @@ +# -*- coding: utf-8 -*- +## +# Copyright 2018 University of Bristol - High Performance Networks Research +# Group +# All Rights Reserved. +# +# Contributors: Anderson Bravalheri, Dimitrios Gkounis, Abubakar Siddique +# Muqaddas, Navdeep Uniyal, Reza Nejabati and Dimitra Simeonidou +# +# Licensed under the Apache License, Version 2.0 (the "License"); you may +# not use this file except in compliance with the License. You may obtain +# a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT +# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the +# License for the specific language governing permissions and limitations +# under the License. +# +# For those usages not covered by the Apache License, Version 2.0 please +# contact with: +# +# Neither the name of the University of Bristol nor the names of its +# contributors may be used to endorse or promote products derived from +# this software without specific prior written permission. +# +# This work has been performed in the context of DCMS UK 5G Testbeds +# & Trials Programme and in the framework of the Metro-Haul project - +# funded by the European Commission under Grant number 761727 through the +# Horizon 2020 and 5G-PPP programmes. +## +import hashlib +import shlex +import unittest +from contextlib import contextmanager +from functools import wraps +from hashlib import md5 +from os import environ, pathsep +from subprocess import STDOUT, check_output +from uuid import UUID + +from MySQLdb import connect + +from ..nfvo_db import nfvo_db + +HOST = environ.get('TEST_DB_HOST', 'localhost') +USER = environ.get('TEST_DB_USER', 'mano') +PASSWORD = environ.get('TEST_DB_PASSWORD', 'manopw') +DATABASE = environ.get('TEST_DB_DATABASE', 'mano_db') + + +def uuid(seed): + """Generates strings with a UUID format in a repeatable way""" + return str(UUID(md5(str(seed)).hexdigest())) + + +def sha1(text): + """Generates SHA1 hash code from a text string""" + return hashlib.sha1(text).hexdigest() + + +def run(*args, **kwargs): + """Run a command inside a subprocess, raising an exception when it fails + + Arguments: + *args: you can pass any number of arquments as separated words in the + shell, or just a single string with the entire command + **kwargs: proxied to subprocess.check_output (by default + ``stderr=STDOUT`` and ``universal_newlines=True`` + """ + if len(args) == 1 and isinstance(args[0], str): + args = shlex.split(args[0]) + + opts = dict(stderr=STDOUT, universal_newlines=True) + opts.update(kwargs) + return check_output(args, **opts) + + +# In order to not mess around, enforce user to explicit set the +# test database in a env variable +@unittest.skipUnless( + environ.get('TEST_DB_HOST'), + 'Test database not available. Please set TEST_DB_HOST env var') +class TestCaseWithDatabase(unittest.TestCase): + """Connect to the database and provide methods to facilitate isolating the + database stored inside it between tests. + + In order to avoid connecting, reconnecting, creating tables and destroying + tables all the time, this class manage the database using class-level + fixtures. This reduce the cost of performing these actions but not + guarantees isolation in the DB state between the tests. + To enforce isolation, please call the ``setup_tables`` and + ``empty_database`` directly, or write one single test per class. + """ + + host = HOST + user = USER + password = PASSWORD + database = DATABASE + + @classmethod + def setup_tables(cls): + """Make sure the database is set up and in the right version, with all the + required tables. + """ + dbutils = environ.get('DBUTILS') + + if dbutils: + environ["PATH"] += pathsep + dbutils + + return run('init_mano_db.sh', + '-u', cls.user, + '-p', cls.password, + '-h', cls.host, + '-d', cls.database) + + @classmethod + def empty_database(cls): + """Clear the database, so one test does not interfere with the other""" + # Create a custom connection not attached to the database, so we can + # destroy and recreate the database itself + connection = connect(cls.host, cls.user, cls.password) + cursor = connection.cursor() + cursor.execute( + "DROP DATABASE {};".format( + connection.escape_string(cls.database))) + cursor.execute( + "CREATE DATABASE {};".format( + connection.escape_string(cls.database))) + cursor.close() + connection.close() + + +class TestCaseWithDatabasePerTest(TestCaseWithDatabase): + """Ensure a connection to the database before and + drop tables after each test runs + """ + + def setUp(self): + self.setup_tables() + self.addCleanup(self.empty_database) + + self.maxDiff = None + + self.db = nfvo_db(self.host, self.user, self.password, self.database) + self.db.connect() + + def populate(self, seeds=None, **kwargs): + """Seed the database with initial values""" + if not seeds: + seeds = [] + if not isinstance(seeds, (list, tuple)): + seeds = [seeds] + if kwargs: + seeds.append(kwargs) + self.db.new_rows(seeds) + + def count(self, table): + """Count number of rows in a table""" + return self.db.get_rows( + SELECT='COUNT(*) as count', FROM=table)[0]['count'] + + @contextmanager + def disable_foreign_keys(self): + """Do the test without checking foreign keys""" + try: + cursor = self.db.con.cursor() + cursor.execute('SET FOREIGN_KEY_CHECKS=0;') + yield + finally: + cursor.execute('SET FOREIGN_KEY_CHECKS=1;') + + +def disable_foreign_keys(test): + """Do the test without checking foreign keys. + To be used together in subclasses of TestCaseWithDatabasePerTest + """ + @wraps(test) + def _no_check(self, *args, **kwargs): + with self.disable_foreign_keys(): + result = test(self, *args, **kwargs) + + return result + + return _no_check