import json
import os
import tomllib
import unittest

from tests.fakes import BASE, HEAD, TEST_CONFIG, FakeGitea, FakeJev, FakeModel, config, diff_for, gitea_file
from tests.test_flow import run
from werkrevisor import dbschema
from werkrevisor.config import ConfigError, build_config
from werkrevisor.diff import parse_unified_diff
from werkrevisor.rules import files_from_gitea
from werkrevisor.verdict import Verdict

SQL = "src/main/resources/db/changelog/5000-x.sql"
MASTER = "src/main/resources/db/changelog/db.changelog-master.yaml"
TABLE = """--changeset a:table endDelimiter:--//
create table x (id int);
--//
"""
FN = """--changeset a:fn runOnChange:true endDelimiter:--//
create or replace function f() returns int language sql as $$ select {n} $$;
--//
"""


def sql(*parts):
    return ("-- liquibase formatted sql\n\n" + "\n-- ============================================================================\n"
            .join(parts)).encode()


def jev_config(**db):
    return config(engine={"mode": "combined"}, gate={"yellow_gate": "block"}, db_schema={"judge": "jev", **db})


def schema_gitea(head, base=None, status="modified"):
    gitea = FakeGitea(files=[gitea_file(SQL, status=status)], diff=diff_for(SQL, added=["create table x (id int);"]))
    gitea.raw_files_at[(SQL, HEAD)] = head
    gitea.raw_files_at[(SQL, BASE)] = base
    return gitea


def jev_for(risk, loss, unit="a:table"):
    return FakeJev(problems={(unit, "deploy_risk"): risk / dbschema.RISK_TOP, (unit, "data_loss"): loss})


def reason_codes(outcome):
    return {r.code: r.verdict for r in outcome.result.reasons}


class ChangesetTest(unittest.TestCase):
    def collect(self, head, base, status="modified", diff=""):
        files = files_from_gitea([gitea_file(SQL, status=status)])
        return dbschema.collect(files, parse_unified_diff(diff), lambda p: head, lambda p: base)

    def test_units_hold_new_and_changed_changesets(self):
        units, released, _ = self.collect(sql(TABLE, FN.format(n=2)), sql(FN.format(n=1)))
        self.assertEqual(released, [])
        self.assertEqual([(u.changeset, u.new) for u in units], [("a:table", True), ("a:fn", False)])
        state = units[1].state()
        self.assertTrue(state.startswith(f"File: {SQL}\nChangeset id: a:fn\nChangeset: modified existing changeset\n\n"))
        self.assertIn("select 2", state.split("Version at the base commit:")[0])
        self.assertIn("select 1", state.split("Version at the base commit:")[1])

    def test_unchanged_changesets_and_separators_are_no_unit(self):
        units, released, _ = self.collect(sql(TABLE, FN.format(n=1)) + b"\n\n", sql(TABLE, FN.format(n=1)))
        self.assertEqual((units, released), ([], []))

    def test_released_changes_are_found(self):
        changed = TABLE.replace("id int", "id bigint")
        repeatable = changed.replace("--changeset a:table endDelimiter:--//",
                                     "--changeset a:table endDelimiter:--//\n--validCheckSum: ANY")
        self.assertEqual(self.collect(sql(changed), sql(TABLE))[1], [f"{SQL}: released changeset a:table changed"])
        self.assertEqual(self.collect(sql(FN.format(n=1)), sql(TABLE, FN.format(n=1)))[1],
                         [f"{SQL}: changeset a:table removed"])
        self.assertEqual(self.collect(b"", sql(TABLE), status="deleted")[1], [f"{SQL} deleted"])
        units, released, _ = self.collect(sql(repeatable), sql(TABLE))
        self.assertEqual((released, [u.changeset for u in units]), ([], ["a:table"]))

    def test_unreadable_base_or_duplicate_ids_are_never_judged_as_new(self):
        _, _, unreadable = self.collect(sql(TABLE), b"\0binary")
        self.assertEqual(unreadable, [f"{SQL} (not UTF-8 text)"])
        files = files_from_gitea([gitea_file(SQL)])

        def failing_base(path):
            raise OSError("timeout")
        units, _, unreadable = dbschema.collect(files, {}, lambda p: sql(TABLE), failing_base)
        self.assertEqual((units, unreadable), ([], [f"{SQL} (base version: OSError: timeout)"]))
        units, released, _ = self.collect(sql(TABLE, TABLE), sql(TABLE))
        self.assertEqual((units, released), ([], [f"{SQL}: duplicate changeset id a:table"]))

    def test_file_without_changesets_is_one_unit(self):
        files = files_from_gitea([gitea_file(MASTER)])
        diff = parse_unified_diff(diff_for(MASTER, added=["  - include: {file: 5000-x.sql}"]))
        units = dbschema.collect(files, diff, lambda p: b"databaseChangeLog:\n", lambda p: b"databaseChangeLog:\n")[0]
        self.assertEqual([(u.path, u.changeset, u.text) for u in units],
                         [(MASTER, "", "+  - include: {file: 5000-x.sql}")])


