Draft: CCLF lambda changes

bcda/cclf/parser_test.go

@@ -293,3 +293,30 @@ func TestGetCCLFMetadata(t *testing.T) {
 		})
 	}
 }
+
+func TestCheckIfAttributionCSVFile(t *testing.T) {
+	tests := []struct {
+		name      string
+		path      string
+		testIsCSV bool
+	}{
+		{name: "Is an Attribution CSV File path", path: "P.PCPB.M2014.D00302.T2420001", testIsCSV: true},
+		{name: "Is not an Attribution CSV File path (incorrect first)", path: "T.PCPB.M2014.D00302.T2420001", testIsCSV: false},
+		{name: "Is not an Attribution CSV File path (incorrect second)", path: "P.BFD.N2014.D00302.T2420001", testIsCSV: false},
+		{name: "Is not an Attribution CSV File path (incorrect third)", path: "P.PCPB.M2014.D00302.T2420001", testIsCSV: false},
+		{name: "Is not an Attribution CSV File path (incorrect fourth)", path: "P.PCPB.M2014.D003022.T2420001", testIsCSV: false},
+		{name: "Is not an Attribution CSV File path (incorrect fifth)", path: "P.PCPB.M2014.D00302.T24200011", testIsCSV: false},
+		{name: "Is not an Attribution CSV File path (CCLF file)", path: "T.BCD.A0001.ZCY18.D181121.T1000000", testIsCSV: false},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(sub *testing.T) {
+			isCSV := CheckIfAttributionCSVFile(test.path)
+			if test.testIsCSV {
+				assert.True(sub, isCSV)
+			} else {
+				assert.False(sub, isCSV)
+			}
+		})
+	}
+}

bcda/lambda/cclf/main.go

@@ -30,11 +30,11 @@ func main() {
 			fmt.Println(res)
 		}
 	} else {
-		lambda.Start(cclfImportHandler)
+		lambda.Start(attributionImportHandler)
 	}
 }
 
-func cclfImportHandler(ctx context.Context, sqsEvent events.SQSEvent) (string, error) {
+func attributionImportHandler(ctx context.Context, sqsEvent events.SQSEvent) (string, error) {
 	env := conf.GetEnv("ENV")
 	appName := conf.GetEnv("APP_NAME")
 	logger := configureLogger(env, appName)
@@ -60,14 +60,56 @@ func cclfImportHandler(ctx context.Context, sqsEvent events.SQSEvent) (string, e
 			// importing the one file that was sent in the trigger.
 			filepath := fmt.Sprintf("%s/%s", e.S3.Bucket.Name, e.S3.Object.Key)
 			logger.Infof("Reading %s event for file %s", e.EventName, filepath)
-			return handleCclfImport(s3AssumeRoleArn, filepath)
+			if cclf.CheckIfAttributionCSVFile(e.S3.Object.Key) {
+				return handleCSVImport(s3AssumeRoleArn, filepath)
+			} else {
+				return handleCclfImport(s3AssumeRoleArn, filepath)
+			}
 		}
 	}
 
 	logger.Info("No ObjectCreated events found, skipping safely.")
 	return "", nil
 }
 
+func handleCSVImport(s3AssumeRoleArn, s3ImportPath string) (string, error) {
+	env := conf.GetEnv("ENV")
+	appName := conf.GetEnv("APP_NAME")
+	logger := configureLogger(env, appName)
+	logger = logger.WithFields(logrus.Fields{"import_filename": s3ImportPath})
+
+	importer := cclf.CSVImporter{
+		Logger: logger,
+		FileProcessor: &cclf.S3FileProcessor{
+			Handler: optout.S3FileHandler{
+				Logger:        logger,
+				Endpoint:      os.Getenv("LOCAL_STACK_ENDPOINT"),
+				AssumeRoleArn: s3AssumeRoleArn,
+			},
+		},
+	}
+
+	success, failure, skipped, err := importer.ImportCSV(s3AssumeRoleArn, s3ImportPath)
+
+	if err != nil {
+		logger.Error("error returned from ImportCSV: ", err)
+		return "", err
+	}
+
+	if failure > 0 || skipped > 0 {
+		result := fmt.Sprintf("Successfully imported %v files.  Failed to import %v files.  Skipped %v files.  See logs for more details.", success, failure, skipped)
+		logger.Error(result)
+
+		err = errors.New("Files skipped or failed import. See logs for more details.")
+		return result, err
+
+	}
+
+	result := fmt.Sprintf("Completed CSV import.  Successfully imported %v files.  Failed to import %v files.  Skipped %v files.  See logs for more details.", success, failure, skipped)
+	logger.Info(result)
+	return result, nil
+}
+
 func loadBfdS3Params() (string, error) {
 	env := conf.GetEnv("ENV")
 

bcda/lambda/cclf/main_test.go

@@ -14,15 +14,15 @@ import (
 	"github.com/stretchr/testify/suite"
 )
 
-type CclfImportMainSuite struct {
+type AttributionImportMainSuite struct {
 	suite.Suite
 }
 
-func TestCclfImportMainSuite(t *testing.T) {
-	suite.Run(t, new(CclfImportMainSuite))
+func TestAttributionImportMainSuite(t *testing.T) {
+	suite.Run(t, new(AttributionImportMainSuite))
 }
 
-func (s *CclfImportMainSuite) TestImportCCLFDirectory() {
+func (s *AttributionImportMainSuite) TestImportCCLFDirectory() {
 	targetACO := "A0001"
 	assert := assert.New(s.T())
 
@@ -60,7 +60,7 @@ func (s *CclfImportMainSuite) TestImportCCLFDirectory() {
 		path, cleanup := testUtils.CopyToS3(s.T(), tc.path)
 		defer cleanup()
 
-		res, err := cclfImportHandler(context.Background(), testUtils.GetSQSEvent(s.T(), path, tc.filename))
+		res, err := attributionImportHandler(context.Background(), testUtils.GetSQSEvent(s.T(), path, tc.filename))
 
 		if tc.err == nil {
 			assert.Nil(err)
@@ -75,8 +75,8 @@ func (s *CclfImportMainSuite) TestImportCCLFDirectory() {
 	}
 }
 
-func (s *CclfImportMainSuite) TestHandlerMissingS3AssumeRoleArn() {
+func (s *AttributionImportMainSuite) TestHandlerMissingS3AssumeRoleArn() {
 	assert := assert.New(s.T())
-	_, err := cclfImportHandler(context.Background(), testUtils.GetSQSEvent(s.T(), "doesn't-matter", "fake_filename"))
+	_, err := attributionImportHandler(context.Background(), testUtils.GetSQSEvent(s.T(), "doesn't-matter", "fake_filename"))
 	assert.Contains(err.Error(), "Error retrieving parameter /cclf-import/bcda/local/bfd-bucket-role-arn from parameter store: ParameterNotFound: Parameter /cclf-import/bcda/local/bfd-bucket-role-arn not found.")
 }