import json
import os
import subprocess
import tempfile
import unittest
from pathlib import Path


REPO_ROOT = Path(__file__).resolve().parents[1]

GO_RUNTIME_TEST = r'''
package automation

import (
    "context"
    "encoding/json"
    "os"
    "path/filepath"
    "reflect"
    "testing"

    kernelconnector "github.com/nexuscrm/kernel/pkg/connector"
    "github.com/nexuscrm/kernel/pkg/formula"
    "github.com/nexuscrm/shared/pkg/constants"
    "github.com/nexuscrm/shared/pkg/models"
    connector "github.com/pingcap-inc/nexuscrm-contract/pkg/connector"
)

func TestL2HubSpotOwnerCorrectionUsesTrustedBeforeSaveImage(t *testing.T) {
    root := os.Getenv("L2_OWNER_CONVERGENCE_ROOT")
    load := func(path string, target any) {
        t.Helper()
        raw, err := os.ReadFile(filepath.Join(root, "packs/PingCAP-HubSpot-Companion", path))
        if err != nil { t.Fatal(err) }
        if err := json.Unmarshal(raw, target); err != nil { t.Fatal(err) }
    }
    var flow models.FlowMetadata
    load("flows/HubSpot_Lead_Contact_Outbound_Update.flow.json", &flow)
    var decision *models.FlowDecision
    var assignment *models.FlowAssignment
    for i := range flow.Decisions {
        if flow.Decisions[i].Name == "SkipConnectorInboundEcho" { decision = &flow.Decisions[i] }
    }
    for i := range flow.Assignments {
        if flow.Assignments[i].Name == "SelectOwnerOnlyHubSpotPayload" { assignment = &flow.Assignments[i] }
    }
    if decision == nil || assignment == nil { t.Fatal("outbound routing metadata missing") }

    mappingJSON, err := os.ReadFile(filepath.Join(root, "packs/PingCAP-HubSpot-Companion/field_mappings/contact_lead_outbound.json"))
    if err != nil { t.Fatal(err) }
    parsedMappings, err := kernelconnector.ParseConnectorFieldMappingInputsFromPayload(mappingJSON, "pingcap-hubspot-companion")
    if err != nil { t.Fatal(err) }
    var mappings []connector.ConnectorFieldMappingInput
    for _, mapping := range parsedMappings {
        if mapping.SourceObject == "Lead" && mapping.ActionType == "PushToConnector" {
            mappings = append(mappings, mapping)
        }
    }
    if len(mappings) < 2 { t.Fatal("full Lead mapping not found") }
    var pushContract struct { Associations []json.RawMessage `json:"associations"` }
    load("connector_push_contracts/lead_contact_outbound.json", &pushContract)
    if len(pushContract.Associations) != 0 { t.Fatal("owner-only route would inherit contract associations") }

    for _, tc := range []struct {
        name string
        before bool
        submitted any
        noOwner, noRemote, foreignSnapshot, shell, spoof, localWrite, changedOld bool
        want string
    }{
        {name:"different remote owner with unchanged local shadow", before:true, submitted:"remote-drift", want:"SelectOwnerOnlyHubSpotPayload"},
        {name:"non-shell inbound mirror correction stays owner-only when prior shadow changed", before:true, submitted:"previous-owner", changedOld:true, want:"SelectOwnerOnlyHubSpotPayload"},
        {name:"cleared remote owner is restored", before:true, submitted:nil, want:"SelectOwnerOnlyHubSpotPayload"},
        {name:"corrected owner on required-field shell", before:true, submitted:"remote-drift", shell:true, want:"SelectOwnerOnlyHubSpotPayload"},
        {name:"matching owner echo terminates", before:true, submitted:"bound-owner"},
        {name:"legacy event without submitted image terminates"},
        {name:"caller supplied snapshot is ignored", spoof:true},
        {name:"another record snapshot is ignored", before:true, submitted:"remote-drift", foreignSnapshot:true},
        {name:"unbound current owner remains suppressed", before:true, submitted:"remote-drift", noOwner:true},
        {name:"missing remote identity remains suppressed", before:true, submitted:"remote-drift", noRemote:true},
        {name:"normal local update retains full outbound route", before:true, submitted:"remote-drift", localWrite:true, want:"PushLeadToHubSpotContact"},
        {name:"legacy owner assignment keeps existing route", changedOld:true, want:"PushLeadToHubSpotContact"},
        {name:"legacy shell owner assignment stays sparse", changedOld:true, shell:true, want:"SelectOwnerOnlyHubSpotPayload"},
    } {
        t.Run(tc.name, func(t *testing.T) {
            record := models.SObject{
                "Id":"lead-current", "OwnerId":"crm-owner-current", "HubSpot_Id__c":"contact-current",
                "HubSpot_Owner_Id__c":"bound-owner", "LastName":"Contact", "Company":"Example",
                "Recent_Source_Detail__c":"Other",
            }
            if tc.noOwner { record["HubSpot_Owner_Id__c"] = "" }
            if tc.noRemote { record["HubSpot_Id__c"] = "" }
            if tc.shell { record["LastName"], record["Company"] = "HubSpot Lead", "HubSpot Lead" }
            old := cloneRuntimeSObject(record)
            if tc.changedOld { old["HubSpot_Owner_Id__c"] = "previous-owner" }
            var before models.SObject
            if tc.before {
                before = cloneRuntimeSObject(record)
                before["HubSpot_Owner_Id__c"] = tc.submitted
                if tc.foreignSnapshot { before["Id"] = "different-lead" }
            }
            savedRecord, savedOld := cloneRuntimeSObject(record), cloneRuntimeSObject(old)
            ctx := context.Background()
            if !tc.localWrite { ctx = context.WithValue(ctx, constants.ContextKeyConnectorSyncOrigin, "hubspot-crm-connector") }
            runtime := NewFlowRuntimeContext(ctx, nil).
                WithRecordTriggerNewSnapshot(record).
                WithRecordTriggerOldSnapshot(old).
                WithRecordBeforeSaveSnapshot(before)
            ctx = withFlowRuntimeContext(ctx, runtime)
            interview := &models.FlowInterview{Id:"owner-policy", ContextData:map[string]interface{}{
                "$Record":record, "$Record__Old":old, "$Record__Prior":old,
            }}
            if tc.spoof {
                interview.ContextData["$Record__BeforeSave"] = models.SObject{"Id":"lead-current", "HubSpot_Owner_Id__c":"spoofed"}
            }
            exec := &FlowExecutor{}
            next, err := exec.executeDecision(ctx, interview, decision, nil)
            if err != nil || next != tc.want { t.Fatalf("next=%q, err=%v; want %q", next, err, tc.want) }
            if next == "SelectOwnerOnlyHubSpotPayload" {
                next, err = exec.executeAssignment(ctx, interview, &flow, assignment, nil)
                if err != nil || next != "PushLeadToHubSpotContact" { t.Fatalf("sparse route next=%q err=%v", next, err) }
                fields, err := parsePushToConnectorPayloadSourceFields(interview.ContextData["outboundPayloadSourceFields"])
                if err != nil { t.Fatal(err) }
                selected, err := selectPushToConnectorPayloadMappings(mappings, fields)
                if err != nil || len(selected) != 1 || selected[0].SourceField != "HubSpot_Owner_Id__c" || selected[0].TargetField != "hubspot_owner_id" {
                    t.Fatalf("selected mappings=%#v err=%v; want only owner", selected, err)
                }
                // This is the shared desired-field builder used by Runtime's
                // outbound intent writer. A stale local marketing value must
                // not reach the provider just because owner binding changed.
                engine := formula.NewEngine()
                mapped, err := buildConnectorFieldMappingUpdatesWithEngine(engine, record, models.SObject{}, selected, nil, false, nil)
                if err != nil || !reflect.DeepEqual(mapped, models.SObject{"hubspot_owner_id":"bound-owner"}) {
                    t.Fatalf("owner-only desired fields=%#v err=%v", mapped, err)
                }
                if tc.changedOld && tc.before {
                    full, err := buildConnectorFieldMappingUpdatesWithEngine(engine, record, models.SObject{}, mappings, nil, false, nil)
                    if err != nil || full["recent_source_details"] != "Other" {
                        t.Fatalf("fixture must expose stale data on the legacy full route: fields=%#v err=%v", full, err)
                    }
                    if old["OwnerId"] != record["OwnerId"] || before["OwnerId"] != record["OwnerId"] {
                        t.Fatal("incident changes only the owner mirror, not the canonical CRM owner")
                    }
                }
            }
            if !reflect.DeepEqual(record, savedRecord) || !reflect.DeepEqual(old, savedOld) {
                t.Fatal("outbound policy changed the current/prior CRM record")
            }
        })
    }
}
'''