class DbSchemaFlowTest(unittest.TestCase):
    def test_never_judge_gives_yellow(self):
        for cfg in (config(gate={"yellow_gate": "block"}),
                    config(engine={"mode": "combined"}, gate={"yellow_gate": "block"}, db_schema={"judge": "never"})):
            with self.subTest(judge=cfg.db_schema.judge):
                jev = FakeJev()
                outcome, _ = run(schema_gitea(sql(TABLE), None, "added"), FakeModel(prompt_files=[SQL]), cfg=cfg, jev=jev)
                self.assertIs(outcome.result.verdict, Verdict.YELLOW)
                self.assertIs(reason_codes(outcome)["db_schema"], Verdict.YELLOW)
                self.assertEqual([q for _, q in jev.requests if "deploy_risk" in q], [])

    def test_jev_answers_decide_the_verdict(self):
        for risk, loss, code, verdict in ((0.2, 0.05, "db_schema_judged", Verdict.GREEN),
                                          (2.0, 0.38, "db_schema_judged", Verdict.GREEN),
                                          (2.99, 0.84, "db_schema_risky", Verdict.YELLOW),
                                          (1.0, 0.70, "db_schema_risky", Verdict.YELLOW)):
            with self.subTest(risk=risk, loss=loss):
                outcome, _ = run(schema_gitea(sql(TABLE), None, "added"), FakeModel(prompt_files=[SQL]),
                                 cfg=jev_config(), jev=jev_for(risk, loss))
                self.assertIs(outcome.result.verdict, verdict)
                self.assertIs(reason_codes(outcome)[code], verdict)
                self.assertNotIn("db_schema", reason_codes(outcome))

    def test_risky_reason_names_the_values(self):
        outcome, _ = run(schema_gitea(sql(TABLE), None, "added"), FakeModel(prompt_files=[SQL]),
                         cfg=jev_config(), jev=jev_for(2.99, 0.84))
        reason = next(r for r in outcome.result.reasons if r.code == "db_schema_risky")
        self.assertEqual(reason.text, f"risky DB changes according to Jev: {SQL} a:table (risk 2.99, data loss 0.84)")

    def test_released_changeset_change_gives_yellow(self):
        jev = FakeJev()
        outcome, _ = run(schema_gitea(sql(TABLE.replace("id int", "id bigint")), sql(TABLE)),
                         FakeModel(prompt_files=[SQL]), cfg=jev_config(), jev=jev)
        self.assertIs(reason_codes(outcome)["db_schema_released_changed"], Verdict.YELLOW)
        self.assertEqual([q for _, q in jev.requests if "deploy_risk" in q], [])

    def test_base_is_the_merge_base(self):
        gitea = schema_gitea(sql(TABLE), sql(FN.format(n=1)))
        gitea.merge_base = "c" * 40
        gitea.raw_files_at[(SQL, "c" * 40)] = sql()
        outcome, _ = run(gitea, FakeModel(prompt_files=[SQL]), cfg=jev_config(), jev=jev_for(0.2, 0.05))
        self.assertNotIn("db_schema_released_changed", reason_codes(outcome))
        self.assertIs(outcome.result.verdict, Verdict.GREEN)

    def test_unjudged_changesets_give_yellow(self):
        big = TABLE.replace("create table x (id int);", "-- x\n" * 20000)
        cases = (("Jev fails", sql(TABLE), FakeJev(error=OSError("down")), {}, "OSError: down"),
                 ("too large", sql(big), FakeJev(), {}, "larger than [classifier] max_state_chars"),
                 ("--no-llm", sql(TABLE), FakeJev(), {"use_llm": False}, "(--no-llm)"))
        for name, head, jev, kw, text in cases:
            with self.subTest(name):
                outcome, _ = run(schema_gitea(head, None, "added"), FakeModel(prompt_files=[SQL]), cfg=jev_config(),
                                 jev=jev, **kw)
                reason = next(r for r in outcome.result.reasons if r.code == "db_schema_not_judged")
                self.assertIs(reason.verdict, Verdict.YELLOW)
                self.assertIn(text, reason.text)
        gitea = schema_gitea(sql(TABLE), None, "added")
        gitea.diff = diff_for(SQL, added=["-----BEGIN " + "PRIVATE KEY-----"])
        jev = FakeJev()
        outcome, _ = run(gitea, FakeModel(prompt_files=[SQL]), cfg=jev_config(), jev=jev)
        self.assertIn("(possible secrets in the diff)",
                      next(r.text for r in outcome.result.reasons if r.code == "db_schema_not_judged"))
        self.assertEqual(jev.requests, [])

    def test_usage_record_holds_the_jev_values(self):
        jev = FakeJev(problems={("a:table", "deploy_risk"): 0.1, ("a:table", "data_loss"): 0.05,
                                ("a:fn", "deploy_risk"): 0.9, ("a:fn", "data_loss"): 0.7})
        gitea = schema_gitea(sql(TABLE, FN.format(n=2)), sql(FN.format(n=1)))
        _, d = run(gitea, FakeModel(prompt_files=[SQL]), cfg=jev_config(), jev=jev)
        record = json.loads(d.out.getvalue().strip())
        self.assertEqual([(r["changeset"], r["new"], r["verdict"]) for r in record["db_schema_judgments"]],
                         [("a:table", True, "GREEN"), ("a:fn", False, "YELLOW")])
        self.assertAlmostEqual(record["db_schema_judgments"][1]["risk"], 2.7)
        self.assertEqual((record["db_schema_judge_calls"], record["db_schema_judge_input_tokens"]), (2, 2000))
        self.assertIn("DB schema judge `jev-test` calls 2", gitea.created_comments[0])


