import unittest
from pathlib import Path


ROOT = Path(__file__).resolve().parents[1]
WORKFLOW = ROOT / ".github" / "workflows" / "integration-test-hubspot.yml"


class HubSpotIntegrationWorkflowTests(unittest.TestCase):
    def test_live_workflow_resolves_and_records_latest_acceptance_main(self):
        workflow = WORKFLOW.read_text(encoding="utf-8")

        self.assertIn(
            "HUBSPOT_ACCEPTANCE_REPOSITORY: pingcap-inc/nexuscrm-hubspot-acceptance",
            workflow,
        )
        self.assertIn("HUBSPOT_ACCEPTANCE_REF: main", workflow)
        self.assertNotIn("inputs.acceptance_ref", workflow)
        self.assertNotRegex(
            workflow,
            r"HUBSPOT_ACCEPTANCE_SHA:\s*[0-9a-f]{40}",
        )
        self.assertIn("repository: ${{ env.HUBSPOT_ACCEPTANCE_REPOSITORY }}", workflow)
        self.assertIn("ref: ${{ env.HUBSPOT_ACCEPTANCE_REF }}", workflow)
        self.assertIn("fetch-depth: 1", workflow)
        self.assertIn("token: ${{ secrets.CROSS_REPO_READ_TOKEN }}", workflow)
        self.assertIn('actual_sha="$(git -C hubspot-acceptance rev-parse HEAD)"', workflow)
        self.assertIn('[[ "${actual_sha}" =~ ^[0-9a-f]{40}$ ]]', workflow)
        self.assertIn(
            'echo "HUBSPOT_ACCEPTANCE_SHA=${actual_sha}" >> "${GITHUB_ENV}"',
            workflow,
        )
        self.assertIn(
            'echo "acceptance_sha=${actual_sha}" >> "${GITHUB_OUTPUT}"',
            workflow,
        )
        self.assertIn("node hubspot-acceptance/scripts/validate-catalog.mjs", workflow)
        self.assertLess(
            workflow.index('name: "Resolve and validate HubSpot acceptance catalog"'),
            workflow.index('name: "Execute HubSpot integration test"'),
        )
        self.assertIn("working-directory: hubspot-acceptance/runner", workflow)
        self.assertIn("NEXUSCRM_L2_ROOT: ${{ github.workspace }}", workflow)
        self.assertNotIn("working-directory: integration-test/runner", workflow)

    def test_live_execution_requires_exact_sandbox_for_runtime_and_runner_tokens(self):
        workflow = WORKFLOW.read_text(encoding="utf-8")

        credential_job = workflow.split(
            "  verify-target-hubspot-runtime-credential:", 1
        )[1].split("\n  restore-staging-hubspot-inbound:", 1)[0]
        self.assertIn("inputs.dry_run != true", credential_job)
        self.assertIn(
            "format('[\"self-hosted\",\"Linux\",\"X64\",\"{0}-ec2\"]'",
            credential_job,
        )
        self.assertIn('[[ "${GITHUB_REF}" == "refs/heads/main" ]]', credential_job)
        self.assertIn("infer_active_d3_slot_from_live_conf", credential_job)
        self.assertIn("/envs/test/${target_env}/env/runtime.env", credential_job)
        self.assertIn("hubspot_credential_admission.py", credential_job)
        self.assertIn('--target-env "${TARGET_ENV}"', credential_job)
        self.assertIn("--verify-only", credential_job)

        integration_job = workflow.split("  integration-test:", 1)[1]
        self.assertIn("- verify-target-hubspot-runtime-credential", integration_job)
        self.assertIn(
            "needs.verify-target-hubspot-runtime-credential.result == 'success'",
            integration_job,
        )
        self.assertIn('HUBSPOT_EXPECTED_PORTAL_ID: "51280875"', integration_job)

        runner_token_step = 'name: "Verify selected HubSpot runner token"'
        execute_step = 'name: "Execute HubSpot integration test"'
        self.assertIn(runner_token_step, integration_job)
        self.assertIn(
            "scripts/verify_staging_hubspot_account.py --token-env HUBSPOT_STAGING_TOKEN",
            integration_job,
        )
        self.assertLess(
            integration_job.index(runner_token_step),
            integration_job.index(execute_step),
        )

    def test_provider_write_smoke_keeps_sandbox_gates_without_crm_module_preflights(self):
        workflow = WORKFLOW.read_text(encoding="utf-8")

        self.assertIn("- provider_write_smoke", workflow)
        self.assertIn(
            'args+=(--case "admin/hubspot/provider-write-smoke")',
            workflow,
        )

        credential_job = workflow.split(
            "  verify-target-hubspot-runtime-credential:", 1
        )[1].split("\n  restore-staging-hubspot-inbound:", 1)[0]
        self.assertNotIn("provider_write_smoke", credential_job)

        restore_job = workflow.split(
            "  restore-staging-hubspot-inbound:", 1
        )[1].split("\n  integration-test:", 1)[0]
        self.assertIn(
            "inputs.suite_mode != 'provider_write_smoke'",
            restore_job,
        )

        integration_job = workflow.split("  integration-test:", 1)[1]
        self.assertIn('name: "Verify selected HubSpot runner token"', integration_job)
        for step_name in (
            'name: "Bind installed Salesforce-compatible OAuth client"',
            'name: "Restore legacy-slot HubSpot inbound webhook secret"',
            'name: "Verify HubSpot inbound webhook endpoint"',
            'name: "Preflight HubSpot live route and connector queue"',
        ):
            step = integration_job.split(step_name, 1)[1].split("\n      - name:", 1)[0]
            self.assertIn(
                "inputs.suite_mode != 'provider_write_smoke'",
                step,
                step_name,
            )

    def test_target_lock_prevents_reset_after_credential_verification(self):
        workflow = WORKFLOW.read_text(encoding="utf-8")
        concurrency = workflow.split("\nconcurrency:\n", 1)[1].split("\njobs:", 1)[0]

        self.assertIn(
            "inputs.suite_mode == 'pressure' && "
            "'deploy-test-multi-slot-environment-test1'",
            concurrency,
        )
        self.assertIn(
            "format('deploy-test-multi-slot-environment-{0}', inputs.target_env)",
            concurrency,
        )
        self.assertNotIn("integration-test-hubspot-", concurrency)

    def test_staging_restore_gates_integration_without_swallowing_cancellation(self):
        workflow = WORKFLOW.read_text(encoding="utf-8")

        self.assertIn(
            "inputs.target_env == 'staging' && "
            "'deploy-staging-blue-green-environment'",
            workflow,
        )
        restore_job = workflow.split(
            "  restore-staging-hubspot-inbound:", 1
        )[1].split("\n  integration-test:", 1)[0]
        self.assertIn("inputs.target_env == 'staging'", restore_job)
        self.assertIn("inputs.suite_mode != 'smoke'", restore_job)
        self.assertIn("inputs.case_id != ''", restore_job)
        self.assertIn("inputs.dry_run != true", restore_job)
        self.assertIn(
            "runs-on: [self-hosted, Linux, X64, staging-ec2]",
            restore_job,
        )
        self.assertIn('[[ "${GITHUB_REF}" == "refs/heads/main" ]]', restore_job)
        self.assertIn("infer_active_d3_slot_from_live_conf", restore_job)
        self.assertIn('runtime_env_file("staging", active_slot)', restore_job)
        self.assertIn(
            "scripts/set_staging_inspection_hubspot_inbound.py",
            restore_job,
        )
        self.assertIn("--mode restore", restore_job)
        self.assertNotIn("rotate_staging_hubspot_webhook.py", restore_job)

        integration_job = workflow.split("  integration-test:", 1)[1]
        self.assertIn("needs:", integration_job)
        self.assertIn("- restore-staging-hubspot-inbound", integration_job)
        job_gate = integration_job.split("    runs-on:", 1)[0]
        self.assertIn("!cancelled() &&", job_gate)
        self.assertNotIn("always()", job_gate)
        self.assertIn(
            "needs.restore-staging-hubspot-inbound.result == 'success'",
            integration_job,
        )
        self.assertIn(
            "needs.restore-staging-hubspot-inbound.result == 'skipped'",
            integration_job,
        )

    def test_direct_sfcrm_binds_the_programmatic_oauth_client_before_execution(self):
        workflow = WORKFLOW.read_text(encoding="utf-8")

        bind_step = 'name: "Bind installed Salesforce-compatible OAuth client"'
        setup_step = 'name: "Setup Node.js"'
        execute_step = 'name: "Execute HubSpot integration test"'

        self.assertIn(bind_step, workflow)
        self.assertLess(workflow.index(bind_step), workflow.index(setup_step))
        self.assertLess(workflow.index(bind_step), workflow.index(execute_step))

        binding = workflow[
            workflow.index(bind_step) : workflow.index(setup_step)
        ]
        self.assertIn(
            "env.STAGING_INSPECTION_TARGET_ENVIRONMENT == 'staging' ||",
            binding,
        )
        self.assertIn(
            "env.STAGING_INSPECTION_TARGET_ENVIRONMENT == 'test2'",
            binding,
        )
        self.assertIn(
            "env.STAGING_INSPECTION_TARGET_ENVIRONMENT == 'test3'",
            binding,
        )
        self.assertIn("release/serviceification-d3-bom.json", binding)
        self.assertIn('.sessionIssuerClass == "salesforce_compat"', binding)
        self.assertIn('index("password") != null', binding)
        self.assertIn('echo "::add-mask::${client_secret}"', binding)
        self.assertIn("STAGING_INSPECTION_OAUTH_CLIENT_ID=${client_id}", binding)
        self.assertIn(
            "STAGING_INSPECTION_OAUTH_CLIENT_SECRET=${client_secret}",
            binding,
        )

    def test_direct_sfcrm_targets_never_call_the_legacy_secret_admin_api(self):
        workflow = WORKFLOW.read_text(encoding="utf-8")

        restore_step = 'name: "Restore legacy-slot HubSpot inbound webhook secret"'
        verify_step = 'name: "Verify HubSpot inbound webhook endpoint"'
        self.assertIn(restore_step, workflow)
        restore = workflow[
            workflow.index(restore_step) : workflow.index(verify_step)
        ]

        self.assertIn("inputs.target_env == 'test1'", restore)
        self.assertNotIn("staging) client_secret=", restore)
        self.assertNotIn("test2)   client_secret=", restore)
        self.assertNotIn("test3)   client_secret=", restore)
        self.assertNotIn("STAGING_INSPECTION_OAUTH_CLIENT_ID", restore)
        self.assertNotIn("STAGING_INSPECTION_OAUTH_CLIENT_SECRET", restore)
        self.assertIn("scripts/rotate_staging_hubspot_webhook.py", restore)


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