package controlplane

import (
	"context"
	"database/sql"
	"os"
	"regexp"
	"strings"
	"testing"

	"github.com/DATA-DOG/go-sqlmock"
	"github.com/nexuscrm/automation"
	"github.com/pingcap-inc/nexuscrm-controlplane/packs"
	"github.com/nexuscrm/shared/pkg/constants"
	connector "github.com/pingcap-inc/nexuscrm-contract/pkg/connector"
)

func TestBackfillConnectorPushLegacyMirrorContractScansDeclaredMirrorTuple(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatalf("sqlmock.New: %v", err)
	}
	defer db.Close()

	contract := connectorPushBackfillTestContract()
	expectConnectorPushBackfillObjectMetadata(mock, contract.SourceObject, "")
	expectConnectorPushBackfillFieldMetadata(mock, contract.SourceObject, contract.RemoteIDField, "")
	expectConnectorPushBackfillFieldMetadata(mock, contract.SourceObject, contract.RemoteIDProvenanceField, "")
	mock.ExpectQuery(regexp.QuoteMeta("SELECT `Id`, `RemoteKey__c`, `RemoteKeyProvenance__c` FROM `Contact` WHERE COALESCE(`RemoteKey__c`, '') <> '' AND COALESCE(`RemoteKeyProvenance__c`, '') <> ''")).
		WillReturnRows(sqlmock.NewRows([]string{"Id", "RemoteKey__c", "RemoteKeyProvenance__c"}).
			AddRow("003-adopt", "remote-1", "link-key-1").
			AddRow("003-unmatched", "remote-2", "manual"))

	var seen []automation.ConnectorPushLegacyMirrorEvidence
	resolveCalls := 0
	result, err := backfillConnectorPushLegacyMirrorContract(context.Background(), db, contract, func(context.Context) ([]connector.ConnectorFieldMappingInput, *automation.ConnectorPushProfile, error) {
		resolveCalls++
		return connectorPushBackfillTestMappings(), connectorPushBackfillTestProfile(), nil
	}, func(_ context.Context, evidence automation.ConnectorPushLegacyMirrorEvidence) (bool, error) {
		seen = append(seen, evidence)
		return evidence.RemoteID == "remote-1", nil
	})
	if err != nil {
		t.Fatalf("backfillConnectorPushLegacyMirrorContract: %v", err)
	}
	if result.ContractsScanned != 1 || result.RowsScanned != 2 || result.RowsAdopted != 1 || result.RowsUnmatched != 1 {
		t.Fatalf("result = %+v, want 1 contract, 2 scanned, 1 adopted, 1 unmatched", result)
	}
	if len(seen) != 2 || seen[0].SourceRecordID != "003-adopt" || seen[0].RemoteIDProvenance != "link-key-1" {
		t.Fatalf("seen = %#v, want declared tuple evidence forwarded", seen)
	}
	if resolveCalls != 1 {
		t.Fatalf("resolveCalls = %d, want 1", resolveCalls)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sql expectations: %v", err)
	}
}

func TestBackfillConnectorPushLegacyMirrorContractUsesMetadataOwnedReadPath(t *testing.T) {
	contract := connectorPushBackfillTestContract()
	contract.RemoteIDField = "LegacyRemoteKey"
	contract.RemoteIDProvenanceField = "LegacyRemoteKeyProvenance"
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatalf("sqlmock.New: %v", err)
	}
	defer db.Close()

	expectConnectorPushBackfillObjectMetadata(mock, contract.SourceObject, `{"kind":"physical_table","physicalTable":"ContactMirror"}`)
	expectConnectorPushBackfillFieldMetadata(mock, contract.SourceObject, contract.RemoteIDField, "LegacyRemoteKeyCol")
	expectConnectorPushBackfillFieldMetadata(mock, contract.SourceObject, contract.RemoteIDProvenanceField, "LegacyRemoteKeyProvCol")
	mock.ExpectQuery(regexp.QuoteMeta("SELECT `Id`, `LegacyRemoteKeyCol`, `LegacyRemoteKeyProvCol` FROM `ContactMirror` WHERE COALESCE(`LegacyRemoteKeyCol`, '') <> '' AND COALESCE(`LegacyRemoteKeyProvCol`, '') <> ''")).
		WillReturnRows(sqlmock.NewRows([]string{"Id", "LegacyRemoteKeyCol", "LegacyRemoteKeyProvCol"}).
			AddRow("003-adopt", "remote-1", "link-key-1"))

	result, err := backfillConnectorPushLegacyMirrorContract(context.Background(), db, contract, func(context.Context) ([]connector.ConnectorFieldMappingInput, *automation.ConnectorPushProfile, error) {
		return nil, connectorPushBackfillTestProfile(), nil
	}, func(context.Context, automation.ConnectorPushLegacyMirrorEvidence) (bool, error) {
		return true, nil
	})
	if err != nil {
		t.Fatalf("backfillConnectorPushLegacyMirrorContract: %v", err)
	}
	if result.RowsScanned != 1 || result.RowsAdopted != 1 {
		t.Fatalf("result = %+v, want 1 scanned and 1 adopted row", result)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sql expectations: %v", err)
	}
}

