AWS S3: Empty exception issue #1589

s3/src/main/scala/akka/stream/alpakka/s3/S3Exception.scala

@@ -4,6 +4,8 @@
 
 package akka.stream.alpakka.s3
 
+import akka.http.scaladsl.model.StatusCode
+
 import scala.util.Try
 import scala.xml.{Elem, XML}
 
@@ -23,5 +25,14 @@ class S3Exception(val code: String, val message: String, val requestId: String,
       )
     )
 
-  override def toString = s"${super.toString} (Code: $code, RequestID: $requestId, HostID: $hostId)"
+  def this(response: String, code: StatusCode) =
+    this(
+      Try(XML.loadString(response)).getOrElse(
+        <Error><Code>{code}</Code><Message>{response}</Message><RequestID>-</RequestID><HostID>-</HostID></Error>
+      )
+    )
+
+  override def toString: String =
+    s"${super.toString} (Code: $code, RequestID: $requestId, HostID: $hostId)"
+
 }

s3/src/main/scala/akka/stream/alpakka/s3/impl/S3Stream.scala

@@ -180,10 +180,10 @@ import akka.util.ByteString
             }
           case HttpResponse(NotFound, _, entity, _) =>
             Source.fromFuture(entity.discardBytes().future().map(_ => None))
-          case HttpResponse(_, _, entity, _) =>
+          case HttpResponse(code, _, entity, _) =>
             Source.fromFuture {
               Unmarshal(entity).to[String].map { err =>
-                throw new S3Exception(err)
+                throw new S3Exception(err, code)
               }
             }
         }
@@ -197,10 +197,10 @@ import akka.util.ByteString
         request(s3Location, HttpMethods.DELETE, versionId = versionId).flatMapConcat {
           case HttpResponse(NoContent, _, entity, _) =>
             Source.fromFuture(entity.discardBytes().future().map(_ => Done))
-          case HttpResponse(_, _, entity, _) =>
+          case HttpResponse(code, _, entity, _) =>
             Source.fromFuture {
               Unmarshal(entity).to[String].map { err =>
-                throw new S3Exception(err)
+                throw new S3Exception(err, code)
               }
             }
         }
@@ -240,10 +240,10 @@ import akka.util.ByteString
                   ObjectMetadata(h :+ `Content-Length`(entity.contentLengthOption.getOrElse(0)))
                 }
               }
-            case HttpResponse(_, _, entity, _) =>
+            case HttpResponse(code, _, entity, _) =>
               Source.fromFuture {
                 Unmarshal(entity).to[String].map { err =>
-                  throw new S3Exception(err)
+                  throw new S3Exception(err, code)
                 }
               }
           }
@@ -331,9 +331,9 @@ import akka.util.ByteString
     response match {
       case HttpResponse(status, _, entity, _) if status.isSuccess() =>
         entity.discardBytes().future()
-      case HttpResponse(_, _, entity, _) =>
+      case HttpResponse(code, _, entity, _) =>
         Unmarshal(entity).to[String].map { err =>
-          throw new S3Exception(err)
+          throw new S3Exception(err, code)
         }
     }
   }
@@ -357,9 +357,9 @@ import akka.util.ByteString
                 case other => throw new IllegalArgumentException(s"received status $other")
               }
           )
-      case HttpResponse(_, _, entity, _) =>
+      case HttpResponse(code, _, entity, _) =>
         Unmarshal(entity).to[String].map { err =>
-          throw new S3Exception(err)
+          throw new S3Exception(err, code)
         }
     }
   }
@@ -391,10 +391,10 @@ import akka.util.ByteString
         signAndRequest(req).flatMapConcat {
           case HttpResponse(status, _, entity, _) if status.isSuccess() =>
             Source.fromFuture(Unmarshal(entity).to[MultipartUpload])
-          case HttpResponse(_, _, entity, _) =>
+          case HttpResponse(code, _, entity, _) =>
             Source.fromFuture {
               Unmarshal(entity).to[String].map { err =>
-                throw new S3Exception(err)
+                throw new S3Exception(err, code)
               }
             }
         }
@@ -685,9 +685,9 @@ import akka.util.ByteString
     resp match {
       case HttpResponse(status, headers, entity, _) if status.isSuccess() && !status.isRedirection() =>
         Future.successful((entity, headers))
-      case HttpResponse(_, _, entity, _) =>
+      case HttpResponse(code, _, entity, _) =>
         Unmarshal(entity).to[String].map { err =>
-          throw new S3Exception(err)
+          throw new S3Exception(err, code)
         }
     }
   }
@@ -761,7 +761,7 @@ import akka.util.ByteString
               case statusCode: StatusCode =>
                 Unmarshal(entity).to[String].map { err =>
                   val response = Option(err).getOrElse(s"Failed to upload part into S3, status code was: $statusCode")
-                  throw new S3Exception(response)
+                  throw new S3Exception(response, statusCode)
                 }
             }
 

s3/src/test/scala/akka/stream/alpakka/s3/scaladsl/S3IntegrationSpec.scala

@@ -8,7 +8,7 @@ import java.util.concurrent.TimeUnit
 
 import akka.{Done, NotUsed}
 import akka.actor.ActorSystem
