"""nexa validate: the gate, and every way it is supposed to close.

`cmd_validate` is the only thing standing between a broken rule and a corpus
that ships it. Everything downstream trusts it: `nexa build` runs it, the
Modelfiles are generated from rules that passed it, and the retrieval index
embeds them.

A validator whose failure modes are untested is a validator that can quietly
stop failing, and nothing else in the repository would notice -- a rule with a
pattern that matches nothing looks exactly like a rule with a pattern that
works. So there is one test per way the gate closes, each breaking exactly one
thing in an otherwise-valid rule.

The fixtures write a corpus directory of their own. Validating the real corpus
is what `nexa validate` does; what is under test here is the gate.
"""
import io
import json

import pytest

import corpus


# A rule that passes every check, so each test can break exactly one thing and
# be sure that is what the error is about.
def good_rule(**overrides):
    rule = {
        "id": "python.a-sample-rule",
        "stack": "odoo",
        "severity": "high",
        # The second axis. `unmeasured` is the default the build gives a rule
        # nothing has placed, and it derives advisory false.
        "likelihood": "unmeasured",
        "title": "A sample rule for the validator tests",
        "rule": "Do the correct thing rather than the incorrect thing.",
        "why": "Because the incorrect thing fails silently and late.",
        "wrong": "value = eval(payload)",
        "correct": "value = json.loads(payload)",
        "tags": ["sample"],
    }
    rule.update(overrides)
    return rule


@pytest.fixture()
def corpus_dir(tmp_path, monkeypatch):
    """Point the loader at a corpus this test owns."""
    path = tmp_path / "rules"
    path.mkdir()
    monkeypatch.setattr(corpus, "RULES_DIR", str(path))

    def write(rules, name="odoo.json"):
        with io.open(str(path / name), "w", encoding="utf-8") as handle:
            handle.write(json.dumps(rules, ensure_ascii=False))

    return write


def failures(capsys):
    """The error lines from the run that just happened."""
    out = capsys.readouterr().out
    return [line.strip() for line in out.splitlines() if line.startswith("  ")]


# ------------------------------------------------------------------ the gate

def test_a_valid_rule_passes(corpus_dir, capsys):
    corpus_dir([good_rule()])
    assert corpus.cmd_validate() == 0
    assert "PASS" in capsys.readouterr().out


def test_an_empty_corpus_is_a_failure_not_a_pass(corpus_dir, capsys):
    """Nothing to validate is not the same as nothing wrong. A build that
    silently produced an empty corpus would pass every check below it."""
    corpus_dir([])
    assert corpus.cmd_validate() == 1
    assert "corpus is empty" in capsys.readouterr().out


@pytest.mark.parametrize("field", ["id", "stack", "severity", "likelihood",
                                   "title", "rule", "why", "wrong", "correct",
                                   "tags"])
def test_a_missing_required_field_fails(corpus_dir, capsys, field):
    rule = good_rule()
    del rule[field]
    corpus_dir([rule])
    assert corpus.cmd_validate() == 1
    assert any("missing required field '{0}'".format(field) in line
               for line in failures(capsys))


def test_an_unknown_field_fails(corpus_dir, capsys):
    """additionalProperties is false, so a typo in a key name is a silent
    no-op rather than the setting somebody thought they wrote."""
    corpus_dir([good_rule(violation_patern=r"eval\(")])
    assert corpus.cmd_validate() == 1
    assert any("unknown field 'violation_patern'" in line
               for line in failures(capsys))


def test_a_duplicate_id_fails(corpus_dir, capsys):
    """Ids are what a model cites. Two rules answering to one id means a
    citation cannot be checked against anything."""
    corpus_dir([good_rule(), good_rule(title="A second rule with the same id")])
    assert corpus.cmd_validate() == 1
    assert any("duplicate id" in line for line in failures(capsys))


