#!/usr/bin/env python3
"""Tests du générateur du DocType porteur `OTO Dossier Vente` (Sprint 4 · CRM).

Stdlib pur (`unittest`) → aucune installation pip requise sur le runner Gitea.
La bibliothèque `jsonschema` sert d'*oracle* quand elle est présente, pour se
prémunir d'un écart entre le validateur maison et draft-07.

Axe central : la CROSS-COHÉRENCE workflow ↔ DocType (le DocType doit porter
exactement l'état, les valeurs de statut, la soumissibilité et les rôles du
pipeline `crm/workflow_vente/`).
"""

from __future__ import annotations

import copy
import json
import os
import subprocess
import sys
import unittest

_HERE = os.path.dirname(os.path.abspath(__file__))
_MODULE = os.path.normpath(os.path.join(_HERE, ".."))
_CRM = os.path.normpath(os.path.join(_MODULE, ".."))
_DELIVERABLES = os.path.normpath(os.path.join(_CRM, ".."))

sys.path.insert(0, _MODULE)
sys.path.insert(0, _CRM)
sys.path.insert(0, os.path.join(_DELIVERABLES, "publiciste"))

from dvlib import builder, frappe  # noqa: E402
from dvlib.builder import _UPDATE_FIELD  # noqa: E402
from workflow_vente.wflib.rbac import RoleResolver  # noqa: E402
from lib import validator as maison  # type: ignore  # noqa: E402
import doctype_dossier_vente_gen as gen  # noqa: E402

try:
    import jsonschema  # type: ignore

    _HAS_JSONSCHEMA = True
except Exception:  # pragma: no cover
    _HAS_JSONSCHEMA = False


def _load(path: str) -> dict:
    with open(path, encoding="utf-8") as fh:
        return json.load(fh)


class BaseFixture(unittest.TestCase):
    @classmethod
    def setUpClass(cls) -> None:
        cls.spec = _load(gen._SPEC_PATH)
        cls.wf_spec = _load(gen._WF_SPEC_PATH)
        cls.schema = _load(gen._SCHEMA_PATH)
        cls.resolver = RoleResolver.from_path()
        cls.bundle = builder.build_bundle(cls.spec, cls.wf_spec, cls.resolver)
        cls.dt = cls.bundle["doctype"]
        cls.manifest = cls.bundle["manifest"]
        cls.fields = {f["fieldname"]: f for f in cls.dt["fields"]}

    def _opts(self, fieldname: str) -> list[str]:
        return [ln for ln in self.fields[fieldname].get("options", "").split("\n") if ln]


class TestSchema(BaseFixture):
    def test_maison_schema_ok(self):
        self.assertEqual(maison.validate(self.bundle, self.schema), [])

    @unittest.skipUnless(_HAS_JSONSCHEMA, "jsonschema absent")
    def test_oracle_jsonschema_ok(self):
        jsonschema.validate(self.bundle, self.schema)

    def test_validate_returns_empty(self):
        self.assertEqual(gen._validate(self.bundle, self.wf_spec, self.resolver), [])


class TestCrossCoherence(BaseFixture):
    def test_name_equals_workflow_document_type(self):
        self.assertEqual(self.dt["name"], self.wf_spec["document_type"])

    def test_custom_matches_workflow(self):
        self.assertEqual(bool(self.dt["custom"]), bool(self.wf_spec["document_type_custom"]))

    def test_state_field_options_equal_state_names(self):
        field = self.wf_spec["workflow_state_field"]
        names = [s["state"] for s in self.wf_spec["states"]]
        self.assertEqual(self.fields[field]["fieldtype"], "Select")
        self.assertEqual(self._opts(field), names)
        self.assertEqual(self.fields[field]["read_only"], 1)

    def test_pipeline_value_field_options_equal_update_values(self):
        values = [s["update_value"] for s in self.wf_spec["states"]]
        self.assertIn(_UPDATE_FIELD, self.fields)
        self.assertEqual(self._opts(_UPDATE_FIELD), values)

    def test_is_submittable_derived_from_docstatus(self):
        max_ds = max(s["doc_status"] for s in self.wf_spec["states"])
        self.assertEqual(self.dt["is_submittable"], 1 if max_ds >= "1" else 0)

    def test_update_field_comes_from_workflow_builder(self):
        # Anti-dérive : le champ de valeur machine est importé du builder voisin,
        # pas réécrit en dur ici.
        self.assertEqual(_UPDATE_FIELD, "statut_pipeline")


class TestPermissions(BaseFixture):
    def setUp(self):
        self.perm = {p["role"]: p for p in self.dt["permissions"]}
        self.docstatus = {s["state"]: s["doc_status"] for s in self.wf_spec["states"]}

    def test_every_workflow_role_present(self):
        role_ids = {s["role_id"] for s in self.wf_spec["states"]}
        role_ids |= {t["role_id"] for t in self.wf_spec["transitions"]}
        for rid in role_ids:
            self.assertIn(self.resolver.erpnext_name(rid), self.perm)

    def test_allow_edit_roles_have_write(self):
        for s in self.wf_spec["states"]:
            p = self.perm[self.resolver.erpnext_name(s["role_id"])]
            self.assertEqual(p.get("write"), 1, s["state"])
            if s["doc_status"] == "0":
                self.assertEqual(p.get("create"), 1, s["state"])

    def test_submit_and_cancel_derivation(self):
        for t in self.wf_spec["transitions"]:
            p = self.perm[self.resolver.erpnext_name(t["role_id"])]
            nxt = self.docstatus[t["next_state"]]
            if nxt == "1":
                self.assertEqual(p.get("submit"), 1, t["action"])
            if nxt == "2":
                self.assertEqual(p.get("cancel"), 1, t["action"])

    def test_no_role_name_hardcoded_in_spec(self):
        # #6 : aucun NOM DE RÔLE Frappe en dur dans le contrat métier ; les rôles
        # ne viennent que du workflow (role_id) résolu via le RBAC.
        blob = json.dumps(self.spec, ensure_ascii=False)
        for rid in self.resolver.known_ids():
            self.assertNotIn(self.resolver.erpnext_name(rid), blob)


