from __future__ import annotations

import copy
from concurrent.futures import ThreadPoolExecutor
from datetime import UTC, datetime, timedelta

import httpx
import pytest
from sqlmodel import Session, SQLModel, create_engine, select

from backend.app.db.models import AuditEvent, JobRun
from backend.app.services import action_execution, deployment_broker, runner_statistics


def snapshot():
    at = datetime.now(UTC).isoformat()
    return {
        "schemaVersion": 1, "status": "healthy", "scope": "ci-fleet", "generatedAt": at,
        "budget": {"cpuLimit": 8, "memoryLimitBytes": 16 * 1024**3, "activeJobLimit": 2,
                   "heavyJobLimit": 1, "reservedCpu": 2, "reservedMemoryBytes": 3 * 1024**3,
                   "activeJobs": 0, "heavyJobs": 0},
        "host": {"cpuBusyPercent": None, "loadRatio": None,
                 "availableMemoryBytes": None, "cacheFreeBytes": None},
        "profiles": [{"id": "checks", "displayName": "ITWorx CI · Checks",
                      "runnerName": "itworx-ci-checks", "labels": ["itworx-ci-checks"],
                      "status": "capacity_wait", "reason": "host_headroom",
                      "cpuLimit": 1, "memoryLimitBytes": 1536 * 1024**2,
                      "queueDepth": 1, "activeJobs": 0, "cpuUsed": None,
                      "memoryUsedBytes": None, "oldestQueuedSeconds": 12,
                      "completedJobs": 0, "p50WaitSeconds": None, "p95WaitSeconds": None,
                      "p50DurationSeconds": None, "p95DurationSeconds": None}],
        "window": {"startAt": at, "endAt": at, "completedJobs": 0, "samples": 1,
                   "coverage": "bounded"},
        "history": [{"at": at, "activeJobs": 0, "queuedJobs": 1,
                     "cpuUsed": None, "memoryUsedBytes": None}],
    }


@pytest.fixture(autouse=True)
def reset_cache(monkeypatch):
    monkeypatch.setattr(runner_statistics, "_cached", None)
    monkeypatch.setattr(runner_statistics, "_cached_at", 0.0)


def test_unknown_measurements_survive_and_private_fields_are_removed():
    payload = snapshot()
    payload["rawJobs"] = [{"secret": "fixture-secret"}]
    payload["profiles"][0]["workspacePath"] = "/private/credentials"
    result = runner_statistics.sanitize_statistics(payload)
    assert result["status"] == "healthy"
    assert result["profiles"][0]["cpuUsed"] is None
    assert result["profiles"][0]["queueDepth"] == 1
    assert "rawJobs" not in result
    assert "workspacePath" not in result["profiles"][0]


def test_unattested_profile_limits_remain_unknown():
    payload = snapshot()
    payload["status"] = "degraded"
    payload["profiles"][0].update(cpuLimit=None, memoryLimitBytes=None)
    result = runner_statistics.sanitize_statistics(payload)
    assert result["status"] == "degraded"
    assert result["profiles"][0]["cpuLimit"] is None
    assert result["profiles"][0]["memoryLimitBytes"] is None


@pytest.mark.parametrize("mutation", [
    lambda p: p.update(schemaVersion=2),
    lambda p: p.update(generatedAt="2026-10-01T12:00:00"),
    lambda p: p.update(generatedAt=(datetime.now(UTC) + timedelta(minutes=1)).isoformat()),
    lambda p: p["profiles"].append(copy.deepcopy(p["profiles"][0])),
    lambda p: p["profiles"][0].update(cpuUsed=float("nan")),
    lambda p: p["profiles"][0].update(activeJobs=True),
    lambda p: p["profiles"][0].update(runnerName="secret-token"),
    lambda p: p["profiles"][0].update(labels=["security-validation"]),
    lambda p: p["profiles"][0].update(reason="Bearer fixture-secret-token"),
    lambda p: p["window"].update(startAt=None),
    lambda p: p["history"].append(copy.deepcopy(p["history"][0])),
    lambda p: p.update(history=[p["history"][0]] * 289),
])
def test_untrusted_or_ambiguous_statistics_fail_to_honest_unknown(mutation):
    payload = snapshot()
    mutation(payload)
    result = runner_statistics.sanitize_statistics(payload)
    assert result["status"] == "unavailable"
    assert result["budget"]["cpuLimit"] is None
    assert result["profiles"] == []


def test_old_sample_is_degraded_without_retimestamping():
    payload = snapshot()
    old = (datetime.now(UTC) - timedelta(minutes=2)).isoformat()
    payload["generatedAt"] = old
    payload["history"][0]["at"] = old
    payload["window"].update(startAt=old, endAt=old)
    result = runner_statistics.sanitize_statistics(payload)
    assert result["status"] == "degraded"
    assert result["generatedAt"] == old


def test_singleflight_cache_and_caller_mutation_isolation(monkeypatch):
    calls = []

    class Broker:
        def is_configured(self):
            return True

        def runner_fleet_statistics(self):
            calls.append(1)
            return snapshot()

    monkeypatch.setattr(runner_statistics, "DeploymentBrokerClient", Broker)
    service = runner_statistics.RunnerStatisticsService()
    with ThreadPoolExecutor(max_workers=8) as pool:
        results = list(pool.map(lambda _: service.snapshot(), range(8)))
    assert calls == [1]
    results[0]["budget"]["cpuLimit"] = 999
    assert service.snapshot()["budget"]["cpuLimit"] == 8


