Fix bug with Scylla protocol

docker-compose.yml

@@ -50,10 +50,9 @@ services:
       context: ./test/docker
       dockerfile: scylladb.dockerfile
       args:
-        SCYLLA_VERSION: ${SCYLLA_VERSION:-5.1.6}
+        SCYLLA_VERSION: ${SCYLLA_VERSION:-5.2}
     ports:
       - "9062:9042"
-      - "10000:10000"
     command:
       - "--smp"
       - "1"

lib/xandra/connection.ex

@@ -384,7 +384,7 @@ defmodule Xandra.Connection do
         connect_timeout: Keyword.fetch!(options, :connect_timeout),
         connection_name: Keyword.get(options, :name),
         cluster_pid: Keyword.get(options, :cluster_pid),
-        protocol_version: Keyword.get(options, :protocol_version),
+        protocol_version: data.protocol_version || Keyword.get(options, :protocol_version),
         options: options,
         backoff:
           data.backoff ||

mix.exs

@@ -75,17 +75,38 @@ defmodule Xandra.Mixfile do
   defp aliases() do
     [
       test: "test --exclude scylla_specific",
-      "test.scylladb": [
-        fn _args ->
-          System.put_env("CASSANDRA_PORT", "9062")
-          System.put_env("CASSANDRA_WITH_AUTH_PORT", "9063")
-        end,
-        "test --exclude cassandra_specific --exclude encryption --include scylla_specific"
-      ],
-      "test.all": fn args ->
-        Mix.Task.run(:test, args)
-        Mix.Task.run(:"test.scylladb", args)
+      "test.cassandra": fn args ->
+        print_header("Running Cassandra tests")
+
+        mix_cmd_with_status_check(
+          ["test", "--exclude", "scylla_specific", ansi_option() | args],
+          [
+            {"CASSANDRA_PORT", "9052"},
+            {"CASSANDRA_WITH_AUTH_PORT", "9053"}
+          ]
+        )
+      end,
+      "test.scylladb": fn args ->
+        print_header("Running ScyllaDB tests")
+
+        mix_cmd_with_status_check(
+          [
+            "test",
+            "--exclude",
+            "cassandra_specific",
+            "--exclude",
+            "encryption",
+            "--include",
+            "scylla_specific",
+            ansi_option() | args
+          ],
+          [
+            {"CASSANDRA_PORT", "9062"},
+            {"CASSANDRA_WITH_AUTH_PORT", "9063"}
+          ]
+        )
       end,
+      "test.all": ["test.cassandra", "test.scylladb"],
       docs: [
         "run pages/generate_telemetry_events_page.exs",
         "docs"
@@ -109,4 +130,52 @@ defmodule Xandra.Mixfile do
       {:toxiproxy_ex, github: "whatyouhide/toxiproxy_ex", only: :test}
     ]
   end
+
+  defp mix_cmd_with_status_check(args, env) do
+    port =
+      Port.open({:spawn_executable, System.find_executable("mix")}, [
+        :binary,
+        :exit_status,
+        args: args,
+        env:
+          Enum.map(env, fn {key, val} ->
+            {String.to_charlist(key), String.to_charlist(val)}
+          end)
+      ])
+
+    # We want a port so that we can shut down the port if we shut down the system.
+    receive_loop(port)
+  end
+
+  defp receive_loop(port) do
+    receive do
+      {^port, {:data, data}} ->
+        :ok = IO.write(data)
+        receive_loop(port)
+
+      {^port, {:exit_status, 0}} ->
+        :ok
+
+      {^port, {:exit_status, status}} ->
+        Mix.raise("Mix failed with exit status #{status}")
+    after
+      10_000 ->
+        Mix.raise("Timed out waiting for Mix to send back any data (after 10s)")
+    end
+  end
+
+  defp ansi_option do
+    if IO.ANSI.enabled?(), do: "--color", else: "--no-color"
+  end
+
+  defp print_header(header) do
+    Mix.shell().info([:cyan, :bright, header, :reset])
+
+    Mix.shell().info([
+      :cyan,
+      :bright,
+      String.duplicate("=", String.length(header)) <> "\n",
+      :reset
+    ])
+  end
 end

test/xandra/cluster_test.exs

@@ -341,16 +341,19 @@
         host: %Host{address: {198, 0, 0, 2}, port: @port, data_center: "local_dc"}
       }
 
+      node1_address = "198.0.0.1:#{@port}"
+      node2_address = "198.0.0.2:#{@port}"
+
       assert_receive {^test_ref, PoolMock, :init_called,
                       %{
                         pool_size: 1,
-                        connection_options: %{nodes: ["198.0.0.1:9052"], cluster_pid: ^pid}
+                        connection_options: %{nodes: [^node1_address], cluster_pid: ^pid}
                       }}
 
       assert_receive {^test_ref, PoolMock, :init_called,
                       %{
                         pool_size: 1,
-                        connection_options: %{nodes: ["198.0.0.2:9052"], cluster_pid: ^pid}
+                        connection_options: %{nodes: [^node2_address], cluster_pid: ^pid}
                       }}
 
       refute_receive {[:xandra, :cluster, :pool, :started], ^telemetry_ref, %{},
@@ -909,7 +912,7 @@
     @tag telemetry_events: [
            [:xandra, :cluster, :pool, :stopped]
          ]
     test "when a single connection goes down", %{base_options: opts, telemetry_ref: telemetry_ref} do
       opts = Keyword.merge(opts, sync_connect: 1000, nodes: ["127.0.0.1:19052"])
       pid = start_link_supervised!({Cluster, opts})
 

test/xandra_toxiproxy_test.exs

@@ -6,6 +6,7 @@
 
   @moduletag :toxiproxy
 
+  @tag :cassandra_specific
   test "execute/3,4 supports a network that slices packets",
        %{start_options: opts, keyspace: keyspace} do
     ToxiproxyEx.get!(:xandra_test_cassandra)
@@ -30,7 +31,7 @@
     end)
   end
 
   test "start_link/1 supports the :connect_timeout option", %{start_options: opts} do
     opts =
       Keyword.merge(opts,
         connect_timeout: 0,