package azblob

import (
	"bytes"
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"io/ioutil"
	"math/rand"
	"net/http"
	"net/url"
	"os"
	"reflect"
	"runtime"
	"strings"
	"testing"
	"time"

	chk "gopkg.in/check.v1"

	"github.com/Azure/azure-pipeline-go/pipeline"
	"github.com/Azure/go-autorest/autorest/adal"
)

// For testing docs, see: https://labix.org/gocheck
// To test a specific test: go test -check.f MyTestSuite

// Hookup to the testing framework
func Test(t *testing.T) { chk.TestingT(t) }

type aztestsSuite struct{}

var _ = chk.Suite(&aztestsSuite{})

func (s *aztestsSuite) TestRetryPolicyRetryReadsFromSecondaryHostField(c *chk.C) {
	_, found := reflect.TypeOf(RetryOptions{}).FieldByName("RetryReadsFromSecondaryHost")
	if !found {
		// Make sure the RetryOption was not erroneously overwritten
		c.Fatal("RetryOption's RetryReadsFromSecondaryHost field must exist in the Blob SDK - uncomment it and make sure the field is returned from the retryReadsFromSecondaryHost() method too!")
	}
}

const (
	containerPrefix             = "go"
	blobPrefix                  = "gotestblob"
	blockBlobDefaultData        = "GoBlockBlobData"
	validationErrorSubstring    = "validation failed"
	invalidHeaderErrorSubstring = "invalid header field" // error thrown by the http client
)

var ctx = context.Background()
var basicHeaders = BlobHTTPHeaders{
	ContentType:        "my_type",
	ContentDisposition: "my_disposition",
	CacheControl:       "control",
	ContentMD5:         nil,
	ContentLanguage:    "my_language",
	ContentEncoding:    "my_encoding",
}

var basicMetadata = Metadata{"foo": "bar"}

type testPipeline struct{}

const testPipelineMessage string = "Test factory invoked"

func (tm testPipeline) Do(ctx context.Context, methodFactory pipeline.Factory, request pipeline.Request) (pipeline.Response, error) {
	return nil, errors.New(testPipelineMessage)
}

// This function generates an entity name by concatenating the passed prefix,
// the name of the test requesting the entity name, and the minute, second, and nanoseconds of the call.
// This should make it easy to associate the entities with their test, uniquely identify
// them, and determine the order in which they were created.
// Note that this imposes a restriction on the length of test names
func generateName(prefix string) string {
	// These next lines up through the for loop are obtaining and walking up the stack
	// trace to extrat the test name, which is stored in name
	pc := make([]uintptr, 10)
	runtime.Callers(0, pc)
	frames := runtime.CallersFrames(pc)
	name := ""
	for f, next := frames.Next(); next; f, next = frames.Next() {
		name = f.Function
		if strings.Contains(name, "Suite") {
			break
		}
	}
	funcNameStart := strings.Index(name, "Test")
	name = name[funcNameStart+len("Test"):] // Just get the name of the test and not any of the garbage at the beginning
	name = strings.ToLower(name)            // Ensure it is a valid resource name
	currentTime := time.Now()
	name = fmt.Sprintf("%s%s%d%d%d", prefix, strings.ToLower(name), currentTime.Minute(), currentTime.Second(), currentTime.Nanosecond())
	return name
}

func generateContainerName() string {
	return generateName(containerPrefix)
}

func generateBlobName() string {
	return generateName(blobPrefix)
}

func getContainerURL(c *chk.C, bsu ServiceURL) (container ContainerURL, name string) {
	name = generateContainerName()
	container = bsu.NewContainerURL(name)

	return container, name
}

func getBlockBlobURL(c *chk.C, container ContainerURL) (blob BlockBlobURL, name string) {
	name = generateBlobName()
	blob = container.NewBlockBlobURL(name)

	return blob, name
}

func getAppendBlobURL(c *chk.C, container ContainerURL) (blob AppendBlobURL, name string) {
	name = generateBlobName()
	blob = container.NewAppendBlobURL(name)

	return blob, name
}

func getPageBlobURL(c *chk.C, container ContainerURL) (blob PageBlobURL, name string) {
	name = generateBlobName()
	blob = container.NewPageBlobURL(name)

	return
}

func getReaderToRandomBytes(n int) *bytes.Reader {
	r, _ := getRandomDataAndReader(n)
	return r
}