@pytest.mark.parametrize("configured,raises,expected", [
    (False, False, "not_configured"), (True, True, "unavailable"),
])
def test_unavailable_backend_never_leaks_error_or_invents_budget(monkeypatch, configured, raises, expected):
    class Broker:
        def is_configured(self):
            return configured

        def runner_fleet_statistics(self):
            assert raises
            raise RuntimeError("Bearer fixture-private-secret /private/key")

    monkeypatch.setattr(runner_statistics, "DeploymentBrokerClient", Broker)
    result = runner_statistics.RunnerStatisticsService().snapshot()
    assert result["status"] == expected
    assert result["budget"]["reservedCpu"] is None
    assert "fixture-private" not in str(result)


def test_broker_statistics_fixed_read_path_has_deadline_and_no_credentials(monkeypatch):
    calls = []
    real_client = httpx.Client

    def handle(request):
        calls.append(request)
        return httpx.Response(200, json=snapshot())

    def client(**kwargs):
        assert kwargs == {"timeout": 5, "verify": True, "follow_redirects": False}
        return real_client(**kwargs, transport=httpx.MockTransport(handle))

    monkeypatch.setattr(deployment_broker.httpx, "Client", client)
    broker = deployment_broker.DeploymentBrokerClient(base_url="https://controller.test", verify_tls=True)
    assert broker.runner_fleet_statistics()["schemaVersion"] == 1
    request = calls[0]
    assert request.method == "GET"
    assert str(request.url) == "https://controller.test/v1/runner-fleet/statistics"
    assert "authorization" not in request.headers
    assert not any(key.startswith("x-itworx") for key in request.headers)
    assert broker._runner_fleet_read("/v1/arbitrary")["reason"] == "runner_fleet_path_invalid"
    assert len(calls) == 1


def test_specialize_broker_caller_cannot_choose_script_revision_or_target(monkeypatch):
    calls = []
    broker = deployment_broker.DeploymentBrokerClient()
    monkeypatch.setattr(broker, "_request", lambda *args, **kwargs: calls.append((args, kwargs)) or {})
    broker.runner_fleet_specialize(reason="x" * 1100)
    args, kwargs = calls[0]
    assert args == ("POST", "/v1/appops/runner-fleet/specialize")
    assert kwargs["payload"] == {"reason": "x" * 1000}


@pytest.mark.parametrize("status,success", [("specialized", True), ("already_applied", True),
                                          ("upgraded", False), ("failed", False)])
def test_specialize_execution_persists_actual_success_or_failure(tmp_path, monkeypatch, status, success):
    engine = create_engine(f"sqlite:///{tmp_path / 'specialize.db'}")
    SQLModel.metadata.create_all(engine)
    monkeypatch.setattr(deployment_broker.DeploymentBrokerClient, "runner_fleet_specialize",
                        lambda self, **kwargs: {"status": status, "revision": "a" * 40})
    with Session(engine) as session:
        result = action_execution.execute_governed_action(
            session, action="runner.fleet.specialize", target="tower/runner-fleet-specialized",
            source="chatgpt-mcp", risk="high", reason="approved fleet specialization",
            original_policy_decision="allowed",
        )
        job = session.get(JobRun, result["jobId"])
        audit = session.get(AuditEvent, result["auditId"])
        assert result["success"] is success
        assert job.status == ("succeeded" if success else "failed")
        assert audit.result == ("success" if success else "failed")
        assert audit.target == "tower/runner-fleet-specialized"
        assert ("verified" in result["message"]) is success


def test_wrong_specialization_target_is_blocked_before_job_or_broker(tmp_path, monkeypatch):
    engine = create_engine(f"sqlite:///{tmp_path / 'blocked.db'}")
    SQLModel.metadata.create_all(engine)
    monkeypatch.setattr(deployment_broker.DeploymentBrokerClient, "runner_fleet_specialize",
                        lambda *args, **kwargs: pytest.fail("must not call broker"))
    with Session(engine) as session:
        result = action_execution.execute_governed_action(
            session, action="runner.fleet.specialize", target="other/runner",
            source="chatgpt-mcp", risk="high", reason="wrong target",
            original_policy_decision="allowed",
        )
        assert result["policyDecision"] == "blocked"
        assert session.exec(select(JobRun)).all() == []


@pytest.mark.parametrize('status,success', [('configured', True), ('already_configured', True),
                                          ('specialized', False), ('failed', False)])
def test_statistics_configuration_persists_actual_status_without_secret(tmp_path, monkeypatch, status, success):
    engine = create_engine(f"sqlite:///{tmp_path / 'binding.db'}")
    SQLModel.metadata.create_all(engine)
    monkeypatch.setattr(deployment_broker.DeploymentBrokerClient, 'runner_statistics_configure',
                        lambda self, **kwargs: {'status': status, 'revision': 'a' * 40, 'projectbrainDeliveryRequired': True})
    with Session(engine) as session:
        result = action_execution.execute_governed_action(
            session, action='runner.statistics.configure', target='tower/projectbrain-runner-statistics',
            source='chatgpt-mcp', risk='high', reason='configure read-only metrics', original_policy_decision='allowed')
        assert result['success'] is success
        assert session.get(JobRun, result['jobId']).status == ('succeeded' if success else 'failed')
        assert session.get(AuditEvent, result['auditId']).target == 'tower/projectbrain-runner-statistics'
        assert 'token' not in result


def test_statistics_configuration_broker_has_fixed_signed_body(monkeypatch):
    calls = []
    broker = deployment_broker.DeploymentBrokerClient()
    monkeypatch.setattr(broker, '_request', lambda *args, **kwargs: calls.append((args, kwargs)) or {})
    broker.runner_statistics_configure(reason='configure read-only stats')
    assert calls[0][0] == ('POST', '/v1/appops/runner-statistics/configure')
    assert calls[0][1]['payload'] == {'reason': 'configure read-only stats'}