Merge lp:~jaypipes/drizzle/replication-testing into lp:~drizzle-trunk/drizzle/development

[Jay Pipes]

This patch adds a testing framework for the new replication pieces. There is currently only a single test case for verifying the information written to the publisher manifest file in the transaction log module.

[Eric Day (eday)]

Looks great! I like the use of the Python unit testing framework, I think I've seen that before... :)

review: Approve

Unmerged revisions

1442. By Jay Pipes <jpipes@serialcoder> on 2010-04-27

Connect replication test suite into main transaction_log test suite

1441. By Jay Pipes <jpipes@serialcoder> on 2010-04-27

Adds a custom test suite and framework for testing the replication pieces. The first test case tests basic functionality of creating and writing a serialized publisher manifest file.

1440. By Jay Pipes <jpipes@serialcoder> on 2010-04-26

Publisher manifest file now written to disk properly (was using incorrect ByteSize() instead of std::string::size(). Also, manifest is read on startup

1439. By Jay Pipes <jpipes@serialcoder> on 2010-04-26

Merge Monty build fixes

1438. By Jay Pipes <jpipes@serialcoder> on 2010-04-23

remove references to /drizzled/message/replication.pb.h

1437. By Jay Pipes <jpipes@serialcoder> on 2010-04-23

Moves replication.proto out of /drizzled/message and into /plugin/transaction_log/ since it is implementation-specific.

1436. By Jay Pipes <jpipes@serialcoder> on 2010-04-20

Remove unused ZeroCopyStream buffer

1435. By Jay Pipes <jpipes@serialcoder> on 2010-04-20

TransactionLogPublisher now writes manifest file to disk and syncs on close

1434. By Jay Pipes <jpipes@serialcoder> on 2010-04-20

Merge trunk and resolve conflicts

1433. By Jay Pipes <jpipes@serialcoder> on 2010-04-19

Merge trunk

=== modified file '.bzrignore'
--- .bzrignore	2010-02-24 00:19:06 +0000
+++ .bzrignore	2010-04-27 17:46:21 +0000
@@ -304,3 +304,4 @@
 *.reject
 config/pandora_vc_revinfo
 drizzled/message/schema_writer
+replication_pb2.py

=== modified file 'plugin/replication_dictionary/tests/r/replication_publisher_channels.result'
--- plugin/replication_dictionary/tests/r/replication_publisher_channels.result	2010-04-27 17:46:20 +0000
+++ plugin/replication_dictionary/tests/r/replication_publisher_channels.result	2010-04-27 17:46:21 +0000
@@ -1,4 +1,5 @@
 use data_dictionary;
+SET @@global.transaction_log_truncate_debug= 1;
 SELECT * FROM data_dictionary.replication_publisher_channels;
 PUBLISHER	CHANNEL_ID	LAST_APPLIED_TRANSACTION_ID	LAST_APPLIED_END_TIMESTAMP
 transaction_log_publisher	1	0	0

=== modified file 'plugin/replication_dictionary/tests/t/replication_publisher_channels.test'
--- plugin/replication_dictionary/tests/t/replication_publisher_channels.test	2010-04-27 17:46:20 +0000
+++ plugin/replication_dictionary/tests/t/replication_publisher_channels.test	2010-04-27 17:46:21 +0000
@@ -2,6 +2,8 @@
 
 use data_dictionary;
 
+SET @@global.transaction_log_truncate_debug= 1;
+
 SELECT * FROM data_dictionary.replication_publisher_channels;
 
 use test;

=== modified file 'plugin/transaction_log/publisher.cc'
--- plugin/transaction_log/publisher.cc	2010-04-27 17:46:20 +0000
+++ plugin/transaction_log/publisher.cc	2010-04-27 17:46:21 +0000
@@ -42,6 +42,8 @@
 #include <utility>
 
 #include <drizzled/internal/my_sys.h> /* for internal::my_sync */
+#include <drizzled/gettext.h>
+#include <drizzled/errmsg_print.h>
 
 using namespace std;
 using namespace drizzled;
@@ -56,14 +58,9 @@
   channel_apply_map(),
   manifest_file(-1)
 {
+  /** @todo handle errors much better here! */
+  (void) readManifestFile();
   manifest_file= open(MANIFEST_FILE_NAME.c_str(), O_CREAT|O_SYNC|O_WRONLY, S_IRWXU);
-  /** @todo read in serialized publisher manifest file */
-  PublisherManifest &pm= getPublisherManifest();
-  PublisherManifest::PublisherApplyInfo *appl_info= pm.add_apply_info();
-  appl_info->set_channel_id(1);
-  appl_info->set_last_applied_transaction_id(0);
-  appl_info->set_last_applied_end_timestamp(0);
-  channel_apply_map.insert(make_pair(1, appl_info));
   pthread_mutex_init(&latch, NULL);
 }
 
@@ -73,21 +70,73 @@
   {
     (void) writeManifestFile();
     (void) syncManifestFile();
+    close(manifest_file);
   }
   catch (...)
   {} /* enforce nothrow guarantee on destructors... */
   pthread_mutex_destroy(&latch);
 }
 
