package objstore

import (
	"context"
	"encoding/json"
	"testing"

	"github.com/aws/aws-sdk-go-v2/aws"
	"github.com/aws/aws-sdk-go-v2/service/iam"
	iamtypes "github.com/aws/aws-sdk-go-v2/service/iam/types"
	awss3 "github.com/aws/aws-sdk-go-v2/service/s3"
	s3types "github.com/aws/aws-sdk-go-v2/service/s3/types"
	"github.com/aws/smithy-go"
)

// TestNewAWSProvisioner is construction-only: it must succeed offline (the
// default credential chain loads without ever making a call), and the
// returned value must satisfy Provisioner.
func TestNewAWSProvisioner(t *testing.T) {
	p, err := NewAWSProvisioner(context.Background(), "eu-west-1", "", "")
	if err != nil {
		t.Fatalf("NewAWSProvisioner: %v", err)
	}
	if p == nil {
		t.Fatal("NewAWSProvisioner returned nil Provisioner")
	}
}

// TestNewAWSProvisionerEndpointOptions checks the four s3compat-mirrored
// options land on the real client when endpoint is set: BaseEndpoint +
// UsePathStyle (addressing) and both checksum-WhenRequired settings (the
// pair that keeps default-on flexible checksums from wrapping bodies in
// aws-chunked framing some S3-compatibles reject — DeleteObjects, EmptyBucket's
// batch delete, is exactly a checksummed request). White-box: reaches into
// the unexported s3 field because this is the only seam that observes what
// awss3.NewFromConfig actually built, short of a live request.
func TestNewAWSProvisionerEndpointOptions(t *testing.T) {
	prov, err := NewAWSProvisioner(context.Background(), "eu-west-1", "http://127.0.0.1:9000", "")
	if err != nil {
		t.Fatal(err)
	}
	p, ok := prov.(*awsProvisioner)
	if !ok {
		t.Fatalf("NewAWSProvisioner returned %T, want *awsProvisioner", prov)
	}
	client, ok := p.s3.(*awss3.Client)
	if !ok {
		t.Fatalf("p.s3 is %T, want *s3.Client", p.s3)
	}
	opts := client.Options()
	if opts.BaseEndpoint == nil || *opts.BaseEndpoint != "http://127.0.0.1:9000" {
		t.Fatalf("BaseEndpoint = %v, want http://127.0.0.1:9000", opts.BaseEndpoint)
	}
	if !opts.UsePathStyle {
		t.Fatal("UsePathStyle = false, want true")
	}
	if opts.RequestChecksumCalculation != aws.RequestChecksumCalculationWhenRequired {
		t.Fatalf("RequestChecksumCalculation = %v, want WhenRequired", opts.RequestChecksumCalculation)
	}
	if opts.ResponseChecksumValidation != aws.ResponseChecksumValidationWhenRequired {
		t.Fatalf("ResponseChecksumValidation = %v, want WhenRequired", opts.ResponseChecksumValidation)
	}
}

// TestNewAWSProvisionerNoEndpointOptions checks the stock-AWS path (empty
// endpoint) leaves those four options untouched, matching s3compat.Options'
// own no-op guarantee for a client that talks to real S3.
func TestNewAWSProvisionerNoEndpointOptions(t *testing.T) {
	prov, err := NewAWSProvisioner(context.Background(), "eu-west-1", "", "")
	if err != nil {
		t.Fatal(err)
	}
	p := prov.(*awsProvisioner)
	client := p.s3.(*awss3.Client)
	opts := client.Options()
	if opts.BaseEndpoint != nil {
		t.Fatalf("BaseEndpoint = %v, want nil", *opts.BaseEndpoint)
	}
	if opts.UsePathStyle {
		t.Fatal("UsePathStyle = true, want false")
	}
}