class HubSpotOwnerConvergenceRuntimeTest(unittest.TestCase):
    def test_current_runtime_executes_owner_convergence_policy(self):
        upstream_value = os.environ.get("NEXUSCRM_ROOT")
        self.assertTrue(upstream_value, "NEXUSCRM_ROOT must identify the Core candidate")
        upstream = Path(upstream_value).resolve()
        package_dir = upstream / "nexuscrm-automation"
        self.assertTrue(package_dir.is_dir())
        expected_head = os.environ.get("NEXUSCRM_EXPECTED_HEAD")
        if expected_head:
            actual_head = subprocess.check_output(
                ["git", "rev-parse", "HEAD"], cwd=upstream, text=True
            ).strip()
            self.assertEqual(actual_head, expected_head)
        with tempfile.TemporaryDirectory(prefix="l2-owner-convergence-") as temporary:
            temporary = Path(temporary)
            source = temporary / "l2_owner_convergence_test.go"
            source.write_text(GO_RUNTIME_TEST)
            overlay = temporary / "overlay.json"
            overlay.write_text(json.dumps({"Replace": {
                str(package_dir / "l2_owner_convergence_test.go"): str(source)
            }}))
            completed = subprocess.run(
                ["go", "test", f"-overlay={overlay}", ".", "-run",
                 "^TestL2HubSpotOwnerCorrectionUsesTrustedBeforeSaveImage$", "-count=1"],
                cwd=package_dir,
                env={**os.environ, "GOWORK": "off", "L2_OWNER_CONVERGENCE_ROOT": str(REPO_ROOT)},
                capture_output=True,
                text=True,
                check=False,
            )
        self.assertEqual(completed.returncode, 0,
                         f"go test failed\n{completed.stdout}\n{completed.stderr}")


if __name__ == "__main__":
    unittest.main()