+bool TransactionLogPublisher::readManifestFile()
+{
+  PublisherManifest &pm= getPublisherManifest();
+  int read_file= open(MANIFEST_FILE_NAME.c_str(), O_RDONLY);
+  if (read_file == -1)
+  {
+    /* 
+     * If file doesn't exist, just create a default manifest message
+     * in memory.  This message will be written to disk into the
+     * manifest file later.
+     */
+    if (errno == ENOENT)
+    {
+      /* Create a default manifest with channel 1 and no applied transactions */
+      PublisherManifest::PublisherApplyInfo *appl_info= pm.add_apply_info();
+      appl_info->set_channel_id(1);
+      appl_info->set_last_applied_transaction_id(0);
+      appl_info->set_last_applied_end_timestamp(0);
+      (void) writeManifestFile();
+      (void) syncManifestFile();
+    }
+    else
+    {
+      errmsg_printf(ERRMSG_LVL_ERROR, _("Unable to read publisher's manifest file!  Got error: %s\n"), strerror(errno));
+      close(read_file);
+      return false;
+    }
+  }
+  else
+  {
+    if (not pm.ParseFromFileDescriptor(read_file))
+    {
+      errmsg_printf(ERRMSG_LVL_ERROR,
+                    _("Unable to parse publisher's manifest file!\n"
+                      "PublisherManifest was initialized, but contains corrupted data:\n%s\n"),
+                    pm.DebugString().c_str());
+      close(read_file);
+      return false;
+    }
+  }
+  uint32_t num_channels= pm.apply_info_size();
+  for (uint32_t x= 0; x < num_channels; ++x)
+  {
+    PublisherManifest::PublisherApplyInfo *apply_info= pm.mutable_apply_info(x);
+    channel_apply_map.insert(make_pair(apply_info->channel_id(), apply_info));
+  }
+  close(read_file);
+  return true;
+}
+
 void TransactionLogPublisher::clear()
 {
-  for (ChannelApplyMap::iterator iter= channel_apply_map.begin();
-       iter != channel_apply_map.end();
-       ++iter)
-  {
-    ((*iter).second)->set_last_applied_transaction_id(0);
-    ((*iter).second)->set_last_applied_end_timestamp(0);
-  }
+  publisher_manifest.Clear();
+  /* Create a default manifest with channel 1 and no applied transactions */
+  PublisherManifest::PublisherApplyInfo *appl_info= publisher_manifest.add_apply_info();
+  appl_info->set_channel_id(1);
+  appl_info->set_last_applied_transaction_id(0);
+  appl_info->set_last_applied_end_timestamp(0);
+  (void) ftruncate(manifest_file, 0);
+  (void) internal::my_sync(manifest_file, 0);
 }
 
 void TransactionLogPublisher::setLastApplied(const message::Transaction &applied)
@@ -130,10 +179,14 @@
   bool result= pm.SerializeToString(&write_buffer);
   if (unlikely(not result))
   {
+    errmsg_printf(ERRMSG_LVL_ERROR,
+                  _("Unable to write publisher's manifest file!\n"
+                    "PublisherManifest contains data:\n%s\n"),
+                  pm.DebugString().c_str());
     unlock();
     return false;
   }
-  uint32_t num_bytes_to_write= pm.ByteSize();
+  uint32_t num_bytes_to_write= write_buffer.size();
   ssize_t written= pwrite(manifest_file, write_buffer.c_str(), num_bytes_to_write, 0);
   result= (written != static_cast<ssize_t>(num_bytes_to_write));
   unlock();

=== modified file 'plugin/transaction_log/publisher.h'
--- plugin/transaction_log/publisher.h	2010-04-27 17:46:20 +0000
+++ plugin/transaction_log/publisher.h	2010-04-27 17:46:21 +0000
@@ -144,6 +144,10 @@
     return publisher_manifest;
   }
   /**
+   * Reads the publisher's manifest file on startup
+   */
+  bool readManifestFile();
+  /**
    * Writes the publisher's manifest file
    */
   bool writeManifestFile();