// ---- stub clients -------------------------------------------------------
//
// Each stub implements the narrow s3API/iamAPI seam with per-method hooks.
// An un-set hook fails the test loudly (via t.Fatalf) rather than panicking
// on a nil func — a test that calls a method it didn't mean to exercise
// should say so.

type stubS3 struct {
	t *testing.T

	createBucket  func(*awss3.CreateBucketInput) (*awss3.CreateBucketOutput, error)
	deleteBucket  func(*awss3.DeleteBucketInput) (*awss3.DeleteBucketOutput, error)
	listObjectsV2 func(*awss3.ListObjectsV2Input) (*awss3.ListObjectsV2Output, error)
	deleteObjects func(*awss3.DeleteObjectsInput) (*awss3.DeleteObjectsOutput, error)
}

func (s *stubS3) CreateBucket(_ context.Context, in *awss3.CreateBucketInput, _ ...func(*awss3.Options)) (*awss3.CreateBucketOutput, error) {
	if s.createBucket == nil {
		s.t.Fatal("unexpected CreateBucket call")
	}
	return s.createBucket(in)
}

func (s *stubS3) DeleteBucket(_ context.Context, in *awss3.DeleteBucketInput, _ ...func(*awss3.Options)) (*awss3.DeleteBucketOutput, error) {
	if s.deleteBucket == nil {
		s.t.Fatal("unexpected DeleteBucket call")
	}
	return s.deleteBucket(in)
}

func (s *stubS3) ListObjectsV2(_ context.Context, in *awss3.ListObjectsV2Input, _ ...func(*awss3.Options)) (*awss3.ListObjectsV2Output, error) {
	if s.listObjectsV2 == nil {
		s.t.Fatal("unexpected ListObjectsV2 call")
	}
	return s.listObjectsV2(in)
}

func (s *stubS3) DeleteObjects(_ context.Context, in *awss3.DeleteObjectsInput, _ ...func(*awss3.Options)) (*awss3.DeleteObjectsOutput, error) {
	if s.deleteObjects == nil {
		s.t.Fatal("unexpected DeleteObjects call")
	}
	return s.deleteObjects(in)
}

type stubIAM struct {
	t *testing.T

	createUser       func(*iam.CreateUserInput) (*iam.CreateUserOutput, error)
	getUser          func(*iam.GetUserInput) (*iam.GetUserOutput, error)
	putUserPolicy    func(*iam.PutUserPolicyInput) (*iam.PutUserPolicyOutput, error)
	deleteUserPolicy func(*iam.DeleteUserPolicyInput) (*iam.DeleteUserPolicyOutput, error)
	deleteUser       func(*iam.DeleteUserInput) (*iam.DeleteUserOutput, error)
	createAccessKey  func(*iam.CreateAccessKeyInput) (*iam.CreateAccessKeyOutput, error)
	deleteAccessKey  func(*iam.DeleteAccessKeyInput) (*iam.DeleteAccessKeyOutput, error)
	listAccessKeys   func(*iam.ListAccessKeysInput) (*iam.ListAccessKeysOutput, error)
}

func (s *stubIAM) CreateUser(_ context.Context, in *iam.CreateUserInput, _ ...func(*iam.Options)) (*iam.CreateUserOutput, error) {
	if s.createUser == nil {
		s.t.Fatal("unexpected CreateUser call")
	}
	return s.createUser(in)
}

func (s *stubIAM) GetUser(_ context.Context, in *iam.GetUserInput, _ ...func(*iam.Options)) (*iam.GetUserOutput, error) {
	if s.getUser == nil {
		s.t.Fatal("unexpected GetUser call")
	}
	return s.getUser(in)
}

func (s *stubIAM) PutUserPolicy(_ context.Context, in *iam.PutUserPolicyInput, _ ...func(*iam.Options)) (*iam.PutUserPolicyOutput, error) {
	if s.putUserPolicy == nil {
		s.t.Fatal("unexpected PutUserPolicy call")
	}
	return s.putUserPolicy(in)
}

