Fix loading of imports, add tests

core/models/imports.go

@@ -50,7 +50,8 @@ type ContactImport struct {
 	CreatedByID UserID              `db:"created_by_id"`
 	FinishedOn  *time.Time          `db:"finished_on"`
 
-	BatchStatuses []string `db:"batch_statuses"`
+	// we fetch unique batch statuses concatenated as a string, see https://github.com/jmoiron/sqlx/issues/168
+	BatchStatuses string `db:"batch_statuses"`
 }
 
 var loadContactImportSQL = `
@@ -59,16 +60,16 @@ SELECT
   	i.org_id AS "org_id",
   	i.status AS "status",
   	i.created_by_id AS "created_by_id",
-	i.finished_on AS "finished_on"
-	array_agg(DISTINCT b.status) AS "batch_statuses"
+	i.finished_on AS "finished_on",
+	array_to_string(array_agg(DISTINCT b.status), '') AS "batch_statuses"
 FROM
 	contacts_contactimport i
 INNER JOIN
 	contacts_contactimportbatch b ON b.contact_import_id = i.id
 WHERE
-	id = $1
+	i.id = $1
 GROUP BY
-	c.id`
+	i.id`
 
 // LoadContactImport loads a contact import by ID
 func LoadContactImport(ctx context.Context, db Queryer, id ContactImportID) (*ContactImport, error) {

core/models/notifications.go

@@ -38,10 +38,11 @@ type Notification struct {
 // NotifyImportFinished logs the the finishing of a contact import
 func NotifyImportFinished(ctx context.Context, db Queryer, imp *ContactImport) error {
 	n := &Notification{
-		OrgID:  imp.OrgID,
-		Type:   NotificationTypeImportFinished,
-		Scope:  fmt.Sprintf("contact:%d", imp.ID),
-		UserID: imp.CreatedByID,
+		OrgID:           imp.OrgID,
+		Type:            NotificationTypeImportFinished,
+		Scope:           fmt.Sprintf("contact:%d", imp.ID),
+		UserID:          imp.CreatedByID,
+		ContactImportID: imp.ID,
 	}
 
 	return insertNotifications(ctx, db, []*Notification{n})

core/tasks/contacts/import_contact_batch_test.go

@@ -1,7 +1,9 @@
 package contacts_test
 
 import (
+	"fmt"
 	"testing"
+	"time"
 
 	_ "github.com/nyaruka/mailroom/core/handlers"
 	"github.com/nyaruka/mailroom/core/tasks/contacts"
@@ -12,19 +14,54 @@ import (
 )
 
 func TestImportContactBatch(t *testing.T) {
-	ctx, rt, db, _ := testsuite.Get()
+	ctx, rt, db, rp := testsuite.Get()
+	rc := rp.Get()
+	defer rc.Close()
 
-	importID := testdata.InsertContactImport(db, testdata.Org1)
-	batchID := testdata.InsertContactImportBatch(db, importID, []byte(`[
+	start := time.Now()
+
+	defer func() {
+		db.MustExec(`DELETE FROM contacts_contactgroup_contacts WHERE contact_id > $1`, testdata.Org2Contact.ID)
+		db.MustExec(`DELETE FROM contacts_contacturn WHERE id > $1`, testdata.Org2Contact.URNID)
+		db.MustExec(`DELETE FROM contacts_contact WHERE id > $1`, testdata.Org2Contact.ID)
+	}()
+
+	importID := testdata.InsertContactImport(db, testdata.Org1, testdata.Admin)
+	batch1ID := testdata.InsertContactImportBatch(db, importID, []byte(`[
 		{"name": "Norbert", "language": "eng", "urns": ["tel:+16055740001"]},
 		{"name": "Leah", "urns": ["tel:+16055740002"]}
 	]`))
+	batch2ID := testdata.InsertContactImportBatch(db, importID, []byte(`[
+		{"name": "Rowan", "language": "spa", "urns": ["tel:+16055740003"]}
+	]`))
 
-	task := &contacts.ImportContactBatchTask{ContactImportBatchID: batchID}
+	rc.Do("setex", fmt.Sprintf("contact_import_batches_remaining:%d", importID), 10, 2)
 
-	err := task.Perform(ctx, rt, testdata.Org1.ID)
+	// perform first batch task...
+	task1 := &contacts.ImportContactBatchTask{ContactImportBatchID: batch1ID}
+	err := task1.Perform(ctx, rt, testdata.Org1.ID)
 	require.NoError(t, err)
 
+	// import is still in progress
+	testsuite.AssertQuery(t, db, `SELECT status FROM contacts_contactimport WHERE id = $1`, importID).Columns(map[string]interface{}{"status": "O"})
+
+	// perform second batch task...
+	task2 := &contacts.ImportContactBatchTask{ContactImportBatchID: batch2ID}
+	err = task2.Perform(ctx, rt, testdata.Org1.ID)
+	require.NoError(t, err)
+
+	testsuite.AssertQuery(t, db, `SELECT count(*) FROM contacts_contact WHERE created_on > $1`, start).Returns(3)
 	testsuite.AssertQuery(t, db, `SELECT count(*) FROM contacts_contact WHERE name = 'Norbert' AND language = 'eng'`).Returns(1)
 	testsuite.AssertQuery(t, db, `SELECT count(*) FROM contacts_contact WHERE name = 'Leah' AND language IS NULL`).Returns(1)
+	testsuite.AssertQuery(t, db, `SELECT count(*) FROM contacts_contact WHERE name = 'Rowan' AND language = 'spa'`).Returns(1)
+
+	// import is now complete and there is a notification for the creator
+	testsuite.AssertQuery(t, db, `SELECT status FROM contacts_contactimport WHERE id = $1`, importID).Columns(map[string]interface{}{"status": "C"})
+	testsuite.AssertQuery(t, db, `SELECT org_id, notification_type, scope, user_id FROM notifications_notification WHERE contact_import_id = $1`, importID).
+		Columns(map[string]interface{}{
+			"org_id":            int64(testdata.Org1.ID),
+			"notification_type": "import:finished",
+			"scope":             fmt.Sprintf("contact:%d", importID),
+			"user_id":           int64(testdata.Admin.ID),
+		})
 }

testsuite/testdata/imports.go

@@ -11,10 +11,10 @@ import (
 )
 
 // InsertContactImport inserts a contact import
-func InsertContactImport(db *sqlx.DB, org *Org) models.ContactImportID {
+func InsertContactImport(db *sqlx.DB, org *Org, createdBy *User) models.ContactImportID {
 	var importID models.ContactImportID
-	must(db.Get(&importID, `INSERT INTO contacts_contactimport(org_id, file, original_filename, headers, mappings, num_records, group_id, started_on, created_on, created_by_id, modified_on, modified_by_id, is_active)
-					          VALUES($1, 'contact_imports/1234.xlsx', 'contacts.xlsx', '{"Name", "URN:Tel"}', '{}', 30, NULL, $2, $2, 1, $2, 1, TRUE) RETURNING id`, org.ID, dates.Now(),
+	must(db.Get(&importID, `INSERT INTO contacts_contactimport(org_id, file, original_filename, mappings, num_records, group_id, started_on, status, created_on, created_by_id, modified_on, modified_by_id, is_active)
+					          VALUES($1, 'contact_imports/1234.xlsx', 'contacts.xlsx', '{}', 30, NULL, $2, 'O', $2, $3, $2, $3, TRUE) RETURNING id`, org.ID, dates.Now(), createdBy.ID,
 	))
 	return importID
 }