[Python-modules-commits] [python-testing.mysqld] 03/07: Include upstream test sutie as patch.
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 7c397a21d5e51d17e2df1d1f0c7d497711d94815
Merge: 773f290 2f2a299
Author: Dominik George <nik at naturalnet.de>
Date: Thu Oct 13 17:00:45 2016 +0200
Include upstream test sutie as patch.
debian/.git-dpm | 4 +-
debian/changelog | 3 +-
...001-From-Dominik-George-nik-naturalnet.de.patch | 298 +++++++++++++++++++++
debian/patches/series | 1 +
tests/test_mysql.py | 279 +++++++++++++++++++
5 files changed, 582 insertions(+), 3 deletions(-)
diff --cc debian/.git-dpm
index 7fffd07,0000000..bb17004
mode 100644,000000..100644
--- a/debian/.git-dpm
+++ b/debian/.git-dpm
@@@ -1,11 -1,0 +1,11 @@@
+# see git-dpm(1) from git-dpm package
- 822180fdae01f081cbe9527f6fc5fa4f3f22d672
- 822180fdae01f081cbe9527f6fc5fa4f3f22d672
++2f2a299f56008f1bcd56efec5aea369ec9bf1b99
++2f2a299f56008f1bcd56efec5aea369ec9bf1b99
+822180fdae01f081cbe9527f6fc5fa4f3f22d672
+822180fdae01f081cbe9527f6fc5fa4f3f22d672
+python-testing.mysqld_1.4.0.orig.tar.gz
+d46fa2328948957bce9d8fd2f78cb13b36707199
+9840
+debianTag="debian/%e%v"
+patchedTag="patched/%e%v"
+upstreamTag="upstream/%e%u"
diff --cc debian/changelog
index 46869bd,0000000..5708d8a
mode 100644,000000..100644
--- a/debian/changelog
+++ b/debian/changelog
@@@ -1,5 -1,0 +1,6 @@@
+python-testing.mysqld (1.4.0-1) unstable; urgency=low
+
+ * Initial release. (Closes: #838581)
++ * Include upstream test sutie as patch.
+
- -- Dominik George <nik at naturalnet.de> Thu, 13 Oct 2016 13:56:54 +0200
++ -- Dominik George <nik at naturalnet.de> Thu, 13 Oct 2016 17:00:45 +0200
diff --cc debian/patches/0001-From-Dominik-George-nik-naturalnet.de.patch
index 0000000,0000000..1e2bbdc
new file mode 100644
--- /dev/null
+++ b/debian/patches/0001-From-Dominik-George-nik-naturalnet.de.patch
@@@ -1,0 -1,0 +1,298 @@@
++From 2f2a299f56008f1bcd56efec5aea369ec9bf1b99 Mon Sep 17 00:00:00 2001
++From: Dominik George <nik at naturalnet.de>
++Date: Thu, 13 Oct 2016 16:59:55 +0200
++Subject: =?UTF-8?q?From:=20Dominik=20George=20<nik at naturalnet.de>=0ASubjec?=
++ =?UTF-8?q?t:=20Include=20upstream=20test=20suite=20missing=20in=20source?=
++ =?UTF-8?q?=20tarball=0ABug:=20https://github.com/tk0miya/testing.mysqld/i?=
++ =?UTF-8?q?ssues/4?=
++
++---
++ tests/test_mysql.py | 279 ++++++++++++++++++++++++++++++++++++++++++++++++++++
++ 1 file changed, 279 insertions(+)
++ create mode 100644 tests/test_mysql.py
++
++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()
diff --cc debian/patches/series
index 0000000,0000000..a460a55
new file mode 100644
--- /dev/null
+++ b/debian/patches/series
@@@ -1,0 -1,0 +1,1 @@@
++0001-From-Dominik-George-nik-naturalnet.de.patch
--
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