func (s *stubIAM) DeleteUserPolicy(_ context.Context, in *iam.DeleteUserPolicyInput, _ ...func(*iam.Options)) (*iam.DeleteUserPolicyOutput, error) {
	if s.deleteUserPolicy == nil {
		s.t.Fatal("unexpected DeleteUserPolicy call")
	}
	return s.deleteUserPolicy(in)
}

func (s *stubIAM) DeleteUser(_ context.Context, in *iam.DeleteUserInput, _ ...func(*iam.Options)) (*iam.DeleteUserOutput, error) {
	if s.deleteUser == nil {
		s.t.Fatal("unexpected DeleteUser call")
	}
	return s.deleteUser(in)
}

func (s *stubIAM) CreateAccessKey(_ context.Context, in *iam.CreateAccessKeyInput, _ ...func(*iam.Options)) (*iam.CreateAccessKeyOutput, error) {
	if s.createAccessKey == nil {
		s.t.Fatal("unexpected CreateAccessKey call")
	}
	return s.createAccessKey(in)
}

func (s *stubIAM) DeleteAccessKey(_ context.Context, in *iam.DeleteAccessKeyInput, _ ...func(*iam.Options)) (*iam.DeleteAccessKeyOutput, error) {
	if s.deleteAccessKey == nil {
		s.t.Fatal("unexpected DeleteAccessKey call")
	}
	return s.deleteAccessKey(in)
}

func (s *stubIAM) ListAccessKeys(_ context.Context, in *iam.ListAccessKeysInput, _ ...func(*iam.Options)) (*iam.ListAccessKeysOutput, error) {
	if s.listAccessKeys == nil {
		s.t.Fatal("unexpected ListAccessKeys call")
	}
	return s.listAccessKeys(in)
}

// ---- idempotence mappings (HARD constraints from Tasks 4-6) -------------

// DeleteKey on an absent key (NoSuchEntityException) must be nil: both
// RotateFinish's PrevKeyID escape hatch and Rotate step-1 re-run this against
// a key that is already gone.
func TestDeleteKeyAbsentIsNil(t *testing.T) {
	p := &awsProvisioner{iam: &stubIAM{t: t,
		deleteAccessKey: func(*iam.DeleteAccessKeyInput) (*iam.DeleteAccessKeyOutput, error) {
			return nil, &iamtypes.NoSuchEntityException{Message: aws.String("gone")}
		},
	}}
	if err := p.DeleteKey(context.Background(), "u1", "AKIAABSENT"); err != nil {
		t.Fatalf("DeleteKey on absent key: %v", err)
	}
}

// DeleteUser on an absent user must be nil, and must delete access keys and
// the inline policy BEFORE DeleteUser — IAM refuses to delete a user that
// still owns either, and this is also where a pending-finish PrevKeyID gets
// reaped during teardown.
func TestDeleteUserAbsentIsNilAndOrdered(t *testing.T) {
	var order []string
	stub := &stubIAM{t: t,
		listAccessKeys: func(*iam.ListAccessKeysInput) (*iam.ListAccessKeysOutput, error) {
			order = append(order, "list")
			return &iam.ListAccessKeysOutput{
				AccessKeyMetadata: []iamtypes.AccessKeyMetadata{
					{AccessKeyId: aws.String("AKIAPENDING")},
				},
				IsTruncated: false,
			}, nil
		},
		deleteAccessKey: func(in *iam.DeleteAccessKeyInput) (*iam.DeleteAccessKeyOutput, error) {
			order = append(order, "deleteKey:"+aws.ToString(in.AccessKeyId))
			return &iam.DeleteAccessKeyOutput{}, nil
		},
		deleteUserPolicy: func(*iam.DeleteUserPolicyInput) (*iam.DeleteUserPolicyOutput, error) {
			order = append(order, "deletePolicy")
			return nil, &iamtypes.NoSuchEntityException{Message: aws.String("gone")}
		},
		deleteUser: func(*iam.DeleteUserInput) (*iam.DeleteUserOutput, error) {
			order = append(order, "deleteUser")
			return nil, &iamtypes.NoSuchEntityException{Message: aws.String("gone")}
		},
	}
	p := &awsProvisioner{iam: stub}
	if err := p.DeleteUser(context.Background(), "u1"); err != nil {
		t.Fatalf("DeleteUser on absent user: %v", err)
	}
	want := []string{"list", "deleteKey:AKIAPENDING", "deletePolicy", "deleteUser"}
	if len(order) != len(want) {
		t.Fatalf("call order = %v, want %v", order, want)
	}
	for i := range want {
		if order[i] != want[i] {
			t.Fatalf("call order = %v, want %v", order, want)
		}
	}
}