def test_an_id_with_an_unknown_origin_prefix_fails(corpus_dir, capsys):
    """The prefix names where a rule came from -- odoo. or python. -- and it is
    what every measurement and every citation is keyed on. A rule whose prefix
    is neither is orphaned from its own classification artifact, silently."""
    corpus_dir([good_rule(id="ruff.a-sample-rule")])
    assert corpus.cmd_validate() == 1
    assert any("id prefix is not one of" in line for line in failures(capsys))


def test_both_origin_prefixes_are_accepted_under_the_one_stack(corpus_dir):
    """python.decimal-for-money-not-float carries stack odoo. The prefix is
    provenance, the stack is selection, and they are allowed to differ."""
    corpus_dir([good_rule(id="odoo.a-sample-rule"),
                good_rule(id="python.another-sample-rule",
                          title="A second sample rule for the validator")])
    assert corpus.cmd_validate() == 0


def test_ui_only_enforcement_must_say_what_bypasses_it(corpus_dir, capsys):
    """A rule enforced only in the UI is a rule every import, XML-RPC call and
    sudo write goes around. Declaring the layer without the escape routes is
    the claim that reads as safest and is least true."""
    corpus_dir([good_rule(enforcement_layer="ui-only", bypassed_by=[])])
    assert corpus.cmd_validate() == 1
    assert any("must list what bypasses it" in line for line in failures(capsys))


def test_identical_wrong_and_correct_samples_fail(corpus_dir, capsys):
    """The pair is the rule's whole explanation. Identical, it teaches nothing
    and every pattern self-test below passes vacuously."""
    corpus_dir([good_rule(wrong="x = 1", correct="x = 1")])
    assert corpus.cmd_validate() == 1
    assert any("identical" in line for line in failures(capsys))


# --------------------------------------------------------- pattern self-tests

def test_a_pattern_that_does_not_compile_fails(corpus_dir, capsys):
    corpus_dir([good_rule(violation_pattern=r"eval\(")])
    assert corpus.cmd_validate() == 0, "the control case compiles and matches"
    corpus_dir([good_rule(violation_pattern=r"eval\((")])
    assert corpus.cmd_validate() == 1
    assert any("does not compile" in line for line in failures(capsys))


def test_a_pattern_that_misses_its_own_wrong_sample_fails(corpus_dir, capsys):
    """It catches nothing, and nothing downstream would ever say so."""
    corpus_dir([good_rule(violation_pattern=r"exec\(")])
    assert corpus.cmd_validate() == 1
    assert any("does not match its own 'wrong' sample" in line
               for line in failures(capsys))


def test_a_pattern_that_fires_on_its_own_correct_sample_fails(corpus_dir, capsys):
    """Worse than no pattern: it fires on good code and trains the reader to
    ignore the checker."""
    corpus_dir([good_rule(violation_pattern=r"payload")])
    assert corpus.cmd_validate() == 1
    assert any("fires on its own 'correct' sample" in line
               for line in failures(capsys))


def test_a_satisfied_pattern_that_clears_the_wrong_sample_fails(corpus_dir, capsys):
    """It clears the violation on the very code the rule is about, so the
    check is inert everywhere."""
    corpus_dir([good_rule(violation_pattern=r"eval\(",
                          satisfied_pattern=r"value")])
    assert corpus.cmd_validate() == 1
    assert any("makes the check inert" in line for line in failures(capsys))


def test_a_satisfied_pattern_alone_fails(corpus_dir, capsys):
    """There is no violation for it to clear, so it does nothing at all."""
    corpus_dir([good_rule(satisfied_pattern=r"json")])
    assert corpus.cmd_validate() == 1
    assert any("has no effect" in line for line in failures(capsys))


# -------------------------------------------------------- semantic self-tests

ODOO_WRONG = (
    "@api.depends('line_ids')\n"
    "def _compute_total(self):\n"
    "    for rec in self:\n"
    "        rec.total = sum(l.price for l in rec.line_ids)"
)
ODOO_CORRECT = (
    "@api.depends('line_ids', 'line_ids.price')\n"
    "def _compute_total(self):\n"
    "    for rec in self:\n"
    "        rec.total = sum(l.price for l in rec.line_ids)"
)


