[Python-modules-commits] [python-testing.mysqld] 05/07: Include upstream test suite missing in source tarball
Dominik George
natureshadow-guest at moszumanska.debian.org
Thu Oct 13 15:20:36 UTC 2016
This is an automated email from the git hooks/post-receive script.
natureshadow-guest pushed a commit to branch master
in repository python-testing.mysqld.
commit 5b00df5e53b49298d887e99ee644f346bf9c5297
Author: Dominik George <nik at naturalnet.de>
Date: Thu Oct 13 16:59:55 2016 +0200
Include upstream test suite missing in source tarball
Bug: https://github.com/tk0miya/testing.mysqld/issues/4
---
tests/test_mysql.py | 279 ++++++++++++++++++++++++++++++++++++++++++++++++++++
1 file changed, 279 insertions(+)
diff --git a/tests/test_mysql.py b/tests/test_mysql.py
new file mode 100644
index 0000000..b9deae2
--- /dev/null
+++ b/tests/test_mysql.py
@@ -0,0 +1,279 @@
+# -*- coding: utf-8 -*-
+
+import os
+import sys
+import signal
+import tempfile
+import testing.mysqld
+from mock import patch
+from time import sleep
+from shutil import rmtree
+from contextlib import closing
+import pymysql
+import sqlalchemy
+
+if sys.version_info < (2, 7):
+ import unittest2 as unittest
+else:
+ import unittest
+
+
+class TestMysqld(unittest.TestCase):
+ def test_basic(self):
+ try:
+ # start mysql server
+ mysqld = testing.mysqld.Mysqld(my_cnf={'skip-networking': None})
+ self.assertIsNotNone(mysqld)
+ self.assertEqual(mysqld.dsn(),
+ dict(unix_socket=mysqld.base_dir + '/tmp/mysql.sock',
+ user='root',
+ db='test'))
+
+ # connect to mysql (w/ pymysql)
+ conn = pymysql.connect(**mysqld.dsn())
+ self.assertIsNotNone(conn)
+ self.assertRegexpMatches(mysqld.read_bootlog(), 'ready for connections')
+
+ # connect to mysql (w/ sqlalchemy)
+ engine = sqlalchemy.create_engine(mysqld.url())
+ self.assertIsNotNone(engine)
+
+ # shutting down
+ pid = mysqld.server_pid
+ self.assertTrue(os.path.exists(mysqld.base_dir + '/tmp/mysql.sock'))
+ self.assertTrue(mysqld.is_alive())
+ finally:
+ mysqld.stop()
+ sleep(1)
+
+ self.assertFalse(os.path.exists(mysqld.base_dir + '/tmp/mysql.sock'))
+ self.assertFalse(mysqld.is_alive())
+ with self.assertRaises(OSError):
+ os.kill(pid, 0) # process is down
+
+ def test_stop(self):
+ # start mysql server
+ mysqld = testing.mysqld.Mysqld(my_cnf={'skip-networking': None})
+ self.assertTrue(os.path.exists(mysqld.base_dir))
+ self.assertTrue(mysqld.is_alive())
+
+ # call stop()
+ mysqld.stop()
+ self.assertFalse(os.path.exists(mysqld.base_dir))
+ self.assertFalse(mysqld.is_alive())
+
+ # call stop() again
+ mysqld.stop()
+ self.assertFalse(os.path.exists(mysqld.base_dir))
+ self.assertFalse(mysqld.is_alive())
+
+ # delete mysqld object after stop()
+ del mysqld
+
+ def test_dsn_and_url(self):
+ mysqld = testing.mysqld.Mysqld(auto_start=0)
+ self.assertEqual({'db': 'test', 'unix_socket': mysqld.my_cnf['socket'], 'user': 'root'},
+ mysqld.dsn())
+ self.assertEqual("mysql+pymysql://root@localhost/test?unix_socket=%s" % mysqld.my_cnf['socket'],
+ mysqld.url())
+ self.assertEqual("mysql+pymysql://root@localhost/test?unix_socket=%s&charset=utf8" % mysqld.my_cnf['socket'],
+ mysqld.url(charset='utf8'))
+ self.assertEqual("mysql+mysqldb://root@localhost/test?unix_socket=%s" % mysqld.my_cnf['socket'],
+ mysqld.url(driver='mysqldb'))
+
+ mysqld = testing.mysqld.Mysqld(my_cnf={'port': 12345}, auto_start=0)
+ self.assertEqual({'db': 'test', 'host': '127.0.0.1', 'port': 12345, 'user': 'root'},
+ mysqld.dsn())
+ self.assertEqual("mysql+pymysql://root@127.0.0.1:12345/test", mysqld.url())
+ self.assertEqual("mysql+pymysql://root@127.0.0.1:12345/test?charset=utf8", mysqld.url(charset='utf8'))
+ self.assertEqual("mysql+mysqldb://root@127.0.0.1:12345/test", mysqld.url(driver='mysqldb'))
+
+ def test_with_mysql(self):
+ with testing.mysqld.Mysqld(my_cnf={'skip-networking': None}) as mysqld:
+ self.assertIsNotNone(mysqld)
+
+ # connect to mysql
+ conn = pymysql.connect(**mysqld.dsn())
+ self.assertIsNotNone(conn)
+ self.assertTrue(mysqld.is_alive())
+
+ self.assertFalse(mysqld.is_alive())
+
+ def test_multiple_mysql(self):
+ mysqld1 = testing.mysqld.Mysqld(my_cnf={'skip-networking': None})
+ mysqld2 = testing.mysqld.Mysqld(my_cnf={'skip-networking': None})
+ self.assertNotEqual(mysqld1.server_pid, mysqld2.server_pid)
+
+ self.assertTrue(mysqld1.is_alive())
+ self.assertTrue(mysqld2.is_alive())
+
+ @patch("testing.mysqld.find_program")
+ def test_mysqld_is_not_found(self, find_program):
+ find_program.side_effect = RuntimeError
+
+ with self.assertRaises(RuntimeError):
+ testing.mysqld.Mysqld(my_cnf={'skip-networking': None})
+
+ def test_fork(self):
+ mysqld = testing.mysqld.Mysqld(my_cnf={'skip-networking': None})
+ if os.fork() == 0:
+ del mysqld
+ mysqld = None
+ os.kill(os.getpid(), signal.SIGTERM) # exit tests FORCELY
+ else:
+ os.wait()
+ sleep(1)
+ self.assertTrue(mysqld.is_alive()) # process is alive (delete mysqld obj in child does not effect)
+
+ def test_stop_on_child_process(self):
+ mysqld = testing.mysqld.Mysqld(my_cnf={'skip-networking': None})
+ if os.fork() == 0:
+ mysqld.stop()
+ os.kill(mysqld.server_pid, 0) # process is alive (calling stop() is ignored)
+ os.kill(os.getpid(), signal.SIGTERM) # exit tests FORCELY
+ else:
+ os.wait()
+ sleep(1)
+ self.assertTrue(mysqld.is_alive()) # process is alive (calling stop() in child is ignored)
+
+ def test_copy_data_from(self):
+ try:
+ tmpdir = tempfile.mkdtemp()
+
+ # create new database
+ with testing.mysqld.Mysqld(my_cnf={'skip-networking': None}, base_dir=tmpdir) as mysqld:
+ conn = pymysql.connect(**mysqld.dsn())
+ cursor = conn.cursor()
+ cursor.execute("CREATE TABLE hello(id int, value varchar(256))")
+ cursor.execute("INSERT INTO hello values(1, 'hello'), (2, 'ciao')")
+ conn.commit()
+
+ # create another database from first one
+ data_dir = os.path.join(tmpdir, 'var')
+ with testing.mysqld.Mysqld(my_cnf={'skip-networking': None}, copy_data_from=data_dir) as mysqld:
+ conn = pymysql.connect(**mysqld.dsn())
+ cursor = conn.cursor()
+ cursor.execute('SELECT * FROM test.hello ORDER BY id')
+
+ self.assertEqual(cursor.fetchall(), ((1, 'hello'), (2, 'ciao')))
+ finally:
+ rmtree(tmpdir)
+
+ def test_copy_data_from_with_passwd(self):
+ try:
+ tmpdir = tempfile.mkdtemp()
+
+ # create new database
+ with testing.mysqld.Mysqld(my_cnf={'skip-networking': None}, base_dir=tmpdir) as mysqld:
+ conn = pymysql.connect(**mysqld.dsn())
+ cursor = conn.cursor()
+ cursor.execute("CREATE TABLE hello(id int, value varchar(256))")
+ cursor.execute("INSERT INTO hello values(1, 'hello'), (2, 'ciao')")
+ cursor.execute("SET PASSWORD FOR 'root'@'localhost' = PASSWORD('secret'); FLUSH PRIVILEGES;")
+ conn.commit()
+
+ # create another database from first one
+ data_dir = os.path.join(tmpdir, 'var')
+ with testing.mysqld.Mysqld(my_cnf={'skip-networking': None},
+ copy_data_from=data_dir, passwd="secret") as mysqld:
+ conn = pymysql.connect(**mysqld.dsn())
+ cursor = conn.cursor()
+ cursor.execute('SELECT * FROM test.hello ORDER BY id')
+
+ self.assertEqual(cursor.fetchall(), ((1, 'hello'), (2, 'ciao')))
+ finally:
+ rmtree(tmpdir)
+
+ def test_skipIfNotInstalled_found(self):
+ @testing.mysqld.skipIfNotInstalled
+ def testcase():
+ pass
+
+ self.assertEqual(False, hasattr(testcase, '__unittest_skip__'))
+ self.assertEqual(False, hasattr(testcase, '__unittest_skip_why__'))
+
+ @patch("testing.mysqld.find_program")
+ def test_skipIfNotInstalled_notfound(self, find_program):
+ find_program.side_effect = RuntimeError
+
+ @testing.mysqld.skipIfNotInstalled
+ def testcase():
+ pass
+
+ self.assertEqual(True, hasattr(testcase, '__unittest_skip__'))
+ self.assertEqual(True, hasattr(testcase, '__unittest_skip_why__'))
+ self.assertEqual(True, testcase.__unittest_skip__)
+ self.assertEqual("mysqld not found", testcase.__unittest_skip_why__)
+
+ def test_skipIfNotInstalled_with_args_found(self):
+ path = testing.mysqld.find_program('mysqld', ['sbin'])
+
+ @testing.mysqld.skipIfNotInstalled(path)
+ def testcase():
+ pass
+
+ self.assertEqual(False, hasattr(testcase, '__unittest_skip__'))
+ self.assertEqual(False, hasattr(testcase, '__unittest_skip_why__'))
+
+ def test_skipIfNotInstalled_with_args_notfound(self):
+ @testing.mysqld.skipIfNotInstalled("/path/to/anywhere")
+ def testcase():
+ pass
+
+ self.assertEqual(True, hasattr(testcase, '__unittest_skip__'))
+ self.assertEqual(True, hasattr(testcase, '__unittest_skip_why__'))
+ self.assertEqual(True, testcase.__unittest_skip__)
+ self.assertEqual("mysqld not found", testcase.__unittest_skip_why__)
+
+ def test_skipIfNotFound_found(self):
+ @testing.mysqld.skipIfNotFound
+ def testcase():
+ pass
+
+ self.assertEqual(False, hasattr(testcase, '__unittest_skip__'))
+ self.assertEqual(False, hasattr(testcase, '__unittest_skip_why__'))
+
+ @patch("testing.mysqld.find_program")
+ def test_skipIfNotFound_notfound(self, find_program):
+ find_program.side_effect = RuntimeError
+
+ @testing.mysqld.skipIfNotFound
+ def testcase():
+ pass
+
+ self.assertEqual(True, hasattr(testcase, '__unittest_skip__'))
+ self.assertEqual(True, hasattr(testcase, '__unittest_skip_why__'))
+ self.assertEqual(True, testcase.__unittest_skip__)
+ self.assertEqual("mysqld not found", testcase.__unittest_skip_why__)
+
+ def test_MysqldFactory(self):
+ Mysqld = testing.mysqld.MysqldFactory(cache_initialized_db=True)
+ with Mysqld() as mysqld1:
+ self.assertTrue(mysqld1.settings['copy_data_from'])
+ copy_data_from1 = mysqld1.settings['copy_data_from']
+ self.assertTrue(os.path.exists(copy_data_from1))
+ with Mysqld() as mysqld2:
+ self.assertEqual(copy_data_from1, mysqld2.settings['copy_data_from'])
+ Mysqld.clear_cache()
+ self.assertFalse(os.path.exists(copy_data_from1))
+
+ def test_MysqldFactory_with_initialized_handler(self):
+ def handler(mysqld):
+ conn = pymysql.connect(**mysqld.dsn())
+ with closing(conn.cursor()) as cursor:
+ cursor.execute("CREATE TABLE hello(id int, value varchar(256))")
+ cursor.execute("INSERT INTO hello values(1, 'hello'), (2, 'ciao')")
+ conn.commit()
+ conn.close()
+
+ Mysqld = testing.mysqld.MysqldFactory(cache_initialized_db=True,
+ on_initialized=handler)
+ try:
+ with Mysqld() as mysqld:
+ conn = pymysql.connect(**mysqld.dsn())
+ with closing(conn.cursor()) as cursor:
+ cursor.execute('SELECT * FROM hello ORDER BY id')
+ self.assertEqual(cursor.fetchall(), ((1, 'hello'), (2, 'ciao')))
+ conn.close()
+ finally:
+ Mysqld.clear_cache()
--
Alioth's /usr/local/bin/git-commit-notice on /srv/git.debian.org/git/python-modules/packages/python-testing.mysqld.git
More information about the Python-modules-commits
mailing list