from unittest.mock import patch

import pytest
from django.db import connection

from finance.models import FormulaDefinition

pytestmark = pytest.mark.django_db


def column_definition(name="收入成本差"):
    return {
        "name": name,
        "meaning": "实际收入减成本",
        "scope": "所选分组与期间",
        "missing": "结果不可用（#N/A）；零分母不可用",
        "expression": {
            "operator": "subtract",
            "left": {"metric": "revenue"},
            "right": {"metric": "cost"},
        },
    }


def _post(client, url, definition, revision=None):
    return client.post(
        url,
        data={"definition": definition, **({"revision": revision} if revision is not None else {})},
        content_type="application/json",
    )


def test_same_name_update_requires_matching_revision(workspace):
    ds, _, client, *_ = workspace
    url = f"/api/datasets/{ds.pk}/formula-definitions"

    saved = _post(client, url, column_definition())
    assert saved.status_code == 201
    item_id, revision = saved.json()["id"], saved.json()["revision"]

    # 缺失或过期的 revision 都不能静默覆盖其他窗口的修改
    assert _post(client, url, column_definition()).status_code == 409
    assert _post(client, url, column_definition(), revision=999).status_code == 409

    updated = _post(client, url, column_definition(), revision=revision)
    assert updated.status_code == 200
    assert updated.json()["revision"] == revision + 1

    # 基于旧版本的再次更新必须被拒绝，两个标签页不互相静默丢失
    stale = _post(client, url, column_definition(), revision=revision)
    assert stale.status_code == 409
    current = FormulaDefinition.objects.get(pk=item_id)
    assert current.revision == revision + 1
    # 列表接口暴露当前 revision 供前端携带
    listed = client.get(url).json()["items"]
    assert listed[0]["revision"] == revision + 1


def test_definition_quota_is_enforced_and_release_allows_new_save(workspace):
    ds, _, client, *_ = workspace
    url = f"/api/datasets/{ds.pk}/formula-definitions"
    for index in range(50):
        FormulaDefinition.objects.create(
            dataset=ds,
            user=ds.created_by,
            normalized_name=f"定义{index:02d}",
            definition=column_definition(f"定义{index:02d}"),
        )

    rejected = _post(client, url, column_definition("第 51 个不同名定义"))
    assert rejected.status_code == 400
    assert "50" in rejected.json()["error"]
    assert FormulaDefinition.objects.filter(dataset=ds).count() == 50

    # 同名更新不受配额限制；删除后名额释放可再保存新名称
    assert _post(client, url, column_definition("定义00"), revision=1).status_code == 200
    FormulaDefinition.objects.filter(normalized_name="定义00").delete()
    assert _post(client, url, column_definition("第 51 个不同名定义")).status_code == 201


def test_audit_failure_rolls_back_definition_change(workspace):
    ds, _, client, *_ = workspace
    url = f"/api/datasets/{ds.pk}/formula-definitions"

    with patch("finance.formula_views.audit", side_effect=RuntimeError("audit unavailable")):
        with pytest.raises(RuntimeError, match="audit unavailable"):
            _post(client, url, column_definition())

    assert FormulaDefinition.objects.filter(dataset=ds).count() == 0


def test_concurrent_distinct_saves_cannot_exceed_quota_on_postgres(workspace):
    if connection.vendor != "postgresql":
        pytest.skip("并发配额依赖 PostgreSQL 行锁语义，生产库单独验证")
    import threading

    from django.test import Client

    ds, member, *_ = workspace
    user = ds.created_by
    for index in range(49):
        FormulaDefinition.objects.create(
            dataset=ds,
            user=user,
            normalized_name=f"定义{index:02d}",
            definition=column_definition(f"定义{index:02d}"),
        )

    barrier = threading.Barrier(4)

    def save(index):
        client = Client()
        assert client.login(username=user.username, password="Finance-TestPassword991!")
        barrier.wait()
        _post(client, f"/api/datasets/{ds.pk}/formula-definitions", column_definition(f"并发{index}"))

    threads = [threading.Thread(target=save, args=(index,)) for index in range(4)]
    for thread in threads:
        thread.start()
    for thread in threads:
        thread.join()

    assert FormulaDefinition.objects.filter(dataset=ds).count() == 50