func getRandomDataAndReader(n int) (*bytes.Reader, []byte) {
	data := make([]byte, n, n)
	rand.Read(data)
	return bytes.NewReader(data), data
}

func createNewContainer(c *chk.C, bsu ServiceURL) (container ContainerURL, name string) {
	container, name = getContainerURL(c, bsu)

	cResp, err := container.Create(ctx, nil, PublicAccessNone)
	c.Assert(err, chk.IsNil)
	c.Assert(cResp.StatusCode(), chk.Equals, 201)
	return container, name
}

func createNewContainerWithSuffix(c *chk.C, bsu ServiceURL, suffix string) (container ContainerURL, name string) {
	// The goal of adding the suffix is to be able to predetermine what order the containers will be in when listed.
	// We still need the container prefix to come first, though, to ensure only containers as a part of this test
	// are listed at all.
	name = generateName(containerPrefix + suffix)
	container = bsu.NewContainerURL(name)

	cResp, err := container.Create(ctx, nil, PublicAccessNone)
	c.Assert(err, chk.IsNil)
	c.Assert(cResp.StatusCode(), chk.Equals, 201)
	return container, name
}

func createNewBlockBlob(c *chk.C, container ContainerURL) (blob BlockBlobURL, name string) {
	blob, name = getBlockBlobURL(c, container)

	cResp, err := blob.Upload(ctx, strings.NewReader(blockBlobDefaultData), BlobHTTPHeaders{}, nil, BlobAccessConditions{}, DefaultAccessTier, nil, ClientProvidedKeyOptions{}, ImmutabilityPolicyOptions{})

	c.Assert(err, chk.IsNil)
	c.Assert(cResp.StatusCode(), chk.Equals, 201)

	return
}

func createNewBlockBlobWithCPK(c *chk.C, container ContainerURL, cpk ClientProvidedKeyOptions) (blob BlockBlobURL, name string) {
	blob, name = getBlockBlobURL(c, container)

	cResp, err := blob.Upload(ctx, strings.NewReader(blockBlobDefaultData), BlobHTTPHeaders{},
		nil, BlobAccessConditions{}, DefaultAccessTier, nil, cpk, ImmutabilityPolicyOptions{})
	c.Assert(err, chk.IsNil)
	c.Assert(cResp.StatusCode(), chk.Equals, 201)
	return
}

func createNewAppendBlob(c *chk.C, container ContainerURL) (blob AppendBlobURL, name string) {
	blob, name = getAppendBlobURL(c, container)

	resp, err := blob.Create(ctx, BlobHTTPHeaders{}, nil, BlobAccessConditions{}, nil, ClientProvidedKeyOptions{}, ImmutabilityPolicyOptions{})

	c.Assert(err, chk.IsNil)
	c.Assert(resp.StatusCode(), chk.Equals, 201)
	return
}

func createNewAppendBlobWithCPK(c *chk.C, container ContainerURL, cpk ClientProvidedKeyOptions) (blob AppendBlobURL, name string) {
	blob, name = getAppendBlobURL(c, container)

	resp, err := blob.Create(ctx, BlobHTTPHeaders{}, nil, BlobAccessConditions{}, nil, cpk, ImmutabilityPolicyOptions{})
	c.Assert(err, chk.IsNil)
	c.Assert(resp.StatusCode(), chk.Equals, 201)
	return
}

func createNewPageBlob(c *chk.C, container ContainerURL) (blob PageBlobURL, name string) {
	blob, name = getPageBlobURL(c, container)

	resp, err := blob.Create(ctx, PageBlobPageBytes*10, 0, BlobHTTPHeaders{}, nil, BlobAccessConditions{}, DefaultPremiumBlobAccessTier, nil, ClientProvidedKeyOptions{}, ImmutabilityPolicyOptions{})
	c.Assert(err, chk.IsNil)
	c.Assert(resp.StatusCode(), chk.Equals, 201)
	return
}

func createNewPageBlobWithSize(c *chk.C, container ContainerURL, sizeInBytes int64) (blob PageBlobURL, name string) {
	blob, name = getPageBlobURL(c, container)

	resp, err := blob.Create(ctx, sizeInBytes, 0, BlobHTTPHeaders{}, nil, BlobAccessConditions{}, DefaultPremiumBlobAccessTier, nil, ClientProvidedKeyOptions{}, ImmutabilityPolicyOptions{})
	c.Assert(err, chk.IsNil)
	c.Assert(resp.StatusCode(), chk.Equals, 201)
	return
}

