package records

// persistence_parent_controlled_op_test.go
//
// Pins S7-Insert and S7-Update from PR #1242: PersistenceService.Insert /
// Update refuse to author an ObjectPermissions row whose SobjectType is a
// parent-controlled child. Closes the generic-CRUD bypass — POST/PATCH
// /sobjects/ObjectPermissions + composite + composite-tree paths all funnel
// through PersistenceService and now hit the same guard.
//
// Test mechanics follow the existing fifteen-site records-package pattern
// (real permissions.NewPermissionService + stub metadata) — no interface
// refactor. The metadata stub is wired via the existing stubUpsertMetaRepo
// with a non-nil `relationships` slice to make
// MetadataService.GetParentControlledObject resolve.

import (
	"context"
	stderrors "errors"
	"strings"
	"testing"

	eventFeature "github.com/nexuscrm/events"
	apperrors "github.com/nexuscrm/kernel/pkg/errors"
	"github.com/nexuscrm/kernel/pkg/formula"
	"github.com/nexuscrm/kernel/pkg/validation"
	metadataFeature "github.com/nexuscrm/metadata"
	"github.com/nexuscrm/permissions"
	"github.com/nexuscrm/shared/pkg/constants"
	"github.com/nexuscrm/shared/pkg/contract"
	"github.com/nexuscrm/shared/pkg/models"
)

// pcOPTestRepo extends stubUpsertRecordRepo with spy counters for the generic
// ObjectPermissions write paths.
type pcOPTestRepo struct {
	stubUpsertRecordRepo
	insertCalls                int
	updateCalls                int
	findOneCalls               int
	bulkInsertCalls            int
	bulkInsertWithLandingCalls int
	insertedTable              string
	insertedRecord             models.SObject
	existingRecords            map[string]bool
	findOneResult              models.SObject
	findOneErr                 error
}

func (r *pcOPTestRepo) Insert(ctx context.Context, t string, rec models.SObject) error {
	r.insertCalls++
	r.insertedTable = t
	r.insertedRecord = rec
	return nil
}

func (r *pcOPTestRepo) Update(ctx context.Context, t string, id string, u models.SObject) error {
	r.updateCalls++
	return nil
}

func (r *pcOPTestRepo) FindOne(ctx context.Context, t string, id string) (models.SObject, error) {
	r.findOneCalls++
	return r.findOneResult, r.findOneErr
}

func (r *pcOPTestRepo) GetLock(ctx context.Context, t string, id string) (models.SObject, error) {
	if r.findOneResult != nil {
		return cloneSObject(r.findOneResult), nil
	}
	if r.existingRecords != nil && r.existingRecords[t+":"+id] {
		return models.SObject{constants.FieldID: id}, nil
	}
	return nil, nil
}

func (r *pcOPTestRepo) Exists(ctx context.Context, t string, id string) (bool, error) {
	if r.existingRecords == nil {
		return false, nil
	}
	return r.existingRecords[t+":"+id], nil
}

func (r *pcOPTestRepo) BulkInsert(ctx context.Context, t string, recs []models.SObject, cols []string, opts interface{}) error {
	r.bulkInsertCalls++
	return nil
}

func (r *pcOPTestRepo) BulkInsertWithLanding(ctx context.Context, t string, recs []models.SObject, cols []string, opts interface{}) ([]bool, error) {
	r.bulkInsertWithLandingCalls++
	landed := make([]bool, len(recs))
	for i := range landed {
		landed[i] = true
	}
	return landed, nil
}