class TestFrappeKnowledge(unittest.TestCase):
    def test_docfield_rejects_unknown_fieldtype(self):
        with self.assertRaises(ValueError):
            frappe.docfield("x", "Rocket")

    def test_permission_row_rejects_unknown_perm(self):
        with self.assertRaises(ValueError):
            frappe.permission_row("OTO X", {"read", "teleport"})

    def test_docfield_only_emits_set_flags(self):
        f = frappe.docfield("x", "Data", flags={"reqd": 1})
        self.assertEqual(f.get("reqd"), 1)
        self.assertNotIn("hidden", f)

    def test_naming_series_field_is_readonly_select(self):
        f = frappe.naming_series_field("OTO-DV-.YYYY.-")
        self.assertEqual(f["fieldtype"], "Select")
        self.assertEqual(f["read_only"], 1)


class TestInvariantsAndStructure(BaseFixture):
    def test_field_order_matches_fields(self):
        self.assertEqual(self.dt["field_order"], [f["fieldname"] for f in self.dt["fields"]])

    def test_fieldnames_unique(self):
        names = [f["fieldname"] for f in self.dt["fields"]]
        self.assertEqual(len(names), len(set(names)))

    def test_no_currency_default(self):
        for f in self.dt["fields"]:
            if f["fieldtype"] == "Currency":
                self.assertNotIn("default", f, f["fieldname"])

    def test_currency_options_reference_devise(self):
        for f in self.dt["fields"]:
            if f["fieldtype"] == "Currency":
                self.assertEqual(f.get("options"), "devise")

    def test_links_declared_in_manifest(self):
        declared = set(self.manifest["doctypes_lies_a_confirmer"])
        for f in self.dt["fields"]:
            if f["fieldtype"] == "Link":
                self.assertIn(f["options"], declared)

    def test_naming_series_present(self):
        self.assertIn("naming_series", self.fields)
        self.assertEqual(self.dt["autoname"], "naming_series:")

    def test_manifest_counts_coherent(self):
        c = self.manifest["counts"]
        self.assertEqual(c["fields"], len(self.dt["fields"]))
        self.assertEqual(c["permissions"], len(self.dt["permissions"]))
        self.assertEqual(c["pipeline_states"], len({s["state"] for s in self.wf_spec["states"]}))


class TestDeterminismAndAntiRegression(BaseFixture):
    def test_build_is_deterministic(self):
        again = builder.build_bundle(self.spec, self.wf_spec, self.resolver)
        self.assertEqual(json.dumps(again, sort_keys=True), json.dumps(self.bundle, sort_keys=True))

    def test_committed_out_matches_regeneration(self):
        out = os.path.join(_MODULE, "out", "doctype_oto_dossier_vente.json")
        man = os.path.join(_MODULE, "out", "MANIFEST.json")
        if not (os.path.exists(out) and os.path.exists(man)):
            self.skipTest("out/ non commité")
        self.assertEqual(_load(out), self.dt)
        self.assertEqual(_load(man), self.manifest)


class TestTamperDetection(BaseFixture):
    def test_break_is_submittable_detected(self):
        bad = copy.deepcopy(self.bundle)
        bad["doctype"]["is_submittable"] = 0
        self.assertTrue(gen._validate(bad, self.wf_spec, self.resolver))

    def test_drop_state_option_detected(self):
        bad = copy.deepcopy(self.bundle)
        for f in bad["doctype"]["fields"]:
            if f["fieldname"] == self.wf_spec["workflow_state_field"]:
                f["options"] = "\n".join(f["options"].split("\n")[:-1])
        self.assertTrue(gen._validate(bad, self.wf_spec, self.resolver))

    def test_rename_doctype_detected(self):
        bad = copy.deepcopy(self.bundle)
        bad["doctype"]["name"] = "Autre Chose"
        self.assertTrue(gen._validate(bad, self.wf_spec, self.resolver))

    def test_inject_currency_default_detected(self):
        bad = copy.deepcopy(self.bundle)
        for f in bad["doctype"]["fields"]:
            if f["fieldtype"] == "Currency":
                f["default"] = "150000"
                break
        self.assertTrue(gen._validate(bad, self.wf_spec, self.resolver))


class TestCLI(unittest.TestCase):
    def test_cli_validate_exit_zero(self):
        r = subprocess.run(
            [sys.executable, os.path.join(_MODULE, "doctype_dossier_vente_gen.py"), "validate"],
            capture_output=True, text=True,
        )
        self.assertEqual(r.returncode, 0, r.stderr)


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