220 lines
12 KiB
Python
220 lines
12 KiB
Python
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()
|