// newPCOPService builds a PersistenceService wired with:
//   - ObjectPermissions schema with a SobjectType string field (the column
//     our guard reads). Minimal so prepareInsertRequest's normalization
//     does not reject the row.
//   - A ChildPC schema declared as ControlledByParent, plus a Parent
//     schema and an IsSharingParent=true relationship row, so
//     MetadataService.GetParentControlledObject returns "Parent" for
//     ChildPC and "" for ChildOK / unknown names.
//   - Permissions repo that grants Create/Edit so prepareOperation passes.
func newPCOPService() (*PersistenceService, *pcOPTestRepo) {
	opSchema := &models.ObjectMetadata{
		APIName: constants.TableObjectPermissions,
		AccessPolicy: &models.ObjectAccessPolicy{
			WriteValidation: &models.RecordWriteValidationAccessPolicy{
				Validators: append(
					append([]models.RecordWriteValidationPolicy{
						{Phase: string(contract.RecordWriteValidationBeforeCreate), Capability: securityWriteValidateObjectPermissionsCapability},
						{Phase: string(contract.RecordWriteValidationBeforeUpdate), Capability: securityWriteValidateObjectPermissionsCapability},
					}, systemProfileOwnedImmutableWriteValidationPolicies()...),
					managedPermissionSetDefinitionWriteValidationPolicies()...,
				),
				OwnerResolution: &models.OwnerResolutionSpec{
					Paths: []models.OwnerResolutionPath{
						{
							Hops: []models.OwnerResolutionHop{
								{ViaField: "ParentId", ToObject: constants.TablePermissionSet, Gate: &models.OwnerPredicate{Field: constants.FieldPermissionSet_IsOwnedByProfile, Is: true}},
								{ViaField: "ProfileId", ToObject: constants.TableProfile, SystemWhen: &models.OwnerPredicate{Field: constants.FieldProfile_IsSystem, Is: true}},
							},
						},
						{
							Hops: []models.OwnerResolutionHop{
								{ViaField: "ParentId", ToObject: constants.TableProfile, SystemWhen: &models.OwnerPredicate{Field: constants.FieldProfile_IsSystem, Is: true}},
							},
						},
					},
				},
			},
		},
		Fields: []models.FieldMetadata{
			{APIName: constants.FieldID, Type: constants.FieldTypeString},
			{APIName: constants.FieldObjectPermissions_SobjectType, Type: constants.FieldTypeString, Createable: true, Updateable: true},
			{APIName: constants.FieldObjectPermissions_ParentID, Type: constants.FieldTypeString, Createable: true, Updateable: true},
			{APIName: constants.FieldObjectPermissions_PermissionsRead, Type: constants.FieldTypeBoolean, Createable: true, Updateable: true},
			{APIName: constants.FieldObjectPermissions_PermissionsCreate, Type: constants.FieldTypeBoolean, Createable: true, Updateable: true},
			{APIName: constants.FieldObjectPermissions_PermissionsEdit, Type: constants.FieldTypeBoolean, Createable: true, Updateable: true},
			{APIName: constants.FieldObjectPermissions_PermissionsDelete, Type: constants.FieldTypeBoolean, Createable: true, Updateable: true},
			{APIName: constants.FieldObjectPermissions_PermissionsViewAllRecords, Type: constants.FieldTypeBoolean, Createable: true, Updateable: true},
			{APIName: constants.FieldObjectPermissions_PermissionsModifyAllRecords, Type: constants.FieldTypeBoolean, Createable: true, Updateable: true},
		},
	}
	// Use master-detail FK on ChildPC so buildFieldDerivedRelationships emits
	// a base relationship with IsSharingParent=true (master-detail auto-flag).
	// Without a field-derived base relationship, the explicit `relationships`
	// override is ignored — overlayRelationshipConfig only updates pre-existing
	// keys, it does not add new ones. (See metadata_relationships.go:65-101.)
	childPC := &models.ObjectMetadata{
		APIName:      "ChildPC",
		SharingModel: constants.SharingModelControlledByParent,
		Fields: []models.FieldMetadata{
			{
				APIName:        "ParentId",
				Type:           constants.FieldTypeMasterDetail,
				IsMasterDetail: true,
				ReferenceTo:    []string{"Parent"},
			},
		},
	}
	childOK := &models.ObjectMetadata{APIName: "ChildOK"}
	parent := &models.ObjectMetadata{APIName: "Parent"}

	rel := &models.Relationship{
		ChildObjectAPIName:  "ChildPC",
		ParentObjectAPIName: "Parent",
		FieldAPIName:        "ParentId",
		IsSharingParent:     true,
	}

	metaRepo := &stubUpsertMetaRepo{
		schemas:       []*models.ObjectMetadata{opSchema, childPC, childOK, parent},
		relationships: []*models.Relationship{rel},
	}
	metaSvc := metadataFeature.NewMetadataService(metaRepo, nil, nil)
	permSvc := permissions.NewPermissionService(
		&stubUpsertPermRepo{grantEdit: true, grantCreate: true},
		metaSvc,
		nil,
	)

	recordRepo := &pcOPTestRepo{}
	ps := &PersistenceService{
		repo:                recordRepo,
		rollup:              noopRollupProcessor{},
		metadata:            metaSvc,
		permissions:         permSvc,
		eventBus:            &eventFeature.NoOpEventBus{},
		validator:           metadataFeature.NewValidationService(formula.NewEngine()),
		schemaValidator:     validation.NewSchemaValidator(),
		txManager:           inlineTxManager{},
		parentLaneAdmission: &fakeParentLaneAdmissionController{},
	}
	ps.registerBuiltInSecurityWriteValidators()
	return installNoOpPermissionStateAuthority(ps), recordRepo
}