// ReapKeys on an absent user (NoSuchEntity from ListAccessKeys) must be nil.
func TestReapKeysAbsentUserIsNil(t *testing.T) {
	p := &awsProvisioner{iam: &stubIAM{t: t,
		listAccessKeys: func(*iam.ListAccessKeysInput) (*iam.ListAccessKeysOutput, error) {
			return nil, &iamtypes.NoSuchEntityException{Message: aws.String("gone")}
		},
	}}
	if err := p.ReapKeys(context.Background(), "ghost", []string{"keep-me"}); err != nil {
		t.Fatalf("ReapKeys on absent user: %v", err)
	}
}

// ReapKeys deletes every key not in keep and leaves the kept ones alone.
func TestReapKeysDeletesUnkept(t *testing.T) {
	deleted := map[string]bool{}
	p := &awsProvisioner{iam: &stubIAM{t: t,
		listAccessKeys: func(*iam.ListAccessKeysInput) (*iam.ListAccessKeysOutput, error) {
			return &iam.ListAccessKeysOutput{
				AccessKeyMetadata: []iamtypes.AccessKeyMetadata{
					{AccessKeyId: aws.String("KEEP")},
					{AccessKeyId: aws.String("REAP1")},
					{AccessKeyId: aws.String("REAP2")},
				},
				IsTruncated: false,
			}, nil
		},
		deleteAccessKey: func(in *iam.DeleteAccessKeyInput) (*iam.DeleteAccessKeyOutput, error) {
			deleted[aws.ToString(in.AccessKeyId)] = true
			return &iam.DeleteAccessKeyOutput{}, nil
		},
	}}
	if err := p.ReapKeys(context.Background(), "u1", []string{"KEEP"}); err != nil {
		t.Fatal(err)
	}
	if !deleted["REAP1"] || !deleted["REAP2"] {
		t.Fatalf("expected REAP1 and REAP2 deleted, got %v", deleted)
	}
	if deleted["KEEP"] {
		t.Fatal("KEEP should not have been deleted")
	}
}

// EmptyBucket on an absent bucket must be nil, including when the NoSuchBucket
// only surfaces on the ListObjectsV2 leg (as a smithy APIError, not the typed
// s3types.NoSuchBucket) — teardown resume after a crash between DeleteBucket
// and DeleteUser re-runs EmptyBucket on a gone bucket.
func TestEmptyBucketAbsentIsNil(t *testing.T) {
	p := &awsProvisioner{s3: &stubS3{t: t,
		listObjectsV2: func(*awss3.ListObjectsV2Input) (*awss3.ListObjectsV2Output, error) {
			return nil, &smithyAPIErrorStub{code: "NoSuchBucket", msg: "The specified bucket does not exist"}
		},
	}}
	if err := p.EmptyBucket(context.Background(), "gone-bucket"); err != nil {
		t.Fatalf("EmptyBucket on absent bucket: %v", err)
	}
}