func createNewPageBlobWithCPK(c *chk.C, container ContainerURL, sizeInBytes int64, cpk ClientProvidedKeyOptions) (blob PageBlobURL, name string) {
	blob, name = getPageBlobURL(c, container)

	resp, err := blob.Create(ctx, sizeInBytes, 0, BlobHTTPHeaders{}, nil, BlobAccessConditions{}, DefaultPremiumBlobAccessTier, nil, cpk, ImmutabilityPolicyOptions{})
	c.Assert(err, chk.IsNil)
	c.Assert(resp.StatusCode(), chk.Equals, 201)
	return
}

func createBlockBlobWithPrefix(c *chk.C, container ContainerURL, prefix string) (blob BlockBlobURL, name string) {
	name = prefix + generateName(blobPrefix)
	blob = container.NewBlockBlobURL(name)

	cResp, err := blob.Upload(ctx, strings.NewReader(blockBlobDefaultData), BlobHTTPHeaders{}, nil, BlobAccessConditions{}, DefaultAccessTier, nil, ClientProvidedKeyOptions{}, ImmutabilityPolicyOptions{})

	c.Assert(err, chk.IsNil)
	c.Assert(cResp.StatusCode(), chk.Equals, 201)
	return
}

func getARMContainerURI(containerName string) (*url.URL, error) {
	// this makes testing harder to reproduce for potential contributors, so I'm not a huge fan, but it's mandatory.
	subscriptionId := os.Getenv("SUBSCRIPTION_ID")
	rgName := os.Getenv("RESOURCE_GROUP_NAME")
	accountName := os.Getenv("ACCOUNT_NAME")

	if subscriptionId == "" || rgName == "" || accountName == "" {
		return nil, errors.New("the SUBSCRIPTION_ID, RESOURCE_GROUP_NAME, or ACCOUNT_NAME environment variable was not specified")
	}

	fURI := fmt.Sprintf("https://management.azure.com/subscriptions/%s/resourceGroups/%s/providers/Microsoft.Storage/storageAccounts/%s/blobServices/default/containers/%s?api-version=2021-04-01",
		subscriptionId,
		rgName,
		accountName,
		containerName,
	)

	return url.Parse(fURI)
}

func createNewContainerWithVersionLevelWORM(c *chk.C, bsu ServiceURL) (ContainerURL, string) {
	cURL, name := getContainerURL(c, bsu)
	deleteContainer(c, cURL, false) // ensure that the container doesn't already exist
	cURI, err := getARMContainerURI(name)
	c.Assert(err, chk.IsNil)

	type j map[string]interface{}

	body, err := json.Marshal(
		j{
			"properties": j{
				"immutableStorageWithVersioning": j{
					"enabled": true,
				},
			},
		},
	)

	buf := bytes.NewBuffer(body)

	req, err := http.NewRequest("PUT", cURI.String(), buf)
	c.Assert(err, chk.IsNil)

	cred, err := getOAuthCredential("", "https://management.azure.com/")
	c.Assert(err, chk.IsNil)

	req.Header["Authorization"] = []string{"Bearer " + cred.Token()}

	resp, err := http.DefaultClient.Do(req)
	c.Assert(err, chk.IsNil)
	defer resp.Body.Close()
	// 200 is bad-- We want a *new* container.
	if resp.StatusCode != 201 {
		buf, err := ioutil.ReadAll(resp.Body)
		c.Assert(err, chk.IsNil)

		c.Log(string(buf))

		c.Assert(resp.StatusCode, chk.Equals, 201)
	}

	return cURL, name
}