-import akka.http.scaladsl.model.ContentTypes
+import akka.http.scaladsl.model.{ContentTypes, StatusCodes}
 import akka.stream.ActorMaterializer
 import akka.stream.alpakka.s3.BucketAccess.{AccessGranted, NotExists}
 import akka.stream.alpakka.s3._
@@ -58,14 +58,14 @@ trait S3IntegrationSpec extends FlatSpecLike with BeforeAndAfterAll with Matcher
     S3Settings().withPathStyleAccess(true).withS3RegionProvider(otherRegionProvider)
   def listBucketVersion1Settings =
     S3Settings().withListBucketApiVersion(ApiVersion.ListBucketVersion1)
+  def invalidCredentials = S3Settings() //Empty settings to be override in MinioSpec
 
   def defaultRegionContentCount = 4
   def otherRegionContentCount = 5
 
   it should "list with real credentials" in {
     val result = S3
       .listBucket(defaultRegionBucket, None)
-      //.addAttributes(S3Attributes.settings(settings))
       .runWith(Sink.seq)
 
     val listingResult = result.futureValue
@@ -138,7 +138,6 @@ trait S3IntegrationSpec extends FlatSpecLike with BeforeAndAfterAll with Matcher
 
   it should "upload multipart with real credentials" in {
     val source: Source[ByteString, Any] = Source(ByteString(objectValue) :: Nil)
-    //val source: Source[ByteString, Any] = FileIO.fromPath(Paths.get("/tmp/IMG_0470.JPG"))
 
     val result =
       source
@@ -240,16 +239,16 @@ trait S3IntegrationSpec extends FlatSpecLike with BeforeAndAfterAll with Matcher
     S3.deleteObject(defaultRegionBucket, objectKey).runWith(Sink.head).futureValue shouldEqual akka.Done
   }
 
-  it should "upload, download and delete with spaces in the key in non us-east-1 zone" in uploadDownloadAndDeteleInOtherRegionCase(
+  it should "upload, download and delete with spaces in the key in non us-east-1 zone" in uploadDownloadAndDeleteInOtherRegionCase(
     "test folder/test file.txt"
   )
 
   // we want ASCII and other UTF-8 characters!
-  it should "upload, download and delete with special characters in the key in non us-east-1 zone" in uploadDownloadAndDeteleInOtherRegionCase(
+  it should "upload, download and delete with special characters in the key in non us-east-1 zone" in uploadDownloadAndDeleteInOtherRegionCase(
     "føldęrü/1234()[]><!? .TXT"
   )
 
-  it should "upload, download and delete with `+` character in the key in non us-east-1 zone" in uploadDownloadAndDeteleInOtherRegionCase(
+  it should "upload, download and delete with `+` character in the key in non us-east-1 zone" in uploadDownloadAndDeleteInOtherRegionCase(
     "1 + 2 = 3"
   )
 
@@ -404,10 +403,23 @@ trait S3IntegrationSpec extends FlatSpecLike with BeforeAndAfterAll with Matcher
     }
   }
 
+  it should "contain error code even if exception in empty" in {
+    val exception = intercept[S3Exception] {
+      val result = S3
+        .getObjectMetadata(defaultRegionBucket, "sample")
+        .withAttributes(S3Attributes.settings(invalidCredentials))
+        .runWith(Sink.head)
+
+      Await.result(result, Duration(1, TimeUnit.MINUTES))
+    }
+
+    exception.code shouldBe StatusCodes.Forbidden.toString()
+  }
+
   private def deleteBucketAfterTest(bucketName: String): Unit =
     Await.result(S3.deleteBucket(bucketName), Duration(1, TimeUnit.MINUTES))
 
-  private def uploadDownloadAndDeteleInOtherRegionCase(objectKey: String): Assertion = {
+  private def uploadDownloadAndDeleteInOtherRegionCase(objectKey: String): Assertion = {
     val source: Source[ByteString, Any] = Source(ByteString(objectValue) :: Nil)
 
     val results = for {
@@ -477,6 +489,10 @@ class MinioS3IntegrationSpec extends S3IntegrationSpec {
     new BasicAWSCredentials(accessKey, secret)
   )
 
+  val invalidCredentialsProvider = new AWSStaticCredentialsProvider(
+    new BasicAWSCredentials(invalidAccessKey, invalidSecret)
+  )
+
   override val defaultRegionContentCount = 0
   override val otherRegionContentCount = 0
 
@@ -500,6 +516,12 @@ class MinioS3IntegrationSpec extends S3IntegrationSpec {
       .withEndpointUrl(endpointUrl)
       .withPathStyleAccess(true)
 
+  override def invalidCredentials: S3Settings =
+    S3Settings()
+      .withCredentialsProvider(invalidCredentialsProvider)
+      .withEndpointUrl(endpointUrl)
+      .withPathStyleAccess(true)
+
   it should "properly set the endpointUrl" in {
     S3Settings().endpointUrl.value shouldEqual endpointUrl
   }
@@ -509,4 +531,7 @@ object MinioS3IntegrationSpec {
   val accessKey = "TESTKEY"
   val secret = "TESTSECRET"
   val endpointUrl = "http://localhost:9000"
+
+  val invalidAccessKey = ""
+  val invalidSecret = ""
 }