from contextlib import contextmanager
from copy import deepcopy
from uuid import uuid4

import pytest

from samples.peripheral.auth import Identity
from samples.peripheral.config import SampleSettings
from samples.peripheral.schemas import TestDataRecord as DataRecordSchema
from samples.peripheral.schemas import WorkOrder
from samples.peripheral.store import PostgresStore


class FakeResult:
    def __init__(self, *, row=None, rows=None):
        self.row = row
        self.rows = rows or []

    def fetchone(self):
        return self.row

    def fetchall(self):
        return self.rows


class ManagedProjectionConnection:
    def __init__(self, table: str, document: dict):
        self.table = table
        self.document = document

    def execute(self, statement, parameters):
        normalized = " ".join(statement.split())
        if "FROM peripheral_sample.scenarios" in normalized:
            return FakeResult(row=None)
        if f"FROM peripheral_sample.{self.table}" in normalized:
            row = {"document": self.document}
            return FakeResult(row=row, rows=[row])
        raise AssertionError(f"unexpected query: {normalized}")


class ManagedProjectionStore(PostgresStore):
    def __init__(self, table: str, document: dict):
        super().__init__(SampleSettings())
        self.fake_connection = ManagedProjectionConnection(table, document)

    @contextmanager
    def connection(self, *, readonly=False, timeseries=False):
        assert readonly is True
        assert timeseries is False
        yield self.fake_connection


def work_order_document() -> dict:
    now = "2026-09-18T00:00:00Z"
    return {
        "work_order_id": str(uuid4()),
        "project_id": str(uuid4()),
        "revision": 2,
        "title": "恢复后的演示工单",
        "summary": "当前行已恢复，但旧实现留下 tombstone 字段。",
        "status": "open",
        "priority": "normal",
        "assignee": None,
        "scheduled_at": None,
        "origin": "synthetic",
        "created_at": now,
        "updated_at": now,
        "created_by": str(uuid4()),
        "updated_by": str(uuid4()),
        "work_order_type": "general",
        "location": None,
        "scenario_code": None,
        "deleted_at": now,
        "deleted_by": str(uuid4()),
    }


def test_active_work_order_projection_removes_only_tombstone_fields():
    document = work_order_document()
    stored_document = deepcopy(document)
    store = ManagedProjectionStore("work_orders", stored_document)
    who = Identity(document["project_id"], str(uuid4()), False)

    result = store.get_work_order(who, document["work_order_id"])

    assert "deleted_at" not in result
    assert "deleted_by" not in result
    assert stored_document == document
    WorkOrder.model_validate(result)


@pytest.mark.parametrize(
    ("table", "id_column", "document", "model"),
    [
        ("work_orders", "work_order_id", work_order_document(), WorkOrder),
        (
            "test_data_records",
            "record_id",
            {
                "record_id": str(uuid4()),
                "project_id": str(uuid4()),
                "revision": 2,
                "label": "恢复后的演示数据",
                "point_id": "DEMO-P001",
                "observed_at": "2026-09-18T00:00:00Z",
                "value": 1.0,
                "unit": "m",
                "quality": "valid",
                "notes": "",
                "origin": "synthetic",
                "created_at": "2026-09-18T00:00:00Z",
                "updated_at": "2026-09-18T00:00:00Z",
                "created_by": str(uuid4()),
                "updated_by": str(uuid4()),
                "deleted_at": "2026-09-18T00:00:00Z",
                "deleted_by": str(uuid4()),
            },
            DataRecordSchema,
        ),
    ],
)
def test_active_managed_list_projection_removes_tombstone_fields(
    table, id_column, document, model
):
    store = ManagedProjectionStore(table, document)
    who = Identity(document["project_id"], str(uuid4()), False)

    page = store._list_managed(who, table, id_column, limit=10, cursor=None)

    assert len(page["items"]) == 1
    assert "deleted_at" not in page["items"][0]
    assert "deleted_by" not in page["items"][0]
    model.model_validate(page["items"][0])


def test_active_projection_does_not_hide_unrelated_fields():
    document = {
        "deleted_at": "2026-09-18T00:00:00Z",
        "deleted_by": str(uuid4()),
        "unexpected_field": "must remain visible to schema validation",
    }

    result = PostgresStore._active_managed_document(document)

    assert result == {"unexpected_field": "must remain visible to schema validation"}
    assert "deleted_at" in document
    assert "deleted_by" in document