+++ /dev/null
-# -*- 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: <highperformance-networks@bristol.ac.uk>
-#
-# 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
projects / osm / RO.git / blobdiff