package controlplane import ( "context" "database/sql" "database/sql/driver" "encoding/json" "errors" "fmt" "os" "path/filepath" "regexp" "strings" "testing" "time" "github.com/DATA-DOG/go-sqlmock" "github.com/go-sql-driver/mysql" "github.com/nexuscrm/db" "github.com/nexuscrm/kernel/pkg/formula" "github.com/nexuscrm/kernel/pkg/formula/spec" "github.com/nexuscrm/metadata" "github.com/nexuscrm/shared/pkg/searchmetadata" "github.com/pingcap-inc/nexuscrm-controlplane/packs" ) const ( fakeLegacyHashV0 = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa" fakePayloadHashV1 = packs.PackPayloadHashV1Prefix + "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb" ) func expectInstalledPackRuntimeDeclarationColumns(mock sqlmock.Sqlmock, present bool) { count := 0 if present { count = 4 } mock.ExpectQuery("SELECT COUNT\\(\\*\\)\\s+FROM INFORMATION_SCHEMA.COLUMNS"). WithArgs("_System_InstalledPack", "RuntimeObjectSpecDeclarations", "RuntimeRelatedListRules", "RuntimeSearchScopeOrder", "RuntimeSearchAllDefaultFields"). WillReturnRows(sqlmock.NewRows([]string{"COUNT(*)"}).AddRow(count)) } func expectNoPackOwnedApprovalProcesses(mock sqlmock.Sqlmock, packName string) { mock.ExpectQuery(regexp.QuoteMeta(` SELECT Id FROM ProcessDefinition WHERE COALESCE(OwnerPack, '') = ? ORDER BY Id LIMIT 1`)). WithArgs(packName). WillReturnRows(sqlmock.NewRows([]string{"Id"})) } func expectNoActiveConnectorProjectionOwnersForCompanion(mock sqlmock.Sqlmock, packName string) { mock.ExpectQuery("SELECT DISTINCT d.DependsOnPackName"). WithArgs(packName). WillReturnRows(sqlmock.NewRows([]string{"DependsOnPackName"})) } func expectInstalledPackHostRequirementsDigestUpdate(mock sqlmock.Sqlmock, packName string) { mock.ExpectQuery("SELECT COUNT\\(\\*\\)\\s+FROM INFORMATION_SCHEMA\\.COLUMNS"). WithArgs("_System_InstalledPack", "HostRequirementsDigest"). WillReturnRows(sqlmock.NewRows([]string{"count"}).AddRow(1)) mock.ExpectExec("UPDATE _System_InstalledPack\\s+SET HostRequirementsDigest = \\?,\\s+LastModifiedDate = \\?\\s+WHERE Name = \\?"). WithArgs(sqlmock.AnyArg(), sqlmock.AnyArg(), packName). WillReturnResult(sqlmock.NewResult(0, 1)) } const updateRuntimeDeclarationFourColumnCASPattern = `UPDATE _System_InstalledPack\s+SET RuntimeObjectSpecDeclarations = \?,\s+RuntimeRelatedListRules = \?,\s+RuntimeSearchScopeOrder = \?,\s+RuntimeSearchAllDefaultFields = \?,\s+RuntimeStartupShallowDeclarations = \?\s+WHERE Name = \?\s+AND Category = \?\s+AND Status = \?\s+AND COALESCE\(IsDeleted, 0\) = 0\s+AND SourceKind = \?\s+AND SourceURI = \?\s+AND SourceRevision <=> \?\s+AND AppliedPayloadHash = \?\s+AND DesiredPayloadHash = \?\s+AND AppliedPayloadHash = DesiredPayloadHash\s+AND RuntimeObjectSpecDeclarations <=> \?\s+AND RuntimeRelatedListRules <=> \?\s+AND RuntimeSearchScopeOrder <=> \?\s+AND RuntimeSearchAllDefaultFields <=> \?\s+AND RuntimeStartupShallowDeclarations <=> \?` const updateRuntimeDeclarationThreeColumnCASPattern = `UPDATE _System_InstalledPack\s+SET RuntimeObjectSpecDeclarations = \?,\s+RuntimeRelatedListRules = \?,\s+RuntimeSearchScopeOrder = \?,\s+RuntimeSearchAllDefaultFields = \?\s+WHERE Name = \?\s+AND Category = \?\s+AND Status = \?\s+AND COALESCE\(IsDeleted, 0\) = 0\s+AND SourceKind = \?\s+AND SourceURI = \?\s+AND SourceRevision <=> \?\s+AND AppliedPayloadHash = \?\s+AND DesiredPayloadHash = \?\s+AND AppliedPayloadHash = DesiredPayloadHash\s+AND RuntimeObjectSpecDeclarations <=> \?\s+AND RuntimeRelatedListRules <=> \?\s+AND RuntimeSearchScopeOrder <=> \?\s+AND RuntimeSearchAllDefaultFields <=> \?` func TestRefreshPersistedRuntimeDeclarationsForActivePacksSkipsSourceLessRows(t *testing.T) { db, mock, err := sqlmock.New() if err != nil { t.Fatalf("sqlmock: %v", err) } defer db.Close() sourceRoot := writeConnectorPackSourceRoot(t) pack, err := packs.ResolveInstalledSourceBackedPack("test-connector", packs.ConnectorPack, packs.PackSourceKindLocalPath, sourceRoot) if err != nil { t.Fatalf("ResolveInstalledSourceBackedPack: %v", err) } payloadHash, err := packs.PackPayloadHash(pack) if err != nil { t.Fatalf("PackPayloadHash: %v", err) } meta := pack.PackHashMetadata() mock.ExpectQuery("SELECT COUNT\\(\\*\\)\\s+FROM INFORMATION_SCHEMA.COLUMNS"). WithArgs("_System_InstalledPack", "RuntimeObjectSpecDeclarations", "RuntimeRelatedListRules", "RuntimeSearchScopeOrder", "RuntimeSearchAllDefaultFields"). WillReturnRows(sqlmock.NewRows([]string{"COUNT(*)"}).AddRow(4)) expectInstalledPackRuntimeStartupShallowDeclarationColumn(mock, false) mock.ExpectQuery("SELECT COUNT\\(\\*\\)\\s+FROM INFORMATION_SCHEMA.COLUMNS"). WithArgs("_System_InstalledPack", "RuntimeObjectSpecDeclarations", "RuntimeRelatedListRules", "RuntimeSearchScopeOrder", "RuntimeSearchAllDefaultFields"). WillReturnRows(sqlmock.NewRows([]string{"COUNT(*)"}).AddRow(4)) expectInstalledPackRuntimeStartupShallowDeclarationColumn(mock, false) mock.ExpectQuery("SELECT Name, Category, Status, PackVersion, ManifestFormatVersion, PlatformCompatibility, DesiredPayloadHash, AppliedPayloadHash, PayloadHashAlgorithm, SourceKind, SourceURI, SourceRevision"). WithArgs(string(packs.AppPack), string(packs.ConnectorPack)). WillReturnRows(sqlmock.NewRows([]string{ "Name", "Category", "Status", "PackVersion", "ManifestFormatVersion", "PlatformCompatibility", "DesiredPayloadHash", "AppliedPayloadHash", "PayloadHashAlgorithm", "SourceKind", "SourceURI", "SourceRevision", "HostRequirementsDigest", "RuntimeObjectSpecDeclarations", "RuntimeRelatedListRules", "RuntimeSearchScopeOrder", "RuntimeSearchAllDefaultFields", }). AddRow( "source-less-connector", string(packs.ConnectorPack), installedPackStatusActive, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, ). AddRow( "test-connector", string(packs.ConnectorPack), installedPackStatusActive, meta.PackVersion, meta.ManifestFormatVersion, meta.PlatformCompatibility, payloadHash, payloadHash, packs.PackPayloadHashAlgorithmV1, packs.PackSourceKindLocalPath, sourceRoot, nil, nil, nil, nil, nil, nil, )) mock.ExpectExec(updateRuntimeDeclarationThreeColumnCASPattern). WithArgs( sqlmock.AnyArg(), sqlmock.AnyArg(), sqlmock.AnyArg(), sqlmock.AnyArg(), "test-connector", string(packs.ConnectorPack), installedPackStatusActive, packs.PackSourceKindLocalPath, sourceRoot, nil, payloadHash, payloadHash, nil, nil, nil, nil, ). WillReturnResult(sqlmock.NewResult(0, 1)) if err := refreshPersistedRuntimeDeclarationsForActivePacks(context.Background(), db); err != nil { t.Fatalf("refreshPersistedRuntimeDeclarationsForActivePacks: %v", err) } if err := mock.ExpectationsWereMet(); err != nil { t.Fatalf("sql expectations: %v", err) } } func TestRefreshPersistedRuntimeDeclarationsForActivePacksUsesCASAndOnlyRuntimeColumns(t *testing.T) { db, mock, err := sqlmock.New() if err != nil { t.Fatalf("sqlmock: %v", err) } defer db.Close() sourceRoot := writeConnectorPackSourceRoot(t) pack, err := packs.ResolveInstalledSourceBackedPack("test-connector", packs.ConnectorPack, packs.PackSourceKindLocalPath, sourceRoot) if err != nil { t.Fatalf("ResolveInstalledSourceBackedPack: %v", err) } payloadHash, err := packs.PackPayloadHash(pack) if err != nil { t.Fatalf("PackPayloadHash: %v", err) } decl, err := packs.BuildPersistedRuntimeDeclaration(pack) if err != nil { t.Fatalf("BuildPersistedRuntimeDeclaration: %v", err) } meta := pack.PackHashMetadata() expectInstalledPackRuntimeDeclarationColumns(mock, true) expectInstalledPackRuntimeStartupShallowDeclarationColumn(mock, true) expectInstalledPackRuntimeDeclarationColumns(mock, true) expectInstalledPackRuntimeStartupShallowDeclarationColumn(mock, true) mock.ExpectQuery("SELECT Name, Category, Status, PackVersion, ManifestFormatVersion, PlatformCompatibility, DesiredPayloadHash, AppliedPayloadHash, PayloadHashAlgorithm, SourceKind, SourceURI, SourceRevision, HostRequirementsDigest, RuntimeObjectSpecDeclarations, RuntimeRelatedListRules, RuntimeSearchScopeOrder, RuntimeSearchAllDefaultFields, RuntimeStartupShallowDeclarations"). WithArgs(string(packs.AppPack), string(packs.ConnectorPack)). WillReturnRows(sqlmock.NewRows([]string{ "Name", "Category", "Status", "PackVersion", "ManifestFormatVersion", "PlatformCompatibility", "DesiredPayloadHash", "AppliedPayloadHash", "PayloadHashAlgorithm", "SourceKind", "SourceURI", "SourceRevision", "HostRequirementsDigest", "RuntimeObjectSpecDeclarations", "RuntimeRelatedListRules", "RuntimeSearchScopeOrder", "RuntimeSearchAllDefaultFields", "RuntimeStartupShallowDeclarations", }).AddRow( "test-connector", string(packs.ConnectorPack), installedPackStatusActive, meta.PackVersion, meta.ManifestFormatVersion, meta.PlatformCompatibility, payloadHash, payloadHash, packs.PackPayloadHashAlgorithmV1, packs.PackSourceKindLocalPath, sourceRoot, nil, nil, `[]`, `[]`, `[]`, `[]`, `[]`, )) mock.ExpectExec(updateRuntimeDeclarationFourColumnCASPattern). WithArgs( decl.ObjectPayloadsJSON(), decl.RelatedListRulesJSON(), decl.SearchScopeOrderJSON(), decl.SearchAllDefaultFieldsJSON(), decl.StartupShallowDeclaredRowsJSON(), "test-connector", string(packs.ConnectorPack), installedPackStatusActive, packs.PackSourceKindLocalPath, sourceRoot, nil, payloadHash, payloadHash, `[]`, `[]`, `[]`, `[]`, `[]`, ). WillReturnResult(sqlmock.NewResult(0, 1)) if err := refreshPersistedRuntimeDeclarationsForActivePacks(context.Background(), db); err != nil { t.Fatalf("refreshPersistedRuntimeDeclarationsForActivePacks: %v", err) } if err := mock.ExpectationsWereMet(); err != nil { t.Fatalf("sql expectations: %v", err) } } func TestRefreshPersistedRuntimeDeclarationsForActivePacksSkipsAlreadyCurrentRuntimeDeclarations(t *testing.T) { db, mock, err := sqlmock.New() if err != nil { t.Fatalf("sqlmock: %v", err) } defer db.Close() sourceRoot := writeConnectorPackSourceRoot(t) pack, err := packs.ResolveInstalledSourceBackedPack("test-connector", packs.ConnectorPack, packs.PackSourceKindLocalPath, sourceRoot) if err != nil { t.Fatalf("ResolveInstalledSourceBackedPack: %v", err) } payloadHash, err := packs.PackPayloadHash(pack) if err != nil { t.Fatalf("PackPayloadHash: %v", err) } decl, err := packs.BuildPersistedRuntimeDeclaration(pack) if err != nil { t.Fatalf("BuildPersistedRuntimeDeclaration: %v", err) } meta := pack.PackHashMetadata() expectInstalledPackRuntimeDeclarationColumns(mock, true) expectInstalledPackRuntimeStartupShallowDeclarationColumn(mock, true) expectInstalledPackRuntimeDeclarationColumns(mock, true) expectInstalledPackRuntimeStartupShallowDeclarationColumn(mock, true) mock.ExpectQuery("SELECT Name, Category, Status, PackVersion, ManifestFormatVersion, PlatformCompatibility, DesiredPayloadHash, AppliedPayloadHash, PayloadHashAlgorithm, SourceKind, SourceURI, SourceRevision, HostRequirementsDigest, RuntimeObjectSpecDeclarations, RuntimeRelatedListRules, RuntimeSearchScopeOrder, RuntimeSearchAllDefaultFields, RuntimeStartupShallowDeclarations"). WithArgs(string(packs.AppPack), string(packs.ConnectorPack)). WillReturnRows(sqlmock.NewRows([]string{ "Name", "Category", "Status", "PackVersion", "ManifestFormatVersion", "PlatformCompatibility", "DesiredPayloadHash", "AppliedPayloadHash", "PayloadHashAlgorithm", "SourceKind", "SourceURI", "SourceRevision", "HostRequirementsDigest", "RuntimeObjectSpecDeclarations", "RuntimeRelatedListRules", "RuntimeSearchScopeOrder", "RuntimeSearchAllDefaultFields", "RuntimeStartupShallowDeclarations", }).AddRow( "test-connector", string(packs.ConnectorPack), installedPackStatusActive, meta.PackVersion, meta.ManifestFormatVersion, meta.PlatformCompatibility, payloadHash, payloadHash, packs.PackPayloadHashAlgorithmV1, packs.PackSourceKindLocalPath, sourceRoot, nil, nil, prettyJSONString(t, decl.ObjectPayloadsJSON()), prettyJSONString(t, decl.RelatedListRulesJSON()), prettyJSONString(t, decl.SearchScopeOrderJSON()), prettyJSONString(t, decl.SearchAllDefaultFieldsJSON()), prettyJSONString(t, decl.StartupShallowDeclaredRowsJSON()), )) if err := refreshPersistedRuntimeDeclarationsForActivePacks(context.Background(), db); err != nil { t.Fatalf("refreshPersistedRuntimeDeclarationsForActivePacks: %v", err) } if err := mock.ExpectationsWereMet(); err != nil { t.Fatalf("sql expectations: %v", err) } } func TestRefreshPersistedRuntimeDeclarationsForActivePacksBackfillsRuntimeFormulaFunctionBundles(t *testing.T) { db, mock, err := sqlmock.New() if err != nil { t.Fatalf("sqlmock: %v", err) } defer db.Close() sourceRoot := writeFormulaFunctionBundlePackSourceRoot(t) pack, err := packs.ResolveInstalledSourceBackedPack("formula-functions", packs.AppPack, packs.PackSourceKindLocalPath, sourceRoot) if err != nil { t.Fatalf("ResolveInstalledSourceBackedPack: %v", err) } payloadHash, err := packs.PackPayloadHash(pack) if err != nil { t.Fatalf("PackPayloadHash: %v", err) } decl, err := packs.BuildPersistedRuntimeDeclaration(pack) if err != nil { t.Fatalf("BuildPersistedRuntimeDeclaration: %v", err) } meta := pack.PackHashMetadata() expectInstalledPackRuntimeDeclarationColumns(mock, true) expectInstalledPackRuntimeStartupShallowAndFormulaBundleDeclarationColumns(mock, true, true) expectInstalledPackRuntimeDeclarationColumns(mock, true) expectInstalledPackRuntimeStartupShallowAndFormulaBundleDeclarationColumns(mock, true, true) mock.ExpectQuery("SELECT Name, Category, Status, PackVersion, ManifestFormatVersion, PlatformCompatibility, DesiredPayloadHash, AppliedPayloadHash, PayloadHashAlgorithm, SourceKind, SourceURI, SourceRevision, HostRequirementsDigest, RuntimeObjectSpecDeclarations, RuntimeRelatedListRules, RuntimeSearchScopeOrder, RuntimeSearchAllDefaultFields, RuntimeFormulaFunctionBundles, RuntimeStartupShallowDeclarations"). WithArgs(string(packs.AppPack), string(packs.ConnectorPack)). WillReturnRows(sqlmock.NewRows([]string{ "Name", "Category", "Status", "PackVersion", "ManifestFormatVersion", "PlatformCompatibility", "DesiredPayloadHash", "AppliedPayloadHash", "PayloadHashAlgorithm", "SourceKind", "SourceURI", "SourceRevision", "HostRequirementsDigest", "RuntimeObjectSpecDeclarations", "RuntimeRelatedListRules", "RuntimeSearchScopeOrder", "RuntimeSearchAllDefaultFields", "RuntimeFormulaFunctionBundles", "RuntimeStartupShallowDeclarations", }).AddRow( "formula-functions", string(packs.AppPack), installedPackStatusActive, meta.PackVersion, meta.ManifestFormatVersion, meta.PlatformCompatibility, payloadHash, payloadHash, packs.PackPayloadHashAlgorithmV1, packs.PackSourceKindLocalPath, sourceRoot, nil, nil, `[]`, `[]`, `[]`, `[]`, nil, `[]`, )) mock.ExpectExec(`UPDATE _System_InstalledPack\s+SET RuntimeObjectSpecDeclarations = \?,\s+RuntimeRelatedListRules = \?,\s+RuntimeSearchScopeOrder = \?,\s+RuntimeSearchAllDefaultFields = \?,\s+RuntimeFormulaFunctionBundles = \?,\s+RuntimeStartupShallowDeclarations = \?[\s\S]*AND RuntimeFormulaFunctionBundles <=> \?[\s\S]*AND RuntimeStartupShallowDeclarations <=> \?`). WithArgs( decl.ObjectPayloadsJSON(), decl.RelatedListRulesJSON(), decl.SearchScopeOrderJSON(), decl.SearchAllDefaultFieldsJSON(), decl.FormulaFunctionBundlesJSON(), decl.StartupShallowDeclaredRowsJSON(), "formula-functions", string(packs.AppPack), installedPackStatusActive, packs.PackSourceKindLocalPath, sourceRoot, nil, payloadHash, payloadHash, `[]`, `[]`, `[]`, `[]`, nil, `[]`, ). WillReturnResult(sqlmock.NewResult(0, 1)) if err := refreshPersistedRuntimeDeclarationsForActivePacks(context.Background(), db); err != nil { t.Fatalf("refreshPersistedRuntimeDeclarationsForActivePacks: %v", err) } if err := mock.ExpectationsWereMet(); err != nil { t.Fatalf("sql expectations: %v", err) } } func TestRefreshPersistedRuntimeDeclarationsForActivePacksUsesActiveFormulaBundlesForDependencyConsumers(t *testing.T) { db, mock, err := sqlmock.New() if err != nil { t.Fatalf("sqlmock: %v", err) } defer db.Close() providerDir := writeFormulaFunctionProviderPackSource(t, "formula-provider-pack") providerPack, err := packs.ResolveInstalledSourceBackedPack( "formula-provider-pack", packs.AppPack, string(packs.PackSourceKindLocalPath), providerDir, ) if err != nil { t.Fatalf("resolve provider source-backed pack: %v", err) } providerHash, err := packs.PackPayloadHash(providerPack) if err != nil { t.Fatalf("provider payload hash: %v", err) } providerDecl, err := packs.BuildPersistedRuntimeDeclaration(providerPack) if err != nil { t.Fatalf("provider runtime declaration: %v", err) } providerMeta := providerPack.PackHashMetadata() consumerDir := writeFormulaFunctionConsumerPackSource(t, "formula-consumer-pack") consumerSourceReport := packs.ValidatePackSource("local_path:" + consumerDir) if consumerSourceReport.HasErrors() { t.Fatalf("consumer source diagnostics:\n%s", consumerSourceReport.Error()) } consumerPack, err := packs.NewSourceBackedPack( consumerSourceReport.Source, packs.WithRuntimeFormulaFunctionBundlesForValidation(providerPack.RuntimeFormulaFunctionBundles()), ) if err != nil { t.Fatalf("construct consumer source-backed pack with active provider bundles: %v", err) } consumerHash, err := packs.PackPayloadHash(consumerPack) if err != nil { t.Fatalf("consumer payload hash: %v", err) } consumerDecl, err := packs.BuildPersistedRuntimeDeclaration(consumerPack) if err != nil { t.Fatalf("consumer runtime declaration: %v", err) } consumerMeta := consumerPack.PackHashMetadata() expectInstalledPackRuntimeDeclarationColumns(mock, true) expectInstalledPackRuntimeStartupShallowAndFormulaBundleDeclarationColumns(mock, true, true) expectInstalledPackRuntimeDeclarationColumns(mock, true) expectInstalledPackRuntimeStartupShallowAndFormulaBundleDeclarationColumns(mock, true, true) mock.ExpectQuery("SELECT Name, Category, Status, PackVersion, ManifestFormatVersion, PlatformCompatibility, DesiredPayloadHash, AppliedPayloadHash, PayloadHashAlgorithm, SourceKind, SourceURI, SourceRevision, HostRequirementsDigest, RuntimeObjectSpecDeclarations, RuntimeRelatedListRules, RuntimeSearchScopeOrder, RuntimeSearchAllDefaultFields, RuntimeFormulaFunctionBundles, RuntimeStartupShallowDeclarations"). WithArgs(string(packs.AppPack), string(packs.ConnectorPack)). WillReturnRows(sqlmock.NewRows([]string{ "Name", "Category", "Status", "PackVersion", "ManifestFormatVersion", "PlatformCompatibility", "DesiredPayloadHash", "AppliedPayloadHash", "PayloadHashAlgorithm", "SourceKind", "SourceURI", "SourceRevision", "HostRequirementsDigest", "RuntimeObjectSpecDeclarations", "RuntimeRelatedListRules", "RuntimeSearchScopeOrder", "RuntimeSearchAllDefaultFields", "RuntimeFormulaFunctionBundles", "RuntimeStartupShallowDeclarations", }).AddRow( consumerPack.Name(), string(packs.AppPack), installedPackStatusActive, consumerMeta.PackVersion, consumerMeta.ManifestFormatVersion, consumerMeta.PlatformCompatibility, consumerHash, consumerHash, packs.PackPayloadHashAlgorithmV1, packs.PackSourceKindLocalPath, consumerDir, nil, nil, consumerDecl.ObjectPayloadsJSON(), consumerDecl.RelatedListRulesJSON(), consumerDecl.SearchScopeOrderJSON(), consumerDecl.SearchAllDefaultFieldsJSON(), consumerDecl.FormulaFunctionBundlesJSON(), consumerDecl.StartupShallowDeclaredRowsJSON(), ).AddRow( providerPack.Name(), string(packs.AppPack), installedPackStatusActive, providerMeta.PackVersion, providerMeta.ManifestFormatVersion, providerMeta.PlatformCompatibility, providerHash, providerHash, packs.PackPayloadHashAlgorithmV1, packs.PackSourceKindLocalPath, providerDir, nil, nil, providerDecl.ObjectPayloadsJSON(), providerDecl.RelatedListRulesJSON(), providerDecl.SearchScopeOrderJSON(), providerDecl.SearchAllDefaultFieldsJSON(), providerDecl.FormulaFunctionBundlesJSON(), providerDecl.StartupShallowDeclaredRowsJSON(), )) if err := refreshPersistedRuntimeDeclarationsForActivePacks(context.Background(), db); err != nil { t.Fatalf("refreshPersistedRuntimeDeclarationsForActivePacks: %v", err) } if err := mock.ExpectationsWereMet(); err != nil { t.Fatalf("sql expectations: %v", err) } } func prettyJSONString(t *testing.T, input string) string { t.Helper() var value any if err := json.Unmarshal([]byte(input), &value); err != nil { t.Fatalf("unmarshal JSON %q: %v", input, err) } data, err := json.MarshalIndent(value, "", " ") if err != nil { t.Fatalf("marshal pretty JSON: %v", err) } return string(data) } func TestRefreshPersistedRuntimeDeclarationsForActivePacksFailsClosedOnCASMiss(t *testing.T) { db, mock, err := sqlmock.New() if err != nil { t.Fatalf("sqlmock: %v", err) } defer db.Close() sourceRoot := writeConnectorPackSourceRoot(t) pack, err := packs.ResolveInstalledSourceBackedPack("test-connector", packs.ConnectorPack, packs.PackSourceKindLocalPath, sourceRoot) if err != nil { t.Fatalf("ResolveInstalledSourceBackedPack: %v", err) } payloadHash, err := packs.PackPayloadHash(pack) if err != nil { t.Fatalf("PackPayloadHash: %v", err) } decl, err := packs.BuildPersistedRuntimeDeclaration(pack) if err != nil { t.Fatalf("BuildPersistedRuntimeDeclaration: %v", err) } meta := pack.PackHashMetadata() expectInstalledPackRuntimeDeclarationColumns(mock, true) expectInstalledPackRuntimeStartupShallowDeclarationColumn(mock, true) expectInstalledPackRuntimeDeclarationColumns(mock, true) expectInstalledPackRuntimeStartupShallowDeclarationColumn(mock, true) mock.ExpectQuery("SELECT Name, Category, Status, PackVersion, ManifestFormatVersion, PlatformCompatibility, DesiredPayloadHash, AppliedPayloadHash, PayloadHashAlgorithm, SourceKind, SourceURI, SourceRevision, HostRequirementsDigest, RuntimeObjectSpecDeclarations, RuntimeRelatedListRules, RuntimeSearchScopeOrder, RuntimeSearchAllDefaultFields, RuntimeStartupShallowDeclarations"). WithArgs(string(packs.AppPack), string(packs.ConnectorPack)). WillReturnRows(sqlmock.NewRows([]string{ "Name", "Category", "Status", "PackVersion", "ManifestFormatVersion", "PlatformCompatibility", "DesiredPayloadHash", "AppliedPayloadHash", "PayloadHashAlgorithm", "SourceKind", "SourceURI", "SourceRevision", "HostRequirementsDigest", "RuntimeObjectSpecDeclarations", "RuntimeRelatedListRules", "RuntimeSearchScopeOrder", "RuntimeSearchAllDefaultFields", "RuntimeStartupShallowDeclarations", }).AddRow( "test-connector", string(packs.ConnectorPack), installedPackStatusActive, meta.PackVersion, meta.ManifestFormatVersion, meta.PlatformCompatibility, payloadHash, payloadHash, packs.PackPayloadHashAlgorithmV1, packs.PackSourceKindLocalPath, sourceRoot, nil, nil, `[]`, `[]`, `[]`, `[]`, `[]`, )) mock.ExpectExec(updateRuntimeDeclarationFourColumnCASPattern). WithArgs( decl.ObjectPayloadsJSON(), decl.RelatedListRulesJSON(), decl.SearchScopeOrderJSON(), decl.SearchAllDefaultFieldsJSON(), decl.StartupShallowDeclaredRowsJSON(), "test-connector", string(packs.ConnectorPack), installedPackStatusActive, packs.PackSourceKindLocalPath, sourceRoot, nil, payloadHash, payloadHash, `[]`, `[]`, `[]`, `[]`, `[]`, ). WillReturnResult(sqlmock.NewResult(0, 0)) err = refreshPersistedRuntimeDeclarationsForActivePacks(context.Background(), db) if err == nil || !strings.Contains(err.Error(), "affected 0 rows") { t.Fatalf("refreshPersistedRuntimeDeclarationsForActivePacks error = %v, want CAS miss failure", err) } if err := mock.ExpectationsWereMet(); err != nil { t.Fatalf("sql expectations: %v", err) } } func TestRefreshPersistedRuntimeDeclarationsForActivePacksRejectsInconsistentActiveHashesBeforeWrite(t *testing.T) { db, mock, err := sqlmock.New() if err != nil { t.Fatalf("sqlmock: %v", err) } defer db.Close() sourceRoot := writeConnectorPackSourceRoot(t) pack, err := packs.ResolveInstalledSourceBackedPack("test-connector", packs.ConnectorPack, packs.PackSourceKindLocalPath, sourceRoot) if err != nil { t.Fatalf("ResolveInstalledSourceBackedPack: %v", err) } payloadHash, err := packs.PackPayloadHash(pack) if err != nil { t.Fatalf("PackPayloadHash: %v", err) } meta := pack.PackHashMetadata() expectInstalledPackRuntimeDeclarationColumns(mock, true) expectInstalledPackRuntimeStartupShallowDeclarationColumn(mock, true) expectInstalledPackRuntimeDeclarationColumns(mock, true) expectInstalledPackRuntimeStartupShallowDeclarationColumn(mock, true) mock.ExpectQuery("SELECT Name, Category, Status, PackVersion, ManifestFormatVersion, PlatformCompatibility, DesiredPayloadHash, AppliedPayloadHash, PayloadHashAlgorithm, SourceKind, SourceURI, SourceRevision, HostRequirementsDigest, RuntimeObjectSpecDeclarations, RuntimeRelatedListRules, RuntimeSearchScopeOrder, RuntimeSearchAllDefaultFields, RuntimeStartupShallowDeclarations"). WithArgs(string(packs.AppPack), string(packs.ConnectorPack)). WillReturnRows(sqlmock.NewRows([]string{ "Name", "Category", "Status", "PackVersion", "ManifestFormatVersion", "PlatformCompatibility", "DesiredPayloadHash", "AppliedPayloadHash", "PayloadHashAlgorithm", "SourceKind", "SourceURI", "SourceRevision", "HostRequirementsDigest", "RuntimeObjectSpecDeclarations", "RuntimeRelatedListRules", "RuntimeSearchScopeOrder", "RuntimeSearchAllDefaultFields", "RuntimeStartupShallowDeclarations", }).AddRow( "test-connector", string(packs.ConnectorPack), installedPackStatusActive, meta.PackVersion, meta.ManifestFormatVersion, meta.PlatformCompatibility, packs.PackPayloadHashV1Prefix+"cccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccc", payloadHash, packs.PackPayloadHashAlgorithmV1, packs.PackSourceKindLocalPath, sourceRoot, nil, nil, `[]`, `[]`, `[]`, `[]`, `[]`, )) err = refreshPersistedRuntimeDeclarationsForActivePacks(context.Background(), db) if err == nil || !strings.Contains(err.Error(), "active hash state is inconsistent") { t.Fatalf("refreshPersistedRuntimeDeclarationsForActivePacks error = %v, want hash-state failure", err) } if err := mock.ExpectationsWereMet(); err != nil { t.Fatalf("sql expectations: %v", err) } } func TestListInstalledSourceBackedPacksLoadsRuntimeStartupShallowWhenColumnPresent(t *testing.T) { db, mock, err := sqlmock.New() if err != nil { t.Fatalf("sqlmock: %v", err) } defer db.Close() expectInstalledPackRuntimeDeclarationColumns(mock, true) expectInstalledPackRuntimeStartupShallowDeclarationColumn(mock, true) mock.ExpectQuery("SELECT Name, Category, Status, PackVersion, ManifestFormatVersion, PlatformCompatibility, DesiredPayloadHash, AppliedPayloadHash, PayloadHashAlgorithm, SourceKind, SourceURI, SourceRevision, HostRequirementsDigest, RuntimeObjectSpecDeclarations, RuntimeRelatedListRules, RuntimeSearchScopeOrder, RuntimeSearchAllDefaultFields, RuntimeStartupShallowDeclarations"). WithArgs(string(packs.AppPack), string(packs.ConnectorPack)). WillReturnRows(sqlmock.NewRows([]string{ "Name", "Category", "Status", "PackVersion", "ManifestFormatVersion", "PlatformCompatibility", "DesiredPayloadHash", "AppliedPayloadHash", "PayloadHashAlgorithm", "SourceKind", "SourceURI", "SourceRevision", "HostRequirementsDigest", "RuntimeObjectSpecDeclarations", "RuntimeRelatedListRules", "RuntimeSearchScopeOrder", "RuntimeSearchAllDefaultFields", "RuntimeStartupShallowDeclarations", }).AddRow( "jira-like-sample", string(packs.AppPack), installedPackStatusActive, "0.1.0", packs.SupportedManifestFormatVersion, nil, fakePayloadHashV1, fakePayloadHashV1, packs.PackPayloadHashAlgorithmV1, packs.PackSourceKindLocalPath, "/tmp/jira-like-sample", nil, nil, `[{"apiName":"Issue"}]`, `[]`, `[]`, `[]`, `[{"kind":"flow","businessKey":"issue-flow"}]`, )) rows, err := listInstalledSourceBackedPacks(context.Background(), db) if err != nil { t.Fatalf("listInstalledSourceBackedPacks: %v", err) } if len(rows) != 1 { t.Fatalf("row count = %d, want 1", len(rows)) } if !rows[0].HasRuntimeStartupDecl { t.Fatal("HasRuntimeStartupDecl = false, want true") } if got := rows[0].RuntimeStartupShallow.String; got != `[{"kind":"flow","businessKey":"issue-flow"}]` { t.Fatalf("RuntimeStartupShallow = %q, want startup payload", got) } if err := mock.ExpectationsWereMet(); err != nil { t.Fatalf("sql expectations: %v", err) } } func TestListInstalledSourceBackedPacksFallsBackWhenStartupShallowColumnAbsent(t *testing.T) { db, mock, err := sqlmock.New() if err != nil { t.Fatalf("sqlmock: %v", err) } defer db.Close() expectInstalledPackRuntimeDeclarationColumns(mock, true) expectInstalledPackRuntimeStartupShallowDeclarationColumn(mock, false) mock.ExpectQuery("SELECT Name, Category, Status, PackVersion, ManifestFormatVersion, PlatformCompatibility, DesiredPayloadHash, AppliedPayloadHash, PayloadHashAlgorithm, SourceKind, SourceURI, SourceRevision, HostRequirementsDigest, RuntimeObjectSpecDeclarations, RuntimeRelatedListRules, RuntimeSearchScopeOrder, RuntimeSearchAllDefaultFields\\s+FROM _System_InstalledPack"). WithArgs(string(packs.AppPack), string(packs.ConnectorPack)). WillReturnRows(sqlmock.NewRows([]string{ "Name", "Category", "Status", "PackVersion", "ManifestFormatVersion", "PlatformCompatibility", "DesiredPayloadHash", "AppliedPayloadHash", "PayloadHashAlgorithm", "SourceKind", "SourceURI", "SourceRevision", "HostRequirementsDigest", "RuntimeObjectSpecDeclarations", "RuntimeRelatedListRules", "RuntimeSearchScopeOrder", "RuntimeSearchAllDefaultFields", }).AddRow( "jira-like-sample", string(packs.AppPack), installedPackStatusActive, "0.1.0", packs.SupportedManifestFormatVersion, nil, fakePayloadHashV1, fakePayloadHashV1, packs.PackPayloadHashAlgorithmV1, packs.PackSourceKindLocalPath, "/tmp/jira-like-sample", nil, nil, `[{"apiName":"Issue"}]`, `[]`, `[]`, `[]`, )) rows, err := listInstalledSourceBackedPacks(context.Background(), db) if err != nil { t.Fatalf("listInstalledSourceBackedPacks: %v", err) } if len(rows) != 1 { t.Fatalf("row count = %d, want 1", len(rows)) } if rows[0].HasRuntimeStartupDecl { t.Fatal("HasRuntimeStartupDecl = true, want false") } if rows[0].RuntimeStartupShallow.Valid { t.Fatal("RuntimeStartupShallow.Valid = true, want false") } if err := mock.ExpectationsWereMet(); err != nil { t.Fatalf("sql expectations: %v", err) } } func TestVerifyInstalledPackHashStateAcceptsV0AndV1(t *testing.T) { pack := fakeInstalledPack{} if err := verifyInstalledPackHashState(installedPackHashRow(fakeLegacyHashV0, fakeLegacyHashV0, installedPackMetadata{}), pack); err != nil { t.Fatalf("verify v0 row: %v", err) } meta := installedPackMetadataFromPack(pack) if err := verifyInstalledPackHashState(installedPackHashRow(fakePayloadHashV1, fakePayloadHashV1, meta), pack); err != nil { t.Fatalf("verify v1 row: %v", err) } } func TestVerifyInstalledPackHashStateRejectsMixedActiveHashes(t *testing.T) { pack := fakeInstalledPack{} row := installedPackHashRow(fakeLegacyHashV0, fakePayloadHashV1, installedPackMetadataFromPack(pack)) err := verifyInstalledPackHashState(row, pack) if err == nil { t.Fatal("verify mixed v0/v1 row = nil, want error") } if !strings.Contains(err.Error(), "active hash state is inconsistent") { t.Fatalf("verify mixed v0/v1 row error = %v, want inconsistent hash state", err) } } func TestVerifyInstalledPackHashStateRejectsV1WithoutMetadata(t *testing.T) { pack := fakeInstalledPack{} row := installedPackHashRow(fakePayloadHashV1, fakePayloadHashV1, installedPackMetadata{}) err := verifyInstalledPackHashState(row, pack) if err == nil { t.Fatal("verify v1 row without metadata = nil, want error") } if !strings.Contains(err.Error(), "hash metadata") { t.Fatalf("verify v1 row without metadata error = %v, want hash metadata", err) } } func TestVerifyInstalledPackHashStateRejectsDriftedHashes(t *testing.T) { pack := fakeInstalledPack{} if err := verifyInstalledPackHashState(installedPackHashRow(strings.Replace(fakeLegacyHashV0, "a", "c", 1), strings.Replace(fakeLegacyHashV0, "a", "c", 1), installedPackMetadata{}), pack); err == nil { t.Fatal("verify drifted v0 row = nil, want error") } driftedV1 := packs.PackPayloadHashV1Prefix + "cccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccc" if err := verifyInstalledPackHashState(installedPackHashRow(driftedV1, driftedV1, installedPackMetadataFromPack(pack)), pack); err == nil { t.Fatal("verify drifted v1 row = nil, want error") } } func TestConnectorInboundWorkerLifecycleVersionUsesWorkerEligibilityProjection(t *testing.T) { db, mock, err := sqlmock.New() if err != nil { t.Fatalf("sqlmock: %v", err) } defer db.Close() expectConnectorInboundWorkerEligibilityRows(mock, []driver.Value{"active-id", "install-v1", "acme-connector", string(packs.ConnectorPack), "active", "hash-v1", 0}, ) expectConnectorInboundProjectionLifecycleRows(mock) got, err := ConnectorInboundWorkerLifecycleVersion(context.Background(), db) if err != nil { t.Fatalf("ConnectorInboundWorkerLifecycleVersion: %v", err) } if got == "" || !strings.HasPrefix(got, "1:") { t.Fatalf("version = %q, want count-prefixed hash", got) } if err := mock.ExpectationsWereMet(); err != nil { t.Fatalf("sql expectations: %v", err) } } func TestConnectorInboundWorkerLifecycleVersionIncludesDrainOnlyEligibility(t *testing.T) { versionWithoutDrain := connectorInboundWorkerLifecycleVersionForEligibilityRows(t) versionWithDrain := connectorInboundWorkerLifecycleVersionForEligibilityRows(t, []driver.Value{"acme-connector", "install-v1", "acme-connector", string(packs.ConnectorPack), "removing", "hash-v1", 1}, ) if versionWithoutDrain == versionWithDrain { t.Fatalf("versions matched %q; want drain-only eligibility to affect reconcile version", versionWithDrain) } } func TestConnectorInboundWorkerLifecycleVersionIncludesCompanionProjectionRows(t *testing.T) { workerRows := [][]driver.Value{ {"active-id", "install-v1", "acme-connector", string(packs.ConnectorPack), "active", "hash-v1", 0}, } versionEmail := connectorInboundWorkerLifecycleVersionForRows(t, workerRows, [][]driver.Value{ {"acme-connector", "install-v1", "acme-companion", "companion-install-v1", "contacts", "email"}, }) versionFirstName := connectorInboundWorkerLifecycleVersionForRows(t, workerRows, [][]driver.Value{ {"acme-connector", "install-v1", "acme-companion", "companion-install-v1", "contacts", "firstname"}, }) if versionEmail == versionFirstName { t.Fatalf("versions matched %q; want companion projection source field to affect reconcile version", versionEmail) } } func TestConnectorInboundWorkerLifecycleVersion_UsesProjectionSnapshotRowsWithoutLiveDependencyStatusSentinel(t *testing.T) { workerRows := [][]driver.Value{ {"active-id", "install-v1", "acme-connector", string(packs.ConnectorPack), "active", "hash-v1", 0}, {"active-id-2", "install-v2", "other-connector", string(packs.ConnectorPack), "active", "hash-v2", 0}, } emptyVersion := connectorInboundWorkerLifecycleVersionForRows(t, workerRows, nil) snapshotVersion := connectorInboundWorkerLifecycleVersionForRows(t, workerRows, [][]driver.Value{ {"acme-connector", "install-v1", "acme-companion", "companion-install-v2", "contacts", "email"}, }) if snapshotVersion == "" { t.Fatal("version = empty, want lifecycle version despite companion mid-apply") } if emptyVersion == snapshotVersion { t.Fatalf("versions matched %q; want projection snapshot rows to affect lifecycle version", snapshotVersion) } } func TestRefreshConnectorInboundProjectionSnapshotsForOwnerPack_SoftDeletesStaleRowsWhenActiveDependencyHasNoMappings(t *testing.T) { db, mock, err := sqlmock.New() if err != nil { t.Fatalf("sqlmock: %v", err) } defer db.Close() now := time.Date(2026, time.June, 6, 18, 6, 0, 0, time.UTC) mock.ExpectQuery("SELECT COALESCE\\(PackInstallId, ''\\)\\s+FROM _System_InstalledPack"). WithArgs("hubspot-crm-connector"). WillReturnRows(sqlmock.NewRows([]string{"PackInstallId"}).AddRow("install-active")) mock.ExpectQuery("SELECT COUNT\\(\\*\\)\\s+FROM _System_ConnectorCalloutProfile"). WithArgs("hubspot-crm-connector", "install-active"). WillReturnRows(sqlmock.NewRows([]string{"COUNT(*)"}).AddRow(1)) mock.ExpectQuery("SELECT DISTINCT COALESCE\\(CompanionPackInstallId, ''\\)\\s+FROM _System_ConnectorEvent"). WithArgs("install-active"). WillReturnRows(sqlmock.NewRows([]string{"CompanionPackInstallId"})) mock.ExpectQuery("SELECT DISTINCT d\\.DependentPackName, COALESCE\\(d\\.DependentPackInstallId, ''\\), m\\.SourceObject, m\\.SourceField"). WithArgs("hubspot-crm-connector"). WillReturnRows(sqlmock.NewRows([]string{"DependentPackName", "DependentPackInstallId", "SourceObject", "SourceField"})) mock.ExpectExec("UPDATE _System_ConnectorInboundProjectionSnapshot"). WithArgs(now, "install-active"). WillReturnResult(sqlmock.NewResult(0, 2)) if err := refreshConnectorInboundProjectionSnapshotsForOwnerPack(context.Background(), db, "hubspot-crm-connector", now); err != nil { t.Fatalf("refreshConnectorInboundProjectionSnapshotsForOwnerPack: %v", err) } if err := mock.ExpectationsWereMet(); err != nil { t.Fatalf("sql expectations: %v", err) } } func TestRefreshConnectorInboundProjectionSnapshotsForActiveConnectorPacksBackfillsProjectionRows(t *testing.T) { db, mock, err := sqlmock.New() if err != nil { t.Fatalf("sqlmock: %v", err) } defer db.Close() now := time.Date(2026, time.June, 22, 9, 15, 0, 0, time.UTC) mock.ExpectQuery("SELECT DISTINCT cp\\.Name\\s+FROM _System_InstalledPack cp"). WillReturnRows(sqlmock.NewRows([]string{"Name"}).AddRow("acme-connector")) mock.ExpectQuery("SELECT COALESCE\\(PackInstallId, ''\\)\\s+FROM _System_InstalledPack"). WithArgs("acme-connector"). WillReturnRows(sqlmock.NewRows([]string{"PackInstallId"}).AddRow("install-active")) mock.ExpectQuery("SELECT COUNT\\(\\*\\)\\s+FROM _System_ConnectorCalloutProfile"). WithArgs("acme-connector", "install-active"). WillReturnRows(sqlmock.NewRows([]string{"COUNT(*)"}).AddRow(1)) mock.ExpectQuery("SELECT DISTINCT COALESCE\\(CompanionPackInstallId, ''\\)\\s+FROM _System_ConnectorEvent"). WithArgs("install-active"). WillReturnRows(sqlmock.NewRows([]string{"CompanionPackInstallId"})) mock.ExpectQuery("SELECT DISTINCT d\\.DependentPackName, COALESCE\\(d\\.DependentPackInstallId, ''\\), m\\.SourceObject, m\\.SourceField"). WithArgs("acme-connector"). WillReturnRows(sqlmock.NewRows([]string{"DependentPackName", "DependentPackInstallId", "SourceObject", "SourceField"}). AddRow("acme-app", "app-install", "contacts", "email")) mock.ExpectExec("INSERT INTO _System_ConnectorInboundProjectionSnapshot"). WithArgs(sqlmock.AnyArg(), "acme-connector", "install-active", "acme-app", "app-install", "contacts", "email", now, now). WillReturnResult(sqlmock.NewResult(0, 1)) mock.ExpectExec("UPDATE _System_ConnectorInboundProjectionSnapshot"). WithArgs(now, "install-active", sqlmock.AnyArg()). WillReturnResult(sqlmock.NewResult(0, 0)) if err := RefreshConnectorInboundProjectionSnapshotsForActiveConnectorPacks(context.Background(), db, now); err != nil { t.Fatalf("RefreshConnectorInboundProjectionSnapshotsForActiveConnectorPacks: %v", err) } if err := mock.ExpectationsWereMet(); err != nil { t.Fatalf("sql expectations: %v", err) } } func TestConnectorInboundProjectionSnapshotUpsertKeepsTimestampOnNoop(t *testing.T) { sourceBytes, err := os.ReadFile("installed_pack_connector_projection.go") if err != nil { t.Fatalf("read installed_pack_connector_projection.go: %v", err) } source := string(sourceBytes) insert := strings.Index(source, "INSERT INTO _System_ConnectorInboundProjectionSnapshot") if insert == -1 { t.Fatal("projection snapshot insert statement missing") } upsert := source[insert:] if !strings.Contains(upsert, "LastModifiedDate = CASE") || !strings.Contains(upsert, "ELSE LastModifiedDate") { t.Fatal("projection snapshot duplicate upsert must preserve LastModifiedDate when the logical row is unchanged") } lastModified := strings.Index(upsert, "LastModifiedDate = CASE") isDeleted := strings.Index(upsert, "IsDeleted = 0`") if lastModified == -1 || isDeleted == -1 { t.Fatalf("projection snapshot upsert markers missing: LastModifiedDate=%d IsDeleted=%d", lastModified, isDeleted) } if lastModified > isDeleted { t.Fatal("LastModifiedDate CASE must run before IsDeleted reset so soft-deleted rows still bump timestamp when revived") } } func TestRefreshConnectorInboundProjectionSnapshotsForOwnerPack_UsesTransactionAfterRowsClosed(t *testing.T) { db, cleanup := newTemporaryConnectorProjectionSnapshotDatabase(t) defer cleanup() ctx := context.Background() createConnectorProjectionSnapshotTestTables(t, ctx, db) seedConnectorProjectionSnapshotTestRows(t, ctx, db) tx, err := db.BeginTx(ctx, nil) if err != nil { t.Fatalf("begin tx: %v", err) } committed := false defer func() { if !committed { _ = tx.Rollback() } }() now := time.Date(2026, time.June, 7, 6, 50, 36, 0, time.UTC) if err := refreshConnectorInboundProjectionSnapshotsForOwnerPack(ctx, tx, "hubspot-crm-connector", now); err != nil { t.Fatalf("refreshConnectorInboundProjectionSnapshotsForOwnerPack: %v", err) } if err := tx.Commit(); err != nil { t.Fatalf("commit tx: %v", err) } committed = true var snapshotCount int if err := db.QueryRowContext(ctx, ` SELECT COUNT(*) FROM _System_ConnectorInboundProjectionSnapshot WHERE OwnerPackName = ? AND CompanionPackName = ? AND ObjectKind = ? AND SourceField IN ('annualrevenue', 'name') AND COALESCE(IsDeleted, 0) = 0`, "hubspot-crm-connector", "pingcap-hubspot-companion", "companies", ).Scan(&snapshotCount); err != nil { t.Fatalf("count projection snapshots: %v", err) } if snapshotCount != 2 { t.Fatalf("snapshot count = %d, want 2", snapshotCount) } } func TestProjectionOnlyRawPayloadLifecycleConvergesPhysicalAndSnapshotState(t *testing.T) { db, cleanup := newTemporaryConnectorProjectionSnapshotDatabase(t) defer cleanup() ctx := context.Background() createConnectorProjectionSnapshotTestTables(t, ctx, db) seedConnectorProjectionSnapshotTestRows(t, ctx, db) if _, err := db.ExecContext(ctx, `INSERT INTO _System_Object (Id, APIName, IsDeleted) VALUES ('lead-object', 'Lead', 0)`); err != nil { t.Fatalf("seed target object: %v", err) } if _, err := db.ExecContext(ctx, `INSERT INTO _System_Field (Id, ObjectId, APIName, Updateable, IsDeleted) VALUES ('legacy-field', 'lead-object', 'Legacy_Source_Id__c', 1, 0)`); err != nil { t.Fatalf("seed target field: %v", err) } const owner = "pingcap-hubspot-companion" projectionPayload := []byte(`[{ "sourceObject":"contacts","targetObject":"Lead","actionType":"syncContact","direction":"inbound", "mappings":[{"sourceField":"legacy_source_id","projectionOnly":true}] }]`) executablePayload := []byte(`[{ "sourceObject":"contacts","targetObject":"Lead","actionType":"syncContact","direction":"inbound", "mappings":[{"sourceField":"legacy_source_id","targetField":"Legacy_Source_Id__c"}] }]`) declared, err := packs.InitializeConnectorFieldMappingsFromPayloadForOwner(db, projectionPayload, owner) if err != nil { t.Fatalf("initialize projection payload: %v", err) } if _, err := db.ExecContext(ctx, ` UPDATE _System_ObjectFieldMapping SET SourceLookupField = 'stale_lookup', IsCustomized = 1 WHERE Source = ? AND SourceField = 'legacy_source_id'`, owner); err != nil { t.Fatalf("seed projection-only admin drift: %v", err) } if _, err := packs.InitializeConnectorFieldMappingsFromPayloadForOwner(db, projectionPayload, owner); err != nil { t.Fatalf("reapply projection payload: %v", err) } if err := packs.VerifyConnectorFieldMappingsFromPayloadForOwner(ctx, db, projectionPayload, owner); err != nil { t.Fatalf("verify projection payload: %v", err) } var sourceLookup sql.NullString var targetField string var isCustomized, projectionOnly bool if err := db.QueryRowContext(ctx, ` SELECT TargetField, SourceLookupField, IsCustomized, ProjectionOnly FROM _System_ObjectFieldMapping WHERE Source = ? AND SourceField = 'legacy_source_id'`, owner). Scan(&targetField, &sourceLookup, &isCustomized, &projectionOnly); err != nil { t.Fatalf("read projection row: %v", err) } if targetField != "" || sourceLookup.Valid || isCustomized || !projectionOnly { t.Fatalf("projection row did not clear write/admin semantics: target=%q lookup=%v customized=%v projection=%v", targetField, sourceLookup, isCustomized, projectionOnly) } now := time.Date(2026, time.July, 18, 12, 0, 0, 0, time.UTC) if err := refreshConnectorInboundProjectionSnapshotsForOwnerPack(ctx, db, "hubspot-crm-connector", now); err != nil { t.Fatalf("refresh projection snapshot: %v", err) } var snapshotCount int if err := db.QueryRowContext(ctx, ` SELECT COUNT(*) FROM _System_ConnectorInboundProjectionSnapshot WHERE OwnerPackName = 'hubspot-crm-connector' AND CompanionPackName = ? AND ObjectKind = 'contacts' AND SourceField = 'legacy_source_id' AND IsDeleted = 0`, owner).Scan(&snapshotCount); err != nil { t.Fatalf("read projection snapshot: %v", err) } if snapshotCount != 1 { t.Fatalf("projection snapshot count = %d, want 1", snapshotCount) } declared, err = packs.InitializeConnectorFieldMappingsFromPayloadForOwner(db, executablePayload, owner) if err != nil { t.Fatalf("convert projection to executable: %v", err) } if err := packs.VerifyConnectorFieldMappingsFromPayloadForOwner(ctx, db, executablePayload, owner); err != nil { t.Fatalf("verify executable conversion: %v", err) } if err := db.QueryRowContext(ctx, ` SELECT TargetField, ProjectionOnly FROM _System_ObjectFieldMapping WHERE Source = ? AND SourceField = 'legacy_source_id'`, owner).Scan(&targetField, &projectionOnly); err != nil { t.Fatalf("read executable conversion: %v", err) } if targetField != "Legacy_Source_Id__c" || projectionOnly { t.Fatalf("executable conversion target=%q projection=%v", targetField, projectionOnly) } if _, err := db.ExecContext(ctx, ` UPDATE _System_ObjectFieldMapping SET ProjectionOnly = 1 WHERE Source = ? AND SourceField = 'legacy_source_id'`, owner); err != nil { t.Fatalf("seed executable projection-state drift: %v", err) } if err := packs.VerifyConnectorFieldMappingsFromPayloadForOwner(ctx, db, executablePayload, owner); err == nil || !strings.Contains(err.Error(), "is missing") { t.Fatalf("executable projection-state drift error = %v", err) } if _, err := packs.InitializeConnectorFieldMappingsFromPayloadForOwner(db, executablePayload, owner); err != nil { t.Fatalf("repair executable projection-state drift: %v", err) } if err := packs.VerifyConnectorFieldMappingsFromPayloadForOwner(ctx, db, executablePayload, owner); err != nil { t.Fatalf("verify repaired executable conversion: %v", err) } declared, err = packs.InitializeConnectorFieldMappingsFromPayloadForOwner(db, projectionPayload, owner) if err != nil { t.Fatalf("convert executable back to projection: %v", err) } if err := packs.VerifyConnectorFieldMappingsFromPayloadForOwner(ctx, db, projectionPayload, owner); err != nil { t.Fatalf("verify projection reconversion: %v", err) } if err := db.QueryRowContext(ctx, ` SELECT TargetField, ProjectionOnly FROM _System_ObjectFieldMapping WHERE Source = ? AND SourceField = 'legacy_source_id'`, owner).Scan(&targetField, &projectionOnly); err != nil { t.Fatalf("read projection reconversion: %v", err) } if targetField != "" || !projectionOnly { t.Fatalf("projection reconversion target=%q projection=%v", targetField, projectionOnly) } if _, err := db.ExecContext(ctx, ` UPDATE _System_ObjectFieldMapping SET ProjectionOnly = 0 WHERE Source = ? AND SourceField = 'legacy_source_id'`, owner); err != nil { t.Fatalf("seed projection executable-state drift: %v", err) } if err := packs.VerifyConnectorFieldMappingsFromPayloadForOwner(ctx, db, projectionPayload, owner); err == nil || !strings.Contains(err.Error(), "projection-only physical row drifted") { t.Fatalf("projection executable-state drift error = %v", err) } declared, err = packs.InitializeConnectorFieldMappingsFromPayloadForOwner(db, projectionPayload, owner) if err != nil { t.Fatalf("repair projection executable-state drift: %v", err) } if err := packs.VerifyConnectorFieldMappingsFromPayloadForOwner(ctx, db, projectionPayload, owner); err != nil { t.Fatalf("verify repaired projection reconversion: %v", err) } if err := packs.SweepConnectorFieldMappingsForOwner(ctx, db, owner, declared); err != nil { t.Fatalf("sweep projection declaration: %v", err) } if err := packs.VerifyNoStaleConnectorFieldMappingsForOwner(ctx, db, owner, declared); err != nil { t.Fatalf("verify projection stale sweep: %v", err) } var activeRows int if err := db.QueryRowContext(ctx, `SELECT COUNT(*) FROM _System_ObjectFieldMapping WHERE Source = ? AND IsActive = 1 AND IsDeleted = 0`, owner).Scan(&activeRows); err != nil { t.Fatalf("count active mappings after sweep: %v", err) } if activeRows != 1 { t.Fatalf("active rows after sweep = %d, want only the declared projection row", activeRows) } } func newTemporaryConnectorProjectionSnapshotDatabase(t *testing.T) (*sql.DB, func()) { t.Helper() host := os.Getenv("TIDB_HOST") if host == "" { host = "127.0.0.1" } port := os.Getenv("TIDB_PORT") if port == "" { port = "4000" } user := os.Getenv("TIDB_USER") if user == "" { user = "root" } password := os.Getenv("TIDB_PASSWORD") adminDSN := fmt.Sprintf("%s:%s@tcp(%s:%s)/?charset=utf8mb4&parseTime=True&loc=Local", user, password, host, port) adminDB, err := sql.Open("mysql", adminDSN) if err != nil { t.Fatalf("open admin TiDB connection: %v", err) } if err := adminDB.Ping(); err != nil { _ = adminDB.Close() t.Skipf("skipping integration test: TiDB is not reachable at %s:%s: %v", host, port, err) } dbName := fmt.Sprintf("nexuscrm_projection_snapshot_%d", time.Now().UnixNano()) if _, err := adminDB.Exec("CREATE DATABASE `" + dbName + "` CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci"); err != nil { _ = adminDB.Close() t.Skipf("skipping integration test: cannot create temporary database %s: %v", dbName, err) } dsn := fmt.Sprintf("%s:%s@tcp(%s:%s)/%s?charset=utf8mb4&parseTime=True&loc=Local", user, password, host, port, dbName) db, err := sql.Open("mysql", dsn) if err != nil { _, _ = adminDB.Exec("DROP DATABASE `" + dbName + "`") _ = adminDB.Close() t.Fatalf("open temp database: %v", err) } cleanup := func() { _ = db.Close() _, _ = adminDB.Exec("DROP DATABASE `" + dbName + "`") _ = adminDB.Close() } return db, cleanup } func createConnectorProjectionSnapshotTestTables(t *testing.T, ctx context.Context, db *sql.DB) { t.Helper() stmts := []string{ `CREATE TABLE _System_InstalledPack ( Id VARCHAR(255) PRIMARY KEY, Name VARCHAR(255) NOT NULL, PackInstallId VARCHAR(255) NOT NULL DEFAULT '', Category VARCHAR(64) NOT NULL, Status VARCHAR(64) NOT NULL, AppliedPayloadHash VARCHAR(128), DesiredPayloadHash VARCHAR(128), IsDeleted TINYINT(1) NOT NULL DEFAULT 0, LastModifiedDate DATETIME DEFAULT CURRENT_TIMESTAMP, UNIQUE KEY uniq_installed_pack_name_install (Name, PackInstallId) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci`, `CREATE TABLE _System_ConnectorCalloutProfile ( Id VARCHAR(255) PRIMARY KEY, OwnerPack VARCHAR(255) NOT NULL, InstalledPackId VARCHAR(255) NOT NULL, BatchSpec JSON, IsDeleted TINYINT(1) NOT NULL DEFAULT 0, LastModifiedDate DATETIME DEFAULT CURRENT_TIMESTAMP ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci`, `CREATE TABLE _System_ConnectorEvent ( Id VARCHAR(255) PRIMARY KEY, InstalledPackId VARCHAR(255) NOT NULL, Status VARCHAR(32) NOT NULL, CompanionPackInstallId VARCHAR(255), LastModifiedDate DATETIME DEFAULT CURRENT_TIMESTAMP ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci`, `CREATE TABLE _System_InstalledPackDependency ( Id VARCHAR(255) PRIMARY KEY, DependentPackName VARCHAR(255) NOT NULL, DependentPackInstallId VARCHAR(255) NOT NULL DEFAULT '', DependsOnPackName VARCHAR(255) NOT NULL, IsDeleted TINYINT(1) NOT NULL DEFAULT 0, LastModifiedDate DATETIME DEFAULT CURRENT_TIMESTAMP ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci`, `CREATE TABLE _System_ObjectFieldMapping ( Id VARCHAR(255) PRIMARY KEY, Source VARCHAR(100) NOT NULL, SourceObject VARCHAR(100) NOT NULL, TargetObject VARCHAR(100) NOT NULL DEFAULT '', ActionType VARCHAR(100) NOT NULL DEFAULT '', SourceField VARCHAR(100) NOT NULL, TargetField VARCHAR(255) NOT NULL DEFAULT '', SourceLookupField VARCHAR(255), DefaultValue TEXT, ValueMap JSON, SortOrder INT NOT NULL DEFAULT 0, IsCustomized TINYINT(1) NOT NULL DEFAULT 0, Direction VARCHAR(32) NOT NULL, TransformExpression TEXT, ConflictPolicy VARCHAR(64), OnUnmappedValue VARCHAR(64), ProjectionOnly TINYINT(1) NOT NULL DEFAULT 0, IsActive TINYINT(1) NOT NULL DEFAULT 1, IsDeleted TINYINT(1) NOT NULL DEFAULT 0, CreatedDate DATETIME DEFAULT CURRENT_TIMESTAMP, LastModifiedDate DATETIME DEFAULT CURRENT_TIMESTAMP, UNIQUE KEY uniq_object_field_mapping_business (Source, SourceObject, TargetObject, ActionType, SourceField, Direction) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci`, `CREATE TABLE _System_Object ( Id VARCHAR(255) PRIMARY KEY, APIName VARCHAR(255) NOT NULL, IsDeleted TINYINT(1) NOT NULL DEFAULT 0 ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci`, `CREATE TABLE _System_Field ( Id VARCHAR(255) PRIMARY KEY, ObjectId VARCHAR(255) NOT NULL, APIName VARCHAR(255) NOT NULL, Updateable TINYINT(1) NOT NULL DEFAULT 1, IsDeleted TINYINT(1) NOT NULL DEFAULT 0 ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci`, `CREATE TABLE _System_ConnectorInboundProjectionSnapshot ( Id VARCHAR(255) PRIMARY KEY, OwnerPackName VARCHAR(255) NOT NULL, OwnerPackInstallId VARCHAR(255) NOT NULL, CompanionPackName VARCHAR(255) NOT NULL, CompanionPackInstallId VARCHAR(255) NOT NULL, ObjectKind VARCHAR(255) NOT NULL, SourceField VARCHAR(255) NOT NULL, IsDeleted TINYINT(1) NOT NULL DEFAULT 0, CreatedDate DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, LastModifiedDate DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, KEY idx_projection_owner_install (OwnerPackInstallId) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci`, } for _, stmt := range stmts { if _, err := db.ExecContext(ctx, stmt); err != nil { t.Fatalf("create projection test table: %v\n%s", err, stmt) } } } func seedConnectorProjectionSnapshotTestRows(t *testing.T, ctx context.Context, db *sql.DB) { t.Helper() stmts := []struct { query string args []any }{ { query: `INSERT INTO _System_InstalledPack (Id, Name, PackInstallId, Category, Status, AppliedPayloadHash, DesiredPayloadHash, IsDeleted) VALUES (?, ?, ?, ?, ?, ?, ?, 0)`, args: []any{"connector-row", "hubspot-crm-connector", "connector-install", string(packs.ConnectorPack), installedPackStatusActive, fakePayloadHashV1, fakePayloadHashV1}, }, { query: `INSERT INTO _System_InstalledPack (Id, Name, PackInstallId, Category, Status, AppliedPayloadHash, DesiredPayloadHash, IsDeleted) VALUES (?, ?, ?, ?, ?, ?, ?, 0)`, args: []any{"companion-row", "pingcap-hubspot-companion", "companion-install", string(packs.AppPack), installedPackStatusActive, fakePayloadHashV1, fakePayloadHashV1}, }, { query: `INSERT INTO _System_ConnectorCalloutProfile (Id, OwnerPack, InstalledPackId, BatchSpec, IsDeleted) VALUES (?, ?, ?, ?, 0)`, args: []any{"profile-row", "hubspot-crm-connector", "connector-install", `{"projectionBodyField":"properties"}`}, }, { query: `INSERT INTO _System_InstalledPackDependency (Id, DependentPackName, DependentPackInstallId, DependsOnPackName, IsDeleted) VALUES (?, ?, ?, ?, 0)`, args: []any{"dep-row", "pingcap-hubspot-companion", "companion-install", "hubspot-crm-connector"}, }, { query: `INSERT INTO _System_ObjectFieldMapping (Id, Source, SourceObject, SourceField, Direction, ProjectionOnly, IsActive, IsDeleted) VALUES (?, ?, ?, ?, ?, 1, 1, 0)`, args: []any{"mapping-annualrevenue", "pingcap-hubspot-companion", "companies", "annualrevenue", "inbound"}, }, { query: `INSERT INTO _System_ObjectFieldMapping (Id, Source, SourceObject, SourceField, Direction, IsActive, IsDeleted) VALUES (?, ?, ?, ?, ?, 1, 0)`, args: []any{"mapping-name", "pingcap-hubspot-companion", "companies", "name", "inbound"}, }, } for _, stmt := range stmts { if _, err := db.ExecContext(ctx, stmt.query, stmt.args...); err != nil { t.Fatalf("seed projection snapshot row: %v\n%s", err, stmt.query) } } } func connectorInboundWorkerLifecycleVersionForEligibilityRows(t *testing.T, eligibilityRows ...[]driver.Value) string { t.Helper() return connectorInboundWorkerLifecycleVersionForRows(t, eligibilityRows, nil) } func connectorInboundWorkerLifecycleVersionForRows(t *testing.T, eligibilityRows [][]driver.Value, projectionRows [][]driver.Value) string { t.Helper() db, mock, err := sqlmock.New() if err != nil { t.Fatalf("sqlmock: %v", err) } defer db.Close() expectConnectorInboundWorkerEligibilityRows(mock, eligibilityRows...) expectConnectorInboundProjectionLifecycleRows(mock, projectionRows...) got, err := ConnectorInboundWorkerLifecycleVersion(context.Background(), db) if err != nil { t.Fatalf("ConnectorInboundWorkerLifecycleVersion: %v", err) } if err := mock.ExpectationsWereMet(); err != nil { t.Fatalf("sql expectations: %v", err) } return got } func expectConnectorInboundWorkerEligibilityRows(mock sqlmock.Sqlmock, values ...[]driver.Value) { rows := sqlmock.NewRows([]string{ "Id", "PackInstallId", "Name", "Category", "Status", "AppliedPayloadHash", "DrainOnly", }) for _, value := range values { rows.AddRow(value...) } mock.ExpectQuery("SELECT Id, PackInstallId, Name, Category, Status, COALESCE\\(AppliedPayloadHash"). WithArgs( installedPackStatusActive, string(packs.ConnectorPack), installedPackStatusActive, installedPackStatusActive, installedPackStatusActive, installedPackStatusActive, ). WillReturnRows(rows) } func expectConnectorInboundProjectionLifecycleRows(mock sqlmock.Sqlmock, values ...[]driver.Value) { rows := sqlmock.NewRows([]string{ "Name", "PackInstallId", "CompanionPackName", "CompanionPackInstallId", "ObjectKind", "SourceField", }) for _, value := range values { rows.AddRow(value...) } mock.ExpectQuery("(?s)SELECT DISTINCT cp\\.Name, cp\\.PackInstallId, s\\.CompanionPackName.*_System_ConnectorInboundProjectionSnapshot s.*_System_ConnectorCalloutProfile profile.*d\\.DependentPackInstallId = s\\.CompanionPackInstallId.*EXISTS\\s*\\(\\s*SELECT 1\\s*FROM _System_InstalledPack p\\s*WHERE p\\.Name = d\\.DependentPackName.*p\\.Status = 'installing'"). WithArgs( string(packs.ConnectorPack), installedPackStatusActive, string(packs.AppPack), installedPackStatusActive, string(packs.AppPack), ). WillReturnRows(rows) } func TestFinalizeAppliedAppPacksDoesNotRewriteAppliedHashBeforePostVerify(t *testing.T) { db, mock, err := sqlmock.New() if err != nil { t.Fatalf("sqlmock: %v", err) } defer db.Close() pack := &postVerifyFailingInstalledPack{finalErr: errors.New("post verify failed")} appPacks := []appliedAppPack{{pack: pack, payloadHash: fakePayloadHashV1}} // markInstalledPackFailed runs a namespace-reservation SELECT before the INSERT. mock.ExpectQuery("SELECT Category, COALESCE\\(IsDeleted, 0\\) FROM _System_InstalledPack WHERE Name"). WithArgs("fake-pack"). WillReturnRows(sqlmock.NewRows([]string{"Category", "IsDeleted"})) expectInstalledPackRuntimeDeclarationColumns(mock, false) mock.ExpectExec("INSERT INTO _System_InstalledPack"). WillReturnResult(sqlmock.NewResult(0, 1)) err = finalizeAppliedAppPacksWith( context.Background(), db, appPacks, func(context.Context, *sql.DB, []packs.Pack) error { return nil }, func(*sql.DB) error { return nil }, nil, ) if err == nil { t.Fatal("finalize applied app packs = nil, want post-verify error") } if !strings.Contains(err.Error(), "post-maintenance verify") { t.Fatalf("finalize error = %v, want post-maintenance verify", err) } if pack.verifyCalls != 2 { t.Fatalf("verify calls = %d, want 2", pack.verifyCalls) } if err := mock.ExpectationsWereMet(); err != nil { t.Fatalf("sql expectations: %v", err) } } func TestFinalizeAppliedAppPacksRunsCandidateFinalVerifyBeforeActivation(t *testing.T) { db, mock, err := sqlmock.New() if err != nil { t.Fatalf("sqlmock: %v", err) } defer db.Close() pack := &postVerifyFailingInstalledPack{} appPacks := []appliedAppPack{{pack: pack, payloadHash: fakePayloadHashV1}} // Candidate verification failure is persisted without entering the active // transaction, so the previous generation remains authoritative. mock.ExpectQuery("SELECT Category, COALESCE\\(IsDeleted, 0\\) FROM _System_InstalledPack WHERE Name"). WithArgs("fake-pack"). WillReturnRows(sqlmock.NewRows([]string{"Category", "IsDeleted"})) expectInstalledPackRuntimeDeclarationColumns(mock, false) mock.ExpectExec("INSERT INTO _System_InstalledPack"). WillReturnResult(sqlmock.NewResult(0, 1)) beforeActiveCalled := false candidateErr := errors.New("candidate programs invalid") err = finalizeAppliedAppPacksWithActiveHook( context.Background(), db, appPacks, func(context.Context, *sql.DB, []packs.Pack) error { return nil }, func(*sql.DB) error { return nil }, nil, func(_ context.Context, candidates []packs.Pack) error { if pack.verifyCalls != 2 { t.Fatalf("candidate verifier ran after %d pack verifies, want 2", pack.verifyCalls) } if len(candidates) != 1 || candidates[0] != pack { t.Fatalf("candidate packs = %#v, want exact candidate", candidates) } return candidateErr }, func(context.Context, *sql.Tx, appliedAppPack) error { beforeActiveCalled = true return nil }, ) if !errors.Is(err, candidateErr) || !strings.Contains(err.Error(), "candidate final verify") { t.Fatalf("finalize error = %v, want candidate verification failure", err) } if beforeActiveCalled { t.Fatal("active transaction hook ran after candidate verification failure") } if err := mock.ExpectationsWereMet(); err != nil { t.Fatalf("sql expectations: %v", err) } } func TestFinalizeAppliedAppPacksMarksFailedWhenMaterializationFails(t *testing.T) { db, mock, err := sqlmock.New() if err != nil { t.Fatalf("sqlmock: %v", err) } defer db.Close() pack := &postVerifyFailingInstalledPack{} appPacks := []appliedAppPack{{pack: pack, payloadHash: fakePayloadHashV1}} // markInstalledPackFailed runs a namespace-reservation SELECT before the INSERT. mock.ExpectQuery("SELECT Category, COALESCE\\(IsDeleted, 0\\) FROM _System_InstalledPack WHERE Name"). WithArgs("fake-pack"). WillReturnRows(sqlmock.NewRows([]string{"Category", "IsDeleted"})) expectInstalledPackRuntimeDeclarationColumns(mock, false) mock.ExpectExec("INSERT INTO _System_InstalledPack"). WillReturnResult(sqlmock.NewResult(0, 1)) err = finalizeAppliedAppPacksWith( context.Background(), db, appPacks, func(context.Context, *sql.DB, []packs.Pack) error { return nil }, func(*sql.DB) error { return nil }, func(context.Context) error { return errors.New("materialize failed") }, ) if err == nil { t.Fatal("finalize applied app packs = nil, want materialization error") } if !strings.Contains(err.Error(), "materialize derived artifacts") { t.Fatalf("finalize error = %v, want materialize derived artifacts", err) } if pack.verifyCalls != 1 { t.Fatalf("verify calls = %d, want only pre-materialization verify", pack.verifyCalls) } if err := mock.ExpectationsWereMet(); err != nil { t.Fatalf("sql expectations: %v", err) } } func TestFinalizeAppliedAppPacksMarksFailedWhenInitialVerifyFails(t *testing.T) { db, mock, err := sqlmock.New() if err != nil { t.Fatalf("sqlmock: %v", err) } defer db.Close() pack := &postVerifyFailingInstalledPack{initialErr: errors.New("pack payload frozen-formula verification disabled by DISABLE_FORMULA_VALIDATION=true")} appPacks := []appliedAppPack{{pack: pack, payloadHash: fakePayloadHashV1}} // markInstalledPackFailed runs a namespace-reservation SELECT before the INSERT. mock.ExpectQuery("SELECT Category, COALESCE\\(IsDeleted, 0\\) FROM _System_InstalledPack WHERE Name"). WithArgs("fake-pack"). WillReturnRows(sqlmock.NewRows([]string{"Category", "IsDeleted"})) expectInstalledPackRuntimeDeclarationColumns(mock, false) mock.ExpectExec("INSERT INTO _System_InstalledPack"). WillReturnResult(sqlmock.NewResult(0, 1)) err = finalizeAppliedAppPacksWith( context.Background(), db, appPacks, func(context.Context, *sql.DB, []packs.Pack) error { return nil }, func(*sql.DB) error { return nil }, nil, ) if err == nil { t.Fatal("finalize applied app packs = nil, want initial verify error") } if !strings.Contains(err.Error(), "pack fake-pack verify") { t.Fatalf("finalize error = %v, want initial verify context", err) } if pack.verifyCalls != 1 { t.Fatalf("verify calls = %d, want 1", pack.verifyCalls) } if err := mock.ExpectationsWereMet(); err != nil { t.Fatalf("sql expectations: %v", err) } } func TestFinalizeAppliedAppPacksPreservesPeerDeclarationOverridesInVerifyContext(t *testing.T) { db, mock, err := sqlmock.New() if err != nil { t.Fatalf("sqlmock: %v", err) } defer db.Close() _, peerPack := writeNamedFieldDeclarationSource(t, "peer-pack", "Issue", "PeerField__c") pack := &contextCheckingInstalledPack{expectedPeerPack: "peer-pack"} appPacks := []appliedAppPack{{pack: pack, payloadHash: fakePayloadHashV1}} expectInstalledPackRuntimeDeclarationColumns(mock, false) mock.ExpectBegin() mock.ExpectExec("UPDATE _System_InstalledPack"). WillReturnResult(sqlmock.NewResult(0, 1)) expectInstalledPackHostRequirementsDigestUpdate(mock, pack.Name()) expectNoActiveConnectorProjectionOwnersForCompanion(mock, pack.Name()) mock.ExpectCommit() ctx := packs.WithPeerDeclarationOverrides(context.Background(), map[string]*packs.SourceBackedPack{ "peer-pack": peerPack, }) err = finalizeAppliedAppPacksWith( ctx, db, appPacks, func(context.Context, *sql.DB, []packs.Pack) error { return nil }, func(*sql.DB) error { return nil }, nil, ) if err != nil { t.Fatalf("finalize applied app packs: %v", err) } if pack.verifyCalls != 2 { t.Fatalf("verify calls = %d, want 2", pack.verifyCalls) } if err := mock.ExpectationsWereMet(); err != nil { t.Fatalf("sql expectations: %v", err) } } func TestListInstalledSourceBackedPacksFallsBackWhenHostRequirementsDigestColumnAbsent(t *testing.T) { db, mock, err := sqlmock.New() if err != nil { t.Fatalf("sqlmock: %v", err) } defer db.Close() cols := []string{ "Name", "Category", "Status", "PackVersion", "ManifestFormatVersion", "PlatformCompatibility", "DesiredPayloadHash", "AppliedPayloadHash", "PayloadHashAlgorithm", "SourceKind", "SourceURI", "SourceRevision", } row := []driver.Value{ "legacy-pack", string(packs.AppPack), installedPackStatusActive, nil, nil, nil, nil, nil, nil, packs.PackSourceKindLocalPath, "/tmp/legacy-pack", nil, } expectInstalledPackRuntimeDeclarationColumns(mock, false) mock.ExpectQuery("SELECT Name, Category, Status[\\s\\S]*HostRequirementsDigest[\\s\\S]*FROM _System_InstalledPack"). WithArgs(string(packs.AppPack), string(packs.ConnectorPack)). WillReturnError(&mysql.MySQLError{Number: 1054, Message: "Unknown column 'HostRequirementsDigest' in 'field list'"}) mock.ExpectQuery("SELECT Name, Category, Status[\\s\\S]*SourceRevision\\s+FROM _System_InstalledPack"). WithArgs(string(packs.AppPack), string(packs.ConnectorPack)). WillReturnRows(sqlmock.NewRows(cols).AddRow(row...)) got, err := listInstalledSourceBackedPacks(context.Background(), db) if err != nil { t.Fatalf("listInstalledSourceBackedPacks: %v", err) } if len(got) != 1 || got[0].Name != "legacy-pack" { t.Fatalf("listInstalledSourceBackedPacks = %#v, want legacy-pack row", got) } if got[0].HostRequirementsDigest.Valid { t.Fatal("HostRequirementsDigest.Valid = true, want false for legacy schema fallback") } if err := mock.ExpectationsWereMet(); err != nil { t.Fatalf("sql expectations: %v", err) } } func TestMarkInstalledPackInstallingPersistsRuntimeDeclarations(t *testing.T) { db, mock, err := sqlmock.New() if err != nil { t.Fatalf("sqlmock: %v", err) } defer db.Close() pack := runtimeDeclInstalledPack{} decl, err := packs.BuildPersistedRuntimeDeclaration(pack) if err != nil { t.Fatalf("BuildPersistedRuntimeDeclaration: %v", err) } mock.ExpectQuery("SELECT Category, COALESCE\\(IsDeleted, 0\\) FROM _System_InstalledPack WHERE Name"). WithArgs(pack.Name()). WillReturnRows(sqlmock.NewRows([]string{"Category", "IsDeleted"})) expectInstalledPackRuntimeDeclarationColumns(mock, true) expectInstalledPackRuntimeStartupShallowDeclarationColumn(mock, true) expectAppPackInstallingUpdate(mock, pack.Name(), 0, 5) expectCompanionAppPackQueuedReapplyGuardLookup(mock, pack.Name(), sqlmock.NewRows([]string{"PackInstallId"})) expectInstalledPackRowExists(mock, pack.Name(), false) mock.ExpectExec("INSERT INTO _System_InstalledPack[\\s\\S]*RuntimeObjectSpecDeclarations[\\s\\S]*RuntimeRelatedListRules[\\s\\S]*RuntimeSearchScopeOrder[\\s\\S]*RuntimeStartupShallowDeclarations"). WithArgs( installedPackID(pack.Name()), pack.Name(), sqlmock.AnyArg(), string(pack.Category()), installedPackStatusInstalling, sqlmock.AnyArg(), sqlmock.AnyArg(), sqlmock.AnyArg(), "hash-v1-install", sqlmock.AnyArg(), sqlmock.AnyArg(), sqlmock.AnyArg(), sqlmock.AnyArg(), decl.ObjectPayloadsJSON(), decl.RelatedListRulesJSON(), decl.SearchScopeOrderJSON(), decl.SearchAllDefaultFieldsJSON(), decl.StartupShallowDeclaredRowsJSON(), sqlmock.AnyArg(), sqlmock.AnyArg(), ). WillReturnResult(sqlmock.NewResult(0, 1)) expectInstalledPackHostRequirementsDigestUpdate(mock, pack.Name()) if err := markInstalledPackInstalling(context.Background(), db, pack, "hash-v1-install"); err != nil { t.Fatalf("markInstalledPackInstalling: %v", err) } if err := mock.ExpectationsWereMet(); err != nil { t.Fatalf("sql expectations: %v", err) } } func TestMarkInstalledPackInstallingUsesThreeColumnSchemaWhenStartupShallowColumnAbsent(t *testing.T) { db, mock, err := sqlmock.New() if err != nil { t.Fatalf("sqlmock: %v", err) } defer db.Close() pack := runtimeDeclInstalledPack{} decl, err := packs.BuildPersistedRuntimeDeclaration(pack) if err != nil { t.Fatalf("BuildPersistedRuntimeDeclaration: %v", err) } mock.ExpectQuery("SELECT Category, COALESCE\\(IsDeleted, 0\\) FROM _System_InstalledPack WHERE Name"). WithArgs(pack.Name()). WillReturnRows(sqlmock.NewRows([]string{"Category", "IsDeleted"})) expectInstalledPackRuntimeDeclarationColumns(mock, true) expectInstalledPackRuntimeStartupShallowDeclarationColumn(mock, false) expectAppPackInstallingUpdate(mock, pack.Name(), 0, 4) expectCompanionAppPackQueuedReapplyGuardLookup(mock, pack.Name(), sqlmock.NewRows([]string{"PackInstallId"})) expectInstalledPackRowExists(mock, pack.Name(), false) mock.ExpectExec("INSERT INTO _System_InstalledPack[\\s\\S]*RuntimeObjectSpecDeclarations[\\s\\S]*RuntimeRelatedListRules[\\s\\S]*RuntimeSearchScopeOrder"). WithArgs( installedPackID(pack.Name()), pack.Name(), sqlmock.AnyArg(), string(pack.Category()), installedPackStatusInstalling, sqlmock.AnyArg(), sqlmock.AnyArg(), sqlmock.AnyArg(), "hash-v1-install", sqlmock.AnyArg(), sqlmock.AnyArg(), sqlmock.AnyArg(), sqlmock.AnyArg(), decl.ObjectPayloadsJSON(), decl.RelatedListRulesJSON(), decl.SearchScopeOrderJSON(), decl.SearchAllDefaultFieldsJSON(), sqlmock.AnyArg(), sqlmock.AnyArg(), ). WillReturnResult(sqlmock.NewResult(0, 1)) expectInstalledPackHostRequirementsDigestUpdate(mock, pack.Name()) if err := markInstalledPackInstalling(context.Background(), db, pack, "hash-v1-install"); err != nil { t.Fatalf("markInstalledPackInstalling: %v", err) } if err := mock.ExpectationsWereMet(); err != nil { t.Fatalf("sql expectations: %v", err) } } func TestMarkInstalledPackInstallingUsesLegacySchemaWhenRuntimeColumnsAbsent(t *testing.T) { db, mock, err := sqlmock.New() if err != nil { t.Fatalf("sqlmock: %v", err) } defer db.Close() pack := runtimeDeclInstalledPack{} mock.ExpectQuery("SELECT Category, COALESCE\\(IsDeleted, 0\\) FROM _System_InstalledPack WHERE Name"). WithArgs(pack.Name()). WillReturnRows(sqlmock.NewRows([]string{"Category", "IsDeleted"})) mock.ExpectQuery("SELECT COUNT\\(\\*\\)\\s+FROM INFORMATION_SCHEMA.COLUMNS"). WithArgs("_System_InstalledPack", "RuntimeObjectSpecDeclarations", "RuntimeRelatedListRules", "RuntimeSearchScopeOrder", "RuntimeSearchAllDefaultFields"). WillReturnRows(sqlmock.NewRows([]string{"COUNT(*)"}).AddRow(0)) expectAppPackInstallingUpdate(mock, pack.Name(), 0, 0) expectCompanionAppPackQueuedReapplyGuardLookup(mock, pack.Name(), sqlmock.NewRows([]string{"PackInstallId"})) expectInstalledPackRowExists(mock, pack.Name(), false) mock.ExpectExec("INSERT INTO _System_InstalledPack"). WithArgs( installedPackID(pack.Name()), pack.Name(), sqlmock.AnyArg(), string(pack.Category()), installedPackStatusInstalling, sqlmock.AnyArg(), sqlmock.AnyArg(), sqlmock.AnyArg(), "hash-v1-install", sqlmock.AnyArg(), sqlmock.AnyArg(), sqlmock.AnyArg(), sqlmock.AnyArg(), sqlmock.AnyArg(), sqlmock.AnyArg(), ). WillReturnResult(sqlmock.NewResult(0, 1)) expectInstalledPackHostRequirementsDigestUpdate(mock, pack.Name()) if err := markInstalledPackInstalling(context.Background(), db, pack, "hash-v1-install"); err != nil { t.Fatalf("markInstalledPackInstalling: %v", err) } if err := mock.ExpectationsWereMet(); err != nil { t.Fatalf("sql expectations: %v", err) } } func TestMarkInstalledPackInstallingFailsClosedBeforeMutationWhenHostRequirementsDigestColumnAbsent(t *testing.T) { db, mock, err := sqlmock.New() if err != nil { t.Fatalf("sqlmock: %v", err) } defer db.Close() pack := minimalPack{ name: "needs-host", category: packs.AppPack, requirements: []packs.HostPrimitiveRequirement{{ Kind: packs.HostPrimitiveKindProductDialect, Name: "test-product-dialect", }}, } mock.ExpectQuery("SELECT Category, COALESCE\\(IsDeleted, 0\\) FROM _System_InstalledPack WHERE Name"). WithArgs(pack.Name()). WillReturnRows(sqlmock.NewRows([]string{"Category", "IsDeleted"})) mock.ExpectQuery("SELECT COUNT\\(\\*\\)\\s+FROM INFORMATION_SCHEMA\\.COLUMNS"). WithArgs("_System_InstalledPack", "HostRequirementsDigest"). WillReturnRows(sqlmock.NewRows([]string{"count"}).AddRow(0)) err = markInstalledPackInstalling(context.Background(), db, pack, "hash-v1-install") if err == nil || !strings.Contains(err.Error(), "HostRequirementsDigest") { t.Fatalf("markInstalledPackInstalling error = %v, want missing HostRequirementsDigest error", err) } if err := mock.ExpectationsWereMet(); err != nil { t.Fatalf("sql expectations: %v", err) } } func TestMarkInstalledPackInstallingFailsClosedWithoutRuntimeDeclarationColumnsForFormulaBundles(t *testing.T) { db, mock, err := sqlmock.New() if err != nil { t.Fatalf("sqlmock: %v", err) } defer db.Close() pack := runtimeDeclFormulaInstalledPack{} mock.ExpectQuery("SELECT Category, COALESCE\\(IsDeleted, 0\\) FROM _System_InstalledPack WHERE Name"). WithArgs(pack.Name()). WillReturnRows(sqlmock.NewRows([]string{"Category", "IsDeleted"})) expectInstalledPackRuntimeDeclarationColumns(mock, false) err = markInstalledPackInstalling(context.Background(), db, pack, "hash-v1-install") if err == nil || !strings.Contains(err.Error(), "persisted runtime declaration columns") { t.Fatalf("markInstalledPackInstalling error = %v, want missing persisted runtime declaration columns error", err) } if err := mock.ExpectationsWereMet(); err != nil { t.Fatalf("sql expectations: %v", err) } } func TestMarkInstalledPackInstallingFailsClosedWithoutRuntimeFormulaFunctionBundlesColumn(t *testing.T) { db, mock, err := sqlmock.New() if err != nil { t.Fatalf("sqlmock: %v", err) } defer db.Close() pack := runtimeDeclFormulaInstalledPack{} mock.ExpectQuery("SELECT Category, COALESCE\\(IsDeleted, 0\\) FROM _System_InstalledPack WHERE Name"). WithArgs(pack.Name()). WillReturnRows(sqlmock.NewRows([]string{"Category", "IsDeleted"})) expectInstalledPackRuntimeDeclarationColumns(mock, true) expectInstalledPackRuntimeStartupShallowAndFormulaBundleDeclarationColumns(mock, true, false) err = markInstalledPackInstalling(context.Background(), db, pack, "hash-v1-install") if err == nil || !strings.Contains(err.Error(), "RuntimeFormulaFunctionBundles") { t.Fatalf("markInstalledPackInstalling error = %v, want RuntimeFormulaFunctionBundles schema error", err) } if err := mock.ExpectationsWereMet(); err != nil { t.Fatalf("sql expectations: %v", err) } } func TestMarkInstalledPackActivePersistsRuntimeDeclarations(t *testing.T) { db, mock, err := sqlmock.New() if err != nil { t.Fatalf("sqlmock: %v", err) } defer db.Close() pack := runtimeDeclInstalledPack{} decl, err := packs.BuildPersistedRuntimeDeclaration(pack) if err != nil { t.Fatalf("BuildPersistedRuntimeDeclaration: %v", err) } expectInstalledPackRuntimeDeclarationColumns(mock, true) expectInstalledPackRuntimeStartupShallowDeclarationColumn(mock, true) mock.ExpectBegin() mock.ExpectExec("UPDATE _System_InstalledPack[\\s\\S]*RuntimeObjectSpecDeclarations = \\?[\\s\\S]*RuntimeRelatedListRules = \\?[\\s\\S]*RuntimeSearchScopeOrder = \\?[\\s\\S]*RuntimeStartupShallowDeclarations = \\?"). WithArgs( installedPackStatusActive, sqlmock.AnyArg(), sqlmock.AnyArg(), sqlmock.AnyArg(), "hash-v1-active", "hash-v1-active", sqlmock.AnyArg(), sqlmock.AnyArg(), sqlmock.AnyArg(), sqlmock.AnyArg(), decl.ObjectPayloadsJSON(), decl.RelatedListRulesJSON(), decl.SearchScopeOrderJSON(), decl.SearchAllDefaultFieldsJSON(), decl.StartupShallowDeclaredRowsJSON(), sqlmock.AnyArg(), sqlmock.AnyArg(), pack.Name(), ). WillReturnResult(sqlmock.NewResult(0, 1)) expectInstalledPackHostRequirementsDigestUpdate(mock, pack.Name()) expectNoActiveConnectorProjectionOwnersForCompanion(mock, pack.Name()) mock.ExpectCommit() if err := markInstalledPackActive(context.Background(), db, pack, "hash-v1-active"); err != nil { t.Fatalf("markInstalledPackActive: %v", err) } if err := mock.ExpectationsWereMet(); err != nil { t.Fatalf("sql expectations: %v", err) } } func TestMarkInstalledPackActiveReconcilesCommittedCommitError(t *testing.T) { db, mock, err := sqlmock.New() if err != nil { t.Fatalf("sqlmock: %v", err) } defer db.Close() pack := runtimeDeclInstalledPack{} decl, err := packs.BuildPersistedRuntimeDeclaration(pack) if err != nil { t.Fatalf("BuildPersistedRuntimeDeclaration: %v", err) } meta := packs.PackHashMetadataForPack(pack) expectInstalledPackRuntimeDeclarationColumns(mock, true) expectInstalledPackRuntimeStartupShallowDeclarationColumn(mock, true) mock.ExpectBegin() mock.ExpectExec("UPDATE _System_InstalledPack[\\s\\S]*RuntimeObjectSpecDeclarations = \\?[\\s\\S]*RuntimeRelatedListRules = \\?[\\s\\S]*RuntimeSearchScopeOrder = \\?[\\s\\S]*RuntimeStartupShallowDeclarations = \\?"). WithArgs( installedPackStatusActive, sqlmock.AnyArg(), sqlmock.AnyArg(), sqlmock.AnyArg(), fakePayloadHashV1, fakePayloadHashV1, sqlmock.AnyArg(), sqlmock.AnyArg(), sqlmock.AnyArg(), sqlmock.AnyArg(), decl.ObjectPayloadsJSON(), decl.RelatedListRulesJSON(), decl.SearchScopeOrderJSON(), decl.SearchAllDefaultFieldsJSON(), decl.StartupShallowDeclaredRowsJSON(), sqlmock.AnyArg(), sqlmock.AnyArg(), pack.Name(), ). WillReturnResult(sqlmock.NewResult(0, 1)) expectInstalledPackHostRequirementsDigestUpdate(mock, pack.Name()) expectNoActiveConnectorProjectionOwnersForCompanion(mock, pack.Name()) commitErr := errors.New("driver lost commit acknowledgement") mock.ExpectCommit().WillReturnError(commitErr) mock.ExpectQuery("(?s)SELECT Id, Name, COALESCE\\(PackInstallId, ''\\).*FROM _System_InstalledPack"). WithArgs(pack.Name()). WillReturnRows(sqlmock.NewRows([]string{ "Id", "Name", "PackInstallId", "Category", "Status", "IsDeleted", "PackVersion", "ManifestFormatVersion", "PlatformCompatibility", "DesiredPayloadHash", "AppliedPayloadHash", "PayloadHashAlgorithm", "SourceKind", "SourceURI", "SourceRevision", "HostRequirementsDigest", }).AddRow( installedPackID(pack.Name()), pack.Name(), "pack-install-1", string(pack.Category()), installedPackStatusActive, 0, meta.PackVersion, meta.ManifestFormatVersion, meta.PlatformCompatibility, fakePayloadHashV1, fakePayloadHashV1, meta.PayloadHashAlgorithm, nil, nil, nil, nil, )) if err := markInstalledPackActive(context.Background(), db, pack, fakePayloadHashV1); err != nil { t.Fatalf("markInstalledPackActive committed ambiguity: %v", err) } if err := mock.ExpectationsWereMet(); err != nil { t.Fatalf("sql expectations: %v", err) } } func TestReconcileInstalledPackActivationCommitClassifiesMismatchAndReadFailure(t *testing.T) { db, _, err := sqlmock.New() if err != nil { t.Fatalf("sqlmock: %v", err) } defer db.Close() pack := fakeInstalledPack{} exact := installedPackHashRow( fakePayloadHashV1, fakePayloadHashV1, installedPackMetadataFromPack(pack), ) exact.PackInstallID = "pack-install-1" err = reconcileInstalledPackActivationCommitWith( context.Background(), db, pack, fakePayloadHashV1, installedPackActivationCommitReconciliationOperations{ readInstalledRow: func(context.Context, *sql.DB, string) (installedPackRow, bool, error) { mismatch := exact mismatch.Status = installedPackStatusInstalling return mismatch, true, nil }, }, ) if !errors.Is(err, errInstalledPackActivationNotCommitted) { t.Fatalf("mismatched durable tuple error = %v, want not committed", err) } readErr := errors.New("activation authority unavailable") err = reconcileInstalledPackActivationCommitWith( context.Background(), db, pack, fakePayloadHashV1, installedPackActivationCommitReconciliationOperations{ readInstalledRow: func(context.Context, *sql.DB, string) (installedPackRow, bool, error) { return installedPackRow{}, false, readErr }, }, ) if !errors.Is(err, readErr) { t.Fatalf("durable tuple read error = %v, want %v", err, readErr) } if errors.Is(err, errInstalledPackActivationNotCommitted) { t.Fatalf("durable tuple read error = %v, must remain indeterminate", err) } } func TestReconcileInstalledPackActivationCommitRequiresSourceGenerationAuthority(t *testing.T) { db, _, err := sqlmock.New() if err != nil { t.Fatalf("sqlmock: %v", err) } defer db.Close() pack := testControlplaneOfficialRepoSourceBackedPack(t, "source-revision-1") payloadHash, err := packs.PackPayloadHash(pack) if err != nil { t.Fatalf("payload hash: %v", err) } meta := packs.PackHashMetadataForPack(pack) hostDigest, err := packs.PackHostRequirementsDigest(pack) if err != nil { t.Fatalf("host requirements digest: %v", err) } exact := installedPackHashRow( payloadHash, payloadHash, installedPackMetadataFromPack(pack), ) exact.Name = pack.Name() exact.PackInstallID = "pack-install-1" exact.SourceKind = sql.NullString{String: meta.SourceKind, Valid: true} exact.SourceURI = sql.NullString{String: meta.SourceURI, Valid: true} exact.SourceRevision = sql.NullString{String: meta.SourceRevision, Valid: true} exact.HostRequirementsDigest = sql.NullString{String: hostDigest, Valid: true} authorityErr := errors.New("immutable generation is missing") verified := 0 err = reconcileInstalledPackActivationCommitWith( context.Background(), db, pack, payloadHash, installedPackActivationCommitReconciliationOperations{ readInstalledRow: func(context.Context, *sql.DB, string) (installedPackRow, bool, error) { return exact, true, nil }, verifySourceBackedGeneration: func( context.Context, *sql.DB, appliedAppPack, ) error { verified++ return authorityErr }, }, ) if !errors.Is(err, authorityErr) { t.Fatalf("missing generation reconciliation error = %v, want %v", err, authorityErr) } if errors.Is(err, errInstalledPackActivationNotCommitted) { t.Fatalf("active row without generation was misclassified as a safe rollback: %v", err) } if verified != 1 { t.Fatalf("generation verification calls = %d, want 1", verified) } } func TestReconcileInstalledPackActivationCommitDoesNotInventLocalPathGeneration(t *testing.T) { db, _, err := sqlmock.New() if err != nil { t.Fatalf("sqlmock: %v", err) } defer db.Close() sourceRoot := writeConnectorPackSourceRoot(t) pack, err := packs.ResolveInstalledSourceBackedPack( "test-connector", packs.ConnectorPack, packs.PackSourceKindLocalPath, sourceRoot, ) if err != nil { t.Fatalf("resolve local_path pack: %v", err) } payloadHash, err := packs.PackPayloadHash(pack) if err != nil { t.Fatalf("payload hash: %v", err) } meta := packs.PackHashMetadataForPack(pack) hostDigest, err := packs.PackHostRequirementsDigest(pack) if err != nil { t.Fatalf("host requirements digest: %v", err) } exact := installedPackHashRow( payloadHash, payloadHash, installedPackMetadataFromPack(pack), ) exact.Name = pack.Name() exact.Category = string(pack.Category()) exact.PackInstallID = "local-install-1" exact.SourceKind = sql.NullString{String: meta.SourceKind, Valid: true} exact.SourceURI = sql.NullString{String: meta.SourceURI, Valid: true} exact.SourceRevision = sql.NullString{ String: meta.SourceRevision, Valid: meta.SourceRevisionValid, } exact.HostRequirementsDigest = sql.NullString{String: hostDigest, Valid: true} generationVerifierCalls := 0 absenceVerifierCalls := 0 err = reconcileInstalledPackActivationCommitWith( context.Background(), db, pack, payloadHash, installedPackActivationCommitReconciliationOperations{ readInstalledRow: func(context.Context, *sql.DB, string) (installedPackRow, bool, error) { return exact, true, nil }, verifySourceBackedGeneration: func( context.Context, *sql.DB, appliedAppPack, ) error { generationVerifierCalls++ return errors.New("local_path must not require generation authority") }, verifyNoMutableGeneration: func( context.Context, *sql.DB, string, ) error { absenceVerifierCalls++ return nil }, }, ) if err != nil { t.Fatalf("local_path commit reconciliation: %v", err) } if generationVerifierCalls != 0 { t.Fatalf("local_path generation verifier calls = %d, want 0", generationVerifierCalls) } if absenceVerifierCalls != 1 { t.Fatalf("local_path generation absence verifier calls = %d, want 1", absenceVerifierCalls) } } func TestMarkInstalledPackActiveFailsClosedBeforeMutationWhenHostRequirementsDigestColumnAbsent(t *testing.T) { db, mock, err := sqlmock.New() if err != nil { t.Fatalf("sqlmock: %v", err) } defer db.Close() pack := minimalPack{ name: "needs-host", category: packs.AppPack, requirements: []packs.HostPrimitiveRequirement{{ Kind: packs.HostPrimitiveKindProductDialect, Name: "test-product-dialect", }}, } expectInstalledPackRuntimeDeclarationColumns(mock, false) mock.ExpectQuery("SELECT COUNT\\(\\*\\)\\s+FROM INFORMATION_SCHEMA\\.COLUMNS"). WithArgs("_System_InstalledPack", "HostRequirementsDigest"). WillReturnRows(sqlmock.NewRows([]string{"count"}).AddRow(0)) err = markInstalledPackActive(context.Background(), db, pack, "hash-v1-active") if err == nil || !strings.Contains(err.Error(), "HostRequirementsDigest") { t.Fatalf("markInstalledPackActive error = %v, want missing HostRequirementsDigest error", err) } if err := mock.ExpectationsWereMet(); err != nil { t.Fatalf("sql expectations: %v", err) } } func TestMarkInstalledPackActiveFailsClosedWithoutRuntimeFormulaFunctionBundlesColumn(t *testing.T) { db, mock, err := sqlmock.New() if err != nil { t.Fatalf("sqlmock: %v", err) } defer db.Close() pack := runtimeDeclFormulaInstalledPack{} expectInstalledPackRuntimeDeclarationColumns(mock, true) expectInstalledPackRuntimeStartupShallowAndFormulaBundleDeclarationColumns(mock, true, false) err = markInstalledPackActive(context.Background(), db, pack, "hash-v1-active") if err == nil || !strings.Contains(err.Error(), "RuntimeFormulaFunctionBundles") { t.Fatalf("markInstalledPackActive error = %v, want RuntimeFormulaFunctionBundles schema error", err) } if err := mock.ExpectationsWereMet(); err != nil { t.Fatalf("sql expectations: %v", err) } } func TestMarkInstalledPackActiveFailsClosedWithoutRuntimeDeclarationColumnsForFormulaBundles(t *testing.T) { db, mock, err := sqlmock.New() if err != nil { t.Fatalf("sqlmock: %v", err) } defer db.Close() pack := runtimeDeclFormulaInstalledPack{} expectInstalledPackRuntimeDeclarationColumns(mock, false) err = markInstalledPackActive(context.Background(), db, pack, "hash-v1-active") if err == nil || !strings.Contains(err.Error(), "persisted runtime declaration columns") { t.Fatalf("markInstalledPackActive error = %v, want missing persisted runtime declaration columns error", err) } if err := mock.ExpectationsWereMet(); err != nil { t.Fatalf("sql expectations: %v", err) } } func TestMarkInstalledPackActivePersistsRuntimeFormulaFunctionBundlesWhenColumnPresent(t *testing.T) { db, mock, err := sqlmock.New() if err != nil { t.Fatalf("sqlmock: %v", err) } defer db.Close() pack := runtimeDeclFormulaInstalledPack{} decl, err := packs.BuildPersistedRuntimeDeclaration(pack) if err != nil { t.Fatalf("BuildPersistedRuntimeDeclaration: %v", err) } if !strings.Contains(decl.FormulaFunctionBundlesJSON(), "PACK_PREFIX") { t.Fatalf("FormulaFunctionBundlesJSON = %q, want PACK_PREFIX payload", decl.FormulaFunctionBundlesJSON()) } expectInstalledPackRuntimeDeclarationColumns(mock, true) expectInstalledPackRuntimeStartupShallowAndFormulaBundleDeclarationColumns(mock, true, true) mock.ExpectBegin() mock.ExpectExec("UPDATE _System_InstalledPack[\\s\\S]*RuntimeObjectSpecDeclarations = \\?[\\s\\S]*RuntimeRelatedListRules = \\?[\\s\\S]*RuntimeSearchScopeOrder = \\?[\\s\\S]*RuntimeFormulaFunctionBundles = \\?[\\s\\S]*RuntimeStartupShallowDeclarations = \\?"). WithArgs( installedPackStatusActive, sqlmock.AnyArg(), sqlmock.AnyArg(), sqlmock.AnyArg(), "hash-v1-active", "hash-v1-active", sqlmock.AnyArg(), sqlmock.AnyArg(), sqlmock.AnyArg(), sqlmock.AnyArg(), decl.ObjectPayloadsJSON(), decl.RelatedListRulesJSON(), decl.SearchScopeOrderJSON(), decl.SearchAllDefaultFieldsJSON(), decl.FormulaFunctionBundlesJSON(), decl.StartupShallowDeclaredRowsJSON(), sqlmock.AnyArg(), sqlmock.AnyArg(), pack.Name(), ). WillReturnResult(sqlmock.NewResult(0, 1)) expectInstalledPackHostRequirementsDigestUpdate(mock, pack.Name()) expectNoActiveConnectorProjectionOwnersForCompanion(mock, pack.Name()) mock.ExpectCommit() if err := markInstalledPackActive(context.Background(), db, pack, "hash-v1-active"); err != nil { t.Fatalf("markInstalledPackActive: %v", err) } if err := mock.ExpectationsWereMet(); err != nil { t.Fatalf("sql expectations: %v", err) } } func TestMarkInstalledPackActiveUsesThreeColumnSchemaWhenStartupShallowColumnAbsent(t *testing.T) { db, mock, err := sqlmock.New() if err != nil { t.Fatalf("sqlmock: %v", err) } defer db.Close() pack := runtimeDeclInstalledPack{} decl, err := packs.BuildPersistedRuntimeDeclaration(pack) if err != nil { t.Fatalf("BuildPersistedRuntimeDeclaration: %v", err) } expectInstalledPackRuntimeDeclarationColumns(mock, true) expectInstalledPackRuntimeStartupShallowDeclarationColumn(mock, false) mock.ExpectBegin() mock.ExpectExec("UPDATE _System_InstalledPack[\\s\\S]*RuntimeObjectSpecDeclarations = \\?[\\s\\S]*RuntimeRelatedListRules = \\?[\\s\\S]*RuntimeSearchScopeOrder = \\?"). WithArgs( installedPackStatusActive, sqlmock.AnyArg(), sqlmock.AnyArg(), sqlmock.AnyArg(), "hash-v1-active", "hash-v1-active", sqlmock.AnyArg(), sqlmock.AnyArg(), sqlmock.AnyArg(), sqlmock.AnyArg(), decl.ObjectPayloadsJSON(), decl.RelatedListRulesJSON(), decl.SearchScopeOrderJSON(), decl.SearchAllDefaultFieldsJSON(), sqlmock.AnyArg(), sqlmock.AnyArg(), pack.Name(), ). WillReturnResult(sqlmock.NewResult(0, 1)) expectInstalledPackHostRequirementsDigestUpdate(mock, pack.Name()) expectNoActiveConnectorProjectionOwnersForCompanion(mock, pack.Name()) mock.ExpectCommit() if err := markInstalledPackActive(context.Background(), db, pack, "hash-v1-active"); err != nil { t.Fatalf("markInstalledPackActive: %v", err) } if err := mock.ExpectationsWereMet(); err != nil { t.Fatalf("sql expectations: %v", err) } } func TestMarkInstalledPackActiveUsesLegacySchemaWhenRuntimeColumnsAbsent(t *testing.T) { db, mock, err := sqlmock.New() if err != nil { t.Fatalf("sqlmock: %v", err) } defer db.Close() pack := runtimeDeclInstalledPack{} mock.ExpectQuery("SELECT COUNT\\(\\*\\)\\s+FROM INFORMATION_SCHEMA.COLUMNS"). WithArgs("_System_InstalledPack", "RuntimeObjectSpecDeclarations", "RuntimeRelatedListRules", "RuntimeSearchScopeOrder", "RuntimeSearchAllDefaultFields"). WillReturnRows(sqlmock.NewRows([]string{"COUNT(*)"}).AddRow(0)) mock.ExpectBegin() mock.ExpectExec("UPDATE _System_InstalledPack"). WithArgs( installedPackStatusActive, sqlmock.AnyArg(), sqlmock.AnyArg(), sqlmock.AnyArg(), "hash-v1-active", "hash-v1-active", sqlmock.AnyArg(), sqlmock.AnyArg(), sqlmock.AnyArg(), sqlmock.AnyArg(), sqlmock.AnyArg(), sqlmock.AnyArg(), pack.Name(), ). WillReturnResult(sqlmock.NewResult(0, 1)) expectInstalledPackHostRequirementsDigestUpdate(mock, pack.Name()) expectNoActiveConnectorProjectionOwnersForCompanion(mock, pack.Name()) mock.ExpectCommit() if err := markInstalledPackActive(context.Background(), db, pack, "hash-v1-active"); err != nil { t.Fatalf("markInstalledPackActive: %v", err) } if err := mock.ExpectationsWereMet(); err != nil { t.Fatalf("sql expectations: %v", err) } } // TestMarkInstalledPackActive_DependencyReconcileFails_RollsBack verifies that // when the dependency-row reconcile fails after the Status=active UPDATE, the // entire transaction is rolled back so the pack is NOT left live with stale or // missing consent rows. func TestMarkInstalledPackActive_DependencyReconcileFails_RollsBack(t *testing.T) { db, mock, err := sqlmock.New() if err != nil { t.Fatalf("sqlmock: %v", err) } defer db.Close() // Use a pack that implements PackDependencyProvider so upsertPackDependencyRows // actually issues SQL (and we can make it fail). pack := &stubPackWithDeps{ name: "companion-pack", deps: []packs.PackDependency{ {Name: "connector-pack", VersionConstraint: ">=1.0.0"}, }, } // Schema check: no runtime declaration columns → use legacy UPDATE path. mock.ExpectQuery("SELECT COUNT\\(\\*\\)\\s+FROM INFORMATION_SCHEMA.COLUMNS"). WithArgs("_System_InstalledPack", "RuntimeObjectSpecDeclarations", "RuntimeRelatedListRules", "RuntimeSearchScopeOrder", "RuntimeSearchAllDefaultFields"). WillReturnRows(sqlmock.NewRows([]string{"COUNT(*)"}).AddRow(0)) mock.ExpectBegin() // The legacy UPDATE _System_InstalledPack SET Status='active' ... succeeds. mock.ExpectExec("UPDATE _System_InstalledPack"). WillReturnResult(sqlmock.NewResult(0, 1)) expectInstalledPackHostRequirementsDigestUpdate(mock, pack.Name()) // upsertPackDependencyRows first reads PackInstallId — simulate failure. mock.ExpectQuery(`SELECT COALESCE\(PackInstallId, ''\)`). WithArgs("companion-pack"). WillReturnError(errors.New("db: connection reset")) // Expect rollback (not commit) because the dep reconcile failed. mock.ExpectRollback() err = markInstalledPackActive(context.Background(), db, pack, "hash-v1-rollback") if err == nil { t.Fatal("markInstalledPackActive = nil, want error from dependency reconcile failure") } if err := mock.ExpectationsWereMet(); err != nil { t.Fatalf("sql expectations (rollback not issued?): %v", err) } } func TestMarkInstalledPackFailedPersistsRuntimeDeclarations(t *testing.T) { db, mock, err := sqlmock.New() if err != nil { t.Fatalf("sqlmock: %v", err) } defer db.Close() pack := runtimeDeclInstalledPack{} decl, err := packs.BuildPersistedRuntimeDeclaration(pack) if err != nil { t.Fatalf("BuildPersistedRuntimeDeclaration: %v", err) } mock.ExpectQuery("SELECT Category, COALESCE\\(IsDeleted, 0\\) FROM _System_InstalledPack WHERE Name"). WithArgs(pack.Name()). WillReturnRows(sqlmock.NewRows([]string{"Category", "IsDeleted"})) mock.ExpectQuery("SELECT COALESCE\\(PackInstallId, ''\\) FROM _System_InstalledPack WHERE Id = \\? LIMIT 1"). WithArgs(installedPackID(pack.Name())). WillReturnRows(sqlmock.NewRows([]string{"PackInstallId"})) expectInstalledPackRuntimeDeclarationColumns(mock, true) expectInstalledPackRuntimeStartupShallowDeclarationColumn(mock, true) mock.ExpectExec("INSERT INTO _System_InstalledPack[\\s\\S]*RuntimeObjectSpecDeclarations[\\s\\S]*RuntimeRelatedListRules[\\s\\S]*RuntimeSearchScopeOrder[\\s\\S]*RuntimeStartupShallowDeclarations"). WithArgs( installedPackID(pack.Name()), pack.Name(), sqlmock.AnyArg(), string(pack.Category()), installedPackStatusFailed, sqlmock.AnyArg(), sqlmock.AnyArg(), sqlmock.AnyArg(), "hash-v1-failed", sqlmock.AnyArg(), sqlmock.AnyArg(), sqlmock.AnyArg(), sqlmock.AnyArg(), decl.ObjectPayloadsJSON(), decl.RelatedListRulesJSON(), decl.SearchScopeOrderJSON(), decl.SearchAllDefaultFieldsJSON(), decl.StartupShallowDeclaredRowsJSON(), "apply failed", sqlmock.AnyArg(), sqlmock.AnyArg(), ). WillReturnResult(sqlmock.NewResult(0, 1)) if err := markInstalledPackFailed(context.Background(), db, pack, "hash-v1-failed", errors.New("apply failed")); err != nil { t.Fatalf("markInstalledPackFailed: %v", err) } if err := mock.ExpectationsWereMet(); err != nil { t.Fatalf("sql expectations: %v", err) } } func TestRefreshPersistedRuntimeDeclarationsForActivePacksPersistsStartupShallowDeclarationsWhenColumnPresent(t *testing.T) { db, mock, err := sqlmock.New() if err != nil { t.Fatalf("sqlmock: %v", err) } defer db.Close() sourceRoot := writeConnectorPackSourceRoot(t) pack, err := packs.ResolveInstalledSourceBackedPack("test-connector", packs.ConnectorPack, packs.PackSourceKindLocalPath, sourceRoot) if err != nil { t.Fatalf("ResolveInstalledSourceBackedPack: %v", err) } payloadHash, err := packs.PackPayloadHash(pack) if err != nil { t.Fatalf("PackPayloadHash: %v", err) } decl, err := packs.BuildPersistedRuntimeDeclaration(pack) if err != nil { t.Fatalf("BuildPersistedRuntimeDeclaration: %v", err) } meta := pack.PackHashMetadata() expectInstalledPackRuntimeDeclarationColumns(mock, true) expectInstalledPackRuntimeStartupShallowDeclarationColumn(mock, true) mock.ExpectQuery("SELECT COUNT\\(\\*\\)\\s+FROM INFORMATION_SCHEMA.COLUMNS"). WithArgs("_System_InstalledPack", "RuntimeObjectSpecDeclarations", "RuntimeRelatedListRules", "RuntimeSearchScopeOrder", "RuntimeSearchAllDefaultFields"). WillReturnRows(sqlmock.NewRows([]string{"COUNT(*)"}).AddRow(4)) expectInstalledPackRuntimeStartupShallowDeclarationColumn(mock, true) mock.ExpectQuery("SELECT Name, Category, Status, PackVersion, ManifestFormatVersion, PlatformCompatibility, DesiredPayloadHash, AppliedPayloadHash, PayloadHashAlgorithm, SourceKind, SourceURI, SourceRevision, HostRequirementsDigest, RuntimeObjectSpecDeclarations, RuntimeRelatedListRules, RuntimeSearchScopeOrder, RuntimeSearchAllDefaultFields, RuntimeStartupShallowDeclarations"). WithArgs(string(packs.AppPack), string(packs.ConnectorPack)). WillReturnRows(sqlmock.NewRows([]string{ "Name", "Category", "Status", "PackVersion", "ManifestFormatVersion", "PlatformCompatibility", "DesiredPayloadHash", "AppliedPayloadHash", "PayloadHashAlgorithm", "SourceKind", "SourceURI", "SourceRevision", "HostRequirementsDigest", "RuntimeObjectSpecDeclarations", "RuntimeRelatedListRules", "RuntimeSearchScopeOrder", "RuntimeSearchAllDefaultFields", "RuntimeStartupShallowDeclarations", }).AddRow( "test-connector", string(packs.ConnectorPack), installedPackStatusActive, meta.PackVersion, meta.ManifestFormatVersion, meta.PlatformCompatibility, payloadHash, payloadHash, packs.PackPayloadHashAlgorithmV1, packs.PackSourceKindLocalPath, sourceRoot, nil, nil, nil, nil, nil, nil, nil, )) mock.ExpectExec(updateRuntimeDeclarationFourColumnCASPattern). WithArgs( decl.ObjectPayloadsJSON(), decl.RelatedListRulesJSON(), decl.SearchScopeOrderJSON(), decl.SearchAllDefaultFieldsJSON(), decl.StartupShallowDeclaredRowsJSON(), "test-connector", string(packs.ConnectorPack), installedPackStatusActive, packs.PackSourceKindLocalPath, sourceRoot, nil, payloadHash, payloadHash, nil, nil, nil, nil, nil, ). WillReturnResult(sqlmock.NewResult(0, 1)) if err := refreshPersistedRuntimeDeclarationsForActivePacks(context.Background(), db); err != nil { t.Fatalf("refreshPersistedRuntimeDeclarationsForActivePacks: %v", err) } if err := mock.ExpectationsWereMet(); err != nil { t.Fatalf("sql expectations: %v", err) } } func TestVerifyInstalledPersistedRuntimeDeclarationStateRejectsDrift(t *testing.T) { pack := runtimeDeclInstalledPack{} decl, err := packs.BuildPersistedRuntimeDeclaration(pack) if err != nil { t.Fatalf("BuildPersistedRuntimeDeclaration: %v", err) } row := installedPackRow{ Name: pack.Name(), RuntimeObjectSpecDecl: sql.NullString{String: decl.ObjectPayloadsJSON(), Valid: true}, RuntimeRelatedLists: sql.NullString{String: decl.RelatedListRulesJSON(), Valid: true}, RuntimeSearchScope: sql.NullString{String: decl.SearchScopeOrderJSON(), Valid: true}, RuntimeSearchAllDefault: sql.NullString{String: decl.SearchAllDefaultFieldsJSON(), Valid: true}, RuntimeStartupShallow: sql.NullString{String: `[{"kind":"layout","businessKey":"Issue::Alt"}]`, Valid: true}, HasRuntimeStartupDecl: true, } err = verifyInstalledPersistedRuntimeDeclarationState(row, pack, true) if err == nil || !strings.Contains(err.Error(), "persisted runtime declaration drifted from source identity") { t.Fatalf("verifyInstalledPersistedRuntimeDeclarationState error = %v, want drift error", err) } } func TestVerifyInstalledPersistedRuntimeDeclarationStateRejectsIncompleteStartupShallow(t *testing.T) { pack := runtimeDeclInstalledPack{} decl, err := packs.BuildPersistedRuntimeDeclaration(pack) if err != nil { t.Fatalf("BuildPersistedRuntimeDeclaration: %v", err) } row := installedPackRow{ Name: pack.Name(), RuntimeObjectSpecDecl: sql.NullString{String: decl.ObjectPayloadsJSON(), Valid: true}, RuntimeRelatedLists: sql.NullString{String: decl.RelatedListRulesJSON(), Valid: true}, RuntimeSearchScope: sql.NullString{String: decl.SearchScopeOrderJSON(), Valid: true}, HasRuntimeStartupDecl: true, } err = verifyInstalledPersistedRuntimeDeclarationState(row, pack, true) if err == nil || !strings.Contains(err.Error(), "persisted runtime declaration is incomplete") { t.Fatalf("verifyInstalledPersistedRuntimeDeclarationState error = %v, want incomplete error", err) } } func TestVerifyInstalledPersistedRuntimeDeclarationStateRejectsCorruptStartupShallowJSON(t *testing.T) { pack := runtimeDeclInstalledPack{} decl, err := packs.BuildPersistedRuntimeDeclaration(pack) if err != nil { t.Fatalf("BuildPersistedRuntimeDeclaration: %v", err) } row := installedPackRow{ Name: pack.Name(), RuntimeObjectSpecDecl: sql.NullString{String: decl.ObjectPayloadsJSON(), Valid: true}, RuntimeRelatedLists: sql.NullString{String: decl.RelatedListRulesJSON(), Valid: true}, RuntimeSearchScope: sql.NullString{String: decl.SearchScopeOrderJSON(), Valid: true}, RuntimeSearchAllDefault: sql.NullString{String: decl.SearchAllDefaultFieldsJSON(), Valid: true}, RuntimeStartupShallow: sql.NullString{String: "{", Valid: true}, HasRuntimeStartupDecl: true, } err = verifyInstalledPersistedRuntimeDeclarationState(row, pack, true) if err == nil || !strings.Contains(err.Error(), "persisted runtime declaration drifted from source identity") { t.Fatalf("verifyInstalledPersistedRuntimeDeclarationState error = %v, want drift error", err) } } func TestVerifyInstalledPersistedRuntimeDeclarationStateRejectsFormulaBundleDrift(t *testing.T) { pack := runtimeDeclFormulaInstalledPack{} decl, err := packs.BuildPersistedRuntimeDeclaration(pack) if err != nil { t.Fatalf("BuildPersistedRuntimeDeclaration: %v", err) } row := installedPackRow{ Name: pack.Name(), RuntimeObjectSpecDecl: sql.NullString{String: decl.ObjectPayloadsJSON(), Valid: true}, RuntimeRelatedLists: sql.NullString{String: decl.RelatedListRulesJSON(), Valid: true}, RuntimeSearchScope: sql.NullString{String: decl.SearchScopeOrderJSON(), Valid: true}, RuntimeSearchAllDefault: sql.NullString{String: decl.SearchAllDefaultFieldsJSON(), Valid: true}, RuntimeFormulaBundles: sql.NullString{String: `[]`, Valid: true}, RuntimeStartupShallow: sql.NullString{String: decl.StartupShallowDeclaredRowsJSON(), Valid: true}, HasRuntimeFormula: true, HasRuntimeStartupDecl: true, } err = verifyInstalledPersistedRuntimeDeclarationState(row, pack, true) if err == nil || !strings.Contains(err.Error(), "persisted runtime declaration drifted from source identity") { t.Fatalf("verifyInstalledPersistedRuntimeDeclarationState error = %v, want formula bundle drift error", err) } } func TestVerifyInstalledPersistedRuntimeDeclarationStateRejectsIncompleteFormulaBundles(t *testing.T) { pack := runtimeDeclFormulaInstalledPack{} decl, err := packs.BuildPersistedRuntimeDeclaration(pack) if err != nil { t.Fatalf("BuildPersistedRuntimeDeclaration: %v", err) } row := installedPackRow{ Name: pack.Name(), RuntimeObjectSpecDecl: sql.NullString{String: decl.ObjectPayloadsJSON(), Valid: true}, RuntimeRelatedLists: sql.NullString{String: decl.RelatedListRulesJSON(), Valid: true}, RuntimeSearchScope: sql.NullString{String: decl.SearchScopeOrderJSON(), Valid: true}, RuntimeSearchAllDefault: sql.NullString{String: decl.SearchAllDefaultFieldsJSON(), Valid: true}, RuntimeStartupShallow: sql.NullString{String: decl.StartupShallowDeclaredRowsJSON(), Valid: true}, HasRuntimeFormula: true, HasRuntimeStartupDecl: true, } err = verifyInstalledPersistedRuntimeDeclarationState(row, pack, true) if err == nil || !strings.Contains(err.Error(), "persisted runtime declaration is incomplete") { t.Fatalf("verifyInstalledPersistedRuntimeDeclarationState error = %v, want incomplete formula bundle error", err) } } func TestVerifyInstalledPersistedRuntimeDeclarationStateAcceptsSemanticallyEquivalentJSON(t *testing.T) { pack := runtimeDeclInstalledPack{} decl, err := packs.BuildPersistedRuntimeDeclaration(pack) if err != nil { t.Fatalf("BuildPersistedRuntimeDeclaration: %v", err) } row := installedPackRow{ Name: pack.Name(), RuntimeObjectSpecDecl: sql.NullString{String: jsonIndentString(t, decl.ObjectPayloadsJSON()), Valid: true}, RuntimeRelatedLists: sql.NullString{String: jsonIndentString(t, decl.RelatedListRulesJSON()), Valid: true}, RuntimeSearchScope: sql.NullString{String: jsonIndentString(t, decl.SearchScopeOrderJSON()), Valid: true}, RuntimeSearchAllDefault: sql.NullString{ String: jsonIndentString(t, decl.SearchAllDefaultFieldsJSON()), Valid: true, }, RuntimeStartupShallow: sql.NullString{String: jsonIndentString(t, decl.StartupShallowDeclaredRowsJSON()), Valid: true}, HasRuntimeStartupDecl: true, } if err := verifyInstalledPersistedRuntimeDeclarationState(row, pack, true); err != nil { t.Fatalf("verifyInstalledPersistedRuntimeDeclarationState: %v", err) } } func TestVerifyInstalledPersistedRuntimeDeclarationStateAcceptsSearchAllDefaultFields(t *testing.T) { pack := runtimeDeclSearchAllInstalledPack{} decl, err := packs.BuildPersistedRuntimeDeclaration(pack) if err != nil { t.Fatalf("BuildPersistedRuntimeDeclaration: %v", err) } row := installedPackRow{ Name: pack.Name(), RuntimeObjectSpecDecl: sql.NullString{String: decl.ObjectPayloadsJSON(), Valid: true}, RuntimeRelatedLists: sql.NullString{String: decl.RelatedListRulesJSON(), Valid: true}, RuntimeSearchScope: sql.NullString{String: decl.SearchScopeOrderJSON(), Valid: true}, RuntimeSearchAllDefault: sql.NullString{String: jsonIndentString(t, decl.SearchAllDefaultFieldsJSON()), Valid: true}, RuntimeStartupShallow: sql.NullString{String: decl.StartupShallowDeclaredRowsJSON(), Valid: true}, HasRuntimeStartupDecl: true, } if err := verifyInstalledPersistedRuntimeDeclarationState(row, pack, true); err != nil { t.Fatalf("verifyInstalledPersistedRuntimeDeclarationState: %v", err) } } func TestVerifyInstalledPersistedRuntimeDeclarationStateIgnoresStartupShallowWhenDisabled(t *testing.T) { pack := runtimeDeclInstalledPack{} decl, err := packs.BuildPersistedRuntimeDeclaration(pack) if err != nil { t.Fatalf("BuildPersistedRuntimeDeclaration: %v", err) } row := installedPackRow{ Name: pack.Name(), RuntimeObjectSpecDecl: sql.NullString{String: decl.ObjectPayloadsJSON(), Valid: true}, RuntimeRelatedLists: sql.NullString{String: decl.RelatedListRulesJSON(), Valid: true}, RuntimeSearchScope: sql.NullString{String: decl.SearchScopeOrderJSON(), Valid: true}, RuntimeSearchAllDefault: sql.NullString{String: decl.SearchAllDefaultFieldsJSON(), Valid: true}, RuntimeStartupShallow: sql.NullString{String: `{`, Valid: true}, HasRuntimeStartupDecl: true, } if err := verifyInstalledPersistedRuntimeDeclarationState(row, pack, false); err != nil { t.Fatalf("verifyInstalledPersistedRuntimeDeclarationState(false): %v", err) } } func TestVerifyInstalledAppPacksDoesNotBackfillLegacyStartupShallowDeclaration(t *testing.T) { sourceRoot := writeObjectSpecListFieldsPackSourceRoot(t) pack, err := packs.ResolveInstalledSourceBackedPack( "object-spec-list-fields-pack", packs.AppPack, string(packs.PackSourceKindLocalPath), sourceRoot, ) if err != nil { t.Fatalf("ResolveInstalledSourceBackedPack: %v", err) } decl, err := packs.BuildPersistedRuntimeDeclaration(pack) if err != nil { t.Fatalf("BuildPersistedRuntimeDeclaration: %v", err) } payloadHash, err := packs.PackPayloadHash(pack) if err != nil { t.Fatalf("PackPayloadHash: %v", err) } meta := pack.PackHashMetadata() rawDB, mock, err := sqlmock.New() if err != nil { t.Fatalf("sqlmock: %v", err) } defer rawDB.Close() expectInstalledPackRuntimeDeclarationColumns(mock, true) expectInstalledPackRuntimeStartupShallowDeclarationColumn(mock, true) mock.ExpectQuery("SELECT Name, Category, Status, PackVersion, ManifestFormatVersion, PlatformCompatibility, DesiredPayloadHash, AppliedPayloadHash, PayloadHashAlgorithm, SourceKind, SourceURI, SourceRevision, HostRequirementsDigest, RuntimeObjectSpecDeclarations, RuntimeRelatedListRules, RuntimeSearchScopeOrder, RuntimeSearchAllDefaultFields, RuntimeStartupShallowDeclarations"). WithArgs(string(packs.AppPack), string(packs.ConnectorPack)). WillReturnRows(sqlmock.NewRows([]string{ "Name", "Category", "Status", "PackVersion", "ManifestFormatVersion", "PlatformCompatibility", "DesiredPayloadHash", "AppliedPayloadHash", "PayloadHashAlgorithm", "SourceKind", "SourceURI", "SourceRevision", "HostRequirementsDigest", "RuntimeObjectSpecDeclarations", "RuntimeRelatedListRules", "RuntimeSearchScopeOrder", "RuntimeSearchAllDefaultFields", "RuntimeStartupShallowDeclarations", }).AddRow( pack.Name(), string(packs.AppPack), installedPackStatusActive, meta.PackVersion, meta.ManifestFormatVersion, meta.PlatformCompatibility, payloadHash, payloadHash, packs.PackPayloadHashAlgorithmV1, packs.PackSourceKindLocalPath, sourceRoot, nil, nil, decl.ObjectPayloadsJSON(), decl.RelatedListRulesJSON(), decl.SearchScopeOrderJSON(), decl.SearchAllDefaultFieldsJSON(), `[]`, )) expectNoPackOwnedApprovalProcesses(mock, pack.Name()) err = verifyInstalledAppPacks(context.Background(), database.NewConnectionFromDB(rawDB), nil, true) if err == nil || !strings.Contains(err.Error(), "persisted runtime declaration drifted from source identity") { t.Fatalf("verifyInstalledAppPacks error = %v, want strict drift without readiness-time backfill", err) } if err := mock.ExpectationsWereMet(); err != nil { t.Fatalf("sql expectations: %v", err) } } func TestVerifyInstalledPersistedRuntimeDeclarationStateIgnoresStartupShallowWhenColumnAbsent(t *testing.T) { pack := runtimeDeclInstalledPack{} decl, err := packs.BuildPersistedRuntimeDeclaration(pack) if err != nil { t.Fatalf("BuildPersistedRuntimeDeclaration: %v", err) } row := installedPackRow{ Name: pack.Name(), RuntimeObjectSpecDecl: sql.NullString{String: decl.ObjectPayloadsJSON(), Valid: true}, RuntimeRelatedLists: sql.NullString{String: decl.RelatedListRulesJSON(), Valid: true}, RuntimeSearchScope: sql.NullString{String: decl.SearchScopeOrderJSON(), Valid: true}, RuntimeSearchAllDefault: sql.NullString{String: decl.SearchAllDefaultFieldsJSON(), Valid: true}, RuntimeStartupShallow: sql.NullString{String: `{`, Valid: true}, HasRuntimeStartupDecl: false, } if err := verifyInstalledPersistedRuntimeDeclarationState(row, pack, true); err != nil { t.Fatalf("verifyInstalledPersistedRuntimeDeclarationState legacy column-absent row: %v", err) } } func jsonIndentString(t *testing.T, raw string) string { t.Helper() var value any if err := json.Unmarshal([]byte(raw), &value); err != nil { t.Fatalf("json.Unmarshal: %v", err) } indented, err := json.MarshalIndent(value, "", " ") if err != nil { t.Fatalf("json.MarshalIndent: %v", err) } return string(indented) } type fakeInstalledPack struct{} func (fakeInstalledPack) Name() string { return "fake-pack" } func (fakeInstalledPack) Category() packs.Category { return packs.AppPack } func (fakeInstalledPack) Apply(context.Context, packs.ApplyDeps) error { return nil } func (fakeInstalledPack) Verify(context.Context, packs.VerifyDeps) error { return nil } func (fakeInstalledPack) PayloadHash() (string, error) { return fakePayloadHashV1, nil } func (fakeInstalledPack) LegacyPayloadHash() (string, error) { return fakeLegacyHashV0, nil } func (fakeInstalledPack) PackHashMetadata() packs.PackHashMetadata { return packs.PackHashMetadata{ PackVersion: packs.BundledPackVersion, ManifestFormatVersion: packs.BundledManifestFormatVersion, PlatformCompatibility: packs.BundledPlatformCompatibilityPayload, PlatformCompatibilityValid: true, PayloadHashAlgorithm: packs.PackPayloadHashAlgorithmV1, } } type postVerifyFailingInstalledPack struct { fakeInstalledPack verifyCalls int initialErr error finalErr error } func (p *postVerifyFailingInstalledPack) Verify(context.Context, packs.VerifyDeps) error { p.verifyCalls++ if p.verifyCalls == 1 && p.initialErr != nil { return p.initialErr } if p.verifyCalls == 2 { return p.finalErr } return nil } type contextCheckingInstalledPack struct { fakeInstalledPack expectedPeerPack string verifyCalls int } func (p *contextCheckingInstalledPack) Verify(ctx context.Context, _ packs.VerifyDeps) error { p.verifyCalls++ if _, ok := packs.PeerDeclarationOverrideFromContext(ctx, p.expectedPeerPack); !ok { return errors.New("peer declaration override missing from verify context") } return nil } type runtimeDeclInstalledPack struct { fakeInstalledPack } type runtimeDeclFormulaInstalledPack struct { runtimeDeclInstalledPack } type runtimeDeclSearchAllInstalledPack struct { runtimeDeclInstalledPack } func writeObjectSpecListFieldsPackSourceRoot(t *testing.T) string { t.Helper() root := filepath.Join(t.TempDir(), "object-spec-list-fields-pack") if err := os.MkdirAll(root, 0o755); err != nil { t.Fatalf("mkdir object spec list fields pack root: %v", err) } manifest := `{ "manifestFormatVersion": "1", "name": "object-spec-list-fields-pack", "displayName": "Object Spec List Fields Pack", "category": "app-pack", "version": "0.1.0", "platformCompatibility": {"min": "0.2.0", "maxExclusive": "0.3.0"}, "dependencies": [], "components": [ { "kind": "object_spec", "businessKey": "object:Lead", "manageability": "locked", "payload": [ { "apiName": "Lead", "label": "Lead", "tableType": "standard_object", "category": "crm", "sharingModel": "Private", "enableHierarchySharing": false, "description": "Lead", "listFields": ["Name", "Company"], "fields": [ {"apiName": "Name", "label": "Name", "type": "string", "required": true, "isNameField": true}, {"apiName": "Company", "label": "Company", "type": "string"} ] } ] } ] }` if err := os.WriteFile(filepath.Join(root, "pack.json"), []byte(manifest), 0o644); err != nil { t.Fatalf("write object spec list fields pack manifest: %v", err) } return root } func writeFormulaFunctionBundlePackSourceRoot(t *testing.T) string { t.Helper() root := filepath.Join(t.TempDir(), "formula-functions") if err := os.MkdirAll(root, 0o755); err != nil { t.Fatalf("mkdir formula function bundle pack root: %v", err) } manifest := `{ "manifestFormatVersion": "1", "name": "formula-functions", "displayName": "Formula Functions", "category": "app-pack", "version": "0.1.0", "platformCompatibility": {"min": "0.2.0", "maxExclusive": "0.3.0"}, "dependencies": [], "components": [ { "kind": "formula_function_bundle", "businessKey": "probe-functions", "manageability": "locked", "payload": { "functions": [ { "name": "PACK_PREFIX", "minArgs": 1, "maxArgs": 1, "return": { "kind": "fixed", "type": "string" }, "args": [{ "type": "string", "nullable": true }], "evaluator": { "kind": "formula_expr_v1", "params": ["value"], "body": "value + \"-probe\"" } } ] } } ] }` if err := os.WriteFile(filepath.Join(root, "pack.json"), []byte(manifest), 0o644); err != nil { t.Fatalf("write formula function bundle pack manifest: %v", err) } return root } func (runtimeDeclInstalledPack) RuntimeObjectSpecPayloads() [][]byte { return [][]byte{[]byte(`{"apiName":"Issue"}`)} } func (runtimeDeclInstalledPack) RuntimeRelatedListRules() []metadata.RelatedListRule { return []metadata.RelatedListRule{{ Key: "issue-comments", ParentObjectAPI: "Issue", ChildObjectAPI: "Comment", LookupField: "IssueId", Bucket: metadata.BucketRecommended, Fields: []string{"Body"}, }} } func (runtimeDeclInstalledPack) RuntimeSearchScopeOrder() []metadata.SearchScopeOrderItem { return []metadata.SearchScopeOrderItem{{ Name: "Issue", Label: "Issue", }} } func (runtimeDeclSearchAllInstalledPack) RuntimeSearchAllDefaultFields() []searchmetadata.SearchAllDefaultFieldsDeclaration { return []searchmetadata.SearchAllDefaultFieldsDeclaration{{ ObjectAPIName: "Issue", Fields: []string{"Name"}, }} } func (runtimeDeclFormulaInstalledPack) RuntimeFormulaFunctionBundles() []formula.FunctionBundle { return []formula.FunctionBundle{{ Functions: []formula.FunctionDefinition{{ Name: "PACK_PREFIX", MinArgs: 1, MaxArgs: 1, Return: formula.FunctionReturnDefinition{Kind: "fixed", Type: spec.TypeString}, Args: []spec.ArgSpec{{Type: spec.TypeString, Nullable: true}}, Evaluator: formula.FunctionEvaluator{ Kind: formula.FormulaExprV1EvaluatorKind, Params: []string{"value"}, Body: `value + "-probe"`, }, }}, }} } func (runtimeDeclInstalledPack) OwnedObjectAPINames() []string { return []string{"Issue"} } type installedPackMetadata struct { packVersion string manifestFormatVersion string platformCompatibility string payloadHashAlgorithm string valid bool } func installedPackMetadataFromPack(pack packs.Pack) installedPackMetadata { meta := packs.PackHashMetadataForPack(pack) return installedPackMetadata{ packVersion: meta.PackVersion, manifestFormatVersion: meta.ManifestFormatVersion, platformCompatibility: meta.PlatformCompatibility, payloadHashAlgorithm: meta.PayloadHashAlgorithm, valid: true, } } func installedPackHashRow(applied, desired string, meta installedPackMetadata) installedPackRow { row := installedPackRow{ Name: "fake-pack", Category: string(packs.AppPack), Status: installedPackStatusActive, AppliedPayloadHash: sql.NullString{String: applied, Valid: applied != ""}, DesiredPayloadHash: sql.NullString{String: desired, Valid: desired != ""}, } if meta.valid { row.PackVersion = sql.NullString{String: meta.packVersion, Valid: true} row.ManifestFormatVersion = sql.NullString{String: meta.manifestFormatVersion, Valid: true} row.PlatformCompatibility = sql.NullString{String: meta.platformCompatibility, Valid: true} row.PayloadHashAlgorithm = sql.NullString{String: meta.payloadHashAlgorithm, Valid: true} } return row } // minimalPack is a minimal packs.Pack implementation for validateNamespaceReservation tests. type minimalPack struct { name string category packs.Category requirements []packs.HostPrimitiveRequirement } func (p minimalPack) Name() string { return p.name } func (p minimalPack) Category() packs.Category { return p.category } func (p minimalPack) Apply(context.Context, packs.ApplyDeps) error { return nil } func (p minimalPack) Verify(context.Context, packs.VerifyDeps) error { return nil } func (p minimalPack) HostRequirements() []packs.HostPrimitiveRequirement { return append([]packs.HostPrimitiveRequirement(nil), p.requirements...) } func expectNamespaceReservationLookup(mock sqlmock.Sqlmock, name string, row []driver.Value) { query := "SELECT Category, COALESCE\\(IsDeleted, 0\\) FROM _System_InstalledPack WHERE Name" if row == nil { mock.ExpectQuery(query). WithArgs(name). WillReturnRows(sqlmock.NewRows([]string{"Category", "IsDeleted"})) return } mock.ExpectQuery(query). WithArgs(name). WillReturnRows(sqlmock.NewRows([]string{"Category", "IsDeleted"}).AddRow(row...)) } func expectConnectorPackQueuedReapplyGuardLookup(mock sqlmock.Sqlmock, packName string, rows *sqlmock.Rows) { mock.ExpectQuery(`SELECT COALESCE\(p\.PackInstallId, ''\)[\s\S]*_System_ConnectorWorkItem[\s\S]*ProjectionKey`). WithArgs(packName, string(packs.ConnectorPack), installedPackStatusActive, installedPackStatusInstalling, installedPackStatusRemoving, installedPackStatusFailed). WillReturnRows(rows) } func expectCompanionAppPackQueuedReapplyGuardLookup(mock sqlmock.Sqlmock, packName string, rows *sqlmock.Rows) { mock.ExpectQuery(`SELECT COALESCE\(p\.PackInstallId, ''\)[\s\S]*_System_ConnectorWorkItem[\s\S]*ProjectionKey`). WithArgs(packName, string(packs.AppPack), installedPackStatusActive, installedPackStatusInstalling, installedPackStatusRemoving, installedPackStatusFailed). WillReturnRows(rows) } func expectConnectorWorkItemSchemaLookup(mock sqlmock.Sqlmock, ready bool) { count := 0 if ready { count = len(connectorInboundWorkItemRequiredColumns) } args := []driver.Value{"_System_ConnectorWorkItem"} for _, column := range connectorInboundWorkItemRequiredColumns { args = append(args, column) } mock.ExpectQuery(`SELECT COUNT\(DISTINCT COLUMN_NAME\)[\s\S]*FROM INFORMATION_SCHEMA\.COLUMNS`). WithArgs(args...). WillReturnRows(sqlmock.NewRows([]string{"COUNT(DISTINCT COLUMN_NAME)"}).AddRow(count)) } func expectAppPackInstallingUpdate(mock sqlmock.Sqlmock, packName string, rowsAffected int64, runtimeArgCount int) { expectConnectorWorkItemSchemaLookup(mock, true) args := []driver.Value{ sqlmock.AnyArg(), // PackInstallId string(packs.AppPack), installedPackStatusInstalling, sqlmock.AnyArg(), // PackVersion sqlmock.AnyArg(), // ManifestFormatVersion sqlmock.AnyArg(), // PlatformCompatibility sqlmock.AnyArg(), // DesiredPayloadHash sqlmock.AnyArg(), // PayloadHashAlgorithm sqlmock.AnyArg(), // SourceKind sqlmock.AnyArg(), // SourceURI sqlmock.AnyArg(), // SourceRevision } for i := 0; i < runtimeArgCount; i++ { args = append(args, sqlmock.AnyArg()) } args = append(args, sqlmock.AnyArg(), // LastModifiedDate packName, string(packs.AppPack), ) mock.ExpectExec(`UPDATE _System_InstalledPack p[\s\S]*_System_ConnectorWorkItem[\s\S]*ProjectionKey`). WithArgs(args...). WillReturnResult(sqlmock.NewResult(0, rowsAffected)) } func expectInstalledPackRowExists(mock sqlmock.Sqlmock, packName string, exists bool) { rows := sqlmock.NewRows([]string{"Id"}) if exists { rows.AddRow(installedPackID(packName)) } mock.ExpectQuery(`SELECT Id FROM _System_InstalledPack WHERE Name`). WithArgs(packName). WillReturnRows(rows) } func expectConnectorPackInstallingUpdate(mock sqlmock.Sqlmock, packName string, packInstallIDArg driver.Value, rowsAffected int64) { expectConnectorPackInstallingUpdateWithSchemaReady(mock, packName, packInstallIDArg, rowsAffected, true) } func expectConnectorPackInstallingUpdateWithSchemaReady(mock sqlmock.Sqlmock, packName string, packInstallIDArg driver.Value, rowsAffected int64, schemaReady bool) { expectConnectorWorkItemSchemaLookup(mock, schemaReady) updatePattern := `UPDATE _System_InstalledPack p[\s\S]*_System_ConnectorEvent e[\s\S]*e\.Status IN \('pending', 'failed', 'processing'\)` if schemaReady { updatePattern = `UPDATE _System_InstalledPack p[\s\S]*_System_ConnectorWorkItem[\s\S]*ProjectionKey` } mock.ExpectExec(updatePattern). WithArgs( packInstallIDArg, // PackInstallId string(packs.ConnectorPack), // Category installedPackStatusInstalling, sqlmock.AnyArg(), // PackVersion sqlmock.AnyArg(), // ManifestFormatVersion sqlmock.AnyArg(), // PlatformCompatibility sqlmock.AnyArg(), // DesiredPayloadHash sqlmock.AnyArg(), // PayloadHashAlgorithm sqlmock.AnyArg(), // SourceKind sqlmock.AnyArg(), // SourceURI sqlmock.AnyArg(), // SourceRevision sqlmock.AnyArg(), // LastModifiedDate packName, string(packs.ConnectorPack), ). WillReturnResult(sqlmock.NewResult(0, rowsAffected)) } func expectInstalledPackRuntimeStartupShallowDeclarationColumn(mock sqlmock.Sqlmock, present bool) { expectInstalledPackRuntimeStartupShallowAndFormulaBundleDeclarationColumns(mock, present, false) } func expectInstalledPackRuntimeStartupShallowAndFormulaBundleDeclarationColumns(mock sqlmock.Sqlmock, startupShallowPresent bool, formulaBundlePresent bool) { count := 0 if startupShallowPresent { count = 1 } mock.ExpectQuery("SELECT COUNT\\(\\*\\)\\s+FROM INFORMATION_SCHEMA.COLUMNS"). WithArgs("_System_InstalledPack", "RuntimeStartupShallowDeclarations"). WillReturnRows(sqlmock.NewRows([]string{"COUNT(*)"}).AddRow(count)) count = 0 if formulaBundlePresent { count = 1 } mock.ExpectQuery("SELECT COUNT\\(\\*\\)\\s+FROM INFORMATION_SCHEMA.COLUMNS"). WithArgs("_System_InstalledPack", "RuntimeFormulaFunctionBundles"). WillReturnRows(sqlmock.NewRows([]string{"COUNT(*)"}).AddRow(count)) } func TestValidateNamespaceReservation(t *testing.T) { cases := []struct { name string packName string packCat packs.Category existingRow []driver.Value // nil = no row wantErr bool wantConflict bool }{ { name: "net-new connector-pack install passes", packName: "noop-connector", packCat: packs.ConnectorPack, existingRow: nil, wantErr: false, }, { name: "net-new app-pack install passes", packName: "test-connector", packCat: packs.AppPack, existingRow: nil, wantErr: false, }, { name: "same-category re-install connector passes", packName: "noop-connector", packCat: packs.ConnectorPack, existingRow: []driver.Value{"connector-pack", 0}, wantErr: false, }, { name: "same-category re-install app-pack passes", packName: "peer-app-pack", packCat: packs.AppPack, existingRow: []driver.Value{"app-pack", 0}, wantErr: false, }, { name: "soft-deleted rebirth connector-to-app passes", packName: "noop-connector", packCat: packs.AppPack, existingRow: []driver.Value{"connector-pack", 1}, wantErr: false, }, { name: "connector to non-connector collision fails", packName: "noop-connector", packCat: packs.AppPack, existingRow: []driver.Value{"connector-pack", 0}, wantErr: true, wantConflict: true, }, { name: "non-connector to connector collision fails", packName: "peer-app-pack", packCat: packs.ConnectorPack, existingRow: []driver.Value{"app-pack", 0}, wantErr: true, wantConflict: true, }, { name: "non-connector cross-category platform-core to app-pack passes", packName: "platform-base", packCat: packs.AppPack, existingRow: []driver.Value{"platform-core", 0}, wantErr: false, }, { name: "empty pack name returns error without hitting DB", packName: "", packCat: packs.ConnectorPack, wantErr: true, }, } for _, tc := range cases { t.Run(tc.name, func(t *testing.T) { db, mock, err := sqlmock.New() if err != nil { t.Fatalf("sqlmock: %v", err) } defer db.Close() if tc.packName != "" { expectNamespaceReservationLookup(mock, tc.packName, tc.existingRow) } // Empty pack name returns error before hitting DB — no expectation needed. pack := minimalPack{name: tc.packName, category: tc.packCat} err = validateNamespaceReservation(context.Background(), db, pack) if tc.wantErr { if err == nil { t.Fatal("validateNamespaceReservation = nil, want error") } if tc.wantConflict && !errors.Is(err, ErrConnectorPackNamespaceConflict) { t.Fatalf("validateNamespaceReservation error = %v, want ErrConnectorPackNamespaceConflict", err) } } else { if err != nil { t.Fatalf("validateNamespaceReservation = %v, want nil", err) } } if tc.packName != "" { if err2 := mock.ExpectationsWereMet(); err2 != nil { t.Fatalf("sql expectations: %v", err2) } } }) } } func TestValidateNamespaceReservationDBError(t *testing.T) { db, mock, err := sqlmock.New() if err != nil { t.Fatalf("sqlmock: %v", err) } defer db.Close() mock.ExpectQuery("SELECT Category, COALESCE\\(IsDeleted, 0\\) FROM _System_InstalledPack WHERE Name"). WithArgs("noop-connector"). WillReturnError(errors.New("db connection lost")) pack := minimalPack{name: "noop-connector", category: packs.ConnectorPack} err = validateNamespaceReservation(context.Background(), db, pack) if err == nil { t.Fatal("validateNamespaceReservation DB error = nil, want error") } if !strings.Contains(err.Error(), "noop-connector") { t.Fatalf("validateNamespaceReservation DB error = %v, want name context", err) } if err2 := mock.ExpectationsWereMet(); err2 != nil { t.Fatalf("sql expectations: %v", err2) } } // captureArg is a sqlmock.Argument that records the value it matched so the // test can inspect it after the call. type captureArg struct { captured *string } func (c captureArg) Match(v driver.Value) bool { if s, ok := v.(string); ok { *c.captured = s return true } return false } // TestMarkInstalledPackInstalling_MintsFreshPackInstallIdOnRebirthWithoutQueue // verifies that markInstalledPackInstalling generates a new PackInstallId for // same-name rebirths when the prior active connector install has no // non-terminal inbound rows to drain. // // The test uses sqlmock to assert: // 1. The INSERT SQL contains PackInstallId in the column list. // 2. The PackInstallId arg passed on the second call differs from the first // when there are no old queue rows requiring in-place reapply identity. func TestMarkInstalledPackInstalling_MintsFreshPackInstallIdOnRebirthWithoutQueue(t *testing.T) { db, mock, err := sqlmock.New() if err != nil { t.Fatalf("sqlmock: %v", err) } defer db.Close() pack := minimalPack{name: "rebirth-connector", category: packs.ConnectorPack} var firstInstallID, secondInstallID string firstCap := captureArg{captured: &firstInstallID} secondCap := captureArg{captured: &secondInstallID} // First install: net-new row (namespace reservation returns empty). mock.ExpectQuery("SELECT Category, COALESCE\\(IsDeleted, 0\\) FROM _System_InstalledPack WHERE Name"). WithArgs("rebirth-connector"). WillReturnRows(sqlmock.NewRows([]string{"Category", "IsDeleted"})) expectConnectorPackInstallingUpdate(mock, "rebirth-connector", firstCap, 0) expectConnectorPackQueuedReapplyGuardLookup(mock, "rebirth-connector", sqlmock.NewRows([]string{"PackInstallId"})) mock.ExpectExec("INSERT INTO _System_InstalledPack"). WithArgs( sqlmock.AnyArg(), // Id (deterministic) sqlmock.AnyArg(), // Name firstCap, // PackInstallId — captured for comparison sqlmock.AnyArg(), // Category sqlmock.AnyArg(), // Status sqlmock.AnyArg(), // PackVersion sqlmock.AnyArg(), // ManifestFormatVersion sqlmock.AnyArg(), // PlatformCompatibility sqlmock.AnyArg(), // DesiredPayloadHash sqlmock.AnyArg(), // PayloadHashAlgorithm sqlmock.AnyArg(), // SourceKind sqlmock.AnyArg(), // SourceURI sqlmock.AnyArg(), // SourceRevision sqlmock.AnyArg(), // CreatedDate sqlmock.AnyArg(), // LastModifiedDate ). WillReturnResult(sqlmock.NewResult(0, 1)) expectInstalledPackHostRequirementsDigestUpdate(mock, pack.Name()) if err := markInstalledPackInstalling(context.Background(), db, pack, "hash-v1-first"); err != nil { t.Fatalf("first markInstalledPackInstalling: %v", err) } if firstInstallID == "" { t.Fatal("first PackInstallId was not captured — captureArg did not match; check arg position") } // Second install (rebirth): namespace reservation returns same-category existing row. mock.ExpectQuery("SELECT Category, COALESCE\\(IsDeleted, 0\\) FROM _System_InstalledPack WHERE Name"). WithArgs("rebirth-connector"). WillReturnRows(sqlmock.NewRows([]string{"Category", "IsDeleted"}).AddRow("connector-pack", 0)) expectConnectorPackInstallingUpdate(mock, "rebirth-connector", secondCap, 1) expectInstalledPackHostRequirementsDigestUpdate(mock, pack.Name()) if err := markInstalledPackInstalling(context.Background(), db, pack, "hash-v1-second"); err != nil { t.Fatalf("second markInstalledPackInstalling: %v", err) } if secondInstallID == "" { t.Fatal("second PackInstallId was not captured") } if firstInstallID == secondInstallID { t.Fatalf("PackInstallId not refreshed on queue-free rebirth: both calls produced %q — each fresh incarnation must mint a distinct UUID", firstInstallID) } if len(firstInstallID) < 8 || len(secondInstallID) < 8 { t.Fatalf("PackInstallId values look too short to be UUIDs: first=%q second=%q", firstInstallID, secondInstallID) } if err2 := mock.ExpectationsWereMet(); err2 != nil { t.Fatalf("sql expectations: %v", err2) } } func TestMarkInstalledPackInstallingFallsBackForIncompleteWorkItemSchema(t *testing.T) { db, mock, err := sqlmock.New() if err != nil { t.Fatalf("sqlmock: %v", err) } defer db.Close() pack := minimalPack{name: "legacy-schema-connector", category: packs.ConnectorPack} mock.ExpectQuery("SELECT Category, COALESCE\\(IsDeleted, 0\\) FROM _System_InstalledPack WHERE Name"). WithArgs(pack.Name()). WillReturnRows(sqlmock.NewRows([]string{"Category", "IsDeleted"}).AddRow("connector-pack", 0)) expectConnectorPackInstallingUpdateWithSchemaReady(mock, pack.Name(), sqlmock.AnyArg(), 1, false) expectInstalledPackHostRequirementsDigestUpdate(mock, pack.Name()) if err := markInstalledPackInstalling(context.Background(), db, pack, "hash-v1-legacy-schema"); err != nil { t.Fatalf("markInstalledPackInstalling: %v", err) } if err := mock.ExpectationsWereMet(); err != nil { t.Fatalf("sql expectations: %v", err) } } // TestMarkInstalledPackInstalling_BlocksActiveReapplyWithNonTerminalInboundRows // verifies that an active connector-pack reapply fails closed while the // existing active install still owns non-terminal inbound rows. The current // single-row installed-pack model cannot represent both a new active generation // and an old drain-only generation, so applying now would either orphan old // rows or process them under new live metadata. func TestMarkInstalledPackInstalling_BlocksActiveReapplyWithNonTerminalInboundRows(t *testing.T) { db, mock, err := sqlmock.New() if err != nil { t.Fatalf("sqlmock: %v", err) } defer db.Close() const existingInstallID = "install-active-old" pack := minimalPack{name: "reapply-connector", category: packs.ConnectorPack} mock.ExpectQuery("SELECT Category, COALESCE\\(IsDeleted, 0\\) FROM _System_InstalledPack WHERE Name"). WithArgs("reapply-connector"). WillReturnRows(sqlmock.NewRows([]string{"Category", "IsDeleted"}).AddRow("connector-pack", 0)) expectConnectorPackInstallingUpdate(mock, "reapply-connector", sqlmock.AnyArg(), 0) expectConnectorPackQueuedReapplyGuardLookup(mock, "reapply-connector", sqlmock.NewRows([]string{"PackInstallId"}).AddRow(existingInstallID)) err = markInstalledPackInstalling(context.Background(), db, pack, "hash-v1-reapply") if err == nil { t.Fatal("markInstalledPackInstalling = nil, want queued inbound drain guard error") } if !strings.Contains(err.Error(), "non-terminal inbound rows") || !strings.Contains(err.Error(), existingInstallID) { t.Fatalf("markInstalledPackInstalling error = %v, want queued drain guard mentioning %q", err, existingInstallID) } if err2 := mock.ExpectationsWereMet(); err2 != nil { t.Fatalf("sql expectations: %v", err2) } } // TestMarkInstalledPackInstalling_BlocksFailedRetryWithNonTerminalInboundRows // verifies the retry path after a queued in-place reapply fails. The failed row // still owns non-terminal inbound rows by PackInstallId, so the next install // attempt must fail closed instead of overwriting the only installed-pack row // with a new generation. func TestMarkInstalledPackInstalling_BlocksFailedRetryWithNonTerminalInboundRows(t *testing.T) { db, mock, err := sqlmock.New() if err != nil { t.Fatalf("sqlmock: %v", err) } defer db.Close() const existingInstallID = "install-failed-old" pack := minimalPack{name: "retry-connector", category: packs.ConnectorPack} mock.ExpectQuery("SELECT Category, COALESCE\\(IsDeleted, 0\\) FROM _System_InstalledPack WHERE Name"). WithArgs("retry-connector"). WillReturnRows(sqlmock.NewRows([]string{"Category", "IsDeleted"}).AddRow("connector-pack", 0)) expectConnectorPackInstallingUpdate(mock, "retry-connector", sqlmock.AnyArg(), 0) expectConnectorPackQueuedReapplyGuardLookup(mock, "retry-connector", sqlmock.NewRows([]string{"PackInstallId"}).AddRow(existingInstallID)) err = markInstalledPackInstalling(context.Background(), db, pack, "hash-v1-retry") if err == nil { t.Fatal("markInstalledPackInstalling = nil, want queued inbound drain guard error") } if !strings.Contains(err.Error(), "non-terminal inbound rows") || !strings.Contains(err.Error(), existingInstallID) { t.Fatalf("markInstalledPackInstalling error = %v, want queued drain guard mentioning %q", err, existingInstallID) } if err2 := mock.ExpectationsWereMet(); err2 != nil { t.Fatalf("sql expectations: %v", err2) } } func TestMarkInstalledPackInstalling_BlocksInFlightLifecycleRowsWithNonTerminalInboundRows(t *testing.T) { cases := []struct { name string packName string installID string }{ { name: "installing lifecycle row", packName: "installing-connector", installID: "install-installing-old", }, { name: "removing lifecycle row", packName: "removing-connector", installID: "install-removing-old", }, } for _, tc := range cases { t.Run(tc.name, func(t *testing.T) { db, mock, err := sqlmock.New() if err != nil { t.Fatalf("sqlmock: %v", err) } defer db.Close() pack := minimalPack{name: tc.packName, category: packs.ConnectorPack} mock.ExpectQuery("SELECT Category, COALESCE\\(IsDeleted, 0\\) FROM _System_InstalledPack WHERE Name"). WithArgs(tc.packName). WillReturnRows(sqlmock.NewRows([]string{"Category", "IsDeleted"}).AddRow("connector-pack", 0)) expectConnectorPackInstallingUpdate(mock, tc.packName, sqlmock.AnyArg(), 0) expectConnectorPackQueuedReapplyGuardLookup(mock, tc.packName, sqlmock.NewRows([]string{"PackInstallId"}).AddRow(tc.installID)) err = markInstalledPackInstalling(context.Background(), db, pack, "hash-v1-inflight") if err == nil { t.Fatal("markInstalledPackInstalling = nil, want queued inbound drain guard error") } if !strings.Contains(err.Error(), "non-terminal inbound rows") || !strings.Contains(err.Error(), tc.installID) { t.Fatalf("markInstalledPackInstalling error = %v, want queued drain guard mentioning %q", err, tc.installID) } if err2 := mock.ExpectationsWereMet(); err2 != nil { t.Fatalf("sql expectations: %v", err2) } }) } } func TestMarkInstalledPackInstalling_BlocksCompanionAppPackReapplyWithNonTerminalInboundRows(t *testing.T) { db, mock, err := sqlmock.New() if err != nil { t.Fatalf("sqlmock: %v", err) } defer db.Close() const existingInstallID = "companion-install-old" pack := minimalPack{name: "companion-app", category: packs.AppPack} mock.ExpectQuery("SELECT Category, COALESCE\\(IsDeleted, 0\\) FROM _System_InstalledPack WHERE Name"). WithArgs("companion-app"). WillReturnRows(sqlmock.NewRows([]string{"Category", "IsDeleted"}).AddRow("app-pack", 0)) expectInstalledPackRuntimeDeclarationColumns(mock, true) expectInstalledPackRuntimeStartupShallowDeclarationColumn(mock, true) expectAppPackInstallingUpdate(mock, "companion-app", 0, 5) expectCompanionAppPackQueuedReapplyGuardLookup(mock, "companion-app", sqlmock.NewRows([]string{"PackInstallId"}).AddRow(existingInstallID)) err = markInstalledPackInstalling(context.Background(), db, pack, "hash-v1-reapply") if err == nil { t.Fatal("markInstalledPackInstalling = nil, want companion inbound drain guard error") } if !strings.Contains(err.Error(), "non-terminal connector inbound rows") || !strings.Contains(err.Error(), existingInstallID) { t.Fatalf("markInstalledPackInstalling error = %v, want companion drain guard mentioning %q", err, existingInstallID) } if err2 := mock.ExpectationsWereMet(); err2 != nil { t.Fatalf("sql expectations: %v", err2) } } // TestMarkInstalledPackFailed_PreservesExistingPackInstallId verifies that // markInstalledPackFailed does NOT re-mint the PackInstallId when a prior row // already carries one. The installing phase mints UUID-A; if the fail-mark // overwrites it with UUID-B, any per-incarnation queue state created during // the installing phase becomes orphaned. // // The test asserts: // 1. The SELECT for the existing PackInstallId is issued. // 2. The INSERT SQL does NOT include `PackInstallId = VALUES(PackInstallId)` // in its ON DUPLICATE KEY UPDATE clause (so a row update never replaces // the stored identity). // 3. The value passed into the INSERT for PackInstallId equals the // pre-existing value (not a fresh UUID). func TestMarkInstalledPackFailed_PreservesExistingPackInstallId(t *testing.T) { db, mock, err := sqlmock.New(sqlmock.QueryMatcherOption(sqlmock.QueryMatcherRegexp)) if err != nil { t.Fatalf("sqlmock: %v", err) } defer db.Close() reconcileCalls := 0 resetReconciler := SetConnectorInboundWorkerReconciler(func(context.Context, *sql.DB) error { reconcileCalls++ return nil }) defer resetReconciler() const existingInstallID = "fixed-token-aaa-existing" pack := minimalPack{name: "rebirth-connector", category: packs.ConnectorPack} // namespace-reservation SELECT (no conflict) expectNamespaceReservationLookup(mock, "rebirth-connector", []driver.Value{"connector-pack", 0}) // SELECT for existing PackInstallId — returns the pre-minted value. mock.ExpectQuery(`SELECT COALESCE\(PackInstallId, ''\) FROM _System_InstalledPack WHERE Id = \? LIMIT 1`). WithArgs(installedPackID("rebirth-connector")). WillReturnRows(sqlmock.NewRows([]string{"PackInstallId"}).AddRow(existingInstallID)) mock.ExpectExec(`UPDATE _System_ConnectorCapabilityDeclaration`). WithArgs("rebirth-connector"). WillReturnResult(sqlmock.NewResult(0, 1)) mock.ExpectExec(`UPDATE _System_ConnectorRateLimit`). WithArgs("rebirth-connector"). WillReturnResult(sqlmock.NewResult(0, 1)) mock.ExpectExec(`UPDATE _System_ConnectorInboundHandlerConfig`). WithArgs("rebirth-connector"). WillReturnResult(sqlmock.NewResult(0, 1)) mock.ExpectExec(`UPDATE _System_ConnectorCalloutProfile`). WithArgs("rebirth-connector"). WillReturnResult(sqlmock.NewResult(0, 1)) // Capture the PackInstallId arg passed to the INSERT. var capturedInstallID string cap := captureArg{captured: &capturedInstallID} expectInstalledPackRuntimeDeclarationColumns(mock, false) mock.ExpectExec(`INSERT INTO _System_InstalledPack`). WithArgs( sqlmock.AnyArg(), // Id sqlmock.AnyArg(), // Name cap, // PackInstallId — must equal existingInstallID sqlmock.AnyArg(), // Category sqlmock.AnyArg(), // Status sqlmock.AnyArg(), // PackVersion sqlmock.AnyArg(), // ManifestFormatVersion sqlmock.AnyArg(), // PlatformCompatibility sqlmock.AnyArg(), // DesiredPayloadHash sqlmock.AnyArg(), // PayloadHashAlgorithm sqlmock.AnyArg(), // SourceKind sqlmock.AnyArg(), // SourceURI sqlmock.AnyArg(), // SourceRevision sqlmock.AnyArg(), // LastError (msg) sqlmock.AnyArg(), // CreatedDate sqlmock.AnyArg(), // LastModifiedDate ). WillReturnResult(sqlmock.NewResult(0, 1)) if err := markInstalledPackFailed(context.Background(), db, pack, "hash-v1-failed", errors.New("install exploded")); err != nil { t.Fatalf("markInstalledPackFailed: %v", err) } if capturedInstallID != existingInstallID { t.Fatalf("markInstalledPackFailed used PackInstallId=%q, want %q — existing incarnation identity must be preserved", capturedInstallID, existingInstallID) } if reconcileCalls != 1 { t.Fatalf("reconcileCalls = %d, want 1", reconcileCalls) } if err2 := mock.ExpectationsWereMet(); err2 != nil { t.Fatalf("sql expectations: %v", err2) } } // TestMarkInstalledPackFailed_MintsWhenNoPriorRow verifies that when no prior // _System_InstalledPack row exists (first-ever failed write for this pack), a // fresh non-empty PackInstallId is minted to satisfy the NOT NULL constraint. func TestMarkInstalledPackFailed_MintsWhenNoPriorRow(t *testing.T) { db, mock, err := sqlmock.New(sqlmock.QueryMatcherOption(sqlmock.QueryMatcherRegexp)) if err != nil { t.Fatalf("sqlmock: %v", err) } defer db.Close() pack := minimalPack{name: "new-connector", category: packs.ConnectorPack} // namespace-reservation SELECT (no existing row) expectNamespaceReservationLookup(mock, "new-connector", nil) // SELECT for existing PackInstallId — returns no row. mock.ExpectQuery(`SELECT COALESCE\(PackInstallId, ''\) FROM _System_InstalledPack WHERE Id = \? LIMIT 1`). WithArgs(installedPackID("new-connector")). WillReturnError(errors.New("sql: no rows in result set")) mock.ExpectExec(`UPDATE _System_ConnectorCapabilityDeclaration`). WithArgs("new-connector"). WillReturnResult(sqlmock.NewResult(0, 0)) mock.ExpectExec(`UPDATE _System_ConnectorRateLimit`). WithArgs("new-connector"). WillReturnResult(sqlmock.NewResult(0, 0)) mock.ExpectExec(`UPDATE _System_ConnectorInboundHandlerConfig`). WithArgs("new-connector"). WillReturnResult(sqlmock.NewResult(0, 0)) mock.ExpectExec(`UPDATE _System_ConnectorCalloutProfile`). WithArgs("new-connector"). WillReturnResult(sqlmock.NewResult(0, 0)) var capturedInstallID string cap := captureArg{captured: &capturedInstallID} expectInstalledPackRuntimeDeclarationColumns(mock, false) mock.ExpectExec(`INSERT INTO _System_InstalledPack`). WithArgs( sqlmock.AnyArg(), // Id sqlmock.AnyArg(), // Name cap, // PackInstallId — must be non-empty fresh UUID sqlmock.AnyArg(), // Category sqlmock.AnyArg(), // Status sqlmock.AnyArg(), // PackVersion sqlmock.AnyArg(), // ManifestFormatVersion sqlmock.AnyArg(), // PlatformCompatibility sqlmock.AnyArg(), // DesiredPayloadHash sqlmock.AnyArg(), // PayloadHashAlgorithm sqlmock.AnyArg(), // SourceKind sqlmock.AnyArg(), // SourceURI sqlmock.AnyArg(), // SourceRevision sqlmock.AnyArg(), // LastError (msg) sqlmock.AnyArg(), // CreatedDate sqlmock.AnyArg(), // LastModifiedDate ). WillReturnResult(sqlmock.NewResult(0, 1)) if err := markInstalledPackFailed(context.Background(), db, pack, "hash-v1-failed", errors.New("first install exploded")); err != nil { t.Fatalf("markInstalledPackFailed: %v", err) } if len(capturedInstallID) < 8 { t.Fatalf("markInstalledPackFailed minted PackInstallId=%q (too short / empty) on first-ever row; want a valid UUID", capturedInstallID) } if err2 := mock.ExpectationsWereMet(); err2 != nil { t.Fatalf("sql expectations: %v", err2) } } func TestMarkInstalledPackFailed_ReturnsConnectorCleanupError(t *testing.T) { db, mock, err := sqlmock.New(sqlmock.QueryMatcherOption(sqlmock.QueryMatcherRegexp)) if err != nil { t.Fatalf("sqlmock: %v", err) } defer db.Close() pack := minimalPack{name: "acme-connector", category: packs.ConnectorPack} expectNamespaceReservationLookup(mock, "acme-connector", []driver.Value{"connector-pack", 0}) mock.ExpectQuery(`SELECT COALESCE\(PackInstallId, ''\) FROM _System_InstalledPack WHERE Id = \? LIMIT 1`). WithArgs(installedPackID("acme-connector")). WillReturnRows(sqlmock.NewRows([]string{"PackInstallId"}).AddRow("pack-install-1")) mock.ExpectExec(`UPDATE _System_ConnectorCapabilityDeclaration`). WithArgs("acme-connector"). WillReturnResult(sqlmock.NewResult(0, 1)) mock.ExpectExec(`UPDATE _System_ConnectorRateLimit`). WithArgs("acme-connector"). WillReturnResult(sqlmock.NewResult(0, 1)) mock.ExpectExec(`UPDATE _System_ConnectorInboundHandlerConfig`). WithArgs("acme-connector"). WillReturnError(errors.New("handler config cleanup write failed")) mock.ExpectExec(`UPDATE _System_ConnectorCalloutProfile`). WithArgs("acme-connector"). WillReturnResult(sqlmock.NewResult(0, 1)) err = markInstalledPackFailed(context.Background(), db, pack, "hash-v1-failed", errors.New("install exploded")) if err == nil { t.Fatal("markInstalledPackFailed = nil, want cleanup error") } if !strings.Contains(err.Error(), "_System_ConnectorInboundHandlerConfig") && !strings.Contains(err.Error(), "handler config cleanup write failed") { t.Fatalf("markInstalledPackFailed error = %v, want handler config cleanup failure", err) } if err2 := mock.ExpectationsWereMet(); err2 != nil { t.Fatalf("sql expectations: %v", err2) } } func TestListInstalledSourceBackedPacksReturnsAppAndConnectorPacks(t *testing.T) { db, mock, err := sqlmock.New() if err != nil { t.Fatalf("sqlmock: %v", err) } defer db.Close() cols := []string{ "Name", "Category", "Status", "PackVersion", "ManifestFormatVersion", "PlatformCompatibility", "DesiredPayloadHash", "AppliedPayloadHash", "PayloadHashAlgorithm", "SourceKind", "SourceURI", "SourceRevision", "HostRequirementsDigest", } appRow := []driver.Value{ "alpha-app", string(packs.AppPack), installedPackStatusActive, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, } connRow := []driver.Value{ "beta-connector", string(packs.ConnectorPack), installedPackStatusActive, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, } expectInstalledPackRuntimeDeclarationColumns(mock, false) mock.ExpectQuery("SELECT Name, Category, Status"). WithArgs(string(packs.AppPack), string(packs.ConnectorPack)). WillReturnRows(sqlmock.NewRows(cols).AddRow(appRow...).AddRow(connRow...)) got, err := listInstalledSourceBackedPacks(context.Background(), db) if err != nil { t.Fatalf("listInstalledSourceBackedPacks: %v", err) } if len(got) != 2 { t.Fatalf("listInstalledSourceBackedPacks len = %d, want 2", len(got)) } if got[0].Name != "alpha-app" { t.Errorf("got[0].Name = %q, want alpha-app", got[0].Name) } if got[1].Name != "beta-connector" { t.Errorf("got[1].Name = %q, want beta-connector", got[1].Name) } if got[0].Category != string(packs.AppPack) { t.Errorf("got[0].Category = %q, want app-pack", got[0].Category) } if got[1].Category != string(packs.ConnectorPack) { t.Errorf("got[1].Category = %q, want connector-pack", got[1].Category) } if err2 := mock.ExpectationsWereMet(); err2 != nil { t.Fatalf("sql expectations: %v", err2) } } func TestListInstalledSourceBackedPacksExcludesPlatformCore(t *testing.T) { db, mock, err := sqlmock.New() if err != nil { t.Fatalf("sqlmock: %v", err) } defer db.Close() cols := []string{ "Name", "Category", "Status", "PackVersion", "ManifestFormatVersion", "PlatformCompatibility", "DesiredPayloadHash", "AppliedPayloadHash", "PayloadHashAlgorithm", "SourceKind", "SourceURI", "SourceRevision", "HostRequirementsDigest", } expectInstalledPackRuntimeDeclarationColumns(mock, false) mock.ExpectQuery("SELECT Name, Category, Status"). WithArgs(string(packs.AppPack), string(packs.ConnectorPack)). WillReturnRows(sqlmock.NewRows(cols)) got, err := listInstalledSourceBackedPacks(context.Background(), db) if err != nil { t.Fatalf("listInstalledSourceBackedPacks empty: %v", err) } if len(got) != 0 { t.Fatalf("listInstalledSourceBackedPacks len = %d, want 0", len(got)) } if err2 := mock.ExpectationsWereMet(); err2 != nil { t.Fatalf("sql expectations: %v", err2) } } // TestVerifyInstalledSourceBackedPacksIncludesConnectorPack confirms that // verifyInstalledAppPacks (the verify lifecycle function) now enumerates // connector-pack rows in addition to app-pack rows. The test drives a // connector-pack row with no source identity through the verify path and // asserts that (a) the DB query uses Category IN (app-pack, connector-pack) // and (b) the function returns an error that names the connector pack, // proving the row was inspected rather than silently skipped. func TestVerifyInstalledSourceBackedPacksIncludesConnectorPack(t *testing.T) { rawDB, mock, err := sqlmock.New() if err != nil { t.Fatalf("sqlmock: %v", err) } defer rawDB.Close() cols := []string{ "Name", "Category", "Status", "PackVersion", "ManifestFormatVersion", "PlatformCompatibility", "DesiredPayloadHash", "AppliedPayloadHash", "PayloadHashAlgorithm", "SourceKind", "SourceURI", "SourceRevision", "HostRequirementsDigest", } // Connector pack with no source identity: SourceKind and SourceURI are NULL. // ResolveInstalledPack returns "has no source identity" for this row, which // proves the enumerate-and-inspect path reached it. expectInstalledPackRuntimeDeclarationColumns(mock, false) mock.ExpectQuery("SELECT Name, Category, Status"). WithArgs(string(packs.AppPack), string(packs.ConnectorPack)). WillReturnRows(sqlmock.NewRows(cols).AddRow( "test-connector", string(packs.ConnectorPack), installedPackStatusActive, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, // SourceKind, SourceURI, SourceRevision, HostRequirementsDigest all NULL )) db := database.NewConnectionFromDB(rawDB) runner := &Runner{DB: db} err = verifyInstalledAppPacks(context.Background(), db, runner, false) if err == nil { t.Fatal("verifyInstalledAppPacks = nil, want error for connector pack with no source identity") } if !strings.Contains(err.Error(), "test-connector") { t.Fatalf("verifyInstalledAppPacks error = %v, want error naming test-connector", err) } if err2 := mock.ExpectationsWereMet(); err2 != nil { t.Fatalf("sql expectations: %v", err2) } } func TestVerifyInstalledAppPacksRejectsUnmanagedOwnedApprovalProcess(t *testing.T) { rawDB, mock, err := sqlmock.New() if err != nil { t.Fatalf("sqlmock: %v", err) } defer rawDB.Close() cols := []string{ "Name", "Category", "Status", "PackVersion", "ManifestFormatVersion", "PlatformCompatibility", "DesiredPayloadHash", "AppliedPayloadHash", "PayloadHashAlgorithm", "SourceKind", "SourceURI", "SourceRevision", "HostRequirementsDigest", } expectInstalledPackRuntimeDeclarationColumns(mock, false) mock.ExpectQuery("SELECT Name, Category, Status"). WithArgs(string(packs.AppPack), string(packs.ConnectorPack)). WillReturnRows(sqlmock.NewRows(cols).AddRow( "salesforce-crm-standard", string(packs.AppPack), installedPackStatusActive, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, )) mock.ExpectQuery(regexp.QuoteMeta(` SELECT Id FROM ProcessDefinition WHERE COALESCE(OwnerPack, '') = ? ORDER BY Id LIMIT 1`)). WithArgs("salesforce-crm-standard"). WillReturnRows(sqlmock.NewRows([]string{"Id"}).AddRow("pd-stale")) db := database.NewConnectionFromDB(rawDB) err = verifyInstalledAppPacks(context.Background(), db, nil, false) if err == nil || !strings.Contains(err.Error(), "managed approval coverage") || !strings.Contains(err.Error(), "no OwnerPack-scoped managed approval_process registration covers ProcessDefinition") { t.Fatalf("verifyInstalledAppPacks error = %v, want unmanaged approval ownership failure", err) } if err := mock.ExpectationsWereMet(); err != nil { t.Fatalf("sql expectations: %v", err) } } func TestVerifyInstalledSourceBackedPacksRejectsInvalidConnectorHandlerSemanticConfig(t *testing.T) { root := writeConnectorPackSourceRoot(t) pack, err := packs.ResolveInstalledSourceBackedPack( "test-connector", packs.ConnectorPack, string(packs.PackSourceKindLocalPath), root, ) if err != nil { t.Fatalf("ResolveInstalledSourceBackedPack: %v", err) } payloadHash, err := packs.PackPayloadHash(pack) if err != nil { t.Fatalf("PackPayloadHash: %v", err) } meta := pack.PackHashMetadata() rawDB, mock, err := sqlmock.New() if err != nil { t.Fatalf("sqlmock: %v", err) } defer rawDB.Close() cols := []string{ "Name", "Category", "Status", "PackVersion", "ManifestFormatVersion", "PlatformCompatibility", "DesiredPayloadHash", "AppliedPayloadHash", "PayloadHashAlgorithm", "SourceKind", "SourceURI", "SourceRevision", "HostRequirementsDigest", } expectInstalledPackRuntimeDeclarationColumns(mock, false) mock.ExpectQuery("SELECT Name, Category, Status"). WithArgs(string(packs.AppPack), string(packs.ConnectorPack)). WillReturnRows(sqlmock.NewRows(cols).AddRow( "test-connector", string(packs.ConnectorPack), installedPackStatusActive, meta.PackVersion, meta.ManifestFormatVersion, meta.PlatformCompatibility, payloadHash, payloadHash, packs.PackPayloadHashAlgorithmV1, packs.PackSourceKindLocalPath, meta.SourceURI, nil, nil, )) mock.ExpectQuery("SELECT p.Name, d.Namespace, d.Kind"). WithArgs("test-connector"). WillReturnRows(sqlmock.NewRows([]string{"Name", "Namespace", "Kind", "PackInstallId"})) mock.ExpectQuery("SELECT c.Namespace, c.CapabilityKind, c.ConfigJSON"). WithArgs("test-connector"). WillReturnRows(sqlmock.NewRows([]string{"Namespace", "CapabilityKind", "ConfigJSON", "InstalledPackId"}). AddRow("test-connector", "signature_verifier", `{"scheme":"hmac-sha256-template","signingTemplate":"{Timestamp}.{RawBody}","timestampHeader":"X-Test-Signature","signatureHeader":"X-Test-Signature","signatureEncoding":"hex","skewWindow":"5m","secretName":"default"}`, "pack-install-1")) db := database.NewConnectionFromDB(rawDB) err = verifyInstalledAppPacks(context.Background(), db, nil, false) if err == nil { t.Fatal("verifyInstalledAppPacks = nil, want invalid connector handler semantic config") } for _, want := range []string{ "installed source-backed pack test-connector live connector inbound handlers", "TimestampHeader and SignatureHeader must differ", } { if !strings.Contains(err.Error(), want) { t.Fatalf("verifyInstalledAppPacks error = %v, want %q", err, want) } } if err2 := mock.ExpectationsWereMet(); err2 != nil { t.Fatalf("sql expectations: %v", err2) } } func TestVerifyInstalledAppPacksRejectsTruncatedStartupShallowDeclarationInNonDeepVerify(t *testing.T) { root := writeConnectorPackSourceRoot(t) pack, err := packs.ResolveInstalledSourceBackedPack( "test-connector", packs.ConnectorPack, string(packs.PackSourceKindLocalPath), root, ) if err != nil { t.Fatalf("ResolveInstalledSourceBackedPack: %v", err) } payloadHash, err := packs.PackPayloadHash(pack) if err != nil { t.Fatalf("PackPayloadHash: %v", err) } meta := pack.PackHashMetadata() decl, err := packs.BuildPersistedRuntimeDeclaration(pack) if err != nil { t.Fatalf("BuildPersistedRuntimeDeclaration: %v", err) } rawDB, mock, err := sqlmock.New() if err != nil { t.Fatalf("sqlmock: %v", err) } defer rawDB.Close() expectInstalledPackRuntimeDeclarationColumns(mock, true) expectInstalledPackRuntimeStartupShallowDeclarationColumn(mock, true) mock.ExpectQuery("SELECT Name, Category, Status, PackVersion, ManifestFormatVersion, PlatformCompatibility, DesiredPayloadHash, AppliedPayloadHash, PayloadHashAlgorithm, SourceKind, SourceURI, SourceRevision, HostRequirementsDigest, RuntimeObjectSpecDeclarations, RuntimeRelatedListRules, RuntimeSearchScopeOrder, RuntimeSearchAllDefaultFields, RuntimeStartupShallowDeclarations"). WithArgs(string(packs.AppPack), string(packs.ConnectorPack)). WillReturnRows(sqlmock.NewRows([]string{ "Name", "Category", "Status", "PackVersion", "ManifestFormatVersion", "PlatformCompatibility", "DesiredPayloadHash", "AppliedPayloadHash", "PayloadHashAlgorithm", "SourceKind", "SourceURI", "SourceRevision", "HostRequirementsDigest", "RuntimeObjectSpecDeclarations", "RuntimeRelatedListRules", "RuntimeSearchScopeOrder", "RuntimeSearchAllDefaultFields", "RuntimeStartupShallowDeclarations", }).AddRow( "test-connector", string(packs.ConnectorPack), installedPackStatusActive, meta.PackVersion, meta.ManifestFormatVersion, meta.PlatformCompatibility, payloadHash, payloadHash, packs.PackPayloadHashAlgorithmV1, packs.PackSourceKindLocalPath, meta.SourceURI, nil, nil, decl.ObjectPayloadsJSON(), decl.RelatedListRulesJSON(), decl.SearchScopeOrderJSON(), decl.SearchAllDefaultFieldsJSON(), `[]`, )) mock.ExpectQuery("SELECT RuntimeStartupShallowDeclarations"). WithArgs("test-connector"). WillReturnRows(sqlmock.NewRows([]string{"RuntimeStartupShallowDeclarations"}).AddRow(`[]`)) mock.ExpectQuery("SELECT Namespace, Kind\\s+FROM _System_ConnectorCapabilityDeclaration"). WithArgs("test-connector"). WillReturnRows(sqlmock.NewRows([]string{"Namespace", "Kind"})) mock.ExpectQuery("SELECT Name\\s+FROM _System_WebhookSecret"). WithArgs("test-connector"). WillReturnRows(sqlmock.NewRows([]string{"Name"})) mock.ExpectQuery("SELECT EndpointClass\\s+FROM _System_ConnectorRateLimit"). WithArgs("test-connector"). WillReturnRows(sqlmock.NewRows([]string{"EndpointClass"})) mock.ExpectQuery("SELECT ProfileName\\s+FROM _System_ConnectorCalloutProfile"). WithArgs("test-connector"). WillReturnRows(sqlmock.NewRows([]string{"ProfileName"})) mock.ExpectQuery("SELECT CapabilityKind\\s+FROM _System_ConnectorInboundHandlerConfig"). WithArgs("test-connector"). WillReturnRows(sqlmock.NewRows([]string{"CapabilityKind"})) mock.ExpectQuery("SELECT r.EventKind\\s+FROM _System_ConnectorInboundRoute r\\s+JOIN _System_InstalledPack p ON p.PackInstallId = r.InstalledPackId"). WithArgs("test-connector", "test-connector"). WillReturnRows(sqlmock.NewRows([]string{"EventKind"})) err = verifyInstalledAppPacks(context.Background(), database.NewConnectionFromDB(rawDB), nil, false) if err == nil { t.Fatal("verifyInstalledAppPacks = nil, want startup-shallow source-identity drift error") } for _, want := range []string{ "installed source-backed pack test-connector", "persisted runtime declaration drifted from source identity", } { if !strings.Contains(err.Error(), want) { t.Fatalf("verifyInstalledAppPacks error = %v, want %q", err, want) } } if err := mock.ExpectationsWereMet(); err != nil { t.Fatalf("sql expectations: %v", err) } } func TestVerifyInstalledAppPacksRunsFullVerifyAfterStartupShallow(t *testing.T) { root := writeConnectorPackSourceRoot(t) pack, err := packs.ResolveInstalledSourceBackedPack( "test-connector", packs.ConnectorPack, string(packs.PackSourceKindLocalPath), root, ) if err != nil { t.Fatalf("ResolveInstalledSourceBackedPack: %v", err) } payloadHash, err := packs.PackPayloadHash(pack) if err != nil { t.Fatalf("PackPayloadHash: %v", err) } meta := pack.PackHashMetadata() decl, err := packs.BuildPersistedRuntimeDeclaration(pack) if err != nil { t.Fatalf("BuildPersistedRuntimeDeclaration: %v", err) } rawDB, mock, err := sqlmock.New() if err != nil { t.Fatalf("sqlmock: %v", err) } defer rawDB.Close() expectInstalledPackRuntimeDeclarationColumns(mock, true) expectInstalledPackRuntimeStartupShallowDeclarationColumn(mock, true) mock.ExpectQuery("SELECT Name, Category, Status, PackVersion, ManifestFormatVersion, PlatformCompatibility, DesiredPayloadHash, AppliedPayloadHash, PayloadHashAlgorithm, SourceKind, SourceURI, SourceRevision, HostRequirementsDigest, RuntimeObjectSpecDeclarations, RuntimeRelatedListRules, RuntimeSearchScopeOrder, RuntimeSearchAllDefaultFields, RuntimeStartupShallowDeclarations"). WithArgs(string(packs.AppPack), string(packs.ConnectorPack)). WillReturnRows(sqlmock.NewRows([]string{ "Name", "Category", "Status", "PackVersion", "ManifestFormatVersion", "PlatformCompatibility", "DesiredPayloadHash", "AppliedPayloadHash", "PayloadHashAlgorithm", "SourceKind", "SourceURI", "SourceRevision", "HostRequirementsDigest", "RuntimeObjectSpecDeclarations", "RuntimeRelatedListRules", "RuntimeSearchScopeOrder", "RuntimeSearchAllDefaultFields", "RuntimeStartupShallowDeclarations", }).AddRow( pack.Name(), string(packs.ConnectorPack), installedPackStatusActive, meta.PackVersion, meta.ManifestFormatVersion, meta.PlatformCompatibility, payloadHash, payloadHash, packs.PackPayloadHashAlgorithmV1, packs.PackSourceKindLocalPath, root, nil, nil, decl.ObjectPayloadsJSON(), decl.RelatedListRulesJSON(), decl.SearchScopeOrderJSON(), decl.SearchAllDefaultFieldsJSON(), decl.StartupShallowDeclaredRowsJSON(), )) mock.ExpectQuery("SELECT RuntimeStartupShallowDeclarations"). WithArgs("test-connector"). WillReturnRows(sqlmock.NewRows([]string{"RuntimeStartupShallowDeclarations"}).AddRow(decl.StartupShallowDeclaredRowsJSON())) mock.ExpectQuery("SELECT Namespace, Kind\\s+FROM _System_ConnectorCapabilityDeclaration"). WithArgs("test-connector"). WillReturnRows(sqlmock.NewRows([]string{"Namespace", "Kind"}).AddRow("test-connector", "signature_verifier")) mock.ExpectQuery("SELECT Name\\s+FROM _System_WebhookSecret"). WithArgs("test-connector"). WillReturnRows(sqlmock.NewRows([]string{"Name"})) mock.ExpectQuery("SELECT EndpointClass\\s+FROM _System_ConnectorRateLimit"). WithArgs("test-connector"). WillReturnRows(sqlmock.NewRows([]string{"EndpointClass"})) mock.ExpectQuery("SELECT ProfileName\\s+FROM _System_ConnectorCalloutProfile"). WithArgs("test-connector"). WillReturnRows(sqlmock.NewRows([]string{"ProfileName"})) mock.ExpectQuery("SELECT CapabilityKind\\s+FROM _System_ConnectorInboundHandlerConfig"). WithArgs("test-connector"). WillReturnRows(sqlmock.NewRows([]string{"CapabilityKind"}).AddRow("signature_verifier")) mock.ExpectQuery("SELECT r.EventKind\\s+FROM _System_ConnectorInboundRoute r\\s+JOIN _System_InstalledPack p ON p.PackInstallId = r.InstalledPackId"). WithArgs("test-connector", "test-connector"). WillReturnRows(sqlmock.NewRows([]string{"EventKind"})) mock.ExpectQuery("SELECT p.Name, d.Namespace, d.Kind, p.PackInstallId"). WithArgs("test-connector"). WillReturnRows(sqlmock.NewRows([]string{"Name", "Namespace", "Kind", "PackInstallId"})) mock.ExpectQuery("SELECT c.Namespace, c.CapabilityKind, c.ConfigJSON, c.InstalledPackId"). WithArgs("test-connector"). WillReturnRows(sqlmock.NewRows([]string{"Namespace", "CapabilityKind", "ConfigJSON", "InstalledPackId"}). AddRow("test-connector", "signature_verifier", `{"scheme":"hmac-sha256-template","signingTemplate":"{Timestamp}.{RawBody}","timestampHeader":"X-Test-Timestamp","signatureHeader":"X-Test-Signature","signatureEncoding":"hex","skewWindow":"5m","secretName":"default"}`, "pack-install-1")) mock.ExpectQuery("SELECT Id, InstalledPackName, Namespace, Kind, PayloadHash"). WithArgs("test-connector", "test-connector", "signature_verifier"). WillReturnRows(sqlmock.NewRows([]string{"Id", "InstalledPackName", "Namespace", "Kind", "PayloadHash", "IsDeleted"})) err = verifyInstalledAppPacks(context.Background(), database.NewConnectionFromDB(rawDB), nil, false) if err == nil { t.Fatal("verifyInstalledAppPacks = nil, want full pack verify drift after startup-shallow check") } for _, want := range []string{ "installed source-backed pack test-connector verify", "connector_capability", "test-connector:signature_verifier", } { if !strings.Contains(err.Error(), want) { t.Fatalf("verifyInstalledAppPacks error = %v, want %q", err, want) } } if err := mock.ExpectationsWereMet(); err != nil { t.Fatalf("sql expectations: %v", err) } }