=== modified file 'plugin/transaction_log/replication.proto'
--- plugin/transaction_log/replication.proto	2010-04-27 17:46:20 +0000
+++ plugin/transaction_log/replication.proto	2010-04-27 17:46:21 +0000
@@ -1,5 +1,3 @@
-option optimize_for = SPEED;
-
 /**
  * A class representing this information
  * a subscriber keeps about the data it

=== added directory 'plugin/transaction_log/tests/inc'
=== added file 'plugin/transaction_log/tests/inc/publisher.manifest'
--- plugin/transaction_log/tests/inc/publisher.manifest	1970-01-01 00:00:00 +0000
+++ plugin/transaction_log/tests/inc/publisher.manifest	2010-04-27 17:46:21 +0000
@@ -0,0 +1,2 @@
+
+

¹è½¢Ã¥¡
\ No newline at end of file

=== modified file 'plugin/transaction_log/tests/r/publisher_manifest.result'
--- plugin/transaction_log/tests/r/publisher_manifest.result	2010-04-27 17:46:20 +0000
+++ plugin/transaction_log/tests/r/publisher_manifest.result	2010-04-27 17:46:21 +0000
@@ -9,3 +9,10 @@
 DROP TABLE t1;
 1var/master-data/publisher.manifest
 SET GLOBAL transaction_log_truncate_debug= true;
+testDefaultStartupVariables (__main__.TestPublisherManifest) ... ok
+testWritePublisherManifest (__main__.TestPublisherManifest) ... ok
+
+----------------------------------------------------------------------
+Ran 2 tests
+
+OK

=== added directory 'plugin/transaction_log/tests/replication'
=== added file 'plugin/transaction_log/tests/replication/COPYING'
--- plugin/transaction_log/tests/replication/COPYING	1970-01-01 00:00:00 +0000
+++ plugin/transaction_log/tests/replication/COPYING	2010-04-27 17:46:21 +0000
@@ -0,0 +1,32 @@
+Drizzle Replication Test Suite
+
+Copyright (C) 2010 Jay Pipes <[e-mail]>
+All rights reserved.
+
+Redistribution and use in source and binary forms, with or without
+modification, are permitted provided that the following conditions are
+met:
+
+    * Redistributions of source code must retain the above copyright
+notice, this list of conditions and the following disclaimer.
+
+    * Redistributions in binary form must reproduce the above
+copyright notice, this list of conditions and the following disclaimer
+in the documentation and/or other materials provided with the
+distribution.
+
+    * The names of its contributors may not be used to endorse or
+promote products derived from this software without specific prior
+written permission.
+
+THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.

=== added directory 'plugin/transaction_log/tests/replication/bootstrap'
=== added file 'plugin/transaction_log/tests/replication/bootstrap/ib_logfile0'
Binary files plugin/transaction_log/tests/replication/bootstrap/ib_logfile0	1970-01-01 00:00:00 +0000 and plugin/transaction_log/tests/replication/bootstrap/ib_logfile0	2010-04-27 17:46:21 +0000 differ
=== added file 'plugin/transaction_log/tests/replication/bootstrap/ib_logfile1'
Binary files plugin/transaction_log/tests/replication/bootstrap/ib_logfile1	1970-01-01 00:00:00 +0000 and plugin/transaction_log/tests/replication/bootstrap/ib_logfile1	2010-04-27 17:46:21 +0000 differ
=== added file 'plugin/transaction_log/tests/replication/bootstrap/ibdata1'
Binary files plugin/transaction_log/tests/replication/bootstrap/ibdata1	1970-01-01 00:00:00 +0000 and plugin/transaction_log/tests/replication/bootstrap/ibdata1	2010-04-27 17:46:21 +0000 differ
=== added file 'plugin/transaction_log/tests/replication/test_publisher_manifest.py'
--- plugin/transaction_log/tests/replication/test_publisher_manifest.py	1970-01-01 00:00:00 +0000
+++ plugin/transaction_log/tests/replication/test_publisher_manifest.py	2010-04-27 17:46:21 +0000
@@ -0,0 +1,262 @@
+#! /usr/bin/python
+#
+# Drizzle Replication Test Suite
+#
+# Copyright (c) 2010 Jay Pipes
+# 
+# All rights reserved.
+#
+# Use and distribution licensed under the BSD license. See the
+# COPYING file in the root project directory for full text.
+
+"""Drizzle Transaction Log Test Suite - Publisher Manifest
+
+This tests the startup process of the Drizzle server and
+the reading of the transaction log's publisher manifest
+"""
+
+__author__ = 'Jay Pipes <[e-mail]>'
+
+# 
+# This file contains a test for reading the transaction
+# log's publisher manifest file on startup.
+# 
+# It is necessary to use this testing suite
+# because the test-run.pl script assumes too many things
+# about the Drizzle test environment, and we need to 
+# initialize both proper and corrupted publisher manifests
+# in order to test the startup process correctly.
+#
+
+import os
+import sys
+import commands
+import unittest
+
+drizzled_root_dir= os.path.abspath(os.path.join(os.path.dirname(__file__), '..', '..', '..', '..'))
+trx_log_root_dir= os.path.join(drizzled_root_dir, 'plugin', 'transaction_log')
+replication_test_root_dir= os.path.join(trx_log_root_dir, 'tests', 'replication')
+
+def compile_python_proto(protodir, protofile):
+  """Compiles a proto into the proto_pb2.py file and puts it into the main test directory"""
+  protocommand= "protoc --python_out=. --proto_path=%s %s" % (protodir, os.path.join(protodir, protofile))
+  (retcode, output)= commands.getstatusoutput(protocommand)
+  if retcode != 0:
+    raise Exception("Failed to compile Python protobuffers for %s\nTried command: %s" % (protofile, protocommand))
+
+compile_python_proto(trx_log_root_dir, 'replication.proto') # proto is in /plugin/transaction_log/
+
+import replication_pb2 as rpb
+
+from testlib.server import DrizzledServer
+from testlib.client import DrizzleClient
+
+class TestPublisherManifest(unittest.TestCase):
+
+  DEFAULT_START_OPTIONS= {
+    'mysql_protocol_port': DrizzledServer.DEFAULT_MYSQL_PROTOCOL_PORT,
+    'drizzle_protocol_port': DrizzledServer.DEFAULT_DRIZZLE_PROTOCOL_PORT,
+    'datadir': os.path.join(replication_test_root_dir, "var"),
+    'basedir': drizzled_root_dir
+  }
+
+  def setUp(self):
+    pass
+
+  def tearDown(self):
+    if self.server and self.server.ping():
+      self.server.stop()
+      self.server.clear()
+    self.resetDataDir()
+
+  def resetDataDir(self):
+    """Copies default InnoDB data files from bootstrap into datadir to speed up starts"""
+    (retcode, ignored)= commands.getstatusoutput("rm -rf %s" % os.path.join(replication_test_root_dir, "var", "*"))
+
+  def bootstrapDataFiles(self):
+    """Copies default InnoDB data files from bootstrap into datadir to speed up starts"""
+    (retcode, ignored)= commands.getstatusoutput("cp -r %s %s" % (os.path.join(replication_test_root_dir, "bootstrap", "*"),
+                                                                  os.path.join(replication_test_root_dir, "var", "*")))
+
+  def testWritePublisherManifest(self):
+    #
+    # Here, we test the scenario of a newly
+    # started server with no data history.  There
+    # should be no publisher manifest file at startup
+    # and after plugin processing, the publisher manifest
+    # file should exist but have no size.  Once a few transactions
+    # are applied on the server, the manifest should show data
+    # fields with the last transaction ID and end timestamp
+    #
+    startup_options= {
+      'transaction-log-enable': '1'
+    }
+    for k, v in TestPublisherManifest.DEFAULT_START_OPTIONS.iteritems():
+      startup_options[k]= v
+
+    self.server= DrizzledServer(startup_options)
+
+    if self.server and self.server.ping():
+      self.server.stop()
+      self.resetDataDir()
+    
+    self.bootstrapDataFiles()
+
+    # Test that the publisher manifest does not exist
+    # before startup.
+
+    manifest_filepath= os.path.join(replication_test_root_dir, "var", "publisher.manifest")
+
+    self.assertFalse(os.path.exists(manifest_filepath))
+
+    self.assertTrue(self.server.start())
+    self.assertTrue(self.server.ping())
+
+    # Test that a publisher manifest exists after startup
+    # but has no size since it has not been written to yet
+
+    self.assertTrue(os.path.exists(manifest_filepath))
+    self.assertEqual(os.path.getsize(manifest_filepath), 0)
+
+    # Now add a single transaction and verify the publisher
+    # manifest contains proper information by reading the
+    # manifest into a protobuffer container class object
+
+    client_options= {
+      'basedir': startup_options['basedir'],
+      'port': startup_options['mysql_protocol_port']
+    }
+    
+    self.client= DrizzleClient(client_options)
+    self.assertTrue(self.client.execute("CREATE SCHEMA test"))
+    self.assertTrue(self.client.execute("CREATE TABLE t1 (id INT NOT NULL PRIMARY KEY)", "test"))
+    self.assertTrue(self.client.execute("INSERT INTO t1 VALUES(1)", "test"))
+
+    # Create the manifest GPB class object
+    # by reading in the publisher manifest, and then
+    # check to ensure manifest data memebers are appropriate
+
+    f= open(manifest_filepath)
+
+    try:
+      manifest= rpb.PublisherManifest()
+      manifest.ParseFromString(f.read())
+    except Exception as e:
+      self.assertFalse(True)
+    
+    f.close()
+
+    self.assertEqual(len(manifest.apply_info), 1)
+    self.assertEqual(manifest.apply_info[0].channel_id, 1)
+    self.assertEqual(manifest.apply_info[0].last_applied_transaction_id, 3)
+    self.assertNotEqual(manifest.apply_info[0].last_applied_end_timestamp, 0)
+
+    self.assertTrue(self.server.stop())
+    self.assertFalse(self.server.ping())
+
+  def testDefaultStartupVariables(self):
+    #
+    # Here, we test the "default" scenario of a newly
+    # started server with no data history. We check to see
+    # if the replication_publisher and replication_publisher_channels
+    # views in the data_dictionary contain proper values 
+    #
+    startup_options= {
+      'transaction-log-enable': '1'
+    }
+    for k, v in TestPublisherManifest.DEFAULT_START_OPTIONS.iteritems():
+      startup_options[k]= v
+      
+    client_options= {
+      'basedir': startup_options['basedir'],
+      'port': startup_options['mysql_protocol_port']
+    }
+
+    self.server= DrizzledServer(startup_options)
+
+    if self.server and self.server.ping():
+      self.server.stop()
+      self.resetDataDir()
+    
+    self.bootstrapDataFiles()
+    self.assertTrue(self.server.start())
+    self.assertTrue(self.server.ping())
+    self.client= DrizzleClient(client_options)
+
+    expected_results_1= [
+      {
+      'PUBLISHER_NAME': 'transaction_log_publisher'
+      }
+    ]
+    results= self.client.fetchAllAssoc("SELECT * FROM DATA_DICTIONARY.REPLICATION_PUBLISHERS")
+    
+    self.assertTrue(len(results) == len(expected_results_1))
+    
+    rowno= 0
+    for row in results:
+      for k, v in row.iteritems():
+        self.assertEqual(str(expected_results_1[rowno][k]), str(v))
+      rowno+= 1
+
+    expected_results_2= [
+      {
+      'PUBLISHER': 'transaction_log_publisher',
+      'CHANNEL_ID': 1,
+      'LAST_APPLIED_TRANSACTION_ID': 0,
+      'LAST_APPLIED_END_TIMESTAMP': 0
+      }
+    ]
+    results= self.client.fetchAllAssoc("SELECT * FROM DATA_DICTIONARY.REPLICATION_PUBLISHER_CHANNELS")
+    
+    self.assertTrue(len(results) == len(expected_results_2))
+    
+    rowno= 0
+    for row in results:
+      for k, v in row.iteritems():
+        self.assertEqual(str(expected_results_2[rowno][k]), str(v))
+      rowno+= 1
+
+    self.assertTrue(self.server.stop())
+    self.assertFalse(self.server.ping())
+
+#
+# Code below taken verbatim from Eric Day's prototest
+# test suite in the mysql_protocol Drizzle plugin
+#
+import optparse
+
+parser = optparse.OptionParser()
+
+parser.add_option("-t", "--test", dest="testcase", default=None,
+                  help="Test case to run", metavar="TESTCASE")
+
+(options, args) = parser.parse_args()
+
+class CustomTestRunner(unittest.TextTestRunner):
+  def run(self, test):
+    result = self._makeResult()
+    test(result)
+    result.printErrors()
+    self.stream.writeln(result.separator2)
+    run = result.testsRun
+    self.stream.writeln("Ran %d test%s" % (run, run != 1 and "s" or ""))
+    self.stream.writeln()
+    if not result.wasSuccessful():
+      self.stream.write("FAILED (")
+      failed, errored = map(len, (result.failures, result.errors))
+      if failed:
+        self.stream.write("failures=%d" % failed)
+      if errored:
+        if failed: self.stream.write(", ")
+        self.stream.write("errors=%d" % errored)
+      self.stream.writeln(")")
+    else:
+      self.stream.writeln("OK")
+    return result
+
+if __name__ == '__main__':
+  if options.testcase is None:
+    suite = unittest.TestLoader().loadTestsFromModule(__import__('__main__'))
+  else:
+    suite = unittest.TestLoader().loadTestsFromTestCase(eval(options.testcase))
+  CustomTestRunner(stream=sys.stdout, verbosity=2).run(suite)

=== added directory 'plugin/transaction_log/tests/replication/testlib'
=== added file 'plugin/transaction_log/tests/replication/testlib/__init__.py'
--- plugin/transaction_log/tests/replication/testlib/__init__.py	1970-01-01 00:00:00 +0000
+++ plugin/transaction_log/tests/replication/testlib/__init__.py	2010-04-27 17:46:21 +0000
@@ -0,0 +1,10 @@
+#! /usr/bin/python
+#
+# Drizzle Replication Test Suite
+#
+# Copyright (c) 2010 Jay Pipes
+# 
+# All rights reserved.
+#
+# Use and distribution licensed under the BSD license. See the
+# COPYING file in the root project directory for full text.

=== added file 'plugin/transaction_log/tests/replication/testlib/client.py'
--- plugin/transaction_log/tests/replication/testlib/client.py	1970-01-01 00:00:00 +0000
+++ plugin/transaction_log/tests/replication/testlib/client.py	2010-04-27 17:46:21 +0000
@@ -0,0 +1,137 @@
+#! /usr/bin/python
+#
+# Drizzle Replication Test Suite
+#
+# Copyright (c) 2010 Jay Pipes
+# 
+# All rights reserved.
+#
+# Use and distribution licensed under the BSD license. See the
+# COPYING file in the root project directory for full text.
+
+"""Defines the adapter for controlling a Drizzle database client"""
+
+import os
+import commands
+import sys
+import time
+import unittest
+
+class DrizzleClientException(Exception):
+  pass
+
+class DrizzleClient:
+  """A class wrapping a Drizzled client"""
+
+  DEFAULT_PORT= 9306
+
+  def __init__(self, options= {}):
+    self.start_options= options
+    self.cleanStartOptions()
+
+  def getPort(self):
+    """Returns port the client connects to"""
+    return self.start_options["port"]
+
+  def cleanStartOptions(self):
+    """Cleans up the startup options dictionary"""
+    clean_start_options= {}
+    for key in self.start_options.keys():
+      # Ensure option names are output as --option_name
+      # and clean the self.start_options array to remove any leading --
+      # and convert - to _
+      value= self.start_options[key]
+      
+      if not key.startswith("--"):
+        clean_key= key.replace('-','_')
+      else:
+        clean_key= key[2:].replace("-","_")
+      clean_start_options[clean_key]= value
+    
+    self.start_options= clean_start_options
+
+  def getStartOptionString(self):
+    """Builds the string of options to pass to drizzled on startup"""
+    out_options= ["%s=%s" % ("--" + key, value) for key, value in self.start_options.iteritems()]
+
+    return " ".join(out_options)
+
+  def execute(self, statement, schema=None):
+    """Executes a supplied SQL statement"""
+    if schema is None:
+      schema_string= ""
+    else:
+      schema_string= "--database=%s" % schema
+    client_cmd= "%s --port=%d %s -e\"%s\"" % (os.path.join(self.start_options["basedir"], "client", "drizzle")
+                                        , self.start_options["port"]
+                                        , schema_string
+                                        , statement)
+    (retcode, output)= commands.getstatusoutput(client_cmd)
+    if retcode != 0:
+      print "Client on port %d failed to execute SQL:\n\"%s\"\nGot error: %s" % (self.start_options["port"], statement, output)
+      return False
+    return True
+
+  def executeFile(self, file):
+    """Executes SQL statements in a supplied file"""
+    client_cmd= "%s --port=%d < %s" % (os.path.join(self.start_options["basedir"], "client", "drizzle")
+                                        , self.start_options["port"]
+                                        , file)
+    (retcode, output)= commands.getstatusoutput(client_cmd)
+    if retcode != 0:
+      print "Client on port %d failed to execute SQL in file:\n\"%s\"\nGot error: %s" % (self.start_options["port"], file, output)
+      return False
+    return True
+
+  def fetchAsDict(self, statement):
+    """Executes a supplied SELECT or SHOW statement and returns a dictionary with first column as keys, second column as values."""
+    client_cmd= "%s --port=%d -e\"%s\"" % (os.path.join(self.start_options["basedir"], "client", "drizzle")
+                                        , self.start_options["port"]
+                                        , statement)
+    (retcode, output)= commands.getstatusoutput(client_cmd)
+    if retcode != 0:
+      print "Client on port %d failed to execute SQL:\n\"%s\"\nGot error: %s" % (self.start_options["port"], statement, output)
+      return {}
+    else:
+      # Now we build the results dictionary...
+      results= {}
+      # The first line is the field names separated by a tab
+      # Every line after is a tab-delimited field values.
+      lines= output.split("\n")
+      fields= lines[0].split("\t")
+      for line in lines[1:]:
+        data_fields= line.split("\t")
+        results[data_fields[0]]= data_fields[1]
+      return results
+
+  def fetchAllAssoc(self, statement):
+    """Executes a supplied SELECT or SHOW statement and returns a sequence with associative dictionary of results."""
+    client_cmd= "%s --port=%d -e\"%s\"" % (os.path.join(self.start_options["basedir"], "client", "drizzle")
+                                        , self.start_options["port"]
+                                        , statement)
+    (retcode, output)= commands.getstatusoutput(client_cmd)
+    if retcode != 0:
+      print "Client on port %d failed to execute SQL:\n\"%s\"\nGot error: %s" % (self.start_options["port"], statement, output)
+      return []
+    else:
+      # Now we build the results dictionary...
+      results= []
+      # The first line is the field names separated by a tab
+      # Every line after is a tab-delimited field values.
+      lines= output.split("\n")
+      fields= lines[0].split("\t")
+      for line in lines[1:]:
+        row= {}
+        data_fields= line.split("\t")
+        fieldno= 0
+        for field in fields:
+          row[field]= data_fields[fieldno]
+          fieldno= fieldno + 1
+        results.append(row)
+      return results
+
+class TestDrizzleClient(unittest.TestCase):
+  pass
+
+if __name__ == '__main__':
+  unittest.main()

=== added file 'plugin/transaction_log/tests/replication/testlib/server.py'
--- plugin/transaction_log/tests/replication/testlib/server.py	1970-01-01 00:00:00 +0000
+++ plugin/transaction_log/tests/replication/testlib/server.py	2010-04-27 17:46:21 +0000
@@ -0,0 +1,152 @@
+#! /usr/bin/python
+#
+# Drizzle Replication Test Suite
+#
+# Copyright (c) 2010 Jay Pipes
+# 
+# All rights reserved.
+#
+# Use and distribution licensed under the BSD license. See the
+# COPYING file in the root project directory for full text.
+
+"""Defines the adapter for controlling a Drizzle database server"""
+
+import os
+import commands
+import sys
+import time
+import unittest
+
+class DrizzledServerStartupException(Exception):
+  pass
+
+class DrizzledServerShutdownException(Exception):
+  pass
+
+class DrizzledServer:
+  """A class responsible for starting, stopping, and pinging a Drizzled server"""
+
+  DEFAULT_MYSQL_PROTOCOL_PORT= 9306
+  DEFAULT_DRIZZLE_PROTOCOL_PORT= 9307
+
+  def __init__(self, options= {}):
+    self.start_options= options
+    self.defaults_file= ""
+    self.cleanStartOptions()
+
+  def getPort(self, protocol="drizzle"):
+    """Returns port the server listens for supplied protocol"""
+    return self.start_options[protocol + "_protocol_port"]
+
+  def cleanStartOptions(self):
+    """Cleans up the startup options dictionary"""
+    clean_start_options= {}
+    for key in self.start_options.keys():
+      # Ensure option names are output as --option_name
+      # and clean the self.start_options array to remove any leading --
+      # and convert - to _
+      value= self.start_options[key]
+      
+      if not key.startswith("--"):
+        clean_key= key.replace('-','_')
+      else:
+        clean_key= key[2:].replace("-","_")
+
+      # The defaults-file must ALWAYS be the first command-line
+      # option given to a drizzled server, so we strip it and
+      # store it locally here...
+      if clean_key == 'defaults_file':
+        self.defaults_file= "--defaults-file=%s" % value
+      else:
+        clean_start_options[clean_key]= value
+    self.start_options= clean_start_options
+
+  def getStartOptionString(self):
+    """Builds the string of options to pass to drizzled on startup"""
+    out_options= ["%s=%s" % ("--" + key, value) for key, value in self.start_options.iteritems()]
+
+    return " ".join(out_options)
+
+  def start(self):
+    """Starts a drizzled server"""
+
+    for x in ("mysql_protocol_port","basedir","datadir"):
+      if x not in self.start_options.keys():
+        raise DrizzledServerStartupException("Attempt to start server without required startup options")
+
+    start_option_string= self.getStartOptionString()
+    start_cmd= "%s %s %s > %s 2>&1 &" % (os.path.join(self.start_options['basedir'], "drizzled", "drizzled"),
+                                         self.defaults_file,
+                                         start_option_string,
+                                         os.path.join(self.start_options['datadir'], 'error.log'))
+    server_output= os.system(start_cmd)
+
+    # Here, we sleep until the server is up and running or until a timeout occurs...
+    timeout= 10
+    timer= 0
+    while not self.ping() and timer != timeout:
+      time.sleep(.25)
+      timer= timer + 1
+
+    return True
+
+  def ping(self, quiet= False):
+    """Pings the server. Returns True if server is up and running, False otherwise."""
+    ping_cmd= "%s --ping --port=%d" % (os.path.join(self.start_options['basedir'], "client", "drizzle"),
+                                       self.start_options['mysql_protocol_port'])
+
+    (retcode, output)= commands.getstatusoutput(ping_cmd)
+
+    return retcode == 0
+
+  def clear(self):
+    """Clears data files for the server."""
+    (retcode, ignored)= commands.getstatusoutput("rm -rf %s; mkdir %s" % (self.start_options['datadir'], self.start_options['datadir']))
+
+  def stop(self):
+    """Stops the server"""
+    shutdown_cmd= "%s --shutdown --port=%d" % (os.path.join(self.start_options['basedir'], "client", "drizzle"),
+                                               self.start_options['mysql_protocol_port'])
+
+    (retcode, output)= commands.getstatusoutput(shutdown_cmd)
+
+    return retcode == 0
+
+  def restart(self):
+    self.stop()
+    return self.start()
+
+  def stopAll(self):
+    """Stops ALL drizzled servers"""
+    (retcode, num_drizzled_running)= commands.getstatusoutput("ps aux | grep drizzled | wc -l")
+    if retcode == 0 and int(num_drizzled_running) > 0:
+      (retcode, ignored)= commands.getstatusoutput("killall -9 drizzled")
+      time.sleep(3)
+
+class TestDrizzledServer(unittest.TestCase):
+
+  DEFAULT_START_OPTIONS= {
+    'mysql_protocol_port': DrizzledServer.DEFAULT_MYSQL_PROTOCOL_PORT,
+    'drizzle_protocol_port': DrizzledServer.DEFAULT_DRIZZLE_PROTOCOL_PORT,
+    'datadir': os.path.join(os.getcwd().partition("plugin")[0], "tests", "var"),
+    'basedir': os.path.join(os.getcwd().partition("plugin")[0], "drizzled")
+  }
+
+  def setUp(self):
+    self.server= DrizzledServer(TestDrizzledServer.DEFAULT_START_OPTIONS)
+    if self.server.ping():
+      self.server.stopAll()
+
+  def tearDown(self):
+    self.server.stop()
+                    
+  def testStartAndPing(self):
+    self.assertTrue(self.server.start())
+    self.assertTrue(self.server.ping())
+                    
+  def testStartAndStop(self):
+    self.assertTrue(self.server.start())
+    self.assertTrue(self.server.stop())
+
+if __name__ == '__main__':
+  unittest.main()

=== added directory 'plugin/transaction_log/tests/replication/var'
=== modified file 'plugin/transaction_log/tests/t/publisher_manifest.test'
--- plugin/transaction_log/tests/t/publisher_manifest.test	2010-04-27 17:46:20 +0000
+++ plugin/transaction_log/tests/t/publisher_manifest.test	2010-04-27 17:46:21 +0000
@@ -14,3 +14,6 @@
 
 # Truncate the log file to reset for the next test
 --source ../plugin/transaction_log/tests/t/truncate_log.inc
+
+# Execute the external test suite for replication
+--exec $DRIZZLE_TEST_DIR/../plugin/transaction_log/tests/replication/test_publisher_manifest.py