Merge pull request #325 from sf2ne/issue311

Notification Framework Agbot Case

README.md

@@ -224,6 +224,11 @@ Now you can disable root by setting `api.root.enabled` to `false` in `/etc/horiz
     - detect if a pattern is updated with service that has userInput w/o default values, and give warning
     - Consider changing all creates to POST, and update (via put/patch) return codes to 200
 
+## Changes in 2.14.0
+
+- Issue 311: Notification Framework Agbot Case
+- Fixes for Scalatest upgrade to 3.1
+
 ## Changes in 2.13.0
 
 - Issue 312: Using only node table's lastUpdated field to filter on (updating lastUpdated in node, policy, and agreement changes)

build.sbt

@@ -71,6 +71,7 @@ lazy val root = (project in file(".")).
       "com.typesafe.akka" %% "akka-stream-testkit"  % akkaVersion     % Test,
 
       "org.scalatest" %% "scalatest" % "latest.release" % "test",
+      "org.scalatestplus" %% "junit-4-12" % "latest.release" % "test",
       "org.scalacheck" %% "scalacheck" % "latest.release" % "test",
       "junit" % "junit" % "latest.release" % "test"
     ),

src/main/resources/config.json

@@ -31,7 +31,8 @@
 			"type": "guava"    // currently guava is the only option
 		},
 		"resourceChanges": {
-			"ttl": 604800    // number of seconds to keep the history records of resource changes - 604800 is 1 week
+			"ttl": 604800,    // number of seconds to keep the history records of resource changes - 604800 is 1 week
+			"maxRecordsCap": 10000 // maximum number of records the notification framework route will return
 		},
 		"db": {
 			"driverClass": "org.postgresql.Driver",

src/main/resources/version.txt

@@ -1 +1 @@
-2.13.0
+2.14.0

src/main/scala/com/horizon/exchangeapi/OrgsRoutes.scala

@@ -100,7 +100,7 @@ class NodeHealthHashElement(var lastHeartbeat: String, var agreements: Map[Strin
 final case class PostNodeHealthResponse(nodes: Map[String,NodeHealthHashElement])
 
 /** Case class for request body for ResourceChanges route */
-final case class ResourceChangesRequest(changeId: Int, lastUpdated: Option[String], maxRecords: Int, ibmAgbot: Option[Boolean]) {
+final case class ResourceChangesRequest(changeId: Int, lastUpdated: Option[String], maxRecords: Int, orgList: Option[List[String]]) {
   def getAnyProblem: Option[String] = None // None means no problems with input
 }
 
@@ -550,6 +550,7 @@ trait OrgsRoutes extends JacksonSupport with AuthenticationSupport {
   "changeId": <number-here>,
   "lastUpdated": "<time-here>", --> optional field, only use if the caller doesn't know what changeId to use
   "maxRecords": <number-here>, --> the maximum number of records the caller wants returned to them, NOT optional
+  "orgList": ["", "", ""] --> just for agbots, this should be the list of orgs the agbot is responsible for
 }
 ```""", required = true, content = Array(new Content(schema = new Schema(implementation = classOf[ResourceChangesRequest])))),
     responses = Array(
@@ -564,38 +565,34 @@ trait OrgsRoutes extends JacksonSupport with AuthenticationSupport {
     exchAuth(TOrg(orgId), Access.READ) { ident =>
       validateWithMsg(reqBody.getAnyProblem) {
         complete({
+          // make sure callers obey maxRecords cap set in config, defaults is 10,000
+          val maxRecordsCap = ExchConfig.getInt("api.resourceChanges.maxRecordsCap")
+          val maxRecords = if (reqBody.maxRecords > maxRecordsCap) maxRecordsCap else reqBody.maxRecords
           // Create a query to get the last changeid currently in the table
           val qMaxChangeId = ResourceChangesTQ.rows.sortBy(_.changeId.desc).take(1).map(_.changeId)
           var maxChangeId = 0
-
+          val orgSet : Set[String] = reqBody.orgList.getOrElse(List("")).toSet
           // Create query to get the rows relevant to this client. We only support either changeId or lastUpdated being specified, but not both
           var qFilter = if (reqBody.lastUpdated.getOrElse("") != "" && reqBody.changeId <= 0) ResourceChangesTQ.rows.filter(_.lastUpdated >= reqBody.lastUpdated.get) else ResourceChangesTQ.rows.filter(_.changeId >= reqBody.changeId)
 
-          qFilter = qFilter.filter(u => (u.orgId === orgId) || (u.orgId =!= orgId && u.public === "true"))
-          //val lastTime = reqBody.lastUpdated.getOrElse(ApiTime.beginningUTC)
-          // filter by lastUpdated and changeId then filter by either it's in the org OR it's not in the same org but is public
-          //var qFilter = ResourceChangesTQ.rows.filter(_.lastUpdated >= lastTime).filter(_.changeId >= reqBody.changeId).filter(u => (u.orgId === orgId) || (u.orgId =!= orgId && u.public === "true"))
           ident match {
             case _: INode =>
               // if its a node calling then it doesn't want information about any other nodes
-              qFilter = qFilter.filter(u => (u.category === "node" && u.id === ident.getIdentity) || u.category =!= "node")
-            case _ => ;
-              // Note: repeating some of the filters in both cases to make the final query less nested for the db
-              //ResourceChangesTQ.rows.filter(_.changeId >= reqBody.changeId).filter(_.lastUpdated >= lastTime).filter(u => (u.orgId === orgId) || (u.orgId =!= orgId && u.public === "true"))
+              qFilter = qFilter.filter(u => (u.orgId === orgId) || (u.orgId =!= orgId && u.public === "true")).filter(u => (u.category === "node" && u.id === ident.getIdentity) || u.category =!= "node")
+            case _: IAgbot =>
+              val wildcard = orgSet.contains("*") || orgSet.contains("")
+              if (ident.isMultiTenantAgbot && !wildcard) { // its an IBM Agbot, get all changes from orgs the agbot covers
+                qFilter = qFilter.filter(_.orgId inSet orgSet)
+                // if the caller agbot sends in the wildcard case then we don't want to filter on orgId at all, so don't add any more filters. that's why there's just no code written for that case
+              } else if (!ident.isMultiTenantAgbot) qFilter = qFilter.filter(u => (u.orgId === orgId) || (u.orgId =!= orgId && u.public === "true")) // if its not an IBM agbot use the general case
+            case _ => qFilter = qFilter.filter(u => (u.orgId === orgId) || (u.orgId =!= orgId && u.public === "true"))
           }
           // sort by changeId and take only maxRecords from the query
-          qFilter = qFilter.sortBy(_.changeId).take(reqBody.maxRecords)
+          qFilter = qFilter.sortBy(_.changeId).take(maxRecords)
 
           logger.debug(s"POST /orgs/$orgId/changes db query: ${qFilter.result.statements}")
           var qResp : scala.Seq[ResourceChangeRow] = null
 
-          /* to put back the table trimming: restore this commented section and remove the 1 line of db.run beneath it
-          // Get the time for trimming rows from the table
-          val timeExpires = ApiTime.pastUTC(ExchConfig.getInt("api.resourceChanges.ttl"))
-          db.run(ResourceChangesTQ.getRowsExpired(timeExpires).delete.flatMap({ xs =>
-            logger.debug("POST /orgs/" + orgId + "/changes number of rows deleted: " + xs.toString)
-            qMaxChangeId.result.asTry
-          }).flatMap({ */
           db.run(qMaxChangeId.result.asTry.flatMap({
             case Success(qMaxChangeIdResp) =>
               maxChangeId = if (qMaxChangeIdResp.nonEmpty) qMaxChangeIdResp.head else 0
@@ -622,7 +619,7 @@ trait OrgsRoutes extends JacksonSupport with AuthenticationSupport {
             case Success(n) =>
               logger.debug(s"POST /orgs/$orgId/changes node/agbot heartbeat result: $n")
               if (n > 0) {
-                val hitMaxRecords = (qResp.size == reqBody.maxRecords) // if they are equal then we hit maxRecords
+                val hitMaxRecords = (qResp.size == maxRecords) // if they are equal then we hit maxRecords
                 if(qResp.nonEmpty) (HttpCode.POST_OK, buildResourceChangesResponse(qResp, hitMaxRecords, reqBody.changeId, maxChangeId))
                 else (HttpCode.POST_OK, ResourceChangesRespObject(List[ChangeEntry](), maxChangeId, hitMaxRecords = false, ExchangeApi.adminVersion()))
               }

src/test/scala/exchangeapi/AdminSuite.scala

@@ -1,9 +1,9 @@
 package exchangeapi
 
-import org.scalatest.FunSuite
+import org.scalatest.funsuite.AnyFunSuite
 
 import org.junit.runner.RunWith
-import org.scalatest.junit.JUnitRunner
+import org.scalatestplus.junit.JUnitRunner
 import scalaj.http._
 import org.json4s._
 //import org.json4s.JsonDSL._
@@ -22,7 +22,7 @@ import com.horizon.exchangeapi._
  * clear and detailed tutorial of FunSuite: http://doc.scalatest.org/1.9.1/index.html#org.scalatest.FunSuite
  */
 @RunWith(classOf[JUnitRunner])
-class AdminSuite extends FunSuite {
+class AdminSuite extends AnyFunSuite {
 
   val urlRoot = sys.env.getOrElse("EXCHANGE_URL_ROOT", "http://localhost:8080")
   val URL = urlRoot+"/v1"

src/test/scala/exchangeapi/AgbotsSuite.scala

@@ -1,9 +1,9 @@
 package exchangeapi
 
-import org.scalatest.FunSuite
+import org.scalatest.funsuite.AnyFunSuite
 
 import org.junit.runner.RunWith
-import org.scalatest.junit.JUnitRunner
+import org.scalatestplus.junit.JUnitRunner
 import scalaj.http._
 import org.json4s._
 //import org.json4s.JsonDSL._
@@ -24,7 +24,7 @@ import java.time._
  * clear and detailed tutorial of FunSuite: http://doc.scalatest.org/1.9.1/index.html#org.scalatest.FunSuite
  */
 @RunWith(classOf[JUnitRunner])
-class AgbotsSuite extends FunSuite {
+class AgbotsSuite extends AnyFunSuite {
 
   val localUrlRoot = "http://localhost:8080"
   val urlRoot = sys.env.getOrElse("EXCHANGE_URL_ROOT", localUrlRoot)
@@ -174,6 +174,17 @@ class AgbotsSuite extends FunSuite {
     assert(parsedBody.changes.exists(y => {(y.id == agbotId) && (y.operation == ResourceChangeConfig.CREATEDMODIFIED)}))
   }
 
+  test("POST /orgs/"+orgid+"/changes - verify " + agbotId + " can call notification framework") {
+    val time = ApiTime.pastUTC(secondsAgo)
+    val input = ResourceChangesRequest(0, Some(time), maxRecords, Some(List(orgid)))
+    val response = Http(URL+"/changes").postData(write(input)).method("post").headers(CONTENT).headers(ACCEPT).headers(AGBOTAUTH).asString
+    info("code: "+response.code)
+    assert(response.code === HttpCode.POST_OK.intValue)
+    assert(!response.body.isEmpty)
+    val parsedBody = parse(response.body).extract[ResourceChangesRespObject]
+    assert(parsedBody.changes.exists(y => {(y.id == agbotId) && (y.operation == ResourceChangeConfig.CREATEDMODIFIED)}))
+  }
+
   /** Update normal agbot as user */
   test("PUT /orgs/"+orgid+"/agbots/"+agbotId+" - normal - as user") {
     val input = PutAgbotsRequest(agbotToken, "agbot"+agbotId+"-normal-user", None, "ABC")
@@ -980,6 +991,52 @@ class AgbotsSuite extends FunSuite {
     assert(parsedBody.maxChangeId > 0)
   }
 
+  test("POST /orgs/"+orgid+"/changes - verify " + agbotId + " does not get wildcard case") {
+    val time = ApiTime.pastUTC(secondsAgo)
+    val input = ResourceChangesRequest(0, Some(time), maxRecords, Some(List("*")))
+    val response = Http(URL+"/changes").postData(write(input)).method("post").headers(CONTENT).headers(ACCEPT).headers(AGBOTAUTH).asString
+    info("code: "+response.code)
+    assert(response.code === HttpCode.POST_OK.intValue)
+    assert(!response.body.isEmpty)
+    val parsedBody = parse(response.body).extract[ResourceChangesRespObject]
+    assert(!parsedBody.changes.exists(y => {y.orgId == "IBM"}))
+  }
+
+  test("PUT /orgs/"+orgid+"/changes - with low maxRecords") {
+    if (runningLocally) {     // changing limits via POST /admin/config does not work in multi-node mode
+      // Get the current config value so we can restore it afterward
+      ExchConfig.load()
+      val origMaxRecords = ExchConfig.getInt("api.resourceChanges.maxRecordsCap")
+      val newMaxRecords = 1
+      // Change the maxNodes config value in the svr
+      var configInput = AdminConfigRequest("api.resourceChanges.maxRecordsCap", newMaxRecords.toString)
+      var response = Http(NOORGURL+"/admin/config").postData(write(configInput)).method("put").headers(CONTENT).headers(ACCEPT).headers(ROOTAUTH).asString
+      info("code: "+response.code+", response.body: "+response.body)
+      assert(response.code === HttpCode.PUT_OK.intValue)
+
+      // Now post to /changes and make sure the size is respected even though maxRecords sent in is much higher
+      // NOTE maxRecords the variable must be larger than newMaxRecords
+      val time = ApiTime.pastUTC(secondsAgo)
+      val input = ResourceChangesRequest(0, Some(time), maxRecords, Some(List(orgid)))
+      response = Http(URL+"/changes").postData(write(input)).method("post").headers(CONTENT).headers(ACCEPT).headers(AGBOTAUTH).asString
+      info("code: "+response.code)
+      assert(response.code === HttpCode.POST_OK.intValue)
+      assert(!response.body.isEmpty)
+      val parsedBody = parse(response.body).extract[ResourceChangesRespObject]
+      info("parsedBody.changes.size: " + parsedBody.changes.size + " maxRecords: " + newMaxRecords)
+      assert(parsedBody.changes.size <= newMaxRecords)
+      assert(parsedBody.hitMaxRecords)
+
+      // Restore the maxNodes config value in the svr
+      configInput = AdminConfigRequest("api.resourceChanges.maxRecordsCap", origMaxRecords.toString)
+      response = Http(NOORGURL+"/admin/config").postData(write(configInput)).method("put").headers(CONTENT).headers(ACCEPT).headers(ROOTAUTH).asString
+      info("code: "+response.code+", response.body: "+response.body)
+      assert(response.code === HttpCode.PUT_OK.intValue)
+      val origMaxRecords2 = ExchConfig.getInt("api.resourceChanges.maxRecordsCap")
+      assert(origMaxRecords == origMaxRecords2)
+    }
+  }
+
   /** Explicit delete of agbot */
   test("DELETE /orgs/"+orgid+"/agbots/"+agbotId+" - as user") {
     var response = Http(URL+"/agbots/"+agbotId).method("delete").headers(ACCEPT).headers(USERAUTH).asString
@@ -1002,6 +1059,58 @@ class AgbotsSuite extends FunSuite {
     assert(parsedBody.changes.exists(y => {(y.id == agbotId) && (y.operation == ResourceChangeConfig.DELETED) && (y.resource == "agbot")}))
   }
 
+  test("POST /orgs/IBM/changes - as IBM agbot (if it exists)") {
+    val ibmAgbotAuth = sys.env.getOrElse("EXCHANGE_AGBOTAUTH", "")
+    val ibmAgbotId = """^[^:]+""".r.findFirstIn(ibmAgbotAuth).getOrElse("")     // get the id before the :
+    info("ibmAgbotAuth="+ibmAgbotAuth+", ibmAgbotId="+ibmAgbotId+".")
+    if (ibmAgbotAuth != "") {
+      val IBMAGBOTAUTH = ("Authorization", "Basic " + ApiUtils.encode("IBM/" + ibmAgbotAuth))
+
+      // Notification Framework Tests
+      val time = ApiTime.pastUTC(secondsAgo)
+      var input = ResourceChangesRequest(0, Some(time), maxRecords, Some(List("*")))
+      var response = Http(urlRoot+"/v1/orgs/IBM/changes").postData(write(input)).method("post").headers(CONTENT).headers(ACCEPT).headers(IBMAGBOTAUTH).asString
+      info(urlRoot+"/v1/orgs/IBM/changes -- wildcard splat")
+      info("code: "+response.code)
+      assert(response.code === HttpCode.POST_OK.intValue)
+      assert(!response.body.isEmpty)
+      var parsedBody = parse(response.body).extract[ResourceChangesRespObject]
+      assert(parsedBody.changes.exists(y => {y.orgId == orgid}))
+      assert(parsedBody.changes.exists(y => {y.orgId == "IBM"}))
+
+      input = ResourceChangesRequest(0, Some(time), maxRecords, Some(List("")))
+      response = Http(urlRoot+"/v1/orgs/IBM/changes").postData(write(input)).method("post").headers(CONTENT).headers(ACCEPT).headers(IBMAGBOTAUTH).asString
+      info(urlRoot+"/v1/orgs/IBM/changes -- wildcard empty string")
+      info("code: "+response.code)
+      assert(response.code === HttpCode.POST_OK.intValue)
+      assert(!response.body.isEmpty)
+      parsedBody = parse(response.body).extract[ResourceChangesRespObject]
+      assert(parsedBody.changes.exists(y => {y.orgId == orgid}))
+      assert(parsedBody.changes.exists(y => {y.orgId == "IBM"}))
+
+      input = ResourceChangesRequest(0, Some(time), maxRecords, Some(List(orgid)))
+      response = Http(urlRoot+"/v1/orgs/IBM/changes").postData(write(input)).method("post").headers(CONTENT).headers(ACCEPT).headers(IBMAGBOTAUTH).asString
+      info(urlRoot+"/v1/orgs/IBM/changes -- orgList: ["+orgid+"]")
+      info("code: "+response.code)
+      assert(response.code === HttpCode.POST_OK.intValue)
+      assert(!response.body.isEmpty)
+      parsedBody = parse(response.body).extract[ResourceChangesRespObject]
+      assert(parsedBody.changes.exists(y => {y.orgId == orgid}))
+      assert(!parsedBody.changes.exists(y => {y.orgId == "IBM"}))
+
+      input = ResourceChangesRequest(0, Some(time), maxRecords, Some(List("IBM")))
+      response = Http(urlRoot+"/v1/orgs/IBM/changes").postData(write(input)).method("post").headers(CONTENT).headers(ACCEPT).headers(IBMAGBOTAUTH).asString
+      info(urlRoot+"/v1/orgs/IBM/changes -- orgList: [\"IBM\"]")
+      info("code: "+response.code)
+      assert(response.code === HttpCode.POST_OK.intValue)
+      assert(!response.body.isEmpty)
+      parsedBody = parse(response.body).extract[ResourceChangesRespObject]
+      assert(!parsedBody.changes.exists(y => {y.orgId == orgid}))
+      assert(parsedBody.changes.exists(y => {y.orgId == "IBM"}))
+
+    }
+  }
+
   // Note: testing of msgs is in NodesSuite.scala
 
   /** Clean up, delete all the test agbots */

src/test/scala/exchangeapi/BusinessSuite.scala

@@ -8,8 +8,8 @@ import org.json4s._
 import org.json4s.jackson.JsonMethods._
 import org.json4s.native.Serialization.write
 import org.junit.runner.RunWith
-import org.scalatest.FunSuite
-import org.scalatest.junit.JUnitRunner
+import org.scalatest.funsuite.AnyFunSuite
+import org.scalatestplus.junit.JUnitRunner
 import scalaj.http._
 
 import scala.collection.immutable._
@@ -23,7 +23,7 @@ import scala.collection.immutable._
  * clear and detailed tutorial of FunSuite: http://doc.scalatest.org/1.9.1/index.html#org.scalatest.FunSuite
  */
 @RunWith(classOf[JUnitRunner])
-class BusinessSuite extends FunSuite {
+class BusinessSuite extends AnyFunSuite {
 
   val localUrlRoot = "http://localhost:8080"
   val urlRoot = sys.env.getOrElse("EXCHANGE_URL_ROOT", localUrlRoot)

src/test/scala/exchangeapi/CatalogSuite.scala

@@ -6,16 +6,16 @@ import org.json4s._
 import org.json4s.jackson.JsonMethods._
 import org.json4s.native.Serialization.write
 import org.junit.runner.RunWith
-import org.scalatest.FunSuite
-import org.scalatest.junit.JUnitRunner
+import org.scalatest.funsuite.AnyFunSuite
+import org.scalatestplus.junit.JUnitRunner
 
 import scala.collection.immutable._
 import scalaj.http._
 
 // Tests the catalog APIs
 
 @RunWith(classOf[JUnitRunner])
-class CatalogSuite extends FunSuite {
+class CatalogSuite extends AnyFunSuite {
 
   implicit val formats = DefaultFormats // Brings in default date formats etc.
 

src/test/scala/exchangeapi/ExchConfigSuite.scala

@@ -1,16 +1,16 @@
 package exchangeapi
 
-import org.scalatest.FunSuite
+import org.scalatest.funsuite.AnyFunSuite
 
 import org.junit.runner.RunWith
-import org.scalatest.junit.JUnitRunner
+import org.scalatestplus.junit.JUnitRunner
 import com.horizon.exchangeapi._
 
 /**
  * Tests for the Version and VersionRange case classes
  */
 @RunWith(classOf[JUnitRunner])
-class ExchConfigSuite extends FunSuite {
+class ExchConfigSuite extends AnyFunSuite {
   test("ExchConfig tests") {
     ExchConfig.load()
     // Note: this test needs to work with the default version of config.json that is in src/main/resources (so that 'make test' in travis works)

src/test/scala/exchangeapi/NodesSuite.scala

@@ -9,8 +9,8 @@ import org.json4s._
 import org.json4s.jackson.JsonMethods._
 import org.json4s.native.Serialization.write
 import org.junit.runner.RunWith
-import org.scalatest.FunSuite
-import org.scalatest.junit.JUnitRunner
+import org.scalatest.funsuite.AnyFunSuite
+import org.scalatestplus.junit.JUnitRunner
 
 import scala.collection.immutable._
 import scalaj.http._
@@ -24,7 +24,7 @@ import scalaj.http._
  * clear and detailed tutorial of FunSuite: http://doc.scalatest.org/1.9.1/index.html#org.scalatest.FunSuite
  */
 @RunWith(classOf[JUnitRunner])
-class NodesSuite extends FunSuite {
+class NodesSuite extends AnyFunSuite {
 
   val localUrlRoot = "http://localhost:8080"
   val urlRoot = sys.env.getOrElse("EXCHANGE_URL_ROOT", localUrlRoot)