type parentOnlyPermRepo struct {
	stubUpsertPermRepo
	parentObject string
	parentPerm   *models.ObjectPermissions
	systemPerms  map[string]bool
}

func (r *parentOnlyPermRepo) LoadEffectiveObjectPermission(ctx context.Context, user *models.UserSession, objectAPIName string) (*models.ObjectPermissions, error) {
	if r != nil && strings.EqualFold(objectAPIName, r.parentObject) {
		return r.parentPerm, nil
	}
	return nil, nil
}

func (r *parentOnlyPermRepo) ListEffectiveObjectPermissionsForUser(ctx context.Context, userID, profileID string) ([]models.ObjectPermissions, error) {
	if r == nil || r.parentPerm == nil {
		return nil, nil
	}
	return []models.ObjectPermissions{*r.parentPerm}, nil
}

func (r *parentOnlyPermRepo) GetEffectiveSystemPermissions(ctx context.Context, userID, profileID string) (map[string]bool, error) {
	if r == nil || r.systemPerms == nil {
		return map[string]bool{}, nil
	}
	out := make(map[string]bool, len(r.systemPerms))
	for key, value := range r.systemPerms {
		out[key] = value
	}
	return out, nil
}

func newPCOpportunityLineItemReproService() (*PersistenceService, *pcOPTestRepo) {
	parent := &models.ObjectMetadata{
		APIName:      "Opportunity",
		SharingModel: constants.SharingModelPrivate,
		Fields: []models.FieldMetadata{
			{APIName: constants.FieldID, Type: constants.FieldTypeString},
			{APIName: constants.FieldOwnerID, Type: constants.FieldTypeString},
		},
	}
	child := &models.ObjectMetadata{
		APIName:      "OpportunityLineItem",
		SharingModel: constants.SharingModelControlledByParent,
		Fields: []models.FieldMetadata{
			{APIName: constants.FieldID, Type: constants.FieldTypeString},
			{
				APIName:        "OpportunityId",
				Type:           constants.FieldTypeMasterDetail,
				Createable:     true,
				Updateable:     true,
				Required:       true,
				IsMasterDetail: true,
				ReferenceTo:    []string{"Opportunity"},
			},
			{APIName: "Quantity", Type: constants.FieldTypeDouble, Createable: true, Updateable: true},
		},
	}
	rel := &models.Relationship{
		ChildObjectAPIName:  "OpportunityLineItem",
		ParentObjectAPIName: "Opportunity",
		FieldAPIName:        "OpportunityId",
		IsSharingParent:     true,
	}

	metaRepo := &stubUpsertMetaRepo{
		schemas:       []*models.ObjectMetadata{parent, child},
		relationships: []*models.Relationship{rel},
	}
	metaSvc := metadataFeature.NewMetadataService(metaRepo, nil, nil)
	permSvc := permissions.NewPermissionService(&parentOnlyPermRepo{
		parentObject: "Opportunity",
		parentPerm: &models.ObjectPermissions{
			SobjectType:               "Opportunity",
			PermissionsRead:           true,
			PermissionsCreate:         true,
			PermissionsEdit:           true,
			PermissionsViewAllRecords: true,
		},
	}, metaSvc, nil)

	recordRepo := &pcOPTestRepo{
		existingRecords: map[string]bool{
			"Opportunity:opp1": true,
		},
		findOneResult: models.SObject{
			constants.FieldID:      "opp1",
			constants.FieldOwnerID: "u1",
		},
	}
	ps := &PersistenceService{
		repo:                recordRepo,
		rollup:              noopRollupProcessor{},
		metadata:            metaSvc,
		permissions:         permSvc,
		eventBus:            &eventFeature.NoOpEventBus{},
		validator:           metadataFeature.NewValidationService(formula.NewEngine()),
		schemaValidator:     validation.NewSchemaValidator(),
		txManager:           inlineTxManager{},
		parentLaneAdmission: &fakeParentLaneAdmissionController{},
	}
	return installNoOpPermissionStateAuthority(ps), recordRepo
}