func TestBackfillConnectorPushLegacyMirrorContractAcceptsBroaderResolvedIdentifiers(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatalf("sqlmock.New: %v", err)
	}
	defer db.Close()

	contract := connectorPushBackfillTestContract()
	contract.RemoteIDField = "_RemoteKey__c"
	contract.RemoteIDProvenanceField = "_RemoteKeyProvenance__c"

	expectConnectorPushBackfillObjectMetadata(mock, contract.SourceObject, "")
	expectConnectorPushBackfillFieldMetadata(mock, contract.SourceObject, contract.RemoteIDField, "")
	expectConnectorPushBackfillFieldMetadata(mock, contract.SourceObject, contract.RemoteIDProvenanceField, "")
	mock.ExpectQuery(regexp.QuoteMeta("SELECT `Id`, `_RemoteKey__c`, `_RemoteKeyProvenance__c` FROM `Contact` WHERE COALESCE(`_RemoteKey__c`, '') <> '' AND COALESCE(`_RemoteKeyProvenance__c`, '') <> ''")).
		WillReturnRows(sqlmock.NewRows([]string{"Id", "_RemoteKey__c", "_RemoteKeyProvenance__c"}))

	if _, err := backfillConnectorPushLegacyMirrorContract(context.Background(), db, contract, func(context.Context) ([]connector.ConnectorFieldMappingInput, *automation.ConnectorPushProfile, error) {
		return nil, connectorPushBackfillTestProfile(), nil
	}, func(context.Context, automation.ConnectorPushLegacyMirrorEvidence) (bool, error) {
		return false, nil
	}); err != nil {
		t.Fatalf("backfillConnectorPushLegacyMirrorContract: %v", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sql expectations: %v", err)
	}
}

func TestBackfillConnectorPushLegacyMirrorContractRejectsUnsafeResolvedPhysicalIdentifiers(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatalf("sqlmock.New: %v", err)
	}
	defer db.Close()

	contract := connectorPushBackfillTestContract()
	expectConnectorPushBackfillObjectMetadata(mock, contract.SourceObject, "")
	expectConnectorPushBackfillFieldMetadata(mock, contract.SourceObject, contract.RemoteIDField, "RemoteKey__c; DROP TABLE Contact")

	_, err = backfillConnectorPushLegacyMirrorContract(context.Background(), db, contract, func(context.Context) ([]connector.ConnectorFieldMappingInput, *automation.ConnectorPushProfile, error) {
		return nil, connectorPushBackfillTestProfile(), nil
	}, func(context.Context, automation.ConnectorPushLegacyMirrorEvidence) (bool, error) {
		return false, nil
	})
	if err == nil {
		t.Fatal("expected invalid physical identifier error")
	}
	if !strings.Contains(err.Error(), "invalid physical column") {
		t.Fatalf("error = %v, want invalid physical column detail", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sql expectations: %v", err)
	}
}