// Same, but via the typed s3types.NoSuchBucket rather than a generic
// smithy.APIError — both forms must be recognized.
func TestEmptyBucketAbsentTypedErrorIsNil(t *testing.T) {
	p := &awsProvisioner{s3: &stubS3{t: t,
		listObjectsV2: func(*awss3.ListObjectsV2Input) (*awss3.ListObjectsV2Output, error) {
			return nil, &s3types.NoSuchBucket{Message: aws.String("gone")}
		},
	}}
	if err := p.EmptyBucket(context.Background(), "gone-bucket"); err != nil {
		t.Fatalf("EmptyBucket on absent bucket (typed): %v", err)
	}
}

// EmptyBucket batches DeleteObjects at <=1000 keys per call and follows
// ListObjectsV2 pagination.
func TestEmptyBucketBatchesAndPaginates(t *testing.T) {
	page1 := make([]s3types.Object, 1500) // exceeds the 1000 batch ceiling
	for i := range page1 {
		page1[i] = s3types.Object{Key: aws.String("obj1500")}
	}
	page2 := []s3types.Object{{Key: aws.String("obj-page2")}}

	var listCalls int
	var deleteBatchSizes []int
	p := &awsProvisioner{s3: &stubS3{t: t,
		listObjectsV2: func(in *awss3.ListObjectsV2Input) (*awss3.ListObjectsV2Output, error) {
			listCalls++
			if listCalls == 1 {
				return &awss3.ListObjectsV2Output{
					Contents:              page1,
					IsTruncated:           aws.Bool(true),
					NextContinuationToken: aws.String("page2"),
				}, nil
			}
			if aws.ToString(in.ContinuationToken) != "page2" {
				t.Fatalf("expected continuation token page2, got %q", aws.ToString(in.ContinuationToken))
			}
			return &awss3.ListObjectsV2Output{
				Contents:    page2,
				IsTruncated: aws.Bool(false),
			}, nil
		},
		deleteObjects: func(in *awss3.DeleteObjectsInput) (*awss3.DeleteObjectsOutput, error) {
			deleteBatchSizes = append(deleteBatchSizes, len(in.Delete.Objects))
			return &awss3.DeleteObjectsOutput{}, nil
		},
	}}
	if err := p.EmptyBucket(context.Background(), "b1"); err != nil {
		t.Fatal(err)
	}
	if listCalls != 2 {
		t.Fatalf("listCalls = %d, want 2", listCalls)
	}
	want := []int{1000, 500, 1}
	if len(deleteBatchSizes) != len(want) {
		t.Fatalf("delete batch sizes = %v, want %v", deleteBatchSizes, want)
	}
	for i := range want {
		if deleteBatchSizes[i] != want[i] {
			t.Fatalf("delete batch sizes = %v, want %v", deleteBatchSizes, want)
		}
	}
}

// DeleteObjects is a 200-with-body API: a per-key failure lands in
// out.Errors while the call itself reports no err. EmptyBucket must surface
// that as an error naming the count and the first failing key, rather than
// silently returning nil with objects still in the bucket — a caller that
// swallowed this would hang forever retrying DeleteBucket's later
// BucketNotEmpty with no clue which object is stuck.
func TestEmptyBucketSurfacesPerKeyDeleteErrors(t *testing.T) {
	p := &awsProvisioner{s3: &stubS3{t: t,
		listObjectsV2: func(*awss3.ListObjectsV2Input) (*awss3.ListObjectsV2Output, error) {
			return &awss3.ListObjectsV2Output{
				Contents: []s3types.Object{
					{Key: aws.String("locked-obj")},
					{Key: aws.String("other-obj")},
				},
				IsTruncated: aws.Bool(false),
			}, nil
		},
		deleteObjects: func(*awss3.DeleteObjectsInput) (*awss3.DeleteObjectsOutput, error) {
			return &awss3.DeleteObjectsOutput{
				Errors: []s3types.Error{
					{Key: aws.String("locked-obj"), Code: aws.String("AccessDenied"), Message: aws.String("object is under legal hold")},
				},
			}, nil
		},
	}}
	err := p.EmptyBucket(context.Background(), "b1")
	if err == nil {
		t.Fatal("expected an error from a DeleteObjects batch with populated Errors, got nil")
	}
	const want = `objstore: delete objects in "b1": 1 failed, first locked-obj: AccessDenied (object is under legal hold)`
	if err.Error() != want {
		t.Fatalf("error = %q, want %q", err.Error(), want)
	}
}