class DbSchemaConfigTest(unittest.TestCase):
    def host(self):
        with open(TEST_CONFIG, "rb") as f:
            return tomllib.load(f)

    def test_db_schema_is_host_only_and_validated(self):
        with self.assertRaisesRegex(ConfigError, r"must not set \[db_schema\] judge: only the host config"):
            build_config(self.host(), {"db_schema": {"judge": "jev"}})
        for values, message in (({"judge": "maybe"}, r"\[db_schema\] judge must be one of never, jev"),
                                ({"risk_min": 4}, r"\[db_schema\] risk_min must be a number between 0 and 3"),
                                ({"data_loss_min": 1.5}, r"\[db_schema\] data_loss_min must be a number between 0 and 1"),
                                ({"judge": "jev", "x": 1}, r"\[db_schema\] has unknown keys: x")):
            with self.subTest(values=values):
                with self.assertRaisesRegex(ConfigError, message):
                    config(db_schema=values)
        host = self.host()
        del host["classifier"]
        host["engine"]["mode"] = "llm"
        host["db_schema"] = {"judge": "jev"}
        with self.assertRaisesRegex(ConfigError, r"judge = 'jev' needs a \[classifier\] section"):
            build_config(host)
        self.assertEqual((config().db_schema.judge, config().db_schema.risk_min, config().db_schema.data_loss_min),
                         ("never", 2.5, 0.6))


class DbSchemaPathsTest(unittest.TestCase):
    def test_the_db_schema_paths_are_configurable(self):
        from werkrevisor.config import parse_repository_config
        from werkrevisor.rules import ChangedFile, evaluate_rules
        with open(TEST_CONFIG, "rb") as f:
            host = tomllib.load(f)
        cfg = build_config(host, parse_repository_config(b'[db_schema]\npaths = ["src/main/resources/db/**"]\n'))
        files = [ChangedFile("tools/report.sql", "modified", 1), ChangedFile("src/main/resources/db/1-a.sql", "modified", 1)]
        rules = evaluate_rules(files, {}, cfg)
        self.assertEqual([f.path for f in rules.schema_files], ["src/main/resources/db/1-a.sql"])
        # Without a repository list, the former built-in paths apply.
        self.assertEqual([f.path for f in evaluate_rules(files, {}, build_config(host)).schema_files],
                         ["tools/report.sql", "src/main/resources/db/1-a.sql"])
        with self.assertRaisesRegex(ConfigError, r"\[db_schema\] paths must be a list of non-empty strings"):
            config(db_schema={"paths": [""]})


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