You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@openwhisk.apache.org by GitBox <gi...@apache.org> on 2018/04/19 00:51:42 UTC
[GitHub] rabbah closed pull request #3517: MemoryArtifactStore for unit
testing and ArtifactStore SPI Validation
rabbah closed pull request #3517: MemoryArtifactStore for unit testing and ArtifactStore SPI Validation
URL: https://github.com/apache/incubator-openwhisk/pull/3517
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git a/common/scala/src/main/scala/whisk/core/database/ArtifactStoreExceptions.scala b/common/scala/src/main/scala/whisk/core/database/ArtifactStoreExceptions.scala
index f8e7ec5f98..6ff97724b8 100644
--- a/common/scala/src/main/scala/whisk/core/database/ArtifactStoreExceptions.scala
+++ b/common/scala/src/main/scala/whisk/core/database/ArtifactStoreExceptions.scala
@@ -28,3 +28,9 @@ case class DocumentTypeMismatchException(message: String) extends ArtifactStoreE
case class DocumentUnreadable(message: String) extends ArtifactStoreException(message)
case class PutException(message: String) extends ArtifactStoreException(message)
+
+sealed abstract class ArtifactStoreRuntimeException(message: String) extends RuntimeException(message)
+
+case class UnsupportedQueryKeys(message: String) extends ArtifactStoreRuntimeException(message)
+
+case class UnsupportedView(message: String) extends ArtifactStoreRuntimeException(message)
diff --git a/common/scala/src/main/scala/whisk/core/database/CouchDbRestStore.scala b/common/scala/src/main/scala/whisk/core/database/CouchDbRestStore.scala
index d55db6505a..a2f63e1c8d 100644
--- a/common/scala/src/main/scala/whisk/core/database/CouchDbRestStore.scala
+++ b/common/scala/src/main/scala/whisk/core/database/CouchDbRestStore.scala
@@ -28,10 +28,9 @@ import akka.stream.scaladsl._
import akka.util.ByteString
import spray.json._
import whisk.common.{Logging, LoggingMarkers, MetricEmitter, TransactionId}
+import whisk.core.database.StoreUtils._
import whisk.core.entity.BulkEntityResult
import whisk.core.entity.DocInfo
-import whisk.core.entity.DocRevision
-import whisk.core.entity.WhiskDocument
import whisk.http.Messages
import whisk.core.entity.DocumentReader
@@ -224,27 +223,7 @@ class CouchDbRestStore[DocumentAbstraction <: DocumentSerializer](dbProtocol: St
e match {
case Right(response) =>
transid.finished(this, start, s"[GET] '$dbName' completed: found document '$doc'")
-
- val asFormat = try {
- docReader.read(ma, response)
- } catch {
- case e: Exception => jsonFormat.read(response)
- }
-
- if (asFormat.getClass != ma.runtimeClass) {
- throw DocumentTypeMismatchException(
- s"document type ${asFormat.getClass} did not match expected type ${ma.runtimeClass}.")
- }
-
- val deserialized = asFormat.asInstanceOf[A]
-
- val responseRev = response.fields("_rev").convertTo[String]
- assert(doc.rev.rev == null || doc.rev.rev == responseRev, "Returned revision should match original argument")
- // FIXME remove mutability from appropriate classes now that it is no longer required by GSON.
- deserialized.asInstanceOf[WhiskDocument].revision(DocRevision(responseRev))
-
- deserialized
-
+ deserialize[A, DocumentAbstraction](doc, response)
case Left(StatusCodes.NotFound) =>
transid.finished(this, start, s"[GET] '$dbName', document: '${doc}'; not found.")
// for compatibility
diff --git a/common/scala/src/main/scala/whisk/core/database/DocumentHandler.scala b/common/scala/src/main/scala/whisk/core/database/DocumentHandler.scala
new file mode 100644
index 0000000000..393cf63d8b
--- /dev/null
+++ b/common/scala/src/main/scala/whisk/core/database/DocumentHandler.scala
@@ -0,0 +1,463 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package whisk.core.database
+
+import spray.json._
+import spray.json.DefaultJsonProtocol._
+import whisk.common.TransactionId
+import whisk.core.entity.{DocId, UserLimits}
+import whisk.core.entity.EntityPath.PATHSEP
+import whisk.utils.JsHelpers
+
+import scala.collection.immutable.Seq
+import scala.concurrent.Future
+import scala.concurrent.ExecutionContext
+
+/**
+ * Simple abstraction allow accessing a document just by _id. This would be used
+ * to perform queries related to join support
+ */
+trait DocumentProvider {
+ protected[database] def get(id: DocId)(implicit transid: TransactionId): Future[Option[JsObject]]
+}
+
+trait DocumentHandler {
+
+ /**
+ * Returns a JsObject having computed fields. This is a substitution for fields
+ * computed in CouchDB views
+ */
+ def computedFields(js: JsObject): JsObject = JsObject.empty
+
+ /**
+ * Returns the set of field names (including sub document field) which needs to be fetched as part of
+ * query made for the given view.
+ */
+ def fieldsRequiredForView(ddoc: String, view: String): Set[String] = Set.empty
+
+ /**
+ * Transforms the query result instance from artifact store as per view requirements. Some view computation
+ * may result in performing a join operation.
+ *
+ * If the passed instance does not confirm to view conditions that transformed result would be None. This
+ * would be the case if view condition cannot be completed handled in query made to artifact store
+ */
+ def transformViewResult(
+ ddoc: String,
+ view: String,
+ startKey: List[Any],
+ endKey: List[Any],
+ includeDocs: Boolean,
+ js: JsObject,
+ provider: DocumentProvider)(implicit transid: TransactionId, ec: ExecutionContext): Future[Seq[JsObject]]
+
+ /**
+ * Determines if the complete document should be fetched even if `includeDocs` is set to true. For some view computation
+ * complete document (including sub documents) may be needed and for them its required that complete document must be
+ * fetched as part of query response
+ */
+ def shouldAlwaysIncludeDocs(ddoc: String, view: String): Boolean = false
+
+ def checkIfTableSupported(table: String): Unit = {
+ if (!supportedTables.contains(table)) {
+ throw UnsupportedView(table)
+ }
+ }
+
+ protected def supportedTables: Set[String]
+}
+
+/**
+ * Base class for handlers which do not perform joins for computing views
+ */
+abstract class SimpleHandler extends DocumentHandler {
+ override def transformViewResult(
+ ddoc: String,
+ view: String,
+ startKey: List[Any],
+ endKey: List[Any],
+ includeDocs: Boolean,
+ js: JsObject,
+ provider: DocumentProvider)(implicit transid: TransactionId, ec: ExecutionContext): Future[Seq[JsObject]] = {
+ //Query result from CouchDB have below object structure with actual result in `value` key
+ //So transform the result to confirm to that structure
+ val viewResult = JsObject(
+ "id" -> js.fields("_id"),
+ "key" -> createKey(ddoc, view, startKey, js),
+ "value" -> computeView(ddoc, view, js))
+
+ val result = if (includeDocs) JsObject(viewResult.fields + ("doc" -> js)) else viewResult
+ Future.successful(Seq(result))
+ }
+
+ /**
+ * Computes the view as per viewName. Its passed either the projected object or actual
+ * object
+ */
+ def computeView(ddoc: String, view: String, js: JsObject): JsObject
+
+ /**
+ * Key is an array which matches the view query key
+ */
+ protected def createKey(ddoc: String, view: String, startKey: List[Any], js: JsObject): JsArray
+}
+
+object ActivationHandler extends SimpleHandler {
+ val NS_PATH = "nspath"
+ private val commonFields =
+ Set("namespace", "name", "version", "publish", "annotations", "activationId", "start", "cause")
+ private val fieldsForView = commonFields ++ Seq("end", "response.statusCode")
+
+ protected val supportedTables =
+ Set("activations/byDate", "whisks-filters.v2.1.0/activations", "whisks.v2.1.0/activations")
+
+ override def computedFields(js: JsObject): JsObject = {
+ val path = js.fields.get("namespace") match {
+ case Some(JsString(namespace)) => JsString(namespace + PATHSEP + pathFilter(js))
+ case _ => JsNull
+ }
+ val deleteLogs = annotationValue(js, "kind", { v =>
+ v.convertTo[String] != "sequence"
+ }, true)
+ dropNull((NS_PATH, path), ("deleteLogs", JsBoolean(deleteLogs)))
+ }
+
+ override def fieldsRequiredForView(ddoc: String, view: String): Set[String] = view match {
+ case "activations" => fieldsForView
+ case _ => throw UnsupportedView(s"$ddoc/$view")
+ }
+
+ def computeView(ddoc: String, view: String, js: JsObject): JsObject = view match {
+ case "activations" => computeActivationView(js)
+ case _ => throw UnsupportedView(s"$ddoc/$view")
+ }
+
+ def createKey(ddoc: String, view: String, startKey: List[Any], js: JsObject): JsArray = {
+ startKey match {
+ case (ns: String) :: Nil => JsArray(Vector(JsString(ns)))
+ case (ns: String) :: _ :: Nil => JsArray(Vector(JsString(ns), js.fields("start")))
+ case _ => throw UnsupportedQueryKeys("$ddoc/$view -> ($startKey, $endKey)")
+ }
+ }
+
+ private def computeActivationView(js: JsObject): JsObject = {
+ val common = js.fields.filterKeys(commonFields)
+
+ val (endTime, duration) = js.getFields("end", "start") match {
+ case Seq(JsNumber(end), JsNumber(start)) if end != 0 => (JsNumber(end), JsNumber(end - start))
+ case _ => (JsNull, JsNull)
+ }
+
+ val statusCode = JsHelpers.getFieldPath(js, "response", "statusCode").getOrElse(JsNull)
+
+ val result = common + ("end" -> endTime) + ("duration" -> duration) + ("statusCode" -> statusCode)
+ JsObject(result.filter(_._2 != JsNull))
+ }
+
+ protected[database] def pathFilter(js: JsObject): String = {
+ val name = js.fields("name").convertTo[String]
+ annotationValue(js, "path", { v =>
+ val p = v.convertTo[String].split(PATHSEP)
+ if (p.length == 3) p(1) + PATHSEP + name else name
+ }, name)
+ }
+
+ /**
+ * Finds and transforms annotation with matching key.
+ *
+ * @param js js object having annotations array
+ * @param key annotation key
+ * @param vtr transformer function to map annotation value
+ * @param default default value to use if no matching annotation found
+ * @return annotation value matching given key
+ */
+ protected[database] def annotationValue[T](js: JsObject, key: String, vtr: JsValue => T, default: T): T = {
+ js.fields.get("annotations") match {
+ case Some(JsArray(e)) =>
+ e.view
+ .map(_.asJsObject.getFields("key", "value"))
+ .collectFirst {
+ case Seq(JsString(`key`), v: JsValue) => vtr(v) //match annotation with given key
+ }
+ .getOrElse(default)
+ case _ => default
+ }
+ }
+
+ private def dropNull(fields: JsField*) = JsObject(fields.filter(_._2 != JsNull): _*)
+}
+
+object WhisksHandler extends SimpleHandler {
+ val ROOT_NS = "rootns"
+ private val commonFields = Set("namespace", "name", "version", "publish", "annotations", "updated")
+ private val actionFields = commonFields ++ Set("limits", "exec.binary")
+ private val packageFields = commonFields ++ Set("binding")
+ private val packagePublicFields = commonFields
+ private val ruleFields = commonFields
+ private val triggerFields = commonFields
+
+ protected val supportedTables = Set(
+ "whisks.v2.1.0/actions",
+ "whisks.v2.1.0/packages",
+ "whisks.v2.1.0/packages-public",
+ "whisks.v2.1.0/rules",
+ "whisks.v2.1.0/triggers")
+
+ override def computedFields(js: JsObject): JsObject = {
+ js.fields.get("namespace") match {
+ case Some(JsString(namespace)) =>
+ val ns = namespace.split(PATHSEP)
+ val rootNS = if (ns.length > 1) ns(0) else namespace
+ JsObject((ROOT_NS, JsString(rootNS)))
+ case _ => JsObject.empty
+ }
+ }
+
+ override def fieldsRequiredForView(ddoc: String, view: String): Set[String] = view match {
+ case "actions" => actionFields
+ case "packages" => packageFields
+ case "packages-public" => packagePublicFields
+ case "rules" => ruleFields
+ case "triggers" => triggerFields
+ case _ => throw UnsupportedView(s"$ddoc/$view")
+ }
+
+ def computeView(ddoc: String, view: String, js: JsObject): JsObject = view match {
+ case "actions" => computeActionView(js)
+ case "packages" => computePackageView(js)
+ case "packages-public" => computePublicPackageView(js)
+ case "rules" => computeRulesView(js)
+ case "triggers" => computeTriggersView(js)
+ case _ => throw UnsupportedView(s"$ddoc/$view")
+ }
+
+ def createKey(ddoc: String, view: String, startKey: List[Any], js: JsObject): JsArray = {
+ startKey match {
+ case (ns: String) :: Nil => JsArray(Vector(JsString(ns)))
+ case (ns: String) :: _ :: Nil => JsArray(Vector(JsString(ns), js.fields("updated")))
+ case _ => throw UnsupportedQueryKeys("$ddoc/$view -> ($startKey, $endKey)")
+ }
+ }
+
+ def getEntityTypeForDesignDoc(ddoc: String, view: String): String = view match {
+ case "actions" => "action"
+ case "rules" => "rule"
+ case "triggers" => "trigger"
+ case "packages" | "packages-public" => "package"
+ case _ => throw UnsupportedView(s"$ddoc/$view")
+ }
+
+ private def computeTriggersView(js: JsObject): JsObject = {
+ JsObject(js.fields.filterKeys(commonFields))
+ }
+
+ private def computePublicPackageView(js: JsObject): JsObject = {
+ JsObject(js.fields.filterKeys(commonFields) + ("binding" -> JsFalse))
+ }
+
+ private def computeRulesView(js: JsObject) = {
+ JsObject(js.fields.filterKeys(ruleFields))
+ }
+
+ private def computePackageView(js: JsObject): JsObject = {
+ val common = js.fields.filterKeys(commonFields)
+ val binding = js.fields.get("binding") match {
+ case Some(x: JsObject) if x.fields.nonEmpty => x
+ case _ => JsFalse
+ }
+ JsObject(common + ("binding" -> binding))
+ }
+
+ private def computeActionView(js: JsObject): JsObject = {
+ val base = js.fields.filterKeys(commonFields ++ Set("limits"))
+ val exec_binary = JsHelpers.getFieldPath(js, "exec", "binary")
+ JsObject(base + ("exec" -> JsObject("binary" -> exec_binary.getOrElse(JsFalse))))
+ }
+}
+
+object SubjectHandler extends DocumentHandler {
+
+ protected val supportedTables = Set("subjects/identities", "namespaceThrottlings/blockedNamespaces")
+
+ override def shouldAlwaysIncludeDocs(ddoc: String, view: String): Boolean = {
+ (ddoc, view) match {
+ case ("subjects", "identities") => true
+ case ("namespaceThrottlings", "blockedNamespaces") => true
+ case _ => throw UnsupportedView(s"$ddoc/$view")
+ }
+ }
+
+ override def transformViewResult(
+ ddoc: String,
+ view: String,
+ startKey: List[Any],
+ endKey: List[Any],
+ includeDocs: Boolean,
+ js: JsObject,
+ provider: DocumentProvider)(implicit transid: TransactionId, ec: ExecutionContext): Future[Seq[JsObject]] = {
+
+ val result = (ddoc, view) match {
+ case ("subjects", "identities") =>
+ require(includeDocs) //For subject/identities includeDocs is always true
+ computeSubjectView(startKey, js, provider)
+ case ("namespaceThrottlings", "blockedNamespaces") =>
+ Future.successful(computeBlacklistedNamespaces(js))
+ case _ =>
+ throw UnsupportedView(s"$ddoc/$view")
+ }
+ result
+ }
+
+ /**
+ * {{{
+ * function (doc) {
+ * if (doc._id.indexOf("/limits") >= 0) {
+ * if (doc.concurrentInvocations === 0 || doc.invocationsPerMinute === 0) {
+ * var namespace = doc._id.replace("/limits", "");
+ * emit(namespace, 1);
+ * }
+ * } else if (doc.subject && doc.namespaces && doc.blocked) {
+ * doc.namespaces.forEach(function(namespace) {
+ * emit(namespace.name, 1);
+ * });
+ * }
+ * }
+ * }}}
+ */
+ private def computeBlacklistedNamespaces(js: JsObject): Seq[JsObject] = {
+ val id = js.fields("_id")
+ val value = JsNumber(1)
+ id match {
+ case JsString(idv) if idv.endsWith("/limits") =>
+ val limits = UserLimits.serdes.read(js)
+ if (limits.concurrentInvocations.contains(0) || limits.invocationsPerMinute.contains(0)) {
+ val ns = idv.substring(0, idv.indexOf("/limits"))
+ Seq(JsObject("id" -> id, "key" -> JsString(ns), "value" -> value))
+ } else Seq.empty
+ case _ =>
+ js.getFields("subject", "namespaces", "blocked") match {
+ case Seq(_, namespaces: JsArray, JsTrue) =>
+ namespaces.elements.map { ns =>
+ val name = ns.asJsObject.fields("name")
+ JsObject("id" -> id, "key" -> name, "value" -> value)
+ }
+ case _ =>
+ Seq.empty
+ }
+ }
+ }
+
+ private def computeSubjectView(startKey: List[Any], js: JsObject, provider: DocumentProvider)(
+ implicit transid: TransactionId,
+ ec: ExecutionContext) = {
+ val subjectOpt = findMatchingSubject(startKey, js)
+ val result = subjectOpt match {
+ case Some(subject) =>
+ val limitDocId = s"${subject.namespace}/limits"
+ val viewJS = JsObject(
+ "_id" -> JsString(limitDocId),
+ "namespace" -> JsString(subject.namespace),
+ "uuid" -> JsString(subject.uuid),
+ "key" -> JsString(subject.key))
+ val result =
+ JsObject("id" -> js.fields("_id"), "key" -> createKey(startKey), "value" -> viewJS, "doc" -> JsNull)
+ if (subject.matchInNamespace) {
+ provider
+ .get(DocId(limitDocId))
+ .map(limits => Seq(JsObject(result.fields + ("doc" -> limits.getOrElse(JsNull)))))
+ } else {
+ Future.successful(Seq(result))
+ }
+ case None =>
+ Future.successful(Seq.empty)
+ }
+ result
+ }
+
+ def findMatchingSubject(startKey: List[Any], js: JsObject): Option[SubjectView] = {
+ startKey match {
+ case (ns: String) :: Nil => findMatchingSubject(js, s => s.namespace == ns && !s.blocked)
+ case (uuid: String) :: (key: String) :: Nil =>
+ findMatchingSubject(js, s => s.uuid == uuid && s.key == key && !s.blocked)
+ case _ => None
+ }
+ }
+
+ private def createKey(startKey: List[Any]): JsArray = {
+ startKey match {
+ case (ns: String) :: Nil => JsArray(Vector(JsString(ns))) //namespace or subject
+ case (uuid: String) :: (key: String) :: Nil => JsArray(Vector(JsString(uuid), JsString(key))) // uuid, key
+ case _ => throw UnsupportedQueryKeys("$ddoc/$view -> ($startKey, $endKey)")
+ }
+ }
+
+ /**
+ * Computes the view as per logic below from (identities/subject) view
+ *
+ * {{{
+ * function (doc) {
+ * if(doc.uuid && doc.key && !doc.blocked) {
+ * var v = {namespace: doc.subject, uuid: doc.uuid, key: doc.key};
+ * emit([doc.subject], v);
+ * emit([doc.uuid, doc.key], v);
+ * }
+ * if(doc.namespaces && !doc.blocked) {
+ * doc.namespaces.forEach(function(namespace) {
+ * var v = {_id: namespace.name + '/limits', namespace: namespace.name, uuid: namespace.uuid, key: namespace.key};
+ * emit([namespace.name], v);
+ * emit([namespace.uuid, namespace.key], v);
+ * });
+ * }
+ * }
+ * }}}
+ *
+ * @param js subject json from db
+ * @param matches match predicate
+ */
+ private def findMatchingSubject(js: JsObject, matches: SubjectView => Boolean): Option[SubjectView] = {
+ val blocked = js.fields.get("blocked") match {
+ case Some(JsTrue) => true
+ case _ => false
+ }
+
+ val r = js.getFields("subject", "uuid", "key") match {
+ case Seq(JsString(ns), JsString(uuid), JsString(key)) => Some(SubjectView(ns, uuid, key, blocked)).filter(matches)
+ case _ => None
+ }
+
+ r.orElse {
+ val namespaces = js.fields.get("namespaces") match {
+ case Some(JsArray(e)) =>
+ e.map(_.asJsObject.getFields("name", "uuid", "key") match {
+ case Seq(JsString(ns), JsString(uuid), JsString(key)) =>
+ Some(SubjectView(ns, uuid, key, blocked, matchInNamespace = true))
+ case _ => None
+ })
+
+ case _ => Seq()
+ }
+ namespaces.flatMap(_.filter(matches)).headOption
+ }
+ }
+
+ case class SubjectView(namespace: String,
+ uuid: String,
+ key: String,
+ blocked: Boolean = false,
+ matchInNamespace: Boolean = false)
+}
diff --git a/common/scala/src/main/scala/whisk/core/database/StoreUtils.scala b/common/scala/src/main/scala/whisk/core/database/StoreUtils.scala
new file mode 100644
index 0000000000..cff9dd9f37
--- /dev/null
+++ b/common/scala/src/main/scala/whisk/core/database/StoreUtils.scala
@@ -0,0 +1,68 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package whisk.core.database
+
+import akka.event.Logging.ErrorLevel
+import spray.json.{JsObject, RootJsonFormat}
+import whisk.common.{Logging, StartMarker, TransactionId}
+import spray.json.DefaultJsonProtocol._
+import whisk.core.entity.{DocInfo, DocRevision, DocumentReader, WhiskDocument}
+
+import scala.concurrent.{ExecutionContext, Future}
+
+private[database] object StoreUtils {
+
+ def reportFailure[T](f: Future[T], start: StartMarker, failureMessage: Throwable => String)(
+ implicit transid: TransactionId,
+ logging: Logging,
+ ec: ExecutionContext): Future[T] = {
+ f.onFailure({
+ case _: ArtifactStoreException => // These failures are intentional and shouldn't trigger the catcher.
+ case x => transid.failed(this, start, failureMessage(x), ErrorLevel)
+ })
+ f
+ }
+
+ def checkDocHasRevision(doc: DocInfo): Unit = {
+ require(doc != null, "doc undefined")
+ require(doc.rev.rev != null, "doc revision must be specified")
+ }
+
+ def deserialize[A <: DocumentAbstraction, DocumentAbstraction](doc: DocInfo, js: JsObject)(
+ implicit docReader: DocumentReader,
+ ma: Manifest[A],
+ jsonFormat: RootJsonFormat[DocumentAbstraction]): A = {
+ val asFormat = try {
+ docReader.read(ma, js)
+ } catch {
+ case _: Exception => jsonFormat.read(js)
+ }
+
+ if (asFormat.getClass != ma.runtimeClass) {
+ throw DocumentTypeMismatchException(
+ s"document type ${asFormat.getClass} did not match expected type ${ma.runtimeClass}.")
+ }
+
+ val deserialized = asFormat.asInstanceOf[A]
+
+ val responseRev = js.fields("_rev").convertTo[String]
+ assert(doc.rev.rev == null || doc.rev.rev == responseRev, "Returned revision should match original argument")
+ // FIXME remove mutability from appropriate classes now that it is no longer required by GSON.
+ deserialized.asInstanceOf[WhiskDocument].revision(DocRevision(responseRev))
+ }
+}
diff --git a/common/scala/src/main/scala/whisk/core/database/memory/MemoryArtifactStore.scala b/common/scala/src/main/scala/whisk/core/database/memory/MemoryArtifactStore.scala
new file mode 100644
index 0000000000..1f4b2b3426
--- /dev/null
+++ b/common/scala/src/main/scala/whisk/core/database/memory/MemoryArtifactStore.scala
@@ -0,0 +1,351 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package whisk.core.database.memory
+
+import akka.actor.ActorSystem
+import akka.http.scaladsl.model.ContentType
+import akka.stream.ActorMaterializer
+import akka.stream.scaladsl.{Keep, Sink, Source}
+import akka.util.{ByteString, ByteStringBuilder}
+import spray.json.{DefaultJsonProtocol, DeserializationException, JsObject, JsString, RootJsonFormat}
+import whisk.common.{Logging, LoggingMarkers, TransactionId}
+import whisk.core.database.StoreUtils._
+import whisk.core.database._
+import whisk.core.entity._
+import whisk.http.Messages
+
+import scala.collection.concurrent.TrieMap
+import scala.concurrent.{ExecutionContext, Future}
+import scala.reflect.ClassTag
+import scala.util.Try
+
+object MemoryArtifactStoreProvider extends ArtifactStoreProvider {
+ override def makeStore[D <: DocumentSerializer: ClassTag](useBatching: Boolean)(
+ implicit jsonFormat: RootJsonFormat[D],
+ docReader: DocumentReader,
+ actorSystem: ActorSystem,
+ logging: Logging,
+ materializer: ActorMaterializer): ArtifactStore[D] = {
+
+ val classTag = implicitly[ClassTag[D]]
+ val (dbName, handler, viewMapper) = handlerAndMapper(classTag)
+
+ new MemoryArtifactStore(dbName, handler, viewMapper)
+ }
+
+ private def handlerAndMapper[D](entityType: ClassTag[D])(
+ implicit actorSystem: ActorSystem,
+ logging: Logging,
+ materializer: ActorMaterializer): (String, DocumentHandler, MemoryViewMapper) = {
+ entityType.runtimeClass match {
+ case x if x == classOf[WhiskEntity] =>
+ ("whisks", WhisksHandler, WhisksViewMapper)
+ case x if x == classOf[WhiskActivation] =>
+ ("activations", ActivationHandler, ActivationViewMapper)
+ case x if x == classOf[WhiskAuth] =>
+ ("subjects", SubjectHandler, SubjectViewMapper)
+ }
+ }
+}
+
+/**
+ * In-memory ArtifactStore implementation to enable test setups without requiring a running CouchDB instance
+ * It also serves as a canonical example of how an ArtifactStore can implemented with all the support for CRUD
+ * operations and Queries etc
+ */
+class MemoryArtifactStore[DocumentAbstraction <: DocumentSerializer](dbName: String,
+ documentHandler: DocumentHandler,
+ viewMapper: MemoryViewMapper)(
+ implicit system: ActorSystem,
+ val logging: Logging,
+ jsonFormat: RootJsonFormat[DocumentAbstraction],
+ materializer: ActorMaterializer,
+ docReader: DocumentReader)
+ extends ArtifactStore[DocumentAbstraction]
+ with DefaultJsonProtocol
+ with DocumentProvider {
+
+ override protected[core] implicit val executionContext: ExecutionContext = system.dispatcher
+
+ private val artifacts = new TrieMap[String, Artifact]
+
+ private val _id = "_id"
+ private val _rev = "_rev"
+
+ override protected[database] def put(d: DocumentAbstraction)(implicit transid: TransactionId): Future[DocInfo] = {
+ val asJson = d.toDocumentRecord
+
+ val id = asJson.fields(_id).convertTo[String].trim
+ require(!id.isEmpty, "document id must be defined")
+
+ val rev: Int = getRevision(asJson)
+ val docinfoStr = s"id: $id, rev: $rev"
+ val start = transid.started(this, LoggingMarkers.DATABASE_SAVE, s"[PUT] '$dbName' saving document: '$docinfoStr'")
+
+ val existing = Artifact(id, rev, asJson)
+ val updated = existing.incrementRev()
+ val t = Try[DocInfo] {
+ if (rev == 0) {
+ artifacts.putIfAbsent(id, updated) match {
+ case Some(_) => throw DocumentConflictException("conflict on 'put'")
+ case None => updated.docInfo
+ }
+ } else if (artifacts.replace(id, existing, updated)) {
+ updated.docInfo
+ } else {
+ throw DocumentConflictException("conflict on 'put'")
+ }
+ }
+
+ val f = Future.fromTry(t)
+
+ f.onFailure({
+ case _: DocumentConflictException =>
+ transid.finished(this, start, s"[PUT] '$dbName', document: '$docinfoStr'; conflict.")
+ })
+
+ f.onSuccess({
+ case _ => transid.finished(this, start, s"[PUT] '$dbName' completed document: '$docinfoStr'")
+ })
+
+ reportFailure(f, start, failure => s"[PUT] '$dbName' internal error, failure: '${failure.getMessage}'")
+ }
+
+ override protected[database] def del(doc: DocInfo)(implicit transid: TransactionId): Future[Boolean] = {
+ checkDocHasRevision(doc)
+
+ val start = transid.started(this, LoggingMarkers.DATABASE_DELETE, s"[DEL] '$dbName' deleting document: '$doc'")
+ val t = Try[Boolean] {
+ if (artifacts.remove(doc.id.id, Artifact(doc))) {
+ transid.finished(this, start, s"[DEL] '$dbName' completed document: '$doc'")
+ true
+ } else if (artifacts.contains(doc.id.id)) {
+ //Indicates that document exist but revision does not match
+ transid.finished(this, start, s"[DEL] '$dbName', document: '${doc}'; conflict.")
+ throw DocumentConflictException("conflict on 'delete'")
+ } else {
+ transid.finished(this, start, s"[DEL] '$dbName', document: '$doc'; not found.")
+ // for compatibility
+ throw NoDocumentException("not found on 'delete'")
+ }
+ }
+
+ val f = Future.fromTry(t)
+
+ reportFailure(f, start, failure => s"[DEL] '$dbName' internal error, doc: '$doc', failure: '${failure.getMessage}'")
+ }
+
+ override protected[database] def get[A <: DocumentAbstraction](doc: DocInfo)(implicit transid: TransactionId,
+ ma: Manifest[A]): Future[A] = {
+ val start = transid.started(this, LoggingMarkers.DATABASE_GET, s"[GET] '$dbName' finding document: '$doc'")
+
+ require(doc != null, "doc undefined")
+
+ val t = Try[A] {
+ artifacts.get(doc.id.id) match {
+ case Some(a) =>
+ //Revision matching is enforced in deserilization logic
+ transid.finished(this, start, s"[GET] '$dbName' completed: found document '$doc'")
+ deserialize[A, DocumentAbstraction](doc, a.doc)
+ case _ =>
+ transid.finished(this, start, s"[GET] '$dbName', document: '$doc'; not found.")
+ // for compatibility
+ throw NoDocumentException("not found on 'get'")
+ }
+ }
+
+ val f = Future.fromTry(t).recoverWith {
+ case _: DeserializationException => throw DocumentUnreadable(Messages.corruptedEntity)
+ }
+
+ reportFailure(f, start, failure => s"[GET] '$dbName' internal error, doc: '$doc', failure: '${failure.getMessage}'")
+ }
+
+ override protected[core] def query(table: String,
+ startKey: List[Any],
+ endKey: List[Any],
+ skip: Int,
+ limit: Int,
+ includeDocs: Boolean,
+ descending: Boolean,
+ reduce: Boolean,
+ stale: StaleParameter)(implicit transid: TransactionId): Future[List[JsObject]] = {
+ require(!(reduce && includeDocs), "reduce and includeDocs cannot both be true")
+ require(!reduce, "Reduce scenario not supported") //TODO Investigate reduce
+ documentHandler.checkIfTableSupported(table)
+
+ val Array(ddoc, viewName) = table.split("/")
+
+ val start = transid.started(this, LoggingMarkers.DATABASE_QUERY, s"[QUERY] '$dbName' searching '$table")
+
+ val s = artifacts.toStream
+ .map(_._2)
+ .filter(a => viewMapper.filter(ddoc, viewName, startKey, endKey, a.doc, a.computed))
+ .map(_.doc)
+ .toList
+
+ val sorted = viewMapper.sort(ddoc, viewName, descending, s)
+
+ val out = if (limit > 0) sorted.slice(skip, skip + limit) else sorted.drop(skip)
+
+ val realIncludeDocs = includeDocs | documentHandler.shouldAlwaysIncludeDocs(ddoc, viewName)
+
+ val r = out.map { js =>
+ documentHandler.transformViewResult(
+ ddoc,
+ viewName,
+ startKey,
+ endKey,
+ realIncludeDocs,
+ js,
+ MemoryArtifactStore.this)
+ }.toList
+
+ val f = Future.sequence(r).map(_.flatten)
+ f.onSuccess({
+ case _ => transid.finished(this, start, s"[QUERY] '$dbName' completed: matched ${out.size}")
+ })
+ reportFailure(f, start, failure => s"[QUERY] '$dbName' internal error, failure: '${failure.getMessage}'")
+
+ }
+
+ override protected[core] def count(table: String,
+ startKey: List[Any],
+ endKey: List[Any],
+ skip: Int,
+ stale: StaleParameter)(implicit transid: TransactionId): Future[Long] = {
+ val f =
+ query(table, startKey, endKey, skip, limit = 0, includeDocs = false, descending = true, reduce = false, stale)
+ f.map(_.size)
+ }
+
+ override protected[core] def readAttachment[T](doc: DocInfo, name: String, sink: Sink[ByteString, Future[T]])(
+ implicit transid: TransactionId): Future[(ContentType, T)] = {
+ //TODO Temporary implementation till MemoryAttachmentStore PR is merged
+ artifacts.get(doc.id.id) match {
+ case Some(a: Artifact) if a.attachments.contains(name) =>
+ val attachment = a.attachments(name)
+ val r = Source.single(attachment.bytes).toMat(sink)(Keep.right).run
+ r.map(t => (attachment.contentType, t))
+ case None =>
+ Future.failed(NoDocumentException("Not found on 'readAttachment'."))
+ }
+ }
+
+ override protected[core] def deleteAttachments[T](doc: DocInfo)(implicit transid: TransactionId): Future[Boolean] = {
+ Future.successful(true)
+ }
+
+ override protected[core] def attach(
+ doc: DocInfo,
+ name: String,
+ contentType: ContentType,
+ docStream: Source[ByteString, _])(implicit transid: TransactionId): Future[DocInfo] = {
+
+ //TODO Temporary implementation till MemoryAttachmentStore PR is merged
+ val f = docStream.runFold(new ByteStringBuilder)((builder, b) => builder ++= b)
+ val g = f
+ .map { b =>
+ artifacts.get(doc.id.id) match {
+ case Some(a) =>
+ val existing = Artifact(doc, a.doc, a.computed)
+ val updated = existing.attach(name, Attachment(b.result().compact, contentType))
+ if (artifacts.replace(doc.id.id, existing, updated)) {
+ updated.docInfo
+ } else {
+ throw DocumentConflictException("conflict on 'put'")
+ }
+ case None =>
+ throw DocumentConflictException("conflict on 'put'")
+ }
+ }
+ g
+ }
+
+ override def shutdown(): Unit = {
+ artifacts.clear()
+ }
+
+ override protected[database] def get(id: DocId)(implicit transid: TransactionId): Future[Option[JsObject]] = {
+ val start = transid.started(this, LoggingMarkers.DATABASE_GET, s"[GET] '$dbName' finding document: '$id'")
+
+ val t = Try {
+ artifacts.get(id.id) match {
+ case Some(a) =>
+ transid.finished(this, start, s"[GET] '$dbName' completed: found document '$id'")
+ Some(a.doc)
+ case _ =>
+ transid.finished(this, start, s"[GET] '$dbName', document: '$id'; not found.")
+ None
+ }
+ }
+
+ val f = Future.fromTry(t)
+
+ reportFailure(f, start, failure => s"[GET] '$dbName' internal error, doc: '$id', failure: '${failure.getMessage}'")
+ }
+
+ private def getRevision(asJson: JsObject) = {
+ asJson.fields.get(_rev) match {
+ case Some(JsString(r)) => r.toInt
+ case _ => 0
+ }
+ }
+
+ //Use curried case class to allow equals support only for id and rev
+ //This allows us to implement atomic replace and remove which check
+ //for id,rev equality only
+ private case class Artifact(id: String, rev: Int)(val doc: JsObject,
+ val computed: JsObject,
+ val attachments: Map[String, Attachment] = Map.empty) {
+ def incrementRev(): Artifact = {
+ val (newRev, updatedDoc) = incrementAndGet()
+ copy(rev = newRev)(updatedDoc, computed, Map.empty) //With Couch attachments are lost post update
+ }
+
+ def docInfo = DocInfo(DocId(id), DocRevision(rev.toString))
+
+ def attach(name: String, attachment: Attachment): Artifact = {
+ val (newRev, updatedDoc) = incrementAndGet()
+ copy(rev = newRev)(updatedDoc, computed, attachments + (name -> attachment))
+ }
+
+ private def incrementAndGet() = {
+ val newRev = rev + 1
+ val updatedDoc = JsObject(doc.fields + (_rev -> JsString(newRev.toString)))
+ (newRev, updatedDoc)
+ }
+ }
+
+ private case class Attachment(bytes: ByteString, contentType: ContentType)
+
+ private object Artifact {
+ def apply(id: String, rev: Int, doc: JsObject): Artifact = {
+ Artifact(id, rev)(doc, documentHandler.computedFields(doc))
+ }
+
+ def apply(info: DocInfo): Artifact = {
+ Artifact(info.id.id, info.rev.rev.toInt)(JsObject.empty, JsObject.empty)
+ }
+
+ def apply(info: DocInfo, doc: JsObject, c: JsObject): Artifact = {
+ Artifact(info.id.id, info.rev.rev.toInt)(doc, c)
+ }
+ }
+
+}
diff --git a/common/scala/src/main/scala/whisk/core/database/memory/MemoryViewMapper.scala b/common/scala/src/main/scala/whisk/core/database/memory/MemoryViewMapper.scala
new file mode 100644
index 0000000000..c6dcbbe35b
--- /dev/null
+++ b/common/scala/src/main/scala/whisk/core/database/memory/MemoryViewMapper.scala
@@ -0,0 +1,253 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package whisk.core.database.memory
+
+import spray.json.{JsArray, JsBoolean, JsNumber, JsObject, JsString, JsTrue}
+import whisk.core.database.{ActivationHandler, UnsupportedQueryKeys, UnsupportedView, WhisksHandler}
+import whisk.core.entity.{UserLimits, WhiskEntityQueries}
+import whisk.utils.JsHelpers
+
+/**
+ * Maps the CouchDB view logic to expressed in javascript to Scala logic so as to enable
+ * performing queries by {{{MemoryArtifactStore}}}. Also serves as an example of what all query usecases
+ * are to be supported by any {{{ArtifactStore}}} implementation
+ */
+trait MemoryViewMapper {
+ protected val TOP: String = WhiskEntityQueries.TOP
+
+ def filter(ddoc: String, view: String, startKey: List[Any], endKey: List[Any], d: JsObject, c: JsObject): Boolean
+
+ def sort(ddoc: String, view: String, descending: Boolean, s: Seq[JsObject]): Seq[JsObject]
+
+ protected def checkKeys(startKey: List[Any], endKey: List[Any]): Unit = {
+ require(startKey.nonEmpty)
+ require(endKey.nonEmpty)
+ require(startKey.head == endKey.head, s"First key should be same => ($startKey) - ($endKey)")
+ }
+
+ protected def equal(js: JsObject, name: String, value: String): Boolean =
+ JsHelpers.getFieldPath(js, name) match {
+ case Some(JsString(v)) => v == value
+ case _ => false
+ }
+
+ protected def isTrue(js: JsObject, name: String): Boolean =
+ JsHelpers.getFieldPath(js, name) match {
+ case Some(JsBoolean(v)) => v
+ case _ => false
+ }
+
+ protected def gte(js: JsObject, name: String, value: Number): Boolean =
+ JsHelpers.getFieldPath(js, name) match {
+ case Some(JsNumber(n)) => n.longValue() >= value.longValue()
+ case _ => false
+ }
+
+ protected def lte(js: JsObject, name: String, value: Number): Boolean =
+ JsHelpers.getFieldPath(js, name) match {
+ case Some(JsNumber(n)) => n.longValue() <= value.longValue()
+ case _ => false
+ }
+
+ protected def numericSort(s: Seq[JsObject], descending: Boolean, name: String): Seq[JsObject] = {
+ val f =
+ (js: JsObject) =>
+ JsHelpers.getFieldPath(js, name) match {
+ case Some(JsNumber(n)) => n.longValue()
+ case _ => 0L
+ }
+ val order = implicitly[Ordering[Long]]
+ val ordering = if (descending) order.reverse else order
+ s.sortBy(f)(ordering)
+ }
+}
+
+private object ActivationViewMapper extends MemoryViewMapper {
+ private val NS = "namespace"
+ private val NS_WITH_PATH = ActivationHandler.NS_PATH
+ private val START = "start"
+
+ override def filter(ddoc: String,
+ view: String,
+ startKey: List[Any],
+ endKey: List[Any],
+ d: JsObject,
+ c: JsObject): Boolean = {
+ checkKeys(startKey, endKey)
+ val nsValue = startKey.head.asInstanceOf[String]
+ view match {
+ //whisks-filters ddoc uses namespace + invoking action path as first key
+ case "activations" if ddoc.startsWith("whisks-filters") =>
+ filterActivation(d, equal(c, NS_WITH_PATH, nsValue), startKey, endKey)
+ //whisks ddoc uses namespace as first key
+ case "activations" if ddoc.startsWith("whisks") => filterActivation(d, equal(d, NS, nsValue), startKey, endKey)
+ case _ => throw UnsupportedView(s"$ddoc/$view")
+ }
+ }
+
+ override def sort(ddoc: String, view: String, descending: Boolean, s: Seq[JsObject]): Seq[JsObject] =
+ view match {
+ case "activations" if ddoc.startsWith("whisks") => numericSort(s, descending, START)
+ case _ => throw UnsupportedView(s"$ddoc/$view")
+ }
+
+ private def filterActivation(d: JsObject, matchNS: Boolean, startKey: List[Any], endKey: List[Any]): Boolean = {
+ val filterResult = (startKey, endKey) match {
+ case (_ :: Nil, _ :: `TOP` :: Nil) =>
+ matchNS
+ case (_ :: (since: Number) :: Nil, _ :: `TOP` :: `TOP` :: Nil) =>
+ matchNS && gte(d, START, since)
+ case (_ :: (since: Number) :: Nil, _ :: (upto: Number) :: `TOP` :: Nil) =>
+ matchNS && gte(d, START, since) && lte(d, START, upto)
+ case _ => throw UnsupportedQueryKeys(s"$startKey, $endKey")
+ }
+ filterResult
+ }
+}
+
+private object WhisksViewMapper extends MemoryViewMapper {
+ private val NS = "namespace"
+ private val ROOT_NS = WhisksHandler.ROOT_NS
+ private val TYPE = "entityType"
+ private val UPDATED = "updated"
+ private val PUBLISH = "publish"
+ private val BINDING = "binding"
+
+ override def filter(ddoc: String,
+ view: String,
+ startKey: List[Any],
+ endKey: List[Any],
+ d: JsObject,
+ c: JsObject): Boolean = {
+ checkKeys(startKey, endKey)
+ val entityType = WhisksHandler.getEntityTypeForDesignDoc(ddoc, view)
+
+ val matchTypeAndView = equal(d, TYPE, entityType) && matchViewConditions(ddoc, view, d)
+ val matchNS = equal(d, NS, startKey.head.asInstanceOf[String])
+ val matchRootNS = equal(c, ROOT_NS, startKey.head.asInstanceOf[String])
+
+ //Here ddocs for actions, rules and triggers use
+ //namespace and namespace/packageName as first key
+
+ val filterResult = (startKey, endKey) match {
+ case (ns :: Nil, _ :: `TOP` :: Nil) =>
+ (matchTypeAndView && matchNS) || (matchTypeAndView && matchRootNS)
+
+ case (ns :: (since: Number) :: Nil, _ :: `TOP` :: `TOP` :: Nil) =>
+ (matchTypeAndView && matchNS && gte(d, UPDATED, since)) ||
+ (matchTypeAndView && matchRootNS && gte(d, UPDATED, since))
+ case (ns :: (since: Number) :: Nil, _ :: (upto: Number) :: `TOP` :: Nil) =>
+ (matchTypeAndView && matchNS && gte(d, UPDATED, since) && lte(d, UPDATED, upto)) ||
+ (matchTypeAndView && matchRootNS && gte(d, UPDATED, since) && lte(d, UPDATED, upto))
+
+ case _ => throw UnsupportedQueryKeys(s"$ddoc/$view -> ($startKey, $endKey)")
+ }
+ filterResult
+ }
+
+ private def matchViewConditions(ddoc: String, view: String, d: JsObject): Boolean = {
+ view match {
+ case "packages-public" if ddoc.startsWith("whisks") =>
+ isTrue(d, PUBLISH) && hasEmptyBinding(d)
+ case _ => true
+ }
+ }
+
+ private def hasEmptyBinding(js: JsObject) = {
+ js.fields.get(BINDING) match {
+ case Some(x: JsObject) if x.fields.nonEmpty => false
+ case _ => true
+ }
+ }
+
+ override def sort(ddoc: String, view: String, descending: Boolean, s: Seq[JsObject]): Seq[JsObject] = {
+ view match {
+ case "actions" | "rules" | "triggers" | "packages" | "packages-public" if ddoc.startsWith("whisks") =>
+ numericSort(s, descending, UPDATED)
+ case _ => throw UnsupportedView(s"$ddoc/$view")
+ }
+ }
+}
+
+private object SubjectViewMapper extends MemoryViewMapper {
+ private val BLOCKED = "blocked"
+ private val SUBJECT = "subject"
+ private val UUID = "uuid"
+ private val KEY = "key"
+ private val NS_NAME = "name"
+
+ override def filter(ddoc: String,
+ view: String,
+ startKey: List[Any],
+ endKey: List[Any],
+ d: JsObject,
+ c: JsObject): Boolean = {
+ require(startKey == endKey, s"startKey: $startKey and endKey: $endKey must be same for $ddoc/$view")
+ (ddoc, view) match {
+ case ("subjects", "identities") =>
+ filterForMatchingSubjectOrNamespace(ddoc, view, startKey, endKey, d)
+ case ("namespaceThrottlings", "blockedNamespaces") =>
+ filterForBlacklistedNamespace(d)
+ case _ =>
+ throw UnsupportedView(s"$ddoc/$view")
+ }
+ }
+
+ private def filterForBlacklistedNamespace(d: JsObject): Boolean = {
+ val id = d.fields("_id")
+ id match {
+ case JsString(idv) if idv.endsWith("/limits") =>
+ val limits = UserLimits.serdes.read(d)
+ limits.concurrentInvocations.contains(0) || limits.invocationsPerMinute.contains(0)
+ case _ =>
+ d.getFields(BLOCKED) match {
+ case Seq(JsTrue) => true
+ case _ => false
+ }
+ }
+ }
+
+ private def filterForMatchingSubjectOrNamespace(ddoc: String,
+ view: String,
+ startKey: List[Any],
+ endKey: List[Any],
+ d: JsObject) = {
+ val notBlocked = !isTrue(d, BLOCKED)
+ startKey match {
+ case (ns: String) :: Nil => notBlocked && (equal(d, SUBJECT, ns) || matchingNamespace(d, equal(_, NS_NAME, ns)))
+ case (uuid: String) :: (key: String) :: Nil =>
+ notBlocked &&
+ (
+ (equal(d, UUID, uuid) && equal(d, KEY, key))
+ || matchingNamespace(d, js => equal(js, UUID, uuid) && equal(js, KEY, key))
+ )
+ case _ => throw UnsupportedQueryKeys(s"$ddoc/$view -> ($startKey, $endKey)")
+ }
+ }
+
+ override def sort(ddoc: String, view: String, descending: Boolean, s: Seq[JsObject]): Seq[JsObject] = {
+ s //No sorting to be done
+ }
+
+ private def matchingNamespace(js: JsObject, matcher: JsObject => Boolean): Boolean = {
+ js.fields.get("namespaces") match {
+ case Some(JsArray(e)) => e.exists(v => matcher(v.asJsObject))
+ case _ => false
+ }
+ }
+}
diff --git a/tests/src/test/scala/whisk/core/cli/test/SequenceMigrationTests.scala b/tests/src/test/scala/whisk/core/cli/test/SequenceMigrationTests.scala
index 8c8bb79259..bdf604ead4 100644
--- a/tests/src/test/scala/whisk/core/cli/test/SequenceMigrationTests.scala
+++ b/tests/src/test/scala/whisk/core/cli/test/SequenceMigrationTests.scala
@@ -52,6 +52,11 @@ class SequenceMigrationTests extends TestHelpers with BeforeAndAfter with DbUtil
val namespace = wsk.namespace.whois()
val allowedActionDuration = 120 seconds
+ override protected def withFixture(test: NoArgTest) = {
+ assume(!isMemoryStore(entityStore))
+ super.withFixture(test)
+ }
+
behavior of "Sequence Migration"
it should "check default namespace '_' is preserved in WhiskAction of old style sequence" in {
diff --git a/tests/src/test/scala/whisk/core/controller/test/ActivationsApiTests.scala b/tests/src/test/scala/whisk/core/controller/test/ActivationsApiTests.scala
index f45b77111b..1e4e684b1d 100644
--- a/tests/src/test/scala/whisk/core/controller/test/ActivationsApiTests.scala
+++ b/tests/src/test/scala/whisk/core/controller/test/ActivationsApiTests.scala
@@ -22,20 +22,15 @@ import java.time.{Clock, Instant}
import akka.http.scaladsl.marshallers.sprayjson.SprayJsonSupport._
import akka.http.scaladsl.model.StatusCodes._
import akka.http.scaladsl.server.Route
-import akka.stream.ActorMaterializer
import org.junit.runner.RunWith
import org.scalatest.junit.JUnitRunner
import spray.json.DefaultJsonProtocol._
import spray.json._
import whisk.core.controller.WhiskActivationsApi
-import whisk.core.database.ArtifactStoreProvider
import whisk.core.entitlement.Collection
import whisk.core.entity._
import whisk.core.entity.size._
import whisk.http.{ErrorResponse, Messages}
-import whisk.spi.SpiLoader
-
-import scala.reflect.classTag
/**
* Tests Activations API.
@@ -553,16 +548,6 @@ class ActivationsApiTests extends ControllerTestCommon with WhiskActivationsApi
}
it should "report proper error when record is corrupted on get" in {
- implicit val materializer = ActorMaterializer()
- val activationStore = SpiLoader
- .get[ArtifactStoreProvider]
- .makeStore[WhiskActivation]()(
- classTag[WhiskActivation],
- WhiskActivation.serdes,
- WhiskDocumentReader,
- system,
- logging,
- materializer)
implicit val tid = transid()
//A bad activation type which breaks the deserialization by removing the subject entry
diff --git a/tests/src/test/scala/whisk/core/database/CouchDBArtifactStoreTests.scala b/tests/src/test/scala/whisk/core/database/CouchDBArtifactStoreTests.scala
new file mode 100644
index 0000000000..a36fd30d38
--- /dev/null
+++ b/tests/src/test/scala/whisk/core/database/CouchDBArtifactStoreTests.scala
@@ -0,0 +1,50 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package whisk.core.database
+
+import org.junit.runner.RunWith
+import org.scalatest.FlatSpec
+import org.scalatest.junit.JUnitRunner
+import whisk.core.database.test.behavior.ArtifactStoreBehavior
+import whisk.core.entity._
+
+import scala.reflect.classTag
+
+@RunWith(classOf[JUnitRunner])
+class CouchDBArtifactStoreTests extends FlatSpec with ArtifactStoreBehavior {
+ override def storeType = "CouchDB"
+
+ override val authStore = {
+ implicit val docReader: DocumentReader = WhiskDocumentReader
+ CouchDbStoreProvider.makeStore[WhiskAuth]()
+ }
+
+ override val entityStore =
+ CouchDbStoreProvider.makeStore[WhiskEntity]()(
+ classTag[WhiskEntity],
+ WhiskEntityJsonFormat,
+ WhiskDocumentReader,
+ actorSystem,
+ logging,
+ materializer)
+
+ override val activationStore = {
+ implicit val docReader: DocumentReader = WhiskDocumentReader
+ CouchDbStoreProvider.makeStore[WhiskActivation]()
+ }
+}
diff --git a/tests/src/test/scala/whisk/core/database/memory/MemoryArtifactStoreTests.scala b/tests/src/test/scala/whisk/core/database/memory/MemoryArtifactStoreTests.scala
new file mode 100644
index 0000000000..9d4643f874
--- /dev/null
+++ b/tests/src/test/scala/whisk/core/database/memory/MemoryArtifactStoreTests.scala
@@ -0,0 +1,50 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package whisk.core.database.memory
+
+import org.junit.runner.RunWith
+import org.scalatest.FlatSpec
+import org.scalatest.junit.JUnitRunner
+import whisk.core.database.test.behavior.ArtifactStoreBehavior
+import whisk.core.entity._
+
+import scala.reflect.classTag
+
+@RunWith(classOf[JUnitRunner])
+class MemoryArtifactStoreTests extends FlatSpec with ArtifactStoreBehavior {
+ override def storeType = "Memory"
+
+ override val authStore = {
+ implicit val docReader: DocumentReader = WhiskDocumentReader
+ MemoryArtifactStoreProvider.makeStore[WhiskAuth]()
+ }
+
+ override val entityStore =
+ MemoryArtifactStoreProvider.makeStore[WhiskEntity]()(
+ classTag[WhiskEntity],
+ WhiskEntityJsonFormat,
+ WhiskDocumentReader,
+ actorSystem,
+ logging,
+ materializer)
+
+ override val activationStore = {
+ implicit val docReader: DocumentReader = WhiskDocumentReader
+ MemoryArtifactStoreProvider.makeStore[WhiskActivation]()
+ }
+}
diff --git a/tests/src/test/scala/whisk/core/database/test/DbUtils.scala b/tests/src/test/scala/whisk/core/database/test/DbUtils.scala
index c7a75b71a1..277e1310ac 100644
--- a/tests/src/test/scala/whisk/core/database/test/DbUtils.scala
+++ b/tests/src/test/scala/whisk/core/database/test/DbUtils.scala
@@ -34,6 +34,7 @@ import spray.json.DefaultJsonProtocol._
import whisk.common.TransactionCounter
import whisk.common.TransactionId
import whisk.core.database._
+import whisk.core.database.memory.MemoryArtifactStore
import whisk.core.entity._
import whisk.core.entity.types.AuthStore
import whisk.core.entity.types.EntityStore
@@ -257,4 +258,7 @@ trait DbUtils extends TransactionCounter {
}
docsToDelete.clear()
}
+
+ def isMemoryStore(store: ArtifactStore[_]): Boolean = store.isInstanceOf[MemoryArtifactStore[_]]
+ def isCouchStore(store: ArtifactStore[_]): Boolean = store.isInstanceOf[CouchDbRestStore[_]]
}
diff --git a/tests/src/test/scala/whisk/core/database/test/DocumentHandlerTests.scala b/tests/src/test/scala/whisk/core/database/test/DocumentHandlerTests.scala
new file mode 100644
index 0000000000..c68fffd75c
--- /dev/null
+++ b/tests/src/test/scala/whisk/core/database/test/DocumentHandlerTests.scala
@@ -0,0 +1,594 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package whisk.core.database.test
+
+import org.junit.runner.RunWith
+import org.scalatest.{FlatSpec, Matchers, OptionValues}
+import org.scalatest.junit.JUnitRunner
+import spray.json._
+import DefaultJsonProtocol._
+import common.WskActorSystem
+import org.scalatest.concurrent.ScalaFutures
+import whisk.common.{TransactionCounter, TransactionId}
+import whisk.core.database.SubjectHandler.SubjectView
+import whisk.core.database.WhisksHandler.ROOT_NS
+import whisk.core.database._
+import whisk.core.entity._
+
+import scala.concurrent.Future
+
+@RunWith(classOf[JUnitRunner])
+class DocumentHandlerTests
+ extends FlatSpec
+ with Matchers
+ with ScalaFutures
+ with OptionValues
+ with TransactionCounter
+ with WskActorSystem {
+ override val instanceOrdinal = 0
+
+ behavior of "WhisksHandler computeFields"
+
+ it should "return empty object when namespace does not exist" in {
+ WhisksHandler.computedFields(JsObject()) shouldBe JsObject.empty
+ }
+
+ it should "return JsObject when namespace is simple name" in {
+ WhisksHandler.computedFields(JsObject(("namespace", JsString("foo")))) shouldBe JsObject((ROOT_NS, JsString("foo")))
+ WhisksHandler.computedFields(newRule("foo").toDocumentRecord) shouldBe JsObject((ROOT_NS, JsString("foo")))
+ }
+
+ it should "return JsObject when namespace is path" in {
+ WhisksHandler.computedFields(JsObject(("namespace", JsString("foo/bar")))) shouldBe
+ JsObject((ROOT_NS, JsString("foo")))
+
+ WhisksHandler.computedFields(newRule("foo/bar").toDocumentRecord) shouldBe JsObject((ROOT_NS, JsString("foo")))
+ }
+
+ private def newRule(ns: String): WhiskRule = {
+ WhiskRule(
+ EntityPath(ns),
+ EntityName("foo"),
+ FullyQualifiedEntityName(EntityPath("system"), EntityName("bar")),
+ FullyQualifiedEntityName(EntityPath("system"), EntityName("bar")))
+ }
+
+ behavior of "WhisksHandler computeView"
+
+ it should "include only common fields in trigger view" in {
+ val js = """{
+ | "namespace" : "foo",
+ | "version" : 5,
+ | "end" : 9,
+ | "cause" : 204
+ |}""".stripMargin.parseJson.asJsObject
+
+ val result = """{
+ | "namespace" : "foo",
+ | "version" : 5
+ |}""".stripMargin.parseJson.asJsObject
+ WhisksHandler.computeView("foo", "triggers", js) shouldBe result
+ }
+
+ it should "include false binding in public package view" in {
+ val js =
+ """{
+ | "namespace" : "foo",
+ | "version" : 5,
+ | "binding" : {"foo" : "bar"},
+ | "cause" : 204
+ |}""".stripMargin.parseJson.asJsObject
+
+ val result =
+ """{
+ | "namespace" : "foo",
+ | "version" : 5,
+ | "binding" : false
+ |}""".stripMargin.parseJson.asJsObject
+ WhisksHandler.computeView("foo", "packages-public", js) shouldBe result
+ }
+
+ it should "include actual binding in package view" in {
+ val js = """{
+ | "namespace" : "foo",
+ | "version" : 5,
+ | "binding" : {"foo" : "bar"},
+ | "cause" : 204
+ |}""".stripMargin.parseJson.asJsObject
+
+ val result = """{
+ | "namespace" : "foo",
+ | "version" : 5,
+ | "binding" : {"foo" : "bar"}
+ |}""".stripMargin.parseJson.asJsObject
+ WhisksHandler.computeView("foo", "packages", js) shouldBe result
+ }
+
+ it should "include limits and binary info in action view" in {
+ val js = """{
+ | "namespace" : "foo",
+ | "version" : 5,
+ | "binding" : {"foo" : "bar"},
+ | "limits" : 204,
+ | "exec" : {"binary" : true }
+ |}""".stripMargin.parseJson.asJsObject
+
+ val result = """{
+ | "namespace" : "foo",
+ | "version" : 5,
+ | "limits" : 204,
+ | "exec" : { "binary" : true }
+ |}""".stripMargin.parseJson.asJsObject
+ WhisksHandler.computeView("foo", "actions", js) shouldBe result
+ }
+
+ it should "include binary as false when exec missing" in {
+ val js = """{
+ | "namespace" : "foo",
+ | "version" : 5,
+ | "binding" : {"foo" : "bar"},
+ | "limits" : 204
+ |}""".stripMargin.parseJson.asJsObject
+
+ val result = """{
+ | "namespace" : "foo",
+ | "version" : 5,
+ | "limits" : 204,
+ | "exec" : { "binary" : false }
+ |}""".stripMargin.parseJson.asJsObject
+ WhisksHandler.computeView("foo", "actions", js) shouldBe result
+ }
+
+ it should "include binary as false when exec does not have binary prop" in {
+ val js = """{
+ | "namespace" : "foo",
+ | "version" : 5,
+ | "binding" : {"foo" : "bar"},
+ | "limits" : 204,
+ | "exec" : { "code" : "stuff" }
+ |}""".stripMargin.parseJson.asJsObject
+
+ val result = """{
+ | "namespace" : "foo",
+ | "version" : 5,
+ | "limits" : 204,
+ | "exec" : { "binary" : false }
+ |}""".stripMargin.parseJson.asJsObject
+ WhisksHandler.computeView("foo", "actions", js) shouldBe result
+ }
+
+ behavior of "WhisksHandler fieldsRequiredForView"
+
+ it should "match the expected field names" in {
+ WhisksHandler.fieldsRequiredForView("foo", "actions") shouldBe
+ Set("namespace", "name", "version", "publish", "annotations", "updated", "limits", "exec.binary")
+
+ WhisksHandler.fieldsRequiredForView("foo", "packages") shouldBe
+ Set("namespace", "name", "version", "publish", "annotations", "updated", "binding")
+
+ WhisksHandler.fieldsRequiredForView("foo", "packages-public") shouldBe
+ Set("namespace", "name", "version", "publish", "annotations", "updated")
+
+ WhisksHandler.fieldsRequiredForView("foo", "rules") shouldBe
+ Set("namespace", "name", "version", "publish", "annotations", "updated")
+
+ WhisksHandler.fieldsRequiredForView("foo", "triggers") shouldBe
+ Set("namespace", "name", "version", "publish", "annotations", "updated")
+
+ intercept[UnsupportedView] {
+ WhisksHandler.fieldsRequiredForView("foo", "unknown") shouldBe Set()
+ }
+ }
+
+ behavior of "ActivationHandler computeFields"
+
+ it should "return default value when no annotation found" in {
+ val js = """{"foo" : "bar"}""".parseJson.asJsObject
+ ActivationHandler.annotationValue(js, "fooKey", { _.convertTo[String] }, "barValue") shouldBe "barValue"
+
+ val js2 = """{"foo" : "bar", "annotations" : "a"}""".parseJson.asJsObject
+ ActivationHandler.annotationValue(js2, "fooKey", { _.convertTo[String] }, "barValue") shouldBe "barValue"
+
+ val js3 = """{"foo" : "bar", "annotations" : [{"key" : "someKey", "value" : "someValue"}]}""".parseJson.asJsObject
+ ActivationHandler.annotationValue(js3, "fooKey", { _.convertTo[String] }, "barValue") shouldBe "barValue"
+ }
+
+ it should "return transformed value when annotation found" in {
+ val js = """{
+ | "foo": "bar",
+ | "annotations": [
+ | {
+ | "key": "fooKey",
+ | "value": "fooValue"
+ | }
+ | ]
+ |}""".stripMargin.parseJson.asJsObject
+ ActivationHandler.annotationValue(js, "fooKey", { _.convertTo[String] + "-x" }, "barValue") shouldBe "fooValue-x"
+ }
+
+ it should "computeFields with deleteLogs true" in {
+ val js = """{
+ | "foo": "bar",
+ | "annotations": [
+ | {
+ | "key": "fooKey",
+ | "value": "fooValue"
+ | }
+ | ]
+ |}""".stripMargin.parseJson.asJsObject
+ ActivationHandler.computedFields(js) shouldBe """{"deleteLogs" : true}""".parseJson
+ }
+
+ it should "computeFields with deleteLogs false for sequence kind" in {
+ val js = """{
+ | "foo": "bar",
+ | "annotations": [
+ | {
+ | "key": "kind",
+ | "value": "sequence"
+ | }
+ | ]
+ |}""".stripMargin.parseJson.asJsObject
+ ActivationHandler.computedFields(js) shouldBe """{"deleteLogs" : false}""".parseJson
+ }
+
+ it should "computeFields with nspath as namespace" in {
+ val js = """{
+ | "namespace": "foons",
+ | "name":"bar",
+ | "annotations": [
+ | {
+ | "key": "kind",
+ | "value": "action"
+ | }
+ | ]
+ |}""".stripMargin.parseJson.asJsObject
+ ActivationHandler.computedFields(js) shouldBe """{"nspath": "foons/bar", "deleteLogs" : true}""".parseJson
+ }
+
+ it should "computeFields with nspath as qualified path" in {
+ val js = """{
+ | "namespace": "foons",
+ | "name":"bar",
+ | "annotations": [
+ | {
+ | "key": "path",
+ | "value": "barns/barpkg/baraction"
+ | }
+ | ]
+ |}""".stripMargin.parseJson.asJsObject
+ ActivationHandler.computedFields(js) shouldBe """{"nspath": "foons/barpkg/bar", "deleteLogs" : true}""".parseJson
+ }
+
+ it should "computeFields with nspath as namespace when path value is simple name" in {
+ val js = """{
+ | "namespace": "foons",
+ | "name":"bar",
+ | "annotations": [
+ | {
+ | "key": "path",
+ | "value": "baraction"
+ | }
+ | ]
+ |}""".stripMargin.parseJson.asJsObject
+ ActivationHandler.computedFields(js) shouldBe """{"nspath": "foons/bar", "deleteLogs" : true}""".parseJson
+ }
+
+ behavior of "ActivationHandler computeActivationView"
+
+ it should "include only listed fields" in {
+ val js = """{
+ | "namespace" : "foo",
+ | "extra" : false,
+ | "cause" : 204
+ |}""".stripMargin.parseJson.asJsObject
+
+ val result = """{
+ | "namespace" : "foo",
+ | "cause" : 204
+ |}""".stripMargin.parseJson.asJsObject
+ ActivationHandler.computeView("foo", "activations", js) shouldBe result
+ }
+
+ it should "include duration when end is non zero" in {
+ val js = """{
+ | "namespace" : "foo",
+ | "start" : 5,
+ | "end" : 9,
+ | "cause" : 204
+ |}""".stripMargin.parseJson.asJsObject
+
+ val result = """{
+ | "namespace" : "foo",
+ | "start" : 5,
+ | "end" : 9,
+ | "duration" : 4,
+ | "cause" : 204
+ |}""".stripMargin.parseJson.asJsObject
+ ActivationHandler.computeView("foo", "activations", js) shouldBe result
+ }
+
+ it should "not include duration when end is zero" in {
+ val js = """{
+ | "namespace" : "foo",
+ | "start" : 5,
+ | "end" : 0,
+ | "cause" : 204
+ |}""".stripMargin.parseJson.asJsObject
+
+ val result = """{
+ | "namespace" : "foo",
+ | "start" : 5,
+ | "cause" : 204
+ |}""".stripMargin.parseJson.asJsObject
+ ActivationHandler.computeView("foo", "activations", js) shouldBe result
+ }
+
+ it should "include statusCode" in {
+ val js = """{
+ | "namespace": "foo",
+ | "response": {"statusCode" : 404}
+ |}""".stripMargin.parseJson.asJsObject
+
+ val result = """{
+ | "namespace": "foo",
+ | "statusCode" : 404
+ |}""".stripMargin.parseJson.asJsObject
+ ActivationHandler.computeView("foo", "activations", js) shouldBe result
+ }
+
+ it should "not include statusCode" in {
+ val js = """{
+ | "namespace": "foo",
+ | "response": {"status" : 404}
+ |}""".stripMargin.parseJson.asJsObject
+
+ val result = """{
+ | "namespace": "foo"
+ |}""".stripMargin.parseJson.asJsObject
+ ActivationHandler.computeView("foo", "activations", js) shouldBe result
+ }
+
+ behavior of "ActivationHandler fieldsRequiredForView"
+
+ it should "match the expected field names" in {
+ ActivationHandler.fieldsRequiredForView("foo", "activations") shouldBe
+ Set(
+ "namespace",
+ "name",
+ "version",
+ "publish",
+ "annotations",
+ "activationId",
+ "start",
+ "cause",
+ "end",
+ "response.statusCode")
+ }
+
+ it should "throw UnsupportedView exception" in {
+ intercept[UnsupportedView] {
+ ActivationHandler.fieldsRequiredForView("foo", "unknown")
+ }
+ }
+
+ behavior of "SubjectHandler computeSubjectView"
+
+ it should "match subject with namespace" in {
+ val js = """{
+ | "subject": "foo",
+ | "uuid": "u1",
+ | "key" : "k1"
+ |}""".stripMargin.parseJson.asJsObject
+ SubjectHandler.findMatchingSubject(List("foo"), js).value shouldBe SubjectView("foo", "u1", "k1")
+ }
+
+ it should "match subject with child namespace" in {
+ val js = """{
+ | "subject": "bar",
+ | "uuid": "u1",
+ | "key" : "k1",
+ | "namespaces" : [
+ | {"name": "foo", "uuid":"u2", "key":"k2"}
+ | ]
+ |}""".stripMargin.parseJson.asJsObject
+ SubjectHandler.findMatchingSubject(List("foo"), js).value shouldBe
+ SubjectView("foo", "u2", "k2", matchInNamespace = true)
+ }
+
+ it should "match subject with uuid and key" in {
+ val js = """{
+ | "subject": "foo",
+ | "uuid": "u1",
+ | "key" : "k1"
+ |}""".stripMargin.parseJson.asJsObject
+ SubjectHandler.findMatchingSubject(List("u1", "k1"), js).value shouldBe
+ SubjectView("foo", "u1", "k1")
+ }
+
+ it should "match subject with child namespace with uuid and key" in {
+ val js = """{
+ | "subject": "bar",
+ | "uuid": "u1",
+ | "key" : "k1",
+ | "namespaces" : [
+ | {"name": "foo", "uuid":"u2", "key":"k2"}
+ | ]
+ |}""".stripMargin.parseJson.asJsObject
+ SubjectHandler.findMatchingSubject(List("u2", "k2"), js).value shouldBe
+ SubjectView("foo", "u2", "k2", matchInNamespace = true)
+ }
+
+ it should "throw exception when namespace match but key missing" in {
+ val js = """{
+ | "subject": "foo",
+ | "uuid": "u1",
+ | "blocked" : true
+ |}""".stripMargin.parseJson.asJsObject
+ SubjectHandler.findMatchingSubject(List("foo"), js) shouldBe empty
+ }
+
+ behavior of "SubjectHandler transformViewResult"
+
+ it should "json should match format of CouchDB response" in {
+ implicit val tid: TransactionId = transid()
+ val js = """{
+ | "_id": "bar",
+ | "subject": "bar",
+ | "uuid": "u1",
+ | "key" : "k1",
+ | "namespaces" : [
+ | {"name": "foo", "uuid":"u2", "key":"k2"}
+ | ]
+ |}""".stripMargin.parseJson.asJsObject
+ val result = """{
+ | "id": "bar",
+ | "key": [
+ | "u2",
+ | "k2"
+ | ],
+ | "value": {
+ | "_id": "foo/limits",
+ | "namespace": "foo",
+ | "uuid": "u2",
+ | "key": "k2"
+ | },
+ | "doc": null
+ |}""".stripMargin.parseJson.asJsObject
+ val queryKey = List("u2", "k2")
+ SubjectHandler
+ .transformViewResult("subjects", "identities", queryKey, queryKey, includeDocs = true, js, TestDocumentProvider())
+ .futureValue shouldBe Seq(result)
+ }
+
+ it should "should return none when passed object does not passes view criteria" in {
+ implicit val tid: TransactionId = transid()
+ val js = """{
+ | "_id": "bar",
+ | "subject": "bar",
+ | "uuid": "u1",
+ | "key" : "k1",
+ | "namespaces" : [
+ | {"name": "foo", "uuid":"u2", "key":"k2"},
+ | {"name": "foo2", "uuid":"u3", "key":"k3"}
+ | ]
+ |}""".stripMargin.parseJson.asJsObject
+
+ val queryKey = List("u2", "k3")
+ SubjectHandler
+ .transformViewResult("subjects", "identities", queryKey, queryKey, includeDocs = true, js, TestDocumentProvider())
+ .futureValue shouldBe empty
+ }
+
+ behavior of "SubjectHandler blacklisted namespaces"
+
+ it should "match limits with 0 concurrentInvocations" in {
+ val js = """{
+ | "_id": "bar/limits",
+ | "concurrentInvocations": 0
+ |}""".stripMargin.parseJson.asJsObject
+ val result = """{
+ | "id": "bar/limits",
+ | "key": "bar",
+ | "value": 1
+ |}""".stripMargin.parseJson.asJsObject
+ blacklistingResults(js) shouldBe Seq(result)
+ }
+
+ it should "match limits with 0 invocationsPerMinute" in {
+ val js = """{
+ | "_id": "bar/limits",
+ | "concurrentInvocations": 10,
+ | "invocationsPerMinute": 0
+ |}""".stripMargin.parseJson.asJsObject
+ val result = """{
+ | "id": "bar/limits",
+ | "key": "bar",
+ | "value": 1
+ |}""".stripMargin.parseJson.asJsObject
+ blacklistingResults(js) shouldBe Seq(result)
+ }
+
+ it should "not match limits with invocationsPerMinute and concurrentInvocations defined" in {
+ val js = """{
+ | "_id": "bar/limits",
+ | "concurrentInvocations": 10,
+ | "invocationsPerMinute": 40
+ |}""".stripMargin.parseJson.asJsObject
+ blacklistingResults(js) shouldBe empty
+ }
+
+ it should "list all namespaces of blocked subject" in {
+ val js = """{
+ | "_id": "bar",
+ | "blocked": true,
+ | "subject": "bar",
+ | "namespaces" : [
+ | {"name": "foo", "uuid":"u2", "key":"k2"},
+ | {"name": "foo2", "uuid":"u3", "key":"k3"}
+ | ]
+ |}""".stripMargin.parseJson.asJsObject
+ val r1 = """{
+ | "id": "bar",
+ | "key": "foo",
+ | "value": 1
+ |}""".stripMargin.parseJson.asJsObject
+ val r2 = """{
+ | "id": "bar",
+ | "key": "foo2",
+ | "value": 1
+ |}""".stripMargin.parseJson.asJsObject
+ blacklistingResults(js).toSet shouldBe Set(r1, r2)
+ }
+
+ it should "list no namespace of unblocked subject" in {
+ val js = """{
+ | "_id": "bar",
+ | "subject": "bar",
+ | "namespaces" : [
+ | {"name": "foo", "uuid":"u2", "key":"k2"},
+ | {"name": "foo2", "uuid":"u3", "key":"k3"}
+ | ]
+ |}""".stripMargin.parseJson.asJsObject
+
+ blacklistingResults(js) shouldBe empty
+ }
+
+ private def blacklistingResults(js: JsObject) = {
+ implicit val tid: TransactionId = transid()
+ SubjectHandler
+ .transformViewResult(
+ "namespaceThrottlings",
+ "blockedNamespaces",
+ List.empty,
+ List.empty,
+ includeDocs = false,
+ js,
+ TestDocumentProvider())
+ .futureValue
+ }
+
+ private case class TestDocumentProvider(js: Option[JsObject]) extends DocumentProvider {
+ override protected[database] def get(id: DocId)(implicit transid: TransactionId) = Future.successful(js)
+ }
+
+ private object TestDocumentProvider {
+ def apply(js: JsObject): DocumentProvider = new TestDocumentProvider(Some(js))
+ def apply(): DocumentProvider = new TestDocumentProvider(None)
+ }
+}
diff --git a/tests/src/test/scala/whisk/core/database/test/behavior/ArtifactStoreActivationsQueryBehaviors.scala b/tests/src/test/scala/whisk/core/database/test/behavior/ArtifactStoreActivationsQueryBehaviors.scala
new file mode 100644
index 0000000000..7f4ffbf4b7
--- /dev/null
+++ b/tests/src/test/scala/whisk/core/database/test/behavior/ArtifactStoreActivationsQueryBehaviors.scala
@@ -0,0 +1,55 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package whisk.core.database.test.behavior
+
+import whisk.common.TransactionId
+import whisk.core.entity.WhiskEntityQueries.TOP
+import whisk.core.entity.{EntityPath, WhiskActivation}
+
+trait ArtifactStoreActivationsQueryBehaviors extends ArtifactStoreBehaviorBase {
+
+ it should "list activations between given times" in {
+ implicit val tid: TransactionId = transid()
+ val ns = newNS()
+ val activations = (1000 until 1100 by 10).map(newActivation(ns.asString, "testact", _))
+ activations foreach (put(activationStore, _))
+
+ val entityPath = s"${ns.asString}/testact"
+ waitOnView(activationStore, EntityPath(entityPath), activations.size, WhiskActivation.filtersView)
+
+ val resultSince = query[WhiskActivation](
+ activationStore,
+ WhiskActivation.filtersView.name,
+ List(entityPath, 1050),
+ List(entityPath, TOP, TOP))
+
+ resultSince.map(_.fields("value")) shouldBe activations.reverse
+ .filter(_.start.toEpochMilli >= 1050)
+ .map(_.summaryAsJson)
+
+ val resultBetween = query[WhiskActivation](
+ activationStore,
+ WhiskActivation.filtersView.name,
+ List(entityPath, 1060),
+ List(entityPath, 1090, TOP))
+
+ resultBetween.map(_.fields("value")) shouldBe activations.reverse
+ .filter(a => a.start.toEpochMilli >= 1060 && a.start.toEpochMilli <= 1090)
+ .map(_.summaryAsJson)
+ }
+}
diff --git a/tests/src/test/scala/whisk/core/database/test/behavior/ArtifactStoreBehavior.scala b/tests/src/test/scala/whisk/core/database/test/behavior/ArtifactStoreBehavior.scala
new file mode 100644
index 0000000000..e19649407a
--- /dev/null
+++ b/tests/src/test/scala/whisk/core/database/test/behavior/ArtifactStoreBehavior.scala
@@ -0,0 +1,26 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package whisk.core.database.test.behavior
+
+trait ArtifactStoreBehavior
+ extends ArtifactStoreBehaviorBase
+ with ArtifactStoreQueryBehaviors
+ with ArtifactStoreCRUDBehaviors
+ with ArtifactStoreSubjectQueryBehaviors
+ with ArtifactStoreWhisksQueryBehaviors
+ with ArtifactStoreActivationsQueryBehaviors
diff --git a/tests/src/test/scala/whisk/core/database/test/behavior/ArtifactStoreBehaviorBase.scala b/tests/src/test/scala/whisk/core/database/test/behavior/ArtifactStoreBehaviorBase.scala
new file mode 100644
index 0000000000..e13e420eef
--- /dev/null
+++ b/tests/src/test/scala/whisk/core/database/test/behavior/ArtifactStoreBehaviorBase.scala
@@ -0,0 +1,143 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package whisk.core.database.test.behavior
+
+import java.time.Instant
+
+import akka.stream.ActorMaterializer
+import common.{StreamLogging, WskActorSystem}
+import org.scalatest.concurrent.{IntegrationPatience, ScalaFutures}
+import org.scalatest.{BeforeAndAfterAll, BeforeAndAfterEach, FlatSpec, Matchers}
+import spray.json.{JsObject, JsValue}
+import whisk.common.{TransactionCounter, TransactionId}
+import whisk.core.database.test.DbUtils
+import whisk.core.database.{ArtifactStore, StaleParameter}
+import whisk.core.entity._
+import whisk.utils.JsHelpers
+
+import scala.util.Random
+
+trait ArtifactStoreBehaviorBase
+ extends FlatSpec
+ with ScalaFutures
+ with TransactionCounter
+ with Matchers
+ with StreamLogging
+ with DbUtils
+ with WskActorSystem
+ with IntegrationPatience
+ with BeforeAndAfterEach
+ with BeforeAndAfterAll {
+
+ override val instanceOrdinal = 0
+
+ //Bring in sync the timeout used by ScalaFutures and DBUtils
+ implicit override val patienceConfig: PatienceConfig = PatienceConfig(timeout = dbOpTimeout)
+
+ protected implicit val materializer: ActorMaterializer = ActorMaterializer()
+
+ protected val prefix = s"artifactTCK_${Random.alphanumeric.take(4).mkString}"
+
+ def authStore: ArtifactStore[WhiskAuth]
+ def entityStore: ArtifactStore[WhiskEntity]
+ def activationStore: ArtifactStore[WhiskActivation]
+
+ def storeType: String
+
+ override def afterEach(): Unit = {
+ cleanup()
+ }
+
+ override def afterAll(): Unit = {
+ println("Shutting down store connections")
+ authStore.shutdown()
+ entityStore.shutdown()
+ activationStore.shutdown()
+ super.afterAll()
+ }
+
+ //~----------------------------------------< utility methods >
+
+ protected def query[A <: WhiskEntity](
+ db: ArtifactStore[A],
+ table: String,
+ startKey: List[Any],
+ endKey: List[Any],
+ skip: Int = 0,
+ limit: Int = 0,
+ includeDocs: Boolean = false,
+ descending: Boolean = true,
+ reduce: Boolean = false,
+ stale: StaleParameter = StaleParameter.No)(implicit transid: TransactionId): List[JsObject] = {
+ db.query(table, startKey, endKey, skip, limit, includeDocs, descending, reduce, stale).futureValue
+ }
+
+ protected def count[A <: WhiskEntity](
+ db: ArtifactStore[A],
+ table: String,
+ startKey: List[Any],
+ endKey: List[Any],
+ skip: Int = 0,
+ stale: StaleParameter = StaleParameter.No)(implicit transid: TransactionId): Long = {
+ db.count(table, startKey, endKey, skip, stale).futureValue
+ }
+
+ protected def getWhiskAuth(doc: DocInfo)(implicit transid: TransactionId) = {
+ authStore.get[WhiskAuth](doc).futureValue
+ }
+
+ protected def newAuth() = {
+ val subject = Subject()
+ val namespaces = Set(wskNS("foo"))
+ WhiskAuth(subject, namespaces)
+ }
+
+ protected def wskNS(name: String) = {
+ WhiskNamespace(EntityName(name), AuthKey())
+ }
+
+ private val exec = BlackBoxExec(ExecManifest.ImageName("image"), None, None, native = false)
+
+ protected def newAction(ns: EntityPath): WhiskAction = {
+ WhiskAction(ns, aname(), exec)
+ }
+
+ protected def newActivation(ns: String, actionName: String, start: Long): WhiskActivation = {
+ WhiskActivation(
+ EntityPath(ns),
+ EntityName(actionName),
+ Subject(),
+ ActivationId.generate(),
+ Instant.ofEpochMilli(start),
+ Instant.ofEpochMilli(start + 1000))
+ }
+
+ protected def aname() = EntityName(s"${prefix}_name_${randomString()}")
+
+ protected def newNS() = EntityPath(s"${prefix}_ns_${randomString()}")
+
+ private def randomString() = Random.alphanumeric.take(5).mkString
+
+ protected def getJsObject(js: JsObject, fields: String*): JsObject = {
+ JsHelpers.getFieldPath(js, fields: _*).get.asJsObject
+ }
+
+ protected def getJsField(js: JsObject, subObject: String, fieldName: String): JsValue = {
+ js.fields(subObject).asJsObject().fields(fieldName)
+ }
+}
diff --git a/tests/src/test/scala/whisk/core/database/test/behavior/ArtifactStoreCRUDBehaviors.scala b/tests/src/test/scala/whisk/core/database/test/behavior/ArtifactStoreCRUDBehaviors.scala
new file mode 100644
index 0000000000..fb3fa62bf2
--- /dev/null
+++ b/tests/src/test/scala/whisk/core/database/test/behavior/ArtifactStoreCRUDBehaviors.scala
@@ -0,0 +1,162 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package whisk.core.database.test.behavior
+
+import java.time.Instant
+
+import whisk.common.TransactionId
+import whisk.core.database.{DocumentConflictException, NoDocumentException}
+import whisk.core.entity._
+
+trait ArtifactStoreCRUDBehaviors extends ArtifactStoreBehaviorBase {
+
+ behavior of s"${storeType}ArtifactStore put"
+
+ it should "put document and get a revision 1" in {
+ implicit val tid: TransactionId = transid()
+ val doc = put(authStore, newAuth())
+ doc.rev.empty shouldBe false
+ }
+
+ it should "put and update document" in {
+ implicit val tid: TransactionId = transid()
+ val auth = newAuth()
+ val doc = put(authStore, auth)
+
+ val auth2 =
+ getWhiskAuth(doc)
+ .copy(namespaces = Set(wskNS("foo1")))
+ .revision[WhiskAuth](doc.rev)
+ val doc2 = put(authStore, auth2)
+
+ doc2.rev should not be doc.rev
+ doc2.rev.empty shouldBe false
+ }
+
+ it should "throw DocumentConflictException when updated with old revision" in {
+ implicit val tid: TransactionId = transid()
+ val auth = newAuth()
+ val doc = put(authStore, auth)
+
+ val auth2 = getWhiskAuth(doc).copy(namespaces = Set(wskNS("foo1"))).revision[WhiskAuth](doc.rev)
+ val doc2 = put(authStore, auth2)
+
+ //Updated with _rev set to older one
+ val auth3 = getWhiskAuth(doc2).copy(namespaces = Set(wskNS("foo2"))).revision[WhiskAuth](doc.rev)
+ intercept[DocumentConflictException] {
+ put(authStore, auth3)
+ }
+ }
+
+ it should "throw DocumentConflictException if document with same id is inserted twice" in {
+ implicit val tid: TransactionId = transid()
+ val auth = newAuth()
+ val doc = put(authStore, auth)
+
+ intercept[DocumentConflictException] {
+ put(authStore, auth)
+ }
+ }
+
+ behavior of s"${storeType}ArtifactStore delete"
+
+ it should "deletes existing document" in {
+ implicit val tid: TransactionId = transid()
+ val doc = put(authStore, newAuth())
+ delete(authStore, doc) shouldBe true
+ }
+
+ it should "throws IllegalArgumentException when deleting without revision" in {
+ intercept[IllegalArgumentException] {
+ implicit val tid: TransactionId = transid()
+ delete(authStore, DocInfo("doc-with-empty-revision"))
+ }
+ }
+
+ it should "throws NoDocumentException when document does not exist" in {
+ intercept[NoDocumentException] {
+ implicit val tid: TransactionId = transid()
+ delete(authStore, DocInfo ! ("non-existing-doc", "42"))
+ }
+ }
+
+ it should "throws DocumentConflictException when revision does not match" in {
+ implicit val tid: TransactionId = transid()
+ val auth = newAuth()
+ val doc = put(authStore, auth)
+
+ val auth2 = getWhiskAuth(doc).copy(namespaces = Set(wskNS("foo1"))).revision[WhiskAuth](doc.rev)
+ val doc2 = put(authStore, auth2)
+
+ intercept[DocumentConflictException] {
+ delete(authStore, doc)
+ }
+ }
+
+ behavior of s"${storeType}ArtifactStore get"
+
+ it should "get existing entity matching id and rev" in {
+ implicit val tid: TransactionId = transid()
+ val auth = newAuth()
+ val doc = put(authStore, auth)
+ val authFromGet = getWhiskAuth(doc)
+ authFromGet shouldBe auth
+ authFromGet.docinfo.rev shouldBe doc.rev
+ }
+
+ it should "get existing entity matching id only" in {
+ implicit val tid: TransactionId = transid()
+ val auth = newAuth()
+ val doc = put(authStore, auth)
+ val authFromGet = getWhiskAuth(doc)
+ authFromGet shouldBe auth
+ }
+
+ it should "get entity with timestamp" in {
+ implicit val tid: TransactionId = transid()
+ val activation = WhiskActivation(
+ EntityPath("testnamespace"),
+ EntityName("activation1"),
+ Subject(),
+ ActivationId.generate(),
+ start = Instant.now,
+ end = Instant.now)
+ val activationDoc = put(activationStore, activation)
+ val activationFromDb = activationStore.get[WhiskActivation](activationDoc).futureValue
+ activationFromDb shouldBe activation
+ }
+
+ it should "throws NoDocumentException when document revision does not match" in {
+ implicit val tid: TransactionId = transid()
+ val auth = newAuth()
+ val doc = put(authStore, auth)
+
+ val auth2 = getWhiskAuth(doc).copy(namespaces = Set(wskNS("foo1"))).revision[WhiskAuth](doc.rev)
+ val doc2 = put(authStore, auth2)
+
+ authStore.get[WhiskAuth](doc).failed.futureValue.getCause shouldBe a[AssertionError]
+
+ val authFromGet = getWhiskAuth(doc2)
+ authFromGet shouldBe auth2
+ }
+
+ it should "throws NoDocumentException when document does not exist" in {
+ implicit val tid: TransactionId = transid()
+ authStore.get[WhiskAuth](DocInfo("non-existing-doc")).failed.futureValue shouldBe a[NoDocumentException]
+ }
+}
diff --git a/tests/src/test/scala/whisk/core/database/test/behavior/ArtifactStoreQueryBehaviors.scala b/tests/src/test/scala/whisk/core/database/test/behavior/ArtifactStoreQueryBehaviors.scala
new file mode 100644
index 0000000000..1f6ed9c942
--- /dev/null
+++ b/tests/src/test/scala/whisk/core/database/test/behavior/ArtifactStoreQueryBehaviors.scala
@@ -0,0 +1,229 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package whisk.core.database.test.behavior
+
+import spray.json.{JsArray, JsNumber, JsObject, JsString}
+import whisk.common.TransactionId
+import whisk.core.entity.WhiskEntityQueries.TOP
+import whisk.core.entity.{EntityPath, WhiskAction, WhiskActivation, WhiskEntity}
+
+trait ArtifactStoreQueryBehaviors extends ArtifactStoreBehaviorBase {
+
+ behavior of s"${storeType}ArtifactStore query"
+
+ it should "find single entity" in {
+ implicit val tid: TransactionId = transid()
+
+ val ns = newNS()
+ val action = newAction(ns)
+ val docInfo = put(entityStore, action)
+
+ waitOnView(entityStore, ns.root, 1, WhiskAction.view)
+ val result = query[WhiskEntity](
+ entityStore,
+ WhiskAction.view.name,
+ List(ns.asString, 0),
+ List(ns.asString, TOP, TOP),
+ includeDocs = true)
+
+ result should have length 1
+
+ def js = result.head
+ js.fields("id") shouldBe JsString(docInfo.id.id)
+ js.fields("key") shouldBe JsArray(JsString(ns.asString), JsNumber(action.updated.toEpochMilli))
+ js.fields.get("value") shouldBe defined
+ js.fields.get("doc") shouldBe defined
+ js.fields("value") shouldBe action.summaryAsJson
+ dropRev(js.fields("doc").asJsObject) shouldBe action.toDocumentRecord
+ }
+
+ it should "not have doc with includeDocs = false" in {
+ implicit val tid: TransactionId = transid()
+
+ val ns = newNS()
+ val action = newAction(ns)
+ val docInfo = put(entityStore, action)
+
+ waitOnView(entityStore, ns.root, 1, WhiskAction.view)
+ val result =
+ query[WhiskEntity](entityStore, WhiskAction.view.name, List(ns.asString, 0), List(ns.asString, TOP, TOP))
+
+ result should have length 1
+
+ def js = result.head
+ js.fields("id") shouldBe JsString(docInfo.id.id)
+ js.fields("key") shouldBe JsArray(JsString(ns.asString), JsNumber(action.updated.toEpochMilli))
+ js.fields.get("value") shouldBe defined
+ js.fields.get("doc") shouldBe None
+ js.fields("value") shouldBe action.summaryAsJson
+ }
+
+ it should "find all entities" in {
+ implicit val tid: TransactionId = transid()
+
+ val ns = newNS()
+ val entities = Seq(newAction(ns), newAction(ns))
+
+ entities foreach {
+ put(entityStore, _)
+ }
+
+ waitOnView(entityStore, ns.root, 2, WhiskAction.view)
+ val result =
+ query[WhiskEntity](entityStore, WhiskAction.view.name, List(ns.asString, 0), List(ns.asString, TOP, TOP))
+
+ result should have length entities.length
+ result.map(_.fields("value")) should contain theSameElementsAs entities.map(_.summaryAsJson)
+ }
+
+ it should "return result in sorted order" in {
+ implicit val tid: TransactionId = transid()
+
+ val ns = newNS()
+ val activations = (1000 until 1100 by 10).map(newActivation(ns.asString, "testact", _))
+ activations foreach (put(activationStore, _))
+
+ val entityPath = s"${ns.asString}/testact"
+ waitOnView(activationStore, EntityPath(entityPath), activations.size, WhiskActivation.filtersView)
+
+ val resultDescending = query[WhiskActivation](
+ activationStore,
+ WhiskActivation.filtersView.name,
+ List(entityPath, 0),
+ List(entityPath, TOP, TOP))
+
+ resultDescending should have length activations.length
+ resultDescending.map(getJsField(_, "value", "start")) shouldBe activations
+ .map(_.summaryAsJson.fields("start"))
+ .reverse
+
+ val resultAscending = query[WhiskActivation](
+ activationStore,
+ WhiskActivation.filtersView.name,
+ List(entityPath, 0),
+ List(entityPath, TOP, TOP),
+ descending = false)
+
+ resultAscending.map(getJsField(_, "value", "start")) shouldBe activations.map(_.summaryAsJson.fields("start"))
+ }
+
+ it should "support skipping results" in {
+ implicit val tid: TransactionId = transid()
+
+ val ns = newNS()
+ val activations = (1000 until 1100 by 10).map(newActivation(ns.asString, "testact", _))
+ activations foreach (put(activationStore, _))
+
+ val entityPath = s"${ns.asString}/testact"
+ waitOnView(activationStore, EntityPath(entityPath), activations.size, WhiskActivation.filtersView)
+ val result = query[WhiskActivation](
+ activationStore,
+ WhiskActivation.filtersView.name,
+ List(entityPath, 0),
+ List(entityPath, TOP, TOP),
+ skip = 5,
+ descending = false)
+
+ result.map(getJsField(_, "value", "start")) shouldBe activations.map(_.summaryAsJson.fields("start")).drop(5)
+ }
+
+ it should "support limiting results" in {
+ implicit val tid: TransactionId = transid()
+
+ val ns = newNS()
+ val activations = (1000 until 1100 by 10).map(newActivation(ns.asString, "testact", _))
+ activations foreach (put(activationStore, _))
+
+ val entityPath = s"${ns.asString}/testact"
+ waitOnView(activationStore, EntityPath(entityPath), activations.size, WhiskActivation.filtersView)
+ val result = query[WhiskActivation](
+ activationStore,
+ WhiskActivation.filtersView.name,
+ List(entityPath, 0),
+ List(entityPath, TOP, TOP),
+ limit = 5,
+ descending = false)
+
+ result.map(getJsField(_, "value", "start")) shouldBe activations.map(_.summaryAsJson.fields("start")).take(5)
+ }
+
+ it should "support including complete docs" in {
+ implicit val tid: TransactionId = transid()
+
+ val ns = newNS()
+ val activations = (1000 until 1100 by 10).map(newActivation(ns.asString, "testact", _))
+ activations foreach (put(activationStore, _))
+
+ val entityPath = s"${ns.asString}/testact"
+ waitOnView(activationStore, EntityPath(entityPath), activations.size, WhiskActivation.filtersView)
+ val result = query[WhiskActivation](
+ activationStore,
+ WhiskActivation.filtersView.name,
+ List(entityPath, 0),
+ List(entityPath, TOP, TOP),
+ includeDocs = true,
+ descending = false)
+
+ //Drop the _rev field as activations do not have that field
+ result.map(js => JsObject(getJsObject(js, "doc").fields - "_rev")) shouldBe activations.map(_.toDocumentRecord)
+ }
+
+ behavior of s"${storeType}ArtifactStore count"
+
+ it should "should match all created activations" in {
+ implicit val tid: TransactionId = transid()
+
+ val ns = newNS()
+ val activations = (1000 until 1100 by 10).map(newActivation(ns.asString, "testact", _))
+ activations foreach (put(activationStore, _))
+
+ val entityPath = s"${ns.asString}/testact"
+ waitOnView(activationStore, EntityPath(entityPath), activations.size, WhiskActivation.filtersView)
+ val result = count[WhiskActivation](
+ activationStore,
+ WhiskActivation.filtersView.name,
+ List(entityPath, 0),
+ List(entityPath, TOP, TOP))
+
+ result shouldBe 10
+ }
+
+ it should "count with skip" ignore {
+ //TODO Skip is not working for CouchDB
+ implicit val tid: TransactionId = transid()
+
+ val ns = newNS()
+ val activations = (1000 until 1100 by 10).map(newActivation(ns.asString, "testact", _))
+ activations foreach (put(activationStore, _))
+
+ val entityPath = s"${ns.asString}/testact"
+ waitOnView(activationStore, EntityPath(entityPath), activations.size, WhiskActivation.filtersView)
+ val result = count[WhiskActivation](
+ activationStore,
+ WhiskActivation.filtersView.name,
+ List(entityPath, 0),
+ List(entityPath, TOP, TOP),
+ skip = 4)
+
+ result shouldBe 10 - 4
+ }
+
+ private def dropRev(js: JsObject): JsObject = {
+ JsObject(js.fields - "_rev")
+ }
+}
diff --git a/tests/src/test/scala/whisk/core/database/test/behavior/ArtifactStoreSubjectQueryBehaviors.scala b/tests/src/test/scala/whisk/core/database/test/behavior/ArtifactStoreSubjectQueryBehaviors.scala
new file mode 100644
index 0000000000..ce78397926
--- /dev/null
+++ b/tests/src/test/scala/whisk/core/database/test/behavior/ArtifactStoreSubjectQueryBehaviors.scala
@@ -0,0 +1,178 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package whisk.core.database.test.behavior
+
+import spray.json.DefaultJsonProtocol._
+import spray.json.{JsBoolean, JsObject}
+import whisk.common.TransactionId
+import whisk.core.database.{NoDocumentException, StaleParameter}
+import whisk.core.entity._
+import whisk.core.entity.types.AuthStore
+import whisk.core.invoker.NamespaceBlacklist
+
+import scala.concurrent.duration.Duration
+
+trait ArtifactStoreSubjectQueryBehaviors extends ArtifactStoreBehaviorBase {
+
+ behavior of s"${storeType}ArtifactStore query subjects"
+
+ it should "find subject by namespace" in {
+ implicit val tid: TransactionId = transid()
+ val ak1 = AuthKey()
+ val ak2 = AuthKey()
+ val ns1 = aname()
+ val ns2 = aname()
+ val subs =
+ Array(WhiskAuth(Subject(), Set(WhiskNamespace(ns1, ak1))), WhiskAuth(Subject(), Set(WhiskNamespace(ns2, ak2))))
+ subs foreach (put(authStore, _))
+
+ waitOnView(authStore, ak1, 1)
+ waitOnView(authStore, ak2, 1)
+
+ val s1 = Identity.get(authStore, ns1).futureValue
+ s1.subject shouldBe subs(0).subject
+
+ val s2 = Identity.get(authStore, ak2).futureValue
+ s2.subject shouldBe subs(1).subject
+ }
+
+ it should "not get blocked subject" in {
+ implicit val tid: TransactionId = transid()
+ val ns1 = aname()
+ val ak1 = AuthKey()
+ val auth = new ExtendedAuth(Subject(), Set(WhiskNamespace(ns1, ak1)), blocked = true)
+ put(authStore, auth)
+
+ Identity.get(authStore, ns1).failed.futureValue shouldBe a[NoDocumentException]
+ }
+
+ it should "not find subject when authKey matches partially" in {
+ implicit val tid: TransactionId = transid()
+ val ak1 = AuthKey()
+ val ak2 = AuthKey()
+ val ns1 = aname()
+ val ns2 = aname()
+
+ val auth = WhiskAuth(
+ Subject(),
+ Set(WhiskNamespace(ns1, AuthKey(ak1.uuid, ak2.key)), WhiskNamespace(ns2, AuthKey(ak2.uuid, ak1.key))))
+
+ put(authStore, auth)
+
+ waitOnView(authStore, AuthKey(ak1.uuid, ak2.key), 1)
+ Identity.get(authStore, ak1).failed.futureValue shouldBe a[NoDocumentException]
+ }
+
+ it should "find subject by namespace with limits" in {
+ implicit val tid: TransactionId = transid()
+ val ak1 = AuthKey()
+ val ak2 = AuthKey()
+ val name1 = aname()
+ val name2 = aname()
+ val subs = Array(
+ WhiskAuth(Subject(), Set(WhiskNamespace(name1, ak1))),
+ WhiskAuth(Subject(), Set(WhiskNamespace(name2, ak2))))
+ subs foreach (put(authStore, _))
+
+ waitOnView(authStore, ak1, 1)
+ waitOnView(authStore, ak2, 1)
+
+ val limits = UserLimits(invocationsPerMinute = Some(7), firesPerMinute = Some(31))
+ put(authStore, new LimitEntity(name1, limits))
+
+ val i = Identity.get(authStore, name1).futureValue
+ i.subject shouldBe subs(0).subject
+ i.limits shouldBe limits
+ }
+
+ it should "find blacklisted namespaces" in {
+ implicit val tid: TransactionId = transid()
+
+ val n1 = aname()
+ val n2 = aname()
+ val n3 = aname()
+ val n4 = aname()
+ val n5 = aname()
+
+ val ak1 = AuthKey()
+ val ak2 = AuthKey()
+
+ //Create 3 limits entry where one has limits > 0 thus non blacklisted
+ //And one blocked subject with 2 namespaces
+ val limitsAndAuths = Seq(
+ new LimitEntity(n1, UserLimits(invocationsPerMinute = Some(0))),
+ new LimitEntity(n2, UserLimits(concurrentInvocations = Some(0))),
+ new LimitEntity(n3, UserLimits(invocationsPerMinute = Some(7), concurrentInvocations = Some(7))),
+ new ExtendedAuth(Subject(), Set(WhiskNamespace(n4, ak1), WhiskNamespace(n5, ak2)), blocked = true))
+
+ limitsAndAuths foreach (put(authStore, _))
+
+ //2 for limits
+ //2 for 2 namespace in user blocked
+ waitOnBlacklistView(authStore, 2 + 2)
+
+ //Use contains assertion to ensure that even if same db is used by other setup
+ //we at least get our expected entries
+ val blacklist = new NamespaceBlacklist(authStore)
+ blacklist
+ .refreshBlacklist()
+ .futureValue should contain allElementsOf Seq(n1, n2, n4, n5).map(_.asString).toSet
+ }
+
+ def waitOnBlacklistView(db: AuthStore, count: Int)(implicit transid: TransactionId, timeout: Duration) = {
+ val success = retry(() => {
+ blacklistCount().map { listCount =>
+ if (listCount != count) {
+ throw RetryOp()
+ } else true
+ }
+ }, timeout)
+ assert(success.isSuccess, "wait aborted after: " + timeout + ": " + success)
+ }
+
+ private def blacklistCount()(implicit transid: TransactionId) = {
+ //NamespaceBlacklist uses StaleParameter.UpdateAfter which would lead to race condition
+ //So use actual call here
+ authStore
+ .query(
+ table = NamespaceBlacklist.view.name,
+ startKey = List.empty,
+ endKey = List.empty,
+ skip = 0,
+ limit = Int.MaxValue,
+ includeDocs = false,
+ descending = true,
+ reduce = false,
+ stale = StaleParameter.No)
+ .map(_.map(_.fields("key").convertTo[String]).toSet)
+ .map(_.size)
+ }
+
+ private class LimitEntity(name: EntityName, limits: UserLimits) extends WhiskAuth(Subject(), Set.empty) {
+ override def docid = DocId(s"${name.name}/limits")
+
+ //There is no api to write limits. So piggy back on WhiskAuth but replace auth json
+ //with limits!
+ override def toJson = UserLimits.serdes.write(limits).asJsObject
+ }
+
+ private class ExtendedAuth(subject: Subject, namespaces: Set[WhiskNamespace], blocked: Boolean)
+ extends WhiskAuth(subject, namespaces) {
+ override def toJson = JsObject(super.toJson.fields + ("blocked" -> JsBoolean(blocked)))
+ }
+}
diff --git a/tests/src/test/scala/whisk/core/database/test/behavior/ArtifactStoreWhisksQueryBehaviors.scala b/tests/src/test/scala/whisk/core/database/test/behavior/ArtifactStoreWhisksQueryBehaviors.scala
new file mode 100644
index 0000000000..7733406698
--- /dev/null
+++ b/tests/src/test/scala/whisk/core/database/test/behavior/ArtifactStoreWhisksQueryBehaviors.scala
@@ -0,0 +1,86 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package whisk.core.database.test.behavior
+
+import java.time.Instant
+
+import whisk.common.TransactionId
+import whisk.core.entity.WhiskEntityQueries.TOP
+import whisk.core.entity._
+
+trait ArtifactStoreWhisksQueryBehaviors extends ArtifactStoreBehaviorBase {
+
+ behavior of s"${storeType}ArtifactStore query public packages"
+
+ it should "only list published entities" in {
+ implicit val tid: TransactionId = transid()
+ val ns = newNS()
+ val n1 = aname()
+ val pkgs = Array(
+ WhiskPackage(ns, aname()),
+ WhiskPackage(ns, aname(), publish = true),
+ WhiskPackage(ns, aname(), publish = true, binding = Some(Binding(aname(), aname()))),
+ WhiskPackage(ns, aname(), binding = Some(Binding(aname(), aname()))))
+
+ pkgs foreach (put(entityStore, _))
+
+ waitOnView(entityStore, WhiskPackage, ns, pkgs.length)
+
+ val result =
+ query[WhiskEntity](entityStore, WhiskPackage.view.name, List(ns.asString, 0), List(ns.asString, TOP, TOP))
+ result.size shouldBe pkgs.length
+
+ val resultPublic =
+ query[WhiskEntity](
+ entityStore,
+ WhiskPackage.publicPackagesView.name,
+ List(ns.asString, 0),
+ List(ns.asString, TOP, TOP))
+
+ resultPublic.size shouldBe 1
+ resultPublic.head.fields("value") shouldBe pkgs(1).summaryAsJson
+ }
+
+ it should "list packages between given times" in {
+ implicit val tid: TransactionId = transid()
+
+ val ns = newNS()
+ val pkgs = (1000 until 1100 by 10).map(new TestWhiskPackage(ns, aname(), _))
+
+ pkgs foreach (put(entityStore, _))
+
+ waitOnView(entityStore, WhiskPackage, ns, pkgs.length)
+
+ val resultSince =
+ query[WhiskEntity](entityStore, WhiskPackage.view.name, List(ns.asString, 1050), List(ns.asString, TOP, TOP))
+
+ resultSince.map(_.fields("value")) shouldBe pkgs.reverse.filter(_.updated.toEpochMilli >= 1050).map(_.summaryAsJson)
+
+ val resultBetween =
+ query[WhiskEntity](entityStore, WhiskPackage.view.name, List(ns.asString, 1050), List(ns.asString, 1090, TOP))
+ resultBetween.map(_.fields("value")) shouldBe pkgs.reverse
+ .filter(p => p.updated.toEpochMilli >= 1050 && p.updated.toEpochMilli <= 1090)
+ .map(_.summaryAsJson)
+ }
+
+ private class TestWhiskPackage(override val namespace: EntityPath, override val name: EntityName, updatedTest: Long)
+ extends WhiskPackage(namespace, name) {
+ //Not possible to easily control the updated so need to use this workaround
+ override val updated = Instant.ofEpochMilli(updatedTest)
+ }
+}
diff --git a/tests/src/test/scala/whisk/core/invoker/test/NamespaceBlacklistTests.scala b/tests/src/test/scala/whisk/core/invoker/test/NamespaceBlacklistTests.scala
index 6ab2457af5..914e307469 100644
--- a/tests/src/test/scala/whisk/core/invoker/test/NamespaceBlacklistTests.scala
+++ b/tests/src/test/scala/whisk/core/invoker/test/NamespaceBlacklistTests.scala
@@ -90,6 +90,11 @@ class NamespaceBlacklistTests
}
}
+ override protected def withFixture(test: NoArgTest) = {
+ assume(isCouchStore(authStore))
+ super.withFixture(test)
+ }
+
override def beforeAll() = {
val documents = identities.map { i =>
(i.namespace.name + "/limits", i.limits.toJson.asJsObject)
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services