Merge pull request #13950 from rabbitmq/fix-13045

Selenium: Verify consumer state when using SAC

Author: michaelklishin

deps/rabbit/src/rabbit_fifo.erl

@@ -1529,9 +1529,12 @@ activate_next_consumer(#?STATE{consumers = Cons0,
             State = State0#?STATE{consumers = Cons,
                                   service_queue = ServiceQueue1,
                                   waiting_consumers = Waiting},
+            Effects1 = consumer_update_active_effects(State, Active,
+                                                      false, waiting,
+                                                      Effects0),                                  
             Effects = consumer_update_active_effects(State, Consumer,
                                                      true, single_active,
-                                                     Effects0),
+                                                     Effects1),
             {State, Effects};
         {{ActiveCKey, ?CONSUMER_PRIORITY(ActivePriority) = Active},
          {_NextCKey, ?CONSUMER_PRIORITY(WaitingPriority)}}
@@ -1829,8 +1832,22 @@ complete_and_checkout(#{} = Meta, MsgIds, ConsumerKey,
                       Effects0, State0) ->
     State1 = complete(Meta, ConsumerKey, MsgIds, Con0, State0),
     %% a completion could have removed the active/quiescing consumer
-    {State2, Effects1} = activate_next_consumer(State1, Effects0),
-    checkout(Meta, State0, State2, Effects1).
+    Effects1 = add_active_effect(Con0, State1, Effects0),
+    {State2, Effects2} = activate_next_consumer(State1, Effects1),
+    checkout(Meta, State0, State2, Effects2).
+
+add_active_effect(#consumer{status = quiescing} = Consumer,
+                  #?STATE{cfg = #cfg{consumer_strategy = single_active},
+                          consumers = Consumers} = State,
+                  Effects) ->
+    case active_consumer(Consumers) of
+        undefined ->
+            consumer_update_active_effects(State, Consumer, false, waiting, Effects);
+        _ ->
+            Effects
+    end;
+add_active_effect(_, _, Effects) ->
+    Effects.
 
 cancel_consumer_effects(ConsumerId,
                         #?STATE{cfg = #cfg{resource = QName}},

deps/rabbitmq_management/priv/www/js/tmpl/consumers.ejs

@@ -1,5 +1,5 @@
 <% if (consumers.length > 0) { %>
-    <table class="list">
+    <table class="list" id="consumers">
       <thead>
         <tr>
 <% if (mode == 'queue') { %>

deps/rabbitmq_management/priv/www/js/tmpl/queue.ejs

@@ -11,10 +11,10 @@
 <% } %>
 
     <h3>Details</h3>
-    <table class="facts facts-l">
+    <table class="facts facts-l" id="details-queue-table">
       <tr>
         <th>Features</th>
-        <td><%= fmt_features(queue) %></td>
+        <td id="details-queue-features"><%= fmt_features(queue) %></td>
       </tr>
 <% if(!disable_stats) { %>
       <tr>
@@ -77,7 +77,7 @@
 <% } %>
 
 <% if(!disable_stats) { %>
-<div class="section-hidden">
+<div class="section-hidden" id="queue-consumers-section">
   <h2 class="updatable">Consumers (<%=(queue.consumer_details.length)%>) </h2>
   <div class="hider updatable">
 <%= format('consumers', {'mode': 'queue', 'consumers': queue.consumer_details}) %>

deps/rabbitmq_management/priv/www/js/tmpl/quorum-queue-stats.ejs

@@ -1,23 +1,24 @@
-   <table class="facts facts-l">
+   <table class="facts facts-l" id="details-queue-stats-table">
+      <tbody>
       <tr>
         <th>State</th>
         <td><%= fmt_object_state(queue) %></td>
       </tr>
       <% if(queue.consumers) { %>
       <tr>
         <th>Consumers</th>
-        <td><%= fmt_string(queue.consumers) %></td>
+        <td id="consumers"><%= fmt_string(queue.consumers) %></td>
       </tr>
       <% } else if(queue.hasOwnProperty('consumer_details')) { %>
       <tr>
         <th>Consumers</th>
-        <td><%= fmt_string(queue.consumer_details.length) %></td>
+        <td id="consumers"><%= fmt_string(queue.consumer_details.length) %></td>
       </tr>
       <% } %>
       <% if(queue.hasOwnProperty('publishers')) { %>
       <tr>
         <th>Publishers</th>
-        <td><%= fmt_string(queue.publishers) %></td>
+        <td id="publishers"><%= fmt_string(queue.publishers) %></td>
       </tr>
       <% } %>
       <tr>
@@ -30,6 +31,7 @@
         <td><%= fmt_string(queue.delivery_limit) %></td>
       </tr>
       <% } %>
+    </tbody>
     </table>
 
     <table class="facts">

selenium/Dockerfile

@@ -1,5 +1,4 @@
-# syntax=docker/dockerfile:1
-FROM atools/jdk-maven-node:mvn3-jdk11-node16 as base
+FROM node:18 as base
 
 WORKDIR /code
 

selenium/package.json

@@ -22,7 +22,8 @@
     "proxy": "^1.0.2",
     "rhea": "^3.0.3",
     "selenium-webdriver": "^4.26.0",
-    "xmlhttprequest": "^1.8.0"
+    "xmlhttprequest": "^1.8.0",
+    "amqplib": "0.8.0"
   },
   "devDependencies": {
     "chai": "^4.3.6",

selenium/test/amqp.js

@@ -7,6 +7,7 @@ var connectionOptions = getConnectionOptions()
 
 function getAmqpConnectionOptions() {
   return {
+    'scheme': process.env.RABBITMQ_AMQP_SCHEME || 'amqp',
     'host': process.env.RABBITMQ_HOSTNAME || 'rabbitmq',
     'port': process.env.RABBITMQ_AMQP_PORT || 5672,
     'username' : process.env.RABBITMQ_AMQP_USERNAME || 'guest',
@@ -39,21 +40,28 @@ function getConnectionOptions() {
   }  
 }
 module.exports = {  
-  
-  open: () => {
+  getAmqpConnectionOptions: () => { return connectionOptions },
+  getAmqpUrl: () => {
+    return connectionOptions.scheme + '://' +
+        connectionOptions.username + ":" + connectionOptions.password + "@" +
+        connectionOptions.host + ":" + connectionOptions.port
+  },
+  open: (queueName = "my-queue") => {
     let promise = new Promise((resolve, reject) => {
       container.on('connection_open', function(context) {
         resolve()
       })
     })
+    console.log("Opening amqp connection using " + JSON.stringify(connectionOptions))
+    
     let connection = container.connect(connectionOptions)
     let receiver = connection.open_receiver({
-      source: 'my-queue',
+      source: queueName,
       target: 'receiver-target',
       name: 'receiver-link'
     })
     let sender = connection.open_sender({
-      target: 'my-queue',
+      target: queueName,
       source: 'sender-source',
       name: 'sender-link'
     })
@@ -64,6 +72,13 @@ module.exports = {
       'sender' : sender
     }
   },
+  openReceiver: (handler, queueName = "my-queue") => {
+      return handler.connection.open_receiver({
+        source: queueName,
+        target: 'receiver-target',
+        name: 'receiver-link'
+      })
+  },
   close: (connection) => {
     if (connection != null) {
       connection.close()

selenium/test/exchanges/management.js

@@ -30,7 +30,8 @@ describe('Exchange management', function () {
     if (!await overview.isLoaded()) {
       throw new Error('Failed to login')
     }
-    overview.clickOnExchangesTab()
+    await overview.selectRefreshOption("Do not refresh")
+    await overview.clickOnExchangesTab()
   })
 
   it('display summary of exchanges', async function () {

selenium/test/mgt-api.js

@@ -114,18 +114,17 @@ module.exports = {
       throw new Error(req.responseText)
     }
   },
-  createVhost: (url, name, description = "", tags = []) => {
+  createVhost: (url, authorization, name, description = "", tags = []) => {
     let vhost = {
       "description": description,
       "tags": tags
     }
     log("Create vhost " + JSON.stringify(vhost) 
       + " with name " + name + " on " + url)
     const req = new XMLHttpRequest()
-    let base64Credentials = btoa('administrator-only' + ":" + 'guest')
     let finalUrl = url + "/api/vhosts/" + encodeURIComponent(name)
     req.open('PUT', finalUrl, false)
-    req.setRequestHeader("Authorization", "Basic " + base64Credentials)
+    req.setRequestHeader("Authorization", authorization)
     req.setRequestHeader('Content-Type', 'application/json')
 
     req.send(JSON.stringify(vhost))
@@ -158,13 +157,12 @@ module.exports = {
       throw new Error(req.responseText)
     }
   },
-  deleteVhost: (url, vhost) => {
+  deleteVhost: (url, authorization, vhost) => {
     log("Deleting vhost " + vhost)
     const req = new XMLHttpRequest()
-    let base64Credentials = btoa('administrator-only' + ":" + 'guest')
     let finalUrl = url + "/api/vhosts/" + encodeURIComponent(vhost)
     req.open('DELETE', finalUrl, false)
-    req.setRequestHeader("Authorization", "Basic " + base64Credentials)
+    req.setRequestHeader("Authorization", authorization)
 
     req.send()
     if (req.status == 200 || req.status == 204) {
@@ -194,21 +192,18 @@ module.exports = {
       throw new Error(req.responseText)
     }
   },
-  createQueue: (url, name, vhost, queueType = "quorum") => {    
+  createQueue: (url, authorization, vhost, name, arguments = {}) => {    
     log("Create queue " + JSON.stringify(name) 
       + " in vhost " + vhost + " on " + url)
     const req = new XMLHttpRequest()
-    let base64Credentials = btoa('administrator-only' + ":" + 'guest')
     let finalUrl = url + "/api/queues/" + encodeURIComponent(vhost) + "/"
       + encodeURIComponent(name)
     req.open('PUT', finalUrl, false)
-    req.setRequestHeader("Authorization", "Basic " + base64Credentials)
+    req.setRequestHeader("Authorization", authorization)
     req.setRequestHeader('Content-Type', 'application/json')
     let payload = {
       "durable": true,
-      "arguments":{
-        "x-queue-type" : queueType
-      }
+      "arguments": arguments
     }
     req.send(JSON.stringify(payload))
     if (req.status == 200 || req.status == 204 || req.status == 201) {
@@ -219,14 +214,13 @@ module.exports = {
       throw new Error(req.responseText)
     }
   },
-  deleteQueue: (url, name, vhost) => {
+  deleteQueue: (url, authorization, vhost, name) => {
     log("Deleting queue " + name + " on vhost " + vhost)
     const req = new XMLHttpRequest()
-    let base64Credentials = btoa('administrator-only' + ":" + 'guest')
     let finalUrl = url + "/api/queues/" + encodeURIComponent(vhost) + "/"
       + encodeURIComponent(name)
     req.open('DELETE', finalUrl, false)
-    req.setRequestHeader("Authorization", "Basic " + base64Credentials)
+    req.setRequestHeader("Authorization", authorization)
 
     req.send()
     if (req.status == 200 || req.status == 204) {

selenium/test/pageobjects/BasePage.js

@@ -34,7 +34,12 @@ module.exports = class BasePage {
     this.interactionDelay = parseInt(process.env.SELENIUM_INTERACTION_DELAY) || 0 // slow down interactions (when rabbit is behind a http proxy)
   }
 
-
+  async goTo(path) {
+    return driver.get(d.baseUrl + path)  
+  }
+  async refresh() {
+    return this.driver.navigate().refresh()
+  }
   async isLoaded () {
     return this.waitForDisplayed(MENU_TABS)
   }
@@ -147,11 +152,63 @@ module.exports = class BasePage {
     const select = await new Select(selectable)
     return select.selectByValue(vhost)
   }
+  async getTableMini(tableLocator) {
+    const table = await this.waitForDisplayed(tableLocator)
+    return this.getTableMiniUsingTableElement(table)    
+  }
+  async getTableMiniUsingTableElement(table) {
+    let tbody = await table.findElement(By.css('tbody'))
+    let rows = await tbody.findElements(By.xpath("./child::*"))
+
+    let table_model = []
+    for (let row of rows) {
+      let columnName = await row.findElement(By.css('th')).getText()
+
+      let columnValue = await row.findElement(By.css('td'))
+      let columnContent = await columnValue.findElement(By.xpath("./child::*"))
+
+      let columnType = await columnContent.getTagName()
+      
+      switch (columnType) {
+        case "table": 
+          table_model.push({
+            "name": columnName, 
+            "value" : await this.getTableMiniUsingTableElement(columnValue)
+          })
+          break
+        default: 
+          table_model.push({
+            "name" : columnName,
+            "value" : await columnContent.getText()
+          })
+      }      
+    }
+    return table_model
+  }
   async getTable(tableLocator, firstNColumns, rowClass) {
     const table = await this.waitForDisplayed(tableLocator)
     const rows = await table.findElements(rowClass == undefined ? 
-        By.css('tbody tr') : By.css('tbody tr.' + rowClass))
+    By.css('tbody tr') : By.css('tbody tr.' + rowClass))
+    let table_model = []
+    
+    for (let row of rows) {
+      let columns = await row.findElements(By.css('td'))
+      let table_row = []
+      for (let column of columns) {
+        if (firstNColumns == undefined || table_row.length < firstNColumns) {
+          table_row.push(await column.getText())
+        }
+      }
+      table_model.push(table_row)
+    }
+    return table_model
+  }
+  async getPlainTable(tableLocator, firstNColumns) {
+    const table = await this.waitForDisplayed(tableLocator)
+    let tbody = await table.findElement(By.css('tbody'))
+    let rows = await tbody.findElements(By.xpath("./child::*"))
     let table_model = []
+    
     for (let row of rows) {
       let columns = await row.findElements(By.css('td'))
       let table_row = []

selenium/test/pageobjects/QueuePage.js

@@ -7,14 +7,36 @@ const QUEUE_NAME = By.css('div#main h1 b')
 
 const DELETE_SECTION = By.css('div#main div#delete')
 const DELETE_BUTTON = By.css('div#main div#delete input[type=submit]')
+const FEATURES_TABLE = By.css('table#details-queue-table td#details-queue-features table.mini')
+const STATS_CONSUMER_COUNT = By.css('table#details-queue-stats-table td#consumers')
+
+const CONSUMERS_SECTION = By.css('div#queue-consumers-section')
+const CONSUMERS_SECTION_TITLE = By.css('div#queue-consumers-section h2')
+const CONSUMERS_TABLE = By.css('div#queue-consumers-section table.list#consumers')
 
 module.exports = class QueuePage extends BasePage {
   async isLoaded() {
     return this.waitForDisplayed(QUEUE_NAME)
   }
+
   async getName() {
     return this.getText(QUEUE_NAME)
   }
+  async getConsumerCount() {
+    return this.getText(STATS_CONSUMER_COUNT)
+  }
+  async getFeatures() {
+    return this.getTableMini(FEATURES_TABLE)
+  }
+  async getConsumersSectionTitle() {
+    return this.getText(CONSUMERS_SECTION_TITLE)
+  }
+  async clickOnConsumerSection() {
+    return this.click(CONSUMERS_SECTION)
+  }
+  async getConsumersTable() {
+    return this.getPlainTable(CONSUMERS_TABLE)
+  }
   async ensureDeleteQueueSectionIsVisible() {    
     await this.click(DELETE_SECTION)
     return this.driver.findElement(DELETE_SECTION).isDisplayed()