def odoo_rule(**overrides):
    fields = {"id": "odoo.a-sample-rule", "stack": "odoo",
              "wrong": ODOO_WRONG, "correct": ODOO_CORRECT}
    fields.update(overrides)
    return good_rule(**fields)


def test_a_semantic_check_self_tests_like_a_pattern(corpus_dir, capsys):
    corpus_dir([odoo_rule(semantic_check="depends_covers_compute_reads")],
               name="odoo.json")
    assert corpus.cmd_validate() == 0, capsys.readouterr().out


def test_a_semantic_check_that_does_not_exist_fails(corpus_dir, capsys):
    """A renamed checker that silently returned clean would turn a critical
    rule into a permanent false pass."""
    corpus_dir([odoo_rule(semantic_check="no_such_checker")], name="odoo.json")
    assert corpus.cmd_validate() == 1
    assert any("no semantic checker named" in line for line in failures(capsys))


def test_a_semantic_check_that_misses_its_own_wrong_sample_fails(corpus_dir,
                                                                 capsys):
    corpus_dir([odoo_rule(wrong=ODOO_CORRECT, correct=ODOO_WRONG,
                          semantic_check="depends_covers_compute_reads")],
               name="odoo.json")
    assert corpus.cmd_validate() == 1
    lines = failures(capsys)
    assert any("does not fire on its own" in line for line in lines)
    assert any("fires on its own 'correct'" in line for line in lines)


def test_a_rule_cannot_carry_both_kinds_of_check(corpus_dir, capsys):
    """The domain layer defers to the semantic layer, so the pattern would be
    silently ignored -- and a pattern nobody runs is one nobody maintains."""
    corpus_dir([odoo_rule(semantic_check="depends_covers_compute_reads",
                          violation_pattern=r"api\.depends")],
               name="odoo.json")
    assert corpus.cmd_validate() == 1
    assert any("both a violation_pattern and a semantic_check" in line
               for line in failures(capsys))


# ------------------------------------------------------------- check_value

def test_a_schema_keyword_this_checker_does_not_know_is_an_error():
    """Deliberately not a general JSON Schema implementation: it fails loudly
    on a keyword it does not understand rather than skipping it silently,
    which would validate a rule against less than the schema says."""
    errors = []
    corpus.check_value("x", {"type": "string", "multipleOf": 2}, "p", errors)
    assert any("unsupported keyword 'multipleOf'" in e for e in errors)


@pytest.mark.parametrize("value, spec, fragment", [
    (5, {"type": "string"}, "expected string"),
    ("z", {"enum": ["a", "b"]}, "not in"),
    ("Nope", {"pattern": r"^[a-z]+$"}, "does not match"),
    ("ab", {"minLength": 5}, "below minimum"),
    ("abcdef", {"maxLength": 3}, "above maximum"),
    ([], {"minItems": 1}, "minimum"),
])
def test_each_keyword_reports_its_own_failure(value, spec, fragment):
    errors = []
    corpus.check_value(value, spec, "p", errors)
    assert errors and any(fragment in e for e in errors), errors


def test_a_bool_is_not_an_integer():
    """True == 1 in Python, so a schema saying integer would accept a boolean
    unless the checker says otherwise."""
    errors = []
    corpus.check_value(True, {"type": "integer"}, "p", errors)
    assert errors
    errors = []
    corpus.check_value(True, {"type": "boolean"}, "p", errors)
    assert not errors


def test_items_are_checked_individually_and_named_by_index():
    errors = []
    corpus.check_value(["ok", 7], {"type": "array", "items": {"type": "string"}},
                       "tags", errors)
    assert any("tags[1]" in e for e in errors)
    assert not any("tags[0]" in e for e in errors)


def test_a_type_failure_stops_the_other_keywords():
    """One error about the value being the wrong type, not five about a string
    keyword failing on an integer."""
    errors = []
    corpus.check_value(5, {"type": "string", "minLength": 3,
                           "pattern": r"^[a-z]+$"}, "p", errors)
    assert len(errors) == 1


# ------------------------------------------- the two axes, and the derived gate