// DeleteBucket on an absent bucket must be nil.
func TestDeleteBucketAbsentIsNil(t *testing.T) {
	p := &awsProvisioner{s3: &stubS3{t: t,
		deleteBucket: func(*awss3.DeleteBucketInput) (*awss3.DeleteBucketOutput, error) {
			return nil, &s3types.NoSuchBucket{Message: aws.String("gone")}
		},
	}}
	if err := p.DeleteBucket(context.Background(), "gone-bucket"); err != nil {
		t.Fatalf("DeleteBucket on absent bucket: %v", err)
	}
}

// EnsureBucket treats BucketAlreadyOwnedByYou as success and answers the
// configured region.
func TestEnsureBucketAlreadyOwnedByYou(t *testing.T) {
	p := &awsProvisioner{region: "eu-west-1", s3: &stubS3{t: t,
		createBucket: func(*awss3.CreateBucketInput) (*awss3.CreateBucketOutput, error) {
			return nil, &s3types.BucketAlreadyOwnedByYou{Message: aws.String("already yours")}
		},
	}}
	region, err := p.EnsureBucket(context.Background(), "b1")
	if err != nil {
		t.Fatalf("EnsureBucket on already-owned bucket: %v", err)
	}
	if region != "eu-west-1" {
		t.Fatalf("region = %q, want eu-west-1", region)
	}
}

// EnsureBucket does NOT set a location constraint for us-east-1 (the region
// with no LocationConstraint value of its own).
func TestEnsureBucketUsEast1NoLocationConstraint(t *testing.T) {
	var gotConstraint bool
	p := &awsProvisioner{region: "us-east-1", s3: &stubS3{t: t,
		createBucket: func(in *awss3.CreateBucketInput) (*awss3.CreateBucketOutput, error) {
			gotConstraint = in.CreateBucketConfiguration != nil
			return &awss3.CreateBucketOutput{}, nil
		},
	}}
	if _, err := p.EnsureBucket(context.Background(), "b1"); err != nil {
		t.Fatal(err)
	}
	if gotConstraint {
		t.Fatal("us-east-1 must not set CreateBucketConfiguration")
	}
}

// EnsureUser: EntityAlreadyExists resolves the ARN via GetUser, and
// PutUserPolicy always runs (idempotent overwrite) regardless of which
// branch produced the ARN.
func TestEnsureUserAlreadyExists(t *testing.T) {
	var putPolicyCalled bool
	var putDoc string
	p := &awsProvisioner{iam: &stubIAM{t: t,
		createUser: func(*iam.CreateUserInput) (*iam.CreateUserOutput, error) {
			return nil, &iamtypes.EntityAlreadyExistsException{Message: aws.String("exists")}
		},
		getUser: func(in *iam.GetUserInput) (*iam.GetUserOutput, error) {
			return &iam.GetUserOutput{User: &iamtypes.User{
				Arn:      aws.String("arn:aws:iam::123456789012:user/carlos-store/u1"),
				UserName: in.UserName,
			}}, nil
		},
		putUserPolicy: func(in *iam.PutUserPolicyInput) (*iam.PutUserPolicyOutput, error) {
			putPolicyCalled = true
			putDoc = aws.ToString(in.PolicyDocument)
			return &iam.PutUserPolicyOutput{}, nil
		},
	}}
	arn, err := p.EnsureUser(context.Background(), "u1", UserPath, "arn:aws:s3:::b1", nil)
	if err != nil {
		t.Fatal(err)
	}
	if arn != "arn:aws:iam::123456789012:user/carlos-store/u1" {
		t.Fatalf("arn = %q", arn)
	}
	if !putPolicyCalled {
		t.Fatal("PutUserPolicy must run unconditionally, even on the already-exists branch")
	}
	assertStorePolicyDocument(t, putDoc, "arn:aws:s3:::b1")
}

