policies: treat SERIAL/LOCAL_SERIAL consistency as LWT for routing

Statements with SERIAL or LOCAL_SERIAL consistency level are serialized
through the Paxos path on the server, but TokenAwarePolicy only checked
is_lwt() (from server prepare metadata) when deciding whether to skip
replica shuffling. This meant serial-consistency reads could be routed
with shuffled replicas instead of the deterministic order needed for
optimal Paxos coordination.

Now TokenAwarePolicy also checks the statement's consistency level and
skips shuffling for SERIAL/LOCAL_SERIAL, matching LWT routing behavior.

Fixes: #886

Author: mykaul

cassandra/policies.py

@@ -514,7 +514,7 @@ def make_query_plan(self, working_keyspace=None, query=None):
         else:
             replicas = self._cluster_metadata.get_replicas(keyspace, query.routing_key)
 
-        if self.shuffle_replicas and not query.is_lwt():
+        if self.shuffle_replicas and not query.is_lwt() and not ConsistencyLevel.is_serial(query.consistency_level):
             shuffle(replicas)
 
         def yield_in_order(hosts):

tests/unit/test_policies.py

@@ -944,6 +944,35 @@ def _assert_shuffle(self, patched_shuffle, cluster, keyspace, routing_key):
             assert patched_shuffle.call_count == 1
 
 
+    @patch('cassandra.policies.shuffle')
+    def test_no_shuffle_for_serial_consistency(self, patched_shuffle):
+        """
+        Test to validate that replicas are not shuffled when the statement
+        has SERIAL or LOCAL_SERIAL consistency level, since such statements
+        should be routed like LWT requests.
+        @jira_ticket PYTHON-1394
+        @expected_result shuffle should not be called for serial consistency
+
+        @test_category policy
+        """
+        for cl in (ConsistencyLevel.SERIAL, ConsistencyLevel.LOCAL_SERIAL):
+            for cluster in (self._prepare_cluster_with_vnodes(), self._prepare_cluster_with_tablets()):
+                patched_shuffle.reset_mock()
+                hosts = cluster.metadata.all_hosts()
+                child_policy = Mock()
+                child_policy.make_query_plan.return_value = hosts
+                child_policy.distance.return_value = HostDistance.LOCAL
+
+                policy = TokenAwarePolicy(child_policy, shuffle_replicas=True)
+                policy.populate(cluster, hosts)
+
+                query = Statement(routing_key='routing_key')
+                query.consistency_level = cl
+                list(policy.make_query_plan('keyspace', query))
+                assert patched_shuffle.call_count == 0, \
+                    "shuffle should not be called for consistency level %s" % cl
+
+
 class ConvictionPolicyTest(unittest.TestCase):
     def test_not_implemented(self):
         """