// systemCtx returns a context with SystemContext so the test bypasses
// CRUD/FLS business gates. The S7 guard is parallel to those gates: it must
// fire regardless of context (admin writes ARE the bypass surface), so we
// use SystemContext only to keep the test focused on the guard.
//
// Rationale for using SystemContext in tests: ObjectPermissions is a system
// table and admin writes through generic CRUD originate from authenticated
// admin sessions whose context typically still has BypassesCRUDandFLS()
// false. The guard sits before that branch and is unconditional on context.
// This matches the design intent — block parent-controlled OP authoring
// from EVERY caller, including the bootstrap path that uses SystemContext.
func systemCtx() context.Context {
	return constants.WithSecurityContext(context.Background(), constants.SystemContext, "test")
}

// TestPersistenceInsertRejectsParentControlledObjectPermissionsRow pins
// S7-Insert: POST /sobjects/ObjectPermissions for a parent-controlled
// SobjectType returns *errors.ValidationError; no record is written.
func TestPersistenceInsertRejectsParentControlledObjectPermissionsRow(t *testing.T) {
	t.Run("parent-controlled child SobjectType is rejected", func(t *testing.T) {
		ps, repo := newPCOPService()
		_, err := ps.Insert(systemCtx(), constants.TableObjectPermissions, models.SObject{
			constants.FieldObjectPermissions_ParentID:    "ps1",
			constants.FieldObjectPermissions_SobjectType: "ChildPC",
		}, &models.UserSession{ID: "admin", ProfileID: constants.ProfileSystemAdmin})
		if err == nil {
			t.Fatal("Insert returned nil; want ValidationError for parent-controlled child")
		}
		var ve *apperrors.ValidationError
		if !stderrors.As(err, &ve) {
			t.Fatalf("want *apperrors.ValidationError, got %T: %v", err, err)
		}
		if repo.insertCalls != 0 {
			t.Fatalf("must not reach repo.Insert; insertCalls = %d", repo.insertCalls)
		}
	})

	t.Run("non-parent-controlled SobjectType reaches insert", func(t *testing.T) {
		ps, repo := newPCOPService()
		_, err := ps.Insert(systemCtx(), constants.TableObjectPermissions, models.SObject{
			constants.FieldObjectPermissions_ParentID:    "ps1",
			constants.FieldObjectPermissions_SobjectType: "ChildOK",
		}, &models.UserSession{ID: "admin", ProfileID: constants.ProfileSystemAdmin})
		if err != nil {
			t.Fatalf("non-PC SobjectType must reach insert without error from the guard; got %v", err)
		}
		if repo.insertCalls != 1 {
			t.Fatalf("want insertCalls == 1, got %d", repo.insertCalls)
		}
	})
}