func TestBackfillConnectorPushLegacyMirrorContractSkipsResolverWhenNoLegacyRowsExist(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatalf("sqlmock.New: %v", err)
	}
	defer db.Close()

	contract := connectorPushBackfillTestContract()
	expectConnectorPushBackfillObjectMetadata(mock, contract.SourceObject, "")
	expectConnectorPushBackfillFieldMetadata(mock, contract.SourceObject, contract.RemoteIDField, "")
	expectConnectorPushBackfillFieldMetadata(mock, contract.SourceObject, contract.RemoteIDProvenanceField, "")
	mock.ExpectQuery(regexp.QuoteMeta("SELECT `Id`, `RemoteKey__c`, `RemoteKeyProvenance__c` FROM `Contact` WHERE COALESCE(`RemoteKey__c`, '') <> '' AND COALESCE(`RemoteKeyProvenance__c`, '') <> ''")).
		WillReturnRows(sqlmock.NewRows([]string{"Id", "RemoteKey__c", "RemoteKeyProvenance__c"}))

	resolveCalls := 0
	result, err := backfillConnectorPushLegacyMirrorContract(context.Background(), db, contract, func(context.Context) ([]connector.ConnectorFieldMappingInput, *automation.ConnectorPushProfile, error) {
		resolveCalls++
		return nil, nil, nil
	}, func(context.Context, automation.ConnectorPushLegacyMirrorEvidence) (bool, error) {
		t.Fatal("adopt should not be called when there are no legacy rows")
		return false, nil
	})
	if err != nil {
		t.Fatalf("backfillConnectorPushLegacyMirrorContract: %v", err)
	}
	if result.RowsScanned != 0 || result.RowsAdopted != 0 || result.RowsUnmatched != 0 {
		t.Fatalf("result = %+v, want zero-row no-op", result)
	}
	if resolveCalls != 0 {
		t.Fatalf("resolveCalls = %d, want 0", resolveCalls)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sql expectations: %v", err)
	}
}

func TestBackfillConnectorPushLegacyMirrorContractSkipsResolverForBlankProvenanceField(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatalf("sqlmock.New: %v", err)
	}
	defer db.Close()

	contract := connectorPushBackfillTestContract()
	contract.RemoteIDProvenanceField = ""
	resolveCalls := 0
	result, err := backfillConnectorPushLegacyMirrorContract(context.Background(), db, contract, func(context.Context) ([]connector.ConnectorFieldMappingInput, *automation.ConnectorPushProfile, error) {
		resolveCalls++
		return nil, nil, nil
	}, func(context.Context, automation.ConnectorPushLegacyMirrorEvidence) (bool, error) {
		t.Fatal("adopt should not be called when provenance is blank")
		return false, nil
	})
	if err != nil {
		t.Fatalf("backfillConnectorPushLegacyMirrorContract: %v", err)
	}
	if result.ContractsScanned != 1 || result.RowsScanned != 0 {
		t.Fatalf("result = %+v, want blank-provenance no-op", result)
	}
	if resolveCalls != 0 {
		t.Fatalf("resolveCalls = %d, want 0", resolveCalls)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sql expectations: %v", err)
	}
}

func TestBackfillConnectorPushLegacyMirrorContractSkipsUnauthorizedCrossPackProfile(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatalf("sqlmock.New: %v", err)
	}
	defer db.Close()

	contract := connectorPushBackfillTestContract()
	expectConnectorPushBackfillObjectMetadata(mock, contract.SourceObject, "")
	expectConnectorPushBackfillFieldMetadata(mock, contract.SourceObject, contract.RemoteIDField, "")
	expectConnectorPushBackfillFieldMetadata(mock, contract.SourceObject, contract.RemoteIDProvenanceField, "")
	mock.ExpectQuery(regexp.QuoteMeta("SELECT `Id`, `RemoteKey__c`, `RemoteKeyProvenance__c` FROM `Contact` WHERE COALESCE(`RemoteKey__c`, '') <> '' AND COALESCE(`RemoteKeyProvenance__c`, '') <> ''")).
		WillReturnRows(sqlmock.NewRows([]string{"Id", "RemoteKey__c", "RemoteKeyProvenance__c"}).
			AddRow("003-unauthorized", "remote-1", "link-key-1"))
	expectConnectorPushBackfillActivePack(mock, contract.ConnectorOwnerKey, "app-pack", "app-install-1", "1.0.0")
	expectConnectorPushBackfillActivePack(mock, contract.ProfileOwnerPack, "connector-pack", "connector-install-1", "2.0.0")
	expectConnectorPushBackfillDependencyMissing(mock, contract.ConnectorOwnerKey, contract.ProfileOwnerPack)

	resolveCalls := 0
	result, err := backfillConnectorPushLegacyMirrorContract(context.Background(), db, contract, func(ctx context.Context) ([]connector.ConnectorFieldMappingInput, *automation.ConnectorPushProfile, error) {
		resolveCalls++
		profile, err := loadConnectorPushBackfillProfile(ctx, db, contract)
		return nil, profile, err
	}, func(context.Context, automation.ConnectorPushLegacyMirrorEvidence) (bool, error) {
		t.Fatal("adopt should not be called for unauthorized cross-pack contract")
		return false, nil
	})
	if err != nil {
		t.Fatalf("backfillConnectorPushLegacyMirrorContract: %v", err)
	}
	if result.ContractsScanned != 1 || result.RowsScanned != 0 || result.RowsAdopted != 0 || result.RowsUnmatched != 0 {
		t.Fatalf("result = %+v, want unauthorized cross-pack no-op", result)
	}
	if resolveCalls != 1 {
		t.Fatalf("resolveCalls = %d, want 1", resolveCalls)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sql expectations: %v", err)
	}
}