// EnsureUser on a fresh user: ARN comes from CreateUser's own response, and
// the policy document matches the brief exactly.
func TestEnsureUserFresh(t *testing.T) {
	var putDoc string
	p := &awsProvisioner{iam: &stubIAM{t: t,
		createUser: func(in *iam.CreateUserInput) (*iam.CreateUserOutput, error) {
			return &iam.CreateUserOutput{User: &iamtypes.User{
				Arn:      aws.String("arn:aws:iam::123456789012:user/carlos-store/u1"),
				UserName: in.UserName,
				Path:     in.Path,
			}}, nil
		},
		putUserPolicy: func(in *iam.PutUserPolicyInput) (*iam.PutUserPolicyOutput, error) {
			if aws.ToString(in.PolicyName) != policyName {
				t.Fatalf("policy name = %q, want %q", aws.ToString(in.PolicyName), policyName)
			}
			putDoc = aws.ToString(in.PolicyDocument)
			return &iam.PutUserPolicyOutput{}, nil
		},
	}}
	arn, err := p.EnsureUser(context.Background(), "u1", UserPath, "arn:aws:s3:::b1", nil)
	if err != nil {
		t.Fatal(err)
	}
	if arn != "arn:aws:iam::123456789012:user/carlos-store/u1" {
		t.Fatalf("arn = %q", arn)
	}
	assertStorePolicyDocument(t, putDoc, "arn:aws:s3:::b1")
}

// assertStorePolicyDocument checks doc matches the brief's policy shape
// exactly: ListBucket on bucketARN, Get/Put/DeleteObject on bucketARN+"/*".
func assertStorePolicyDocument(t *testing.T, doc, bucketARN string) {
	t.Helper()
	var parsed struct {
		Version   string `json:"Version"`
		Statement []struct {
			Effect   string   `json:"Effect"`
			Action   []string `json:"Action"`
			Resource string   `json:"Resource"`
		} `json:"Statement"`
	}
	if err := json.Unmarshal([]byte(doc), &parsed); err != nil {
		t.Fatalf("policy document not valid JSON: %v\n%s", err, doc)
	}
	if parsed.Version != "2012-10-17" {
		t.Fatalf("Version = %q", parsed.Version)
	}
	// Two Allows, then the two anti-forgery Denys that every store carries
	// whether or not enforcement is on. This helper checks the Allow half
	// exactly; assertNoPutObjectTagging (enforce_test.go) owns the Deny half,
	// including the x-amz-tagging route that an action-name search cannot see.
	if len(parsed.Statement) != 4 {
		t.Fatalf("Statement has %d entries, want 4 (two Allows, two anti-forgery Denys):\n%s", len(parsed.Statement), doc)
	}
	list, objects := parsed.Statement[0], parsed.Statement[1]
	if list.Effect != "Allow" || len(list.Action) != 1 || list.Action[0] != "s3:ListBucket" || list.Resource != bucketARN {
		t.Fatalf("ListBucket statement = %+v", list)
	}
	// s3:GetObjectTagging is here so an app can read the scan verdict on its
	// own object. s3:PutObjectTagging is deliberately absent and must stay
	// absent — but absence alone was never the whole property, and believing
	// it was left the x-amz-tagging upload route open. storePolicyJSON's
	// comment has the story; the Denys are what actually close it.
	wantActions := map[string]bool{
		"s3:GetObject": true, "s3:PutObject": true, "s3:DeleteObject": true,
		"s3:GetObjectTagging": true,
	}
	if objects.Effect != "Allow" || len(objects.Action) != 4 || objects.Resource != bucketARN+"/*" {
		t.Fatalf("object statement = %+v", objects)
	}
	for _, a := range objects.Action {
		if !wantActions[a] {
			t.Fatalf("unexpected action %q in object statement", a)
		}
	}
	assertNoPutObjectTagging(t, doc)
}