// TestPersistenceInsertSafeOnMissingObjectPermissionsSobjectType pins
// S7-Insert-Safety (a) and (b): a missing or non-string SobjectType MUST
// NOT panic. The comma-ok short-circuit yields s == "" and the helper
// returns nil; the existing downstream validation path emits the canonical
// required-field / type-mismatch error (or the row reaches Insert if the
// stub validator permits — depends on schema completeness, but the load-
// bearing safety property is "no panic and no wrong-shape error from the
// PC guard itself").
func TestPersistenceInsertSafeOnMissingObjectPermissionsSobjectType(t *testing.T) {
	t.Run("missing SobjectType does not panic", func(t *testing.T) {
		ps, _ := newPCOPService()
		defer func() {
			if r := recover(); r != nil {
				t.Fatalf("panic on missing SobjectType: %v", r)
			}
		}()
		_, err := ps.Insert(systemCtx(), constants.TableObjectPermissions, models.SObject{
			constants.FieldObjectPermissions_ParentID: "ps1",
		}, &models.UserSession{ID: "admin", ProfileID: constants.ProfileSystemAdmin})
		// The guard must NOT emit its own ValidationError for missing SobjectType
		// (would be the wrong error shape, complaining about parent-controlled
		// when the actual issue is the malformed payload). If err is non-nil,
		// it should NOT carry the PC guard's hallmark field tag.
		if err != nil {
			var ve *apperrors.ValidationError
			if stderrors.As(err, &ve) && ve.Field == constants.FieldObjectPermissions_SobjectType {
				// Distinguish: our guard's ValidationError carries Field=SobjectType
				// AND a message about parent-controlled. The schema validator's
				// required-field ValidationError may ALSO carry Field=SobjectType,
				// so check message body for our guard's signature.
				if msg := ve.Message; contains(msg, "parent-controlled") {
					t.Fatalf("PC guard emitted ValidationError on missing SobjectType (wrong-shape error): %v", err)
				}
			}
		}
	})

	t.Run("non-string SobjectType does not panic", func(t *testing.T) {
		ps, _ := newPCOPService()
		defer func() {
			if r := recover(); r != nil {
				t.Fatalf("panic on non-string SobjectType: %v", r)
			}
		}()
		_, err := ps.Insert(systemCtx(), constants.TableObjectPermissions, models.SObject{
			constants.FieldObjectPermissions_ParentID:    "ps1",
			constants.FieldObjectPermissions_SobjectType: 42, // int — comma-ok yields "" and the guard short-circuits
		}, &models.UserSession{ID: "admin", ProfileID: constants.ProfileSystemAdmin})
		if err != nil {
			var ve *apperrors.ValidationError
			if stderrors.As(err, &ve) && contains(ve.Message, "parent-controlled") {
				t.Fatalf("PC guard fired on non-string SobjectType (wrong-shape error): %v", err)
			}
		}
	})
}