func TestConnectorPushBackfillPackSourceIdentityRequiresStoredIdentity(t *testing.T) {
	cases := []struct {
		name       string
		sourceKind string
		sourceURI  string
		wantKind   string
		wantURI    string
		wantOK     bool
	}{
		{name: "blank legacy"},
		{name: "blank official repo uri", sourceKind: "official_repo"},
		{name: "uri only is pack name", sourceURI: "crm-pack"},
		{name: "official repo preserved", sourceKind: "official_repo", sourceURI: "crm-pack", wantKind: "official_repo", wantURI: "crm-pack", wantOK: true},
		{name: "local path preserved", sourceKind: "local_path", sourceURI: "/tmp/crm-pack", wantKind: "local_path", wantURI: "/tmp/crm-pack", wantOK: true},
	}
	for _, tc := range cases {
		t.Run(tc.name, func(t *testing.T) {
			gotKind, gotURI, gotOK := connectorPushBackfillPackSourceIdentity("crm-pack", tc.sourceKind, tc.sourceURI)
			if gotOK != tc.wantOK {
				t.Fatalf("ok = %v, want %v", gotOK, tc.wantOK)
			}
			if gotKind != tc.wantKind || gotURI != tc.wantURI {
				t.Fatalf("source identity = %s/%s, want %s/%s", gotKind, gotURI, tc.wantKind, tc.wantURI)
			}
		})
	}
}

func TestConnectorPushLegacyMirrorBackfillRunsAfterSourceConvergence(t *testing.T) {
	source, err := os.ReadFile("operations.go")
	if err != nil {
		t.Fatalf("read operations.go: %v", err)
	}
	text := string(source)
	if strings.Contains(text, `trackStepFatal("BackfillConnectorPushLegacyMirrorAdoptions"`) {
		t.Fatal("legacy mirror backfill must not run in early RunRepair before source identity convergence")
	}
	reconcile := strings.Index(text, `"reconcile-active-installed-pack-sources"`)
	backfill := strings.Index(text, `"backfill-connector-push-legacy-mirrors-after-source-convergence"`)
	if reconcile < 0 || backfill < 0 {
		t.Fatalf("operations.go missing source convergence or legacy mirror backfill phase: reconcile=%d backfill=%d", reconcile, backfill)
	}
	if backfill < reconcile {
		t.Fatalf("legacy mirror backfill phase appears before source convergence: backfill=%d reconcile=%d", backfill, reconcile)
	}
	marker := strings.Index(text, "MarkConnectorPushLegacyMirrorBackfillCompleted")
	if marker < 0 {
		t.Fatal("legacy mirror backfill phase must stamp the warm-upgrade completion marker")
	}
	if marker < backfill {
		t.Fatalf("legacy mirror backfill marker is stamped before the backfill phase: marker=%d backfill=%d", marker, backfill)
	}
}