func deleteContainer(c *chk.C, container ContainerURL, hasImmutability bool) {
	boolv := func(b *bool) bool {
		if b == nil {
			return false
		}

		return *b
	}

	if !hasImmutability {
		_, err := container.Delete(ctx, ContainerAccessConditions{})
		if err != nil {
			if stgErr, ok := err.(StorageError); ok {
				if stgErr.ServiceCode() == ServiceCodeContainerNotFound {
					return
				}
			}

			c.Assert(err, chk.IsNil)
		}
	} else {
		// Kill immutability policies
		var m Marker
		for m.NotDone() {
			resp, err := container.ListBlobsFlatSegment(ctx, m, ListBlobsSegmentOptions{Details: BlobListingDetails{LegalHold: true, ImmutabilityPolicy: true}})

			if stgErr, ok := err.(StorageError); ok {
				if stgErr.ServiceCode() == ServiceCodeContainerNotFound {
					return
				}
			}

			c.Assert(err, chk.IsNil)

			for _, v := range resp.Segment.BlobItems {
				if boolv(v.Properties.LegalHold) {
					_, err := container.NewBlobURL(v.Name).SetLegalHold(ctx, false)
					c.Assert(err, chk.IsNil)
				}
				if v.Properties.ImmutabilityPolicyMode != BlobImmutabilityPolicyModeMutable {
					_, err := container.NewBlobURL(v.Name).DeleteImmutabilityPolicy(ctx)
					c.Assert(err, chk.IsNil)
				}

				_, err := container.NewBlobURL(v.Name).Delete(ctx, DeleteSnapshotsOptionInclude, BlobAccessConditions{})
				c.Assert(err, chk.IsNil)
			}

			m = resp.NextMarker
		}

		// While I would definitely prefer not to write my own REST API requests... azure-sdk-for-go's armcore tries to import now nonexistent APIs.
		cred, err := getOAuthCredential("", "https://management.azure.com/")
		c.Assert(err, chk.IsNil)

		cURLParts := NewBlobURLParts(container.URL())
		armURI, err := getARMContainerURI(cURLParts.ContainerName)
		c.Assert(err, chk.IsNil)

		req, err := http.NewRequest("DELETE", armURI.String(), nil)
		c.Assert(err, chk.IsNil)

		// set authorization
		req.Header["Authorization"] = []string{"Bearer " + cred.Token()}

		resp, err := http.DefaultClient.Do(req)
		c.Assert(err, chk.IsNil)
		defer resp.Body.Close()

		if !(resp.StatusCode == 200 || resp.StatusCode == 204) {
			buf, err := ioutil.ReadAll(resp.Body)
			c.Assert(err, chk.IsNil)

			c.Log(string(buf))

			c.Assert(resp.StatusCode == 200 || resp.StatusCode == 204, chk.Equals, true)
		}
	}
}

func getGenericCredential(accountType string) (*SharedKeyCredential, error) {
	accountNameEnvVar := accountType + "ACCOUNT_NAME"
	accountKeyEnvVar := accountType + "ACCOUNT_KEY"
	accountName, accountKey := os.Getenv(accountNameEnvVar), os.Getenv(accountKeyEnvVar)
	if accountName == "" || accountKey == "" {
		return nil, errors.New(accountNameEnvVar + " and/or " + accountKeyEnvVar + " environment variables not specified.")
	}
	return NewSharedKeyCredential(accountName, accountKey)
}

//getOAuthCredential can intake a OAuth credential from environment variables in one of the following ways:
//Direct: Supply a ADAL OAuth token in OAUTH_TOKEN and application ID in APPLICATION_ID to refresh the supplied token.
//Client secret: Supply a client secret in CLIENT_SECRET and application ID in APPLICATION_ID for SPN auth.
//TENANT_ID is optional and will be inferred as common if it is not explicitly defined.
func getOAuthCredential(accountType string, resource string) (TokenCredential, error) {
	oauthTokenEnvVar := accountType + "OAUTH_TOKEN"
	clientSecretEnvVar := accountType + "CLIENT_SECRET"
	applicationIdEnvVar := accountType + "APPLICATION_ID"
	tenantIdEnvVar := accountType + "TENANT_ID"
	oauthToken, appId, tenantId, clientSecret := []byte(os.Getenv(oauthTokenEnvVar)), os.Getenv(applicationIdEnvVar), os.Getenv(tenantIdEnvVar), os.Getenv(clientSecretEnvVar)
	if (len(oauthToken) == 0 && clientSecret == "") || appId == "" {
		return nil, errors.New("(" + oauthTokenEnvVar + " OR " + clientSecretEnvVar + ") and/or " + applicationIdEnvVar + " environment variables not specified.")
	}
	if tenantId == "" {
		tenantId = "common"
	}
	if resource == "" {
		resource = "https://storage.azure.com"
	}

	var Token adal.Token
	if len(oauthToken) != 0 {
		if err := json.Unmarshal(oauthToken, &Token); err != nil {
			return nil, err
		}
	}

	var spt *adal.ServicePrincipalToken

	oauthConfig, err := adal.NewOAuthConfig("https://login.microsoftonline.com", tenantId)
	if err != nil {
		return nil, err
	}

	if len(oauthToken) == 0 {
		spt, err = adal.NewServicePrincipalToken(
			*oauthConfig,
			appId,
			clientSecret,
			resource)
		if err != nil {
			return nil, err
		}
	} else {
		spt, err = adal.NewServicePrincipalTokenFromManualToken(*oauthConfig,
			appId,
			resource,
			Token,
		)
		if err != nil {
			return nil, err
		}
	}

	err = spt.Refresh()
	if err != nil {
		return nil, err
	}

	tc := NewTokenCredential(spt.Token().AccessToken, func(tc TokenCredential) time.Duration {
		_ = spt.Refresh()
		return time.Until(spt.Token().Expires())
	})

	return tc, nil
}