// TestPersistenceUpdateRejectsParentControlledObjectPermissionsRow pins
// S7-Update: PATCH paths through PersistenceService.Update use the registered
// final-image write validator instead of an ObjectPermissions-specific preflight.
// Three sub-cases:
//  1. Payload includes SobjectType=ChildPC → reject after the parent-lock planning
//     read and the locked old-row read.
//  2. Payload omits SobjectType, locked old row is ChildPC → reject after those reads.
//  3. Payload omits SobjectType, locked old row is ChildOK → success after those reads.
func TestPersistenceUpdateRejectsParentControlledObjectPermissionsRow(t *testing.T) {
	t.Run("payload SobjectType=ChildPC is rejected without FindOne", func(t *testing.T) {
		ps, repo := newPCOPService()
		repo.findOneResult = models.SObject{
			constants.FieldID: "row1",
			constants.FieldObjectPermissions_SobjectType: "ChildOK",
		}
		err := ps.Update(systemCtx(), constants.TableObjectPermissions, "row1", models.SObject{
			constants.FieldObjectPermissions_SobjectType: "ChildPC",
		}, &models.UserSession{ID: "admin", ProfileID: constants.ProfileSystemAdmin})
		if err == nil {
			t.Fatal("Update returned nil; want ValidationError")
		}
		var ve *apperrors.ValidationError
		if !stderrors.As(err, &ve) {
			t.Fatalf("want *apperrors.ValidationError, got %T: %v", err, err)
		}
		if !contains(ve.Message, "inherited from Parent") {
			t.Fatalf("validation error = %v, want parent-controlled child rejection", err)
		}
		if repo.findOneCalls != 2 {
			t.Fatalf("payload path old-row reads = %d, want planning plus locked reads", repo.findOneCalls)
		}
		if repo.updateCalls != 0 {
			t.Fatalf("must not reach repo.Update; updateCalls = %d", repo.updateCalls)
		}
	})

	t.Run("payload omits SobjectType, locked old row is ChildPC -> rejected by validator", func(t *testing.T) {
		ps, repo := newPCOPService()
		repo.findOneResult = models.SObject{
			constants.FieldID: "row1",
			constants.FieldObjectPermissions_SobjectType: "ChildPC",
		}
		err := ps.Update(systemCtx(), constants.TableObjectPermissions, "row1", models.SObject{
			constants.FieldObjectPermissions_PermissionsEdit: false,
		}, &models.UserSession{ID: "admin", ProfileID: constants.ProfileSystemAdmin})
		if err == nil {
			t.Fatal("Update returned nil; want ValidationError")
		}
		var ve *apperrors.ValidationError
		if !stderrors.As(err, &ve) {
			t.Fatalf("want *apperrors.ValidationError, got %T: %v", err, err)
		}
		if !contains(ve.Message, "inherited from Parent") {
			t.Fatalf("validation error = %v, want parent-controlled child rejection", err)
		}
		if repo.findOneCalls != 2 {
			t.Fatalf("write validator old-row reads = %d, want planning plus locked reads", repo.findOneCalls)
		}
		if repo.updateCalls != 0 {
			t.Fatalf("must not reach repo.Update; updateCalls = %d", repo.updateCalls)
		}
	})

	t.Run("payload omits SobjectType, locked old row is ChildOK -> guard passes", func(t *testing.T) {
		ps, repo := newPCOPService()
		repo.findOneResult = models.SObject{
			constants.FieldID: "row1",
			constants.FieldObjectPermissions_SobjectType: "ChildOK",
		}
		// Pass an empty updates map. The guard sits before any field validation;
		// if it passes, control flows to prepareUpdateRequest which is downstream
		// of our concern. Load-bearing assertion: the guard's specific error
		// (ValidationError with "parent-controlled" in the message) is NOT raised.
		err := ps.Update(systemCtx(), constants.TableObjectPermissions, "row1", models.SObject{},
			&models.UserSession{ID: "admin", ProfileID: constants.ProfileSystemAdmin})
		if err != nil {
			var ve *apperrors.ValidationError
			if stderrors.As(err, &ve) && contains(ve.Message, "inherited from Parent") {
				t.Fatalf("PC guard fired on non-PC existing row: %v", err)
			}
			// Any other downstream error is fine — we only care the guard let us pass.
		}
		if repo.findOneCalls != 2 {
			t.Fatalf("write validator old-row reads = %d, want planning plus locked reads", repo.findOneCalls)
		}
	})
}