def test_a_likelihood_outside_the_vocabulary_fails(corpus_dir, capsys):
    corpus_dir([good_rule(likelihood="probably-fine")])
    assert corpus.cmd_validate() == 1
    assert any("likelihood" in line for line in failures(capsys))


def test_advisory_disagreeing_with_likelihood_fails(corpus_dir, capsys):
    """advisory is derived from likelihood by the build. The two can only
    disagree if the JSON was hand-edited, and a rule that gates while grading
    itself a preference gates on something nobody decided."""
    corpus_dir([good_rule(likelihood="preference", advisory=False)])
    assert corpus.cmd_validate() == 1
    assert any("derived, not authored" in line for line in failures(capsys))


def test_advisory_claimed_by_a_measured_rule_fails(corpus_dir, capsys):
    """The other direction: a `defect` rule cannot opt out of gating by setting
    the flag, which is exactly how a one-bit axis used to be overridable."""
    corpus_dir([good_rule(likelihood="defect", advisory=True)])
    assert corpus.cmd_validate() == 1
    assert any("derived, not authored" in line for line in failures(capsys))


def test_both_advisory_grades_derive_the_gate(corpus_dir, capsys):
    for grade in corpus.ADVISORY_LIKELIHOODS:
        corpus_dir([good_rule(likelihood=grade, advisory=True)])
        assert corpus.cmd_validate() == 0, capsys.readouterr().out


def test_likelihood_rank_orders_the_vocabulary():
    """The enum is in rank order and report.py reads findings in it, so the
    order is part of the contract rather than an accident of the literal."""
    ranked = sorted(corpus.LIKELIHOOD_ORDER, key=corpus.likelihood_rank)
    assert ranked == list(corpus.LIKELIHOOD_ORDER)
    assert corpus.likelihood_rank("defect") < corpus.likelihood_rank("construct")
    assert corpus.likelihood_rank("construct") < corpus.likelihood_rank("preference")


def test_an_unknown_grade_sorts_last_rather_than_raising():
    """Report ordering is not the place to fail a run; validate has already
    refused an unknown grade against the schema enum."""
    assert corpus.likelihood_rank("nonsense") == len(corpus.LIKELIHOOD_ORDER)


# --------------------------------------------- the order findings are read in

class _Hit(object):
    """A stand-in for layers.SourceHit: rank_findings reads three attributes."""

    def __init__(self, rule, line, likelihood):
        self.rule = rule
        self.line = line
        self.likelihood = likelihood


def test_findings_are_ranked_by_likelihood_before_line():
    """History has an opinion about which rule is worth reading. sudo landed on
    a line somebody changed 13 times in 41; prefer-sql did it none in 49. Line
    order tells the reader nothing about that."""
    prefer_sql = _Hit("odoo.prefer-sql-constraint", 12, "preference")
    sudo = _Hit("odoo.sudo-bypasses-record-rules", 900, "defect")
    ranked = corpus.rank_findings([prefer_sql, sudo])
    assert [h.rule for h in ranked] == [sudo.rule, prefer_sql.rule]


def test_line_order_survives_inside_a_grade():
    """So a run of equally likely findings still reads down the file."""
    late = _Hit("odoo.a-rule", 90, "defect")
    early = _Hit("odoo.a-rule", 4, "defect")
    assert [h.line for h in corpus.rank_findings([late, early])] == [4, 90]


def test_every_grade_sorts_in_the_documented_order():
    hits = [_Hit("r." + g, 1, g) for g in reversed(corpus.LIKELIHOOD_ORDER)]
    ranked = [h.likelihood for h in corpus.rank_findings(hits)]
    assert ranked == list(corpus.LIKELIHOOD_ORDER)


def test_a_finding_with_no_grade_still_ranks():
    """Ordering a report is not the place to raise. A hit from something that
    predates the axis sorts last rather than taking the run down."""
    class Bare(object):
        rule, line = "odoo.a-rule", 3

    ranked = corpus.rank_findings([Bare(), _Hit("odoo.b-rule", 99, "defect")])
    assert ranked[0].rule == "odoo.b-rule"