// smithyAPIErrorStub is a minimal smithy.APIError for exercising the code
// path where ListObjectsV2 answers NoSuchBucket as a generic API error
// rather than the typed s3types.NoSuchBucket.
type smithyAPIErrorStub struct {
	code string
	msg  string
}

func (e *smithyAPIErrorStub) Error() string                 { return e.code + ": " + e.msg }
func (e *smithyAPIErrorStub) ErrorCode() string             { return e.code }
func (e *smithyAPIErrorStub) ErrorMessage() string          { return e.msg }
func (e *smithyAPIErrorStub) ErrorFault() smithy.ErrorFault { return smithy.FaultUnknown }

var _ smithy.APIError = (*smithyAPIErrorStub)(nil)

// TestEnsureUserStampsThePermissionsBoundary: the boundary is the only thing
// standing between "this role may create IAM users" and "this role may create
// an IAM user with any policy at all", so it has to reach CreateUser on the
// exact call that makes the principal — not be attached afterwards, where a
// crash in between leaves an unbounded user holding a live access key.
func TestEnsureUserStampsThePermissionsBoundary(t *testing.T) {
	const boundary = "arn:aws:iam::1:policy/carlos-store-boundary"
	var seen *string
	iamStub := &stubIAM{
		t: t,
		createUser: func(in *iam.CreateUserInput) (*iam.CreateUserOutput, error) {
			seen = in.PermissionsBoundary
			return &iam.CreateUserOutput{User: &iamtypes.User{
				Arn: aws.String("arn:aws:iam::1:user/carlos-store/carlos-store-bab-titogo"),
			}}, nil
		},
		putUserPolicy: func(*iam.PutUserPolicyInput) (*iam.PutUserPolicyOutput, error) {
			return &iam.PutUserPolicyOutput{}, nil
		},
	}
	p := &awsProvisioner{iam: iamStub, region: "eu-west-1", boundary: boundary}
	if _, err := p.EnsureUser(context.Background(), "carlos-store-bab-titogo", UserPath, "arn:aws:s3:::carlos-bab-titogo", nil); err != nil {
		t.Fatalf("EnsureUser: %v", err)
	}
	if seen == nil {
		t.Fatal("CreateUser was called with NO permissions boundary — the deployment's IAM will refuse this call, and if it ever stops refusing, the user is unbounded")
	}
	if *seen != boundary {
		t.Fatalf("PermissionsBoundary = %q, want %q", *seen, boundary)
	}
}

// An empty boundary must send no field at all rather than an empty string:
// MemProvisioner and the contract tests construct provisioners with no IAM,
// and CreateUser rejects an empty PermissionsBoundary outright.
func TestEnsureUserOmitsAnEmptyBoundary(t *testing.T) {
	sent := true
	iamStub := &stubIAM{
		t: t,
		createUser: func(in *iam.CreateUserInput) (*iam.CreateUserOutput, error) {
			sent = in.PermissionsBoundary != nil
			return &iam.CreateUserOutput{User: &iamtypes.User{Arn: aws.String("arn:aws:iam::1:user/x")}}, nil
		},
		putUserPolicy: func(*iam.PutUserPolicyInput) (*iam.PutUserPolicyOutput, error) {
			return &iam.PutUserPolicyOutput{}, nil
		},
	}
	p := &awsProvisioner{iam: iamStub, region: "eu-west-1"}
	if _, err := p.EnsureUser(context.Background(), "u", UserPath, "arn:aws:s3:::b", nil); err != nil {
		t.Fatalf("EnsureUser: %v", err)
	}
	if sent {
		t.Fatal("PermissionsBoundary was set from an empty boundary; want the field omitted")
	}
}