func TestLoadConnectorPushLegacyMirrorBackfillContractsFailsClosedOnActiveSourceDrift(t *testing.T) {
	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatalf("sqlmock.New: %v", err)
	}
	defer db.Close()

	storedHash := "nexus-pack-sha256-v1:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
	mock.ExpectQuery(regexp.QuoteMeta(`SELECT Name, Category, COALESCE(ManifestFormatVersion, ''), COALESCE(SourceKind, ''),
	       COALESCE(SourceURI, ''), COALESCE(SourceRevision, ''), COALESCE(AppliedPayloadHash, ''),
	       COALESCE(DesiredPayloadHash, '')
	  FROM _System_InstalledPack
	 WHERE Status = 'active'
	   AND Category IN (?, ?)
	   AND COALESCE(IsDeleted, 0) = 0
	   AND COALESCE(AppliedPayloadHash, '') <> ''
	   AND COALESCE(AppliedPayloadHash, '') = COALESCE(DesiredPayloadHash, '')
	 ORDER BY Name`)).
		WithArgs("app-pack", "connector-pack").
		WillReturnRows(sqlmock.NewRows([]string{
			"Name",
			"Category",
			"ManifestFormatVersion",
			"SourceKind",
			"SourceURI",
			"SourceRevision",
			"AppliedPayloadHash",
			"DesiredPayloadHash",
		}).AddRow("drifted-app", "app-pack", "v1", "local_path", "/missing/source/tree", "", storedHash, storedHash))

	contracts, err := loadConnectorPushLegacyMirrorBackfillContracts(context.Background(), db)
	if err == nil {
		t.Fatalf("contracts = %#v, want active source drift to fail closed", contracts)
	}
	if !strings.Contains(err.Error(), "resolve installed pack drifted-app") {
		t.Fatalf("error = %v, want installed-pack resolution context", err)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sql expectations: %v", err)
	}
}

func TestLoadConnectorPushLegacyMirrorBackfillContractsUsesDBSchemaResolverForActiveConnectorAssociationLookup(t *testing.T) {
	root := writeConnectorAssociationSchemaResolverPackSourceRoot(t)
	opts := packs.DefaultPackSourceOptions()
	opts.ObjectSchemaResolver = testSourcePackSchemaResolver()
	activePack, err := packs.ResolveInstalledSourceBackedPackForInstalledRowAtSourceRevisionWithPackSourceOptions(
		"schema-resolver-app",
		packs.AppPack,
		packs.SupportedManifestFormatVersion,
		string(packs.PackSourceKindLocalPath),
		root,
		"",
		opts,
	)
	if err != nil {
		t.Fatalf("resolve active schema resolver pack: %v", err)
	}
	activeHash, err := packs.PackPayloadHash(activePack)
	if err != nil {
		t.Fatalf("active payload hash: %v", err)
	}

	db, mock, err := sqlmock.New()
	if err != nil {
		t.Fatalf("sqlmock.New: %v", err)
	}
	defer db.Close()
	mock.MatchExpectationsInOrder(false)

	mock.ExpectQuery(regexp.QuoteMeta(`SELECT Name, Category, COALESCE(ManifestFormatVersion, ''), COALESCE(SourceKind, ''),
	       COALESCE(SourceURI, ''), COALESCE(SourceRevision, ''), COALESCE(AppliedPayloadHash, ''),
	       COALESCE(DesiredPayloadHash, '')
	  FROM _System_InstalledPack
	 WHERE Status = 'active'
	   AND Category IN (?, ?)
	   AND COALESCE(IsDeleted, 0) = 0
	   AND COALESCE(AppliedPayloadHash, '') <> ''
	   AND COALESCE(AppliedPayloadHash, '') = COALESCE(DesiredPayloadHash, '')
	 ORDER BY Name`)).
		WithArgs("app-pack", "connector-pack").
		WillReturnRows(sqlmock.NewRows([]string{
			"Name",
			"Category",
			"ManifestFormatVersion",
			"SourceKind",
			"SourceURI",
			"SourceRevision",
			"AppliedPayloadHash",
			"DesiredPayloadHash",
		}).AddRow(activePack.Name(), string(activePack.Category()), packs.SupportedManifestFormatVersion, string(packs.PackSourceKindLocalPath), root, "", activeHash, activeHash))
	expectWorkItemAssociationLookupSchemaLoad(mock)

	contracts, err := loadConnectorPushLegacyMirrorBackfillContracts(context.Background(), db)
	if err != nil {
		t.Fatalf("loadConnectorPushLegacyMirrorBackfillContracts rejected resolver-backed connector association lookup: %v", err)
	}
	if len(contracts) != 2 {
		t.Fatalf("contracts = %#v, want both connector push contracts from active pack", contracts)
	}
	foundAssociationContract := false
	for _, contract := range contracts {
		if contract.SourceObject == "WorkItem__c" && len(contract.Associations) == 1 && contract.Associations[0].SourceLookupField == "Account__c" {
			foundAssociationContract = true
		}
	}
	if !foundAssociationContract {
		t.Fatalf("contracts = %#v, want WorkItem__c association contract loaded", contracts)
	}
	if err := mock.ExpectationsWereMet(); err != nil {
		t.Fatalf("sql expectations: %v", err)
	}
}