func TestPersistenceBulkInsertRejectsParentControlledObjectPermissionsRows(t *testing.T) {
	t.Run("parent-controlled child SobjectType is rejected before repo bulk insert", func(t *testing.T) {
		ps, repo := newPCOPService()
		result, err := ps.BulkInsert(systemCtx(), constants.TableObjectPermissions, []models.SObject{{
			constants.FieldObjectPermissions_ParentID:    "ps1",
			constants.FieldObjectPermissions_SobjectType: "ChildPC",
		}}, &models.UserSession{ID: "admin", ProfileID: constants.ProfileSystemAdmin}, BulkInsertOptions{})
		if err == nil {
			t.Fatal("BulkInsert returned nil; want all-records-failed ValidationError")
		}
		var ve *apperrors.ValidationError
		if !stderrors.As(err, &ve) {
			t.Fatalf("want *apperrors.ValidationError, got %T: %v", err, err)
		}
		if result.FailedCount != 1 || len(result.Errors) != 1 ||
			!contains(result.Errors[0], "Access to ChildPC is inherited from Parent") ||
			!contains(result.Errors[0], "update the Parent object permissions") {
			t.Fatalf("unexpected bulk result: %#v", result)
		}
		if repo.bulkInsertCalls != 0 || repo.bulkInsertWithLandingCalls != 0 {
			t.Fatalf("must not reach repo bulk insert; bulk=%d landing=%d", repo.bulkInsertCalls, repo.bulkInsertWithLandingCalls)
		}
	})

	t.Run("non-parent-controlled SobjectType reaches repo bulk insert", func(t *testing.T) {
		ps, repo := newPCOPService()
		result, err := ps.BulkInsert(systemCtx(), constants.TableObjectPermissions, []models.SObject{{
			constants.FieldObjectPermissions_ParentID:    "ps1",
			constants.FieldObjectPermissions_SobjectType: "ChildOK",
		}}, &models.UserSession{ID: "admin", ProfileID: constants.ProfileSystemAdmin}, BulkInsertOptions{})
		if err != nil {
			t.Fatalf("BulkInsert returned error for non-PC row: %v", err)
		}
		if result.SuccessCount != 1 || result.FailedCount != 0 {
			t.Fatalf("unexpected bulk result: %#v", result)
		}
		if repo.bulkInsertCalls != 0 ||
			repo.bulkInsertWithLandingCalls != 1 {
			t.Fatalf(
				"strong bulk path calls = bulk %d landing %d, want 0/1",
				repo.bulkInsertCalls,
				repo.bulkInsertWithLandingCalls,
			)
		}
	})
}

func TestPersistenceInsertOpportunityLineItemUsesParentCRUDWithoutChildObjectPermissionsRow(t *testing.T) {
	ps, repo := newPCOpportunityLineItemReproService()
	user := &models.UserSession{ID: "u1", ProfileID: "p1"}

	inserted, err := ps.Insert(context.Background(), "OpportunityLineItem", models.SObject{
		"OpportunityId": "opp1",
		"Quantity":      float64(2),
	}, user)
	if err != nil {
		t.Fatalf("Insert OpportunityLineItem with only parent Opportunity ObjectPermissions row returned error: %v", err)
	}
	if repo.insertCalls != 1 {
		t.Fatalf("want exactly one repo.Insert call, got %d", repo.insertCalls)
	}
	if repo.insertedTable != "OpportunityLineItem" {
		t.Fatalf("inserted table = %q, want OpportunityLineItem", repo.insertedTable)
	}
	if got := repo.insertedRecord["OpportunityId"]; got != "opp1" {
		t.Fatalf("inserted OpportunityId = %v, want opp1", got)
	}
	if id, _ := inserted[constants.FieldID].(string); id == "" {
		t.Fatalf("inserted record missing generated Id: %#v", inserted)
	}
}

// contains is a tiny helper kept local to this test file (so we don't pull
// strings.Contains into the public test surface).
func contains(s, substr string) bool {
	for i := 0; i+len(substr) <= len(s); i++ {
		if s[i:i+len(substr)] == substr {
			return true
		}
	}
	return false
}