func getGenericBSU(accountType string) (ServiceURL, error) {
	credential, err := getGenericCredential(accountType)
	if err != nil {
		return ServiceURL{}, err
	}

	pipeline := NewPipeline(credential, PipelineOptions{})
	blobPrimaryURL, _ := url.Parse("https://" + credential.AccountName() + ".blob.core.windows.net/")
	return NewServiceURL(*blobPrimaryURL, pipeline), nil
}

func getBSU() ServiceURL {
	bsu, err := getGenericBSU("")

	if err != nil {
		fmt.Println(err)
	}

	return bsu
}

func getAlternateBSU() (ServiceURL, error) {
	return getGenericBSU("SECONDARY_")
}

func getPremiumBSU() (ServiceURL, error) {
	return getGenericBSU("PREMIUM_")
}

func getBlobStorageBSU() (ServiceURL, error) {
	return getGenericBSU("BLOB_STORAGE_")
}

func validateStorageError(c *chk.C, err error, code ServiceCodeType) {
	serr, _ := err.(StorageError)
	c.Assert(serr.ServiceCode(), chk.Equals, code)
}

func getRelativeTimeGMT(amount time.Duration) time.Time {
	currentTime := time.Now().In(time.FixedZone("GMT", 0))
	currentTime = currentTime.Add(amount * time.Second)
	return currentTime
}

func generateCurrentTimeWithModerateResolution() time.Time {
	highResolutionTime := time.Now().UTC()
	return time.Date(highResolutionTime.Year(), highResolutionTime.Month(), highResolutionTime.Day(), highResolutionTime.Hour(), highResolutionTime.Minute(),
		highResolutionTime.Second(), 0, highResolutionTime.Location())
}

// Some tests require setting service properties. It can take up to 30 seconds for the new properties to be reflected across all FEs.
// We will enable the necessary property and try to run the test implementation. If it fails with an error that should be due to
// those changes not being reflected yet, we will wait 30 seconds and try the test again. If it fails this time for any reason,
// we fail the test. It is the responsibility of the the testImplFunc to determine which error string indicates the test should be retried.
// There can only be one such string. All errors that cannot be due to this detail should be asserted and not returned as an error string.
func runTestRequiringServiceProperties(c *chk.C, bsu ServiceURL, code string,
	enableServicePropertyFunc func(*chk.C, ServiceURL),
	testImplFunc func(*chk.C, ServiceURL) error,
	disableServicePropertyFunc func(*chk.C, ServiceURL)) {
	enableServicePropertyFunc(c, bsu)
	defer disableServicePropertyFunc(c, bsu)
	err := testImplFunc(c, bsu)
	// We cannot assume that the error indicative of slow update will necessarily be a StorageError. As in ListBlobs.
	if err != nil && err.Error() == code {
		time.Sleep(time.Second * 30)
		err = testImplFunc(c, bsu)
		c.Assert(err, chk.IsNil)
	}
}

func enableSoftDelete(c *chk.C, bsu ServiceURL) {
	days := int32(1)
	_, err := bsu.SetProperties(ctx, StorageServiceProperties{DeleteRetentionPolicy: &RetentionPolicy{Enabled: true, Days: &days}})
	c.Assert(err, chk.IsNil)
}

func disableSoftDelete(c *chk.C, bsu ServiceURL) {
	_, err := bsu.SetProperties(ctx, StorageServiceProperties{DeleteRetentionPolicy: &RetentionPolicy{Enabled: false}})
	c.Assert(err, chk.IsNil)
}

func validateUpload(c *chk.C, blobURL BlockBlobURL) {
	resp, err := blobURL.Download(ctx, 0, 0, BlobAccessConditions{}, false, ClientProvidedKeyOptions{})
	c.Assert(err, chk.IsNil)
	data, _ := ioutil.ReadAll(resp.Response().Body)
	c.Assert(data, chk.HasLen, 0)
}