func connectorPushBackfillTestContract() connector.ConnectorPushDurableMetadataContract {
	return connector.ConnectorPushDurableMetadataContract{
		ConnectorOwnerKey:       "crm-flow-pack",
		ActionSlotID:            "contact-current",
		PreviousActionSlotIDs:   []string{"contact-v1"},
		ActionType:              constants.ActionTypePushToConnector,
		FlowDeveloperName:       "Contact_To_Remote",
		FlowActionName:          "PushContact",
		SourceObject:            "Contact",
		DestinationSlotID:       "remote-contact",
		ProfileOwnerPack:        "connector-pack",
		ProfileName:             "default",
		TargetObject:            "remoteContact",
		CreatePathTemplate:      "/objects",
		UpdatePathTemplate:      "/objects/{remoteId}",
		ResponseIDPath:          "id",
		RemoteIDField:           "RemoteKey__c",
		RemoteIDProvenanceField: "RemoteKeyProvenance__c",
	}
}

func connectorPushBackfillTestMappings() []connector.ConnectorFieldMappingInput {
	return []connector.ConnectorFieldMappingInput{{
		Source:       "crm-flow-pack",
		SourceObject: "Contact",
		TargetObject: "remoteContact",
		ActionType:   constants.ActionTypePushToConnector,
		SourceField:  "Email",
		TargetField:  "email",
		Direction:    connector.DirectionOutbound,
	}}
}

func connectorPushBackfillTestProfile() *automation.ConnectorPushProfile {
	return &automation.ConnectorPushProfile{
		OwnerPack:            "connector-pack",
		ProfileName:          "default",
		CreatedPackInstallID: "install-1",
		CredentialRef:        "cred",
		EndpointClass:        "default",
		RetryPolicy: automation.ConnectorPushRetryPolicy{
			MaxAttempts: 1,
		},
	}
}

func expectConnectorPushBackfillObjectMetadata(mock sqlmock.Sqlmock, objectAPIName, readSource string) {
	mock.ExpectQuery(regexp.QuoteMeta(connectorPushBackfillObjectMetadataQuery)).
		WithArgs(objectAPIName).
		WillReturnRows(sqlmock.NewRows([]string{"ApiName", "ReadSource"}).
			AddRow(objectAPIName, readSource))
}

func expectConnectorPushBackfillFieldMetadata(mock sqlmock.Sqlmock, objectAPIName, fieldAPIName, physicalName string) {
	mock.ExpectQuery(regexp.QuoteMeta(connectorPushBackfillFieldMetadataQuery)).
		WithArgs(objectAPIName, fieldAPIName).
		WillReturnRows(sqlmock.NewRows([]string{"ApiName", "PhysicalName"}).
			AddRow(fieldAPIName, physicalName))
}

func expectConnectorPushBackfillActivePack(mock sqlmock.Sqlmock, packName, category, installID, version string) {
	rows := sqlmock.NewRows([]string{"PackInstallId", "PackVersion"}).
		AddRow(installID, version)
	mock.ExpectQuery(regexp.QuoteMeta(connectorPushBackfillActivePackInfoQuery)).
		WithArgs(packName, category).
		WillReturnRows(rows)
}

func expectConnectorPushBackfillDependencyMissing(mock sqlmock.Sqlmock, companionPackName, namespace string) {
	mock.ExpectQuery(regexp.QuoteMeta(`
SELECT COALESCE(DependentPackInstallId, ''), VersionConstraint
  FROM _System_InstalledPackDependency
 WHERE DependentPackName = ?
   AND DependsOnPackName = ?
   AND COALESCE(IsDeleted, 0) = 0
 LIMIT 1`)).
		WithArgs(companionPackName, namespace).
		WillReturnError(sql.ErrNoRows)
}