Fix consistency issues with query index tests (#155)

test/query_index_manager_test.rb

@@ -25,27 +25,16 @@ def setup
       return if use_caves?
 
       connect
-      @bucket_name = 'query-idx-test-bucket'
+      @bucket_name = "query-idx-#{SecureRandom.uuid[0..5]}"
       @cluster.buckets.create_bucket(
         Couchbase::Management::BucketSettings.new do |s|
           s.name = @bucket_name
           s.ram_quota_mb = 256
         end
       )
       env.consistency.wait_until_bucket_present(@bucket_name)
-      retry_for_duration(expected_errors: [Error::BucketNotFound]) do
-        @bucket = @cluster.bucket(@bucket_name)
-      end
-
-      # Add a scope in the bucket to verify it has been created
-      if env.server_version.supports_collections?
-        retry_for_duration(expected_errors: [Error::BucketNotFound]) do
-          @bucket.collections.create_scope("test-scope")
-        end
-      end
-
+      env.consistency.wait_until_bucket_present_in_indexes(@bucket_name)
       @idx_mgr = @cluster.query_indexes
-      sleep(2)
     end
 
     def teardown

test/utils/consistency_helper.rb

@@ -32,7 +32,7 @@ class ConsistencyHelper
       attr_reader :management_hosts
 
       RETRY_DELAY_SECS = 0.5
-      DEFAULT_TIMEOUT_SECS = 10
+      DEFAULT_TIMEOUT_SECS = 30
 
       def initialize(management_endpoint, username, password)
         @username = username
@@ -42,6 +42,7 @@ def initialize(management_endpoint, username, password)
 
       def fetch_hosts(management_endpoint, timeout: DEFAULT_TIMEOUT_SECS)
         @management_hosts = []
+        @query_hosts = []
 
         # If no management endpoint is configured, run the tests without consistency checks
         return if management_endpoint.nil?
@@ -51,14 +52,23 @@ def fetch_hosts(management_endpoint, timeout: DEFAULT_TIMEOUT_SECS)
           uri = URI("#{management_endpoint}/pools/nodes")
           req = Net::HTTP::Get.new(uri)
           req.basic_auth(@username, @password)
-          resp = Net::HTTP.start(uri.hostname, uri.port) { |http| http.request(req) }
+          config_resp = Net::HTTP.start(uri.hostname, uri.port) { |http| http.request(req) }
+
+          uri = URI("#{management_endpoint}/pools/default/nodeServices")
+          req = Net::HTTP::Get.new(uri)
+          req.basic_auth(@username, @password)
+          node_services_resp = Net::HTTP.start(uri.hostname, uri.port) { |http| http.request(req) }
 
           # Retry if it was not possible to retrieve the cluster config - if the timeout is exceeded, consistency checks will be disabled
-          next unless resp.code == "200"
+          next unless config_resp.code == "200" && node_services_resp.code == "200"
+
+          config = JSON.parse(config_resp.body)
+          node_services = JSON.parse(node_services_resp.body)
 
-          resp_body = JSON.parse(resp.body)
-          resp_body["nodes"].each do |node|
+          config["nodes"].zip(node_services["nodesExt"]).each do |node, node_ext|
             @management_hosts << node["configuredHostname"]
+            host = node["configuredHostname"].split(":")[0]
+            @query_hosts << "#{host}:#{node_ext['services']['n1ql']}" if node_ext["services"].key?("n1ql")
           end
           break
         end
@@ -119,6 +129,25 @@ def wait_until_collection_dropped(bucket_name, scope_name, collection_name, time
         end
       end
 
+      def wait_until_bucket_present_in_indexes(name, timeout: DEFAULT_TIMEOUT_SECS)
+        wait_until(timeout, "Bucket `#{name}` is not present in the query service in all nodes") do
+          @query_hosts.all? do |host|
+            uri = URI("http://#{host}/query/service")
+            puts "Checking that bucket is present in indexes at #{uri}"
+            req = Net::HTTP::Post.new(uri)
+            req.basic_auth(@username, @password)
+            req.content_type = "application/json"
+            req.body = JSON.generate({
+              "statement" => "SELECT COUNT(*) as count FROM system:keyspaces where `name`=$bucket",
+              "$bucket" => name,
+            })
+            resp = Net::HTTP.start(uri.hostname, uri.port) { |http| http.request(req) }
+            body = JSON.parse(resp.body)
+            (body["results"][0]["count"]).positive?
+          end
+        end
+      end
+
       private
 
       def wait_until(timeout, error_msg)