Add client functions for Operator Policy endpoints

src/main/java/com/rabbitmq/http/client/Client.java

@@ -583,6 +583,11 @@ public void declarePolicy(String vhost, String name, PolicyInfo info) {
     this.httpLayer.put(uri, info);
   }
 
+  public void declareOperatorPolicy(String vhost, String name, PolicyInfo info) {
+    final URI uri = uri().withEncodedPath("./operator-policies").withPath(vhost).withPath(name).get();
+    this.httpLayer.put(uri, info);
+  }
+
   public void declareQueue(String vhost, String name, QueueInfo info) {
     final URI uri = uri().withEncodedPath("./queues").withPath(vhost).withPath(name).get();
     this.httpLayer.put(uri, info);
@@ -684,6 +689,10 @@ public void deletePolicy(String vhost, String name) {
     this.deleteIgnoring404(uri().withEncodedPath("./policies").withPath(vhost).withPath(name).get());
   }
 
+  public void deleteOperatorPolicy(String vhost, String name) {
+    this.deleteIgnoring404(uri().withEncodedPath("./operator-policies").withPath(vhost).withPath(name).get());
+  }
+
   public List<UserInfo> getUsers() {
     final URI uri = uriWithPath("./users/");
     return Arrays.asList(this.httpLayer.get(uri, UserInfo[].class));
@@ -777,6 +786,17 @@ public List<PolicyInfo> getPolicies(String vhost) {
     return asListOrNull(result);
   }
 
+  public List<PolicyInfo> getOperatorPolicies() {
+    final URI uri = uriWithPath("./operator-policies/");
+    return Arrays.asList(this.httpLayer.get(uri, PolicyInfo[].class));
+  }
+
+  public List<PolicyInfo> getOperatorPolicies(String vhost) {
+    final URI uri = uri().withEncodedPath("./operator-policies").withPath(vhost).get();
+    final PolicyInfo[] result = this.getForObjectReturningNullOn404(uri, PolicyInfo[].class);
+    return asListOrNull(result);
+  }
+
   public List<BindingInfo> getBindings() {
     final URI uri = uriWithPath("./bindings/");
     return Arrays.asList(this.httpLayer.get(uri, BindingInfo[].class));

src/main/java/com/rabbitmq/http/client/ReactorNettyClient.java

@@ -187,6 +187,10 @@ public Mono<HttpResponse> declarePolicy(String vhost, String name, PolicyInfo in
         return doPut(info, "policies", encodePath(vhost), encodePath(name));
     }
 
+    public Mono<HttpResponse> declareOperatorPolicy(String vhost, String name, PolicyInfo info) {
+        return doPut(info, "operator-policies", encodePath(vhost), encodePath(name));
+    }
+
     public Flux<PolicyInfo> getPolicies() {
         return doGetFlux(PolicyInfo.class, "policies");
     }
@@ -195,10 +199,22 @@ public Flux<PolicyInfo> getPolicies(String vhost) {
         return doGetFlux(PolicyInfo.class, "policies", encodePath(vhost));
     }
 
+    public Flux<PolicyInfo> getOperatorPolicies() {
+        return doGetFlux(PolicyInfo.class, "operator-policies");
+    }
+
+    public Flux<PolicyInfo> getOperatorPolicies(String vhost) {
+        return doGetFlux(PolicyInfo.class, "operator-policies", encodePath(vhost));
+    }
+
     public Mono<HttpResponse> deletePolicy(String vhost, String name) {
         return doDelete("policies", encodePath(vhost), encodePath(name));
     }
 
+    public Mono<HttpResponse> deleteOperatorPolicy(String vhost, String name) {
+        return doDelete("operator-policies", encodePath(vhost), encodePath(name));
+    }
+
     public Flux<ChannelInfo> getChannels() {
         return doGetFlux(ChannelInfo.class, "channels");
     }

src/test/groovy/com/rabbitmq/http/client/ClientSpec.groovy

@@ -1156,6 +1156,33 @@ class ClientSpec extends Specification {
 
   }
 
+  def "PUT /api/operator-policies/{vhost}/{name}"() {
+    given: "vhost / and definition"
+    def v = "/"
+    def d = new HashMap<String, Object>()
+    d.put("ha-mode", "all")
+
+    when: "client declares an operator policy hop.test"
+    def s = "hop.test"
+    client.declareOperatorPolicy(v, s, new PolicyInfo(".*", 1, null, d))
+
+    and: "client lists operator policies in vhost /"
+    List<PolicyInfo> ps = client.getOperatorPolicies(v)
+
+    then: "hop.test is listed"
+    PolicyInfo p = ps.find { (it.name == s) }
+    p != null
+    p.vhost == v
+    p.name == s
+    p.priority == 1
+    p.applyTo == "all"
+    p.definition == d
+
+    cleanup:
+    client.deleteOperatorPolicy(v, s)
+
+  }
+
 
   def "PUT /api/queues/{vhost}/{name} when vhost DOES NOT exist"() {
     given: "vhost lolwut which does not exist"
@@ -2205,6 +2232,63 @@ class ClientSpec extends Specification {
 
   }
 
+  def "GET /api/operator-policies"() {
+    given: "at least one operator policy was declared"
+    def v = "/"
+    def s = "hop.test"
+    def d = new HashMap<String, Object>()
+    def p = ".*"
+    d.put("ha-mode", "all")
+    client.declareOperatorPolicy(v, s, new PolicyInfo(p, 0, null, d))
+
+    when: "client lists policies"
+    PolicyInfo[] xs = awaitEventPropagation({ client.getOperatorPolicies() }) as PolicyInfo[]
+
+    then: "a list of policies is returned"
+    def x = xs.first()
+    verifyPolicyInfo(x)
+
+    cleanup:
+    client.deleteOperatorPolicy(v, s)
+
+  }
+
+
+  def "GET /api/operator-policies/{vhost} when vhost exists"() {
+    given: "at least one operator  was declared in vhost /"
+    def v = "/"
+    def s = "hop.test"
+    def d = new HashMap<String, Object>()
+    def p = ".*"
+    d.put("ha-mode", "all")
+    client.declareOperatorPolicy(v, s, new PolicyInfo(p, 0, null, d))
+
+    when: "client lists policies"
+    PolicyInfo[] xs = awaitEventPropagation({ client.getOperatorPolicies("/") }) as PolicyInfo[]
+
+    then: "a list of queues is returned"
+    def x = xs.first()
+    verifyPolicyInfo(x)
+
+    cleanup:
+    client.deleteOperatorPolicy(v, s)
+
+  }
+
+
+  def "GET /api/operator-policies/{vhost} when vhost DOES NOT exists"() {
+    given: "vhost lolwut DOES not exist"
+    def v = "lolwut"
+    client.deleteVhost(v)
+
+    when: "client lists operator policies"
+    def xs = awaitEventPropagation({ client.getOperatorPolicies(v) })
+
+    then: "null is returned"
+    xs == null
+
+  }
+
 
   def "GET /api/aliveness-test/{vhost}"() {
     when: "client performs aliveness check for the / vhost"

src/test/groovy/com/rabbitmq/http/client/ReactorNettyClientSpec.groovy

@@ -1109,6 +1109,59 @@ class ReactorNettyClientSpec extends Specification {
         exception.status() == 404
     }
 
+    def "GET /api/operator-policies"() {
+        given: "at least one operator policy was declared"
+        def v = "/"
+        def s = "hop.test"
+        def d = new HashMap<String, Object>()
+        def p = ".*"
+        d.put("ha-mode", "all")
+        client.declareOperatorPolicy(v, s, new PolicyInfo(p, 0, null, d)).block()
+
+        when: "client lists policies"
+        def xs = awaitEventPropagation({ client.getOperatorPolicies() })
+
+        then: "a list of policies is returned"
+        def x = xs.blockFirst()
+        verifyPolicyInfo(x)
+
+        cleanup:
+        client.deleteOperatorPolicy(v, s).block()
+    }
+
+    def "GET /api/operator-policies/{vhost} when vhost exists"() {
+        given: "at least one operator policy was declared in vhost /"
+        def v = "/"
+        def s = "hop.test"
+        def d = new HashMap<String, Object>()
+        def p = ".*"
+        d.put("ha-mode", "all")
+        client.declareOperatorPolicy(v, s, new PolicyInfo(p, 0, null, d)).block()
+
+        when: "client lists policies"
+        def xs = awaitEventPropagation({ client.getOperatorPolicies("/") })
+
+        then: "a list of queues is returned"
+        def x = xs.blockFirst()
+        verifyPolicyInfo(x)
+
+        cleanup:
+        client.deleteOperatorPolicy(v, s).block()
+    }
+
+    def "GET /api/operator-policies/{vhost} when vhost DOES NOT exists"() {
+        given: "vhost lolwut DOES not exist"
+        def v = "lolwut"
+        client.deleteVhost(v).block()
+
+        when: "client lists operator policies"
+        awaitEventPropagation({ client.getOperatorPolicies(v) })
+
+        then: "exception is thrown"
+        def exception = thrown(HttpClientException.class)
+        exception.status() == 404
+    }
+
     def "GET /api/aliveness-test/{vhost}"() {
         when: "client performs aliveness check for the / vhost"
         def hasSucceeded = client.alivenessTest("/").block().isSuccessful()