Author: aconway
Date: Mon May 13 18:46:23 2013
New Revision: 1482023
URL: http://svn.apache.org/r1482023
Log:
NO-JIRA: HA moved ha_store_tests.py into ha_tests.py test suite.
It was too easy to forget the store tests in a separate file.
Removed:
qpid/trunk/qpid/cpp/src/tests/ha_store_tests.py
Modified:
qpid/trunk/qpid/cpp/src/tests/CMakeLists.txt
qpid/trunk/qpid/cpp/src/tests/Makefile.am
qpid/trunk/qpid/cpp/src/tests/ha_tests.py
Modified: qpid/trunk/qpid/cpp/src/tests/CMakeLists.txt
URL:
http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/tests/CMakeLists.txt?rev=1482023&r1=1482022&r2=1482023&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/tests/CMakeLists.txt (original)
+++ qpid/trunk/qpid/cpp/src/tests/CMakeLists.txt Mon May 13 18:46:23 2013
@@ -334,9 +334,6 @@ if (PYTHON_EXECUTABLE)
if (BUILD_AMQP)
add_test (interlink_tests ${test_wrap} ${PYTHON_EXECUTABLE}
${CMAKE_CURRENT_SOURCE_DIR}/interlink_tests.py)
endif (BUILD_AMQP)
- if (BUILD_LEGACYSTORE)
- add_test (ha_store_tests ${test_wrap} ${PYTHON_EXECUTABLE}
${CMAKE_CURRENT_SOURCE_DIR}/ha_store_tests.py)
- endif (BUILD_LEGACYSTORE)
add_test (ipv6_test ${shell}
${CMAKE_CURRENT_SOURCE_DIR}/ipv6_test${test_script_suffix})
add_test (federation_tests ${shell}
${CMAKE_CURRENT_SOURCE_DIR}/run_federation_tests${test_script_suffix})
add_test (federation_sys_tests ${shell}
${CMAKE_CURRENT_SOURCE_DIR}/run_federation_sys_tests${test_script_suffix})
Modified: qpid/trunk/qpid/cpp/src/tests/Makefile.am
URL:
http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/tests/Makefile.am?rev=1482023&r1=1482022&r2=1482023&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/tests/Makefile.am (original)
+++ qpid/trunk/qpid/cpp/src/tests/Makefile.am Mon May 13 18:46:23 2013
@@ -350,7 +350,6 @@ EXTRA_DIST +=
\
run_interlink_tests \
interlink_tests.py \
brokertest.py
\
- ha_store_tests.py \
test_env.ps1.in
check_LTLIBRARIES += libdlclose_noop.la
Modified: qpid/trunk/qpid/cpp/src/tests/ha_tests.py
URL:
http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/tests/ha_tests.py?rev=1482023&r1=1482022&r2=1482023&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/tests/ha_tests.py (original)
+++ qpid/trunk/qpid/cpp/src/tests/ha_tests.py Mon May 13 18:46:23 2013
@@ -1206,6 +1206,94 @@ class ConfigurationTests(HaBrokerTest):
b = start("none", "none")
check(b, "", "")
+
+class StoreTests(BrokerTest):
+ """Test for HA with persistence."""
+ def check_skip(self):
+ if not BrokerTest.store_lib:
+ print "WARNING: skipping HA+store tests, no store lib found."
+ return not BrokerTest.store_lib
+
+ def test_store_recovery(self):
+ """Verify basic store and recover functionality"""
+ if self.check_skip(): return
+ cluster = HaCluster(self, 1)
+ sn = cluster[0].connect().session()
+ # Create queue qq, exchange exx and binding between them
+ s = sn.sender("qq;{create:always,node:{durable:true}}")
+ sk = sn.sender("exx/k;{create:always,node:{type:topic, durable:true,
x-declare:{type:'direct'}, x-bindings:[{exchange:exx,key:k,queue:qq}]}}")
+ for m in ["foo", "bar", "baz"]: s.send(Message(m, durable=True))
+ r = cluster[0].connect().session().receiver("qq")
+ self.assertEqual(r.fetch().content, "foo")
+ r.session.acknowledge()
+ # FIXME aconway 2012-09-21: sending this message is an ugly hack to
flush
+ # the dequeue operation on qq.
+ s.send(Message("flush", durable=True))
+
+ def verify(broker, x_count):
+ sn = broker.connect().session()
+ assert_browse(sn, "qq", [ "bar", "baz", "flush" ]+ (x_count)*["x"])
+ sn.sender("exx/k").send(Message("x", durable=True))
+ assert_browse(sn, "qq", [ "bar", "baz", "flush" ]+
(x_count+1)*["x"])
+
+ verify(cluster[0], 0) # Sanity check
+ cluster.bounce(0)
+ cluster[0].wait_status("active")
+ verify(cluster[0], 1) # Loaded from store
+ cluster.start()
+ cluster[1].wait_status("ready")
+ cluster.kill(0)
+ cluster[1].wait_status("active")
+ verify(cluster[1], 2)
+ cluster.bounce(1, promote_next=False)
+ cluster[1].promote()
+ cluster[1].wait_status("active")
+ verify(cluster[1], 3)
+
+ def test_catchup_store(self):
+ """Verify that a backup erases queue data from store recovery before
+ doing catch-up from the primary."""
+ if self.check_skip(): return
+ cluster = HaCluster(self, 2)
+ sn = cluster[0].connect().session()
+ s1 = sn.sender("q1;{create:always,node:{durable:true}}")
+ for m in ["foo","bar"]: s1.send(Message(m, durable=True))
+ s2 = sn.sender("q2;{create:always,node:{durable:true}}")
+ sk2 = sn.sender("ex/k2;{create:always,node:{type:topic, durable:true,
x-declare:{type:'direct'}, x-bindings:[{exchange:ex,key:k2,queue:q2}]}}")
+ sk2.send(Message("hello", durable=True))
+ # Wait for backup to catch up.
+ cluster[1].assert_browse_backup("q1", ["foo","bar"])
+ cluster[1].assert_browse_backup("q2", ["hello"])
+
+ # Make changes that the backup doesn't see
+ cluster.kill(1, promote_next=False)
+ r1 = cluster[0].connect().session().receiver("q1")
+ for m in ["foo", "bar"]: self.assertEqual(r1.fetch().content, m)
+ r1.session.acknowledge()
+ for m in ["x","y","z"]: s1.send(Message(m, durable=True))
+ # Use old connection to unbind
+ us = cluster[0].connect_old().session(str(uuid4()))
+ us.exchange_unbind(exchange="ex", binding_key="k2", queue="q2")
+ us.exchange_bind(exchange="ex", binding_key="k1", queue="q1")
+ # Restart both brokers from store to get inconsistent sequence
numbering.
+ cluster.bounce(0, promote_next=False)
+ cluster[0].promote()
+ cluster[0].wait_status("active")
+ cluster.restart(1)
+ cluster[1].wait_status("ready")
+
+ # Verify state
+ cluster[0].assert_browse("q1", ["x","y","z"])
+ cluster[1].assert_browse_backup("q1", ["x","y","z"])
+ sn = cluster[0].connect().session() # FIXME aconway 2012-09-25: should
fail over!
+ sn.sender("ex/k1").send("boo")
+ cluster[0].assert_browse_backup("q1", ["x","y","z", "boo"])
+ cluster[1].assert_browse_backup("q1", ["x","y","z", "boo"])
+ sn.sender("ex/k2").send("hoo") # q2 was unbound so this should be
dropped.
+ sn.sender("q2").send("end") # mark the end of the queue for
assert_browse
+ cluster[0].assert_browse("q2", ["hello", "end"])
+ cluster[1].assert_browse_backup("q2", ["hello", "end"])
+
if __name__ == "__main__":
shutil.rmtree("brokertest.tmp", True)
qpid_ha = os.getenv("QPID_HA_EXEC")
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
