1 # -*- coding: utf-8 -*-
3 # Copyright 2018 University of Bristol - High Performance Networks Research
7 # Contributors: Anderson Bravalheri, Dimitrios Gkounis, Abubakar Siddique
8 # Muqaddas, Navdeep Uniyal, Reza Nejabati and Dimitra Simeonidou
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
14 # http://www.apache.org/licenses/LICENSE-2.0
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
22 # For those usages not covered by the Apache License, Version 2.0 please
23 # contact with: <highperformance-networks@bristol.ac.uk>
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.
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.
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
44 from MySQLdb
import connect
46 from ..nfvo_db
import nfvo_db
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')
55 """Generates strings with a UUID format in a repeatable way"""
56 return str(UUID(md5(str(seed
)).hexdigest()))
60 """Generates SHA1 hash code from a text string"""
61 return hashlib
.sha1(text
).hexdigest()
64 def run(*args
, **kwargs
):
65 """Run a command inside a subprocess, raising an exception when it fails
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``
73 if len(args
) == 1 and isinstance(args
[0], str):
74 args
= shlex
.split(args
[0])
76 opts
= dict(stderr
=STDOUT
, universal_newlines
=True)
78 return check_output(args
, **opts
)
81 # In order to not mess around, enforce user to explicit set the
82 # test database in a env variable
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.
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.
104 def setup_tables(cls
):
105 """Make sure the database is set up and in the right version, with all the
108 dbutils
= environ
.get('DBUTILS')
111 environ
["PATH"] += pathsep
+ dbutils
113 return run('init_mano_db.sh',
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()
127 "DROP DATABASE {};".format(
128 connection
.escape_string(cls
.database
)))
130 "CREATE DATABASE {};".format(
131 connection
.escape_string(cls
.database
)))
136 class TestCaseWithDatabasePerTest(TestCaseWithDatabase
):
137 """Ensure a connection to the database before and
138 drop tables after each test runs
143 self
.addCleanup(self
.empty_database
)
147 self
.db
= nfvo_db(self
.host
, self
.user
, self
.password
, self
.database
)
150 def populate(self
, seeds
=None, **kwargs
):
151 """Seed the database with initial values"""
154 if not isinstance(seeds
, (list, tuple)):
158 self
.db
.new_rows(seeds
)
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']
166 def disable_foreign_keys(self
):
167 """Do the test without checking foreign keys"""
169 cursor
= self
.db
.con
.cursor()
170 cursor
.execute('SET FOREIGN_KEY_CHECKS=0;')
173 cursor
.execute('SET FOREIGN_KEY_CHECKS=1;')
176 def disable_foreign_keys(test
):
177 """Do the test without checking foreign keys.
178 To be used together in subclasses of TestCaseWithDatabasePerTest
181 def _no_check(self
, *args
, **kwargs
):
182 with self
.disable_foreign_keys():
183 result
= test(self
, *args
, **kwargs
)