365 lines
16 KiB
Python
365 lines
16 KiB
Python
"""Tests for version-coupling detection.
|
|
|
|
These cover the failure mode the tool exists to catch: not "the release was
|
|
bad" but "half the matched set got flashed". All offline.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import unittest
|
|
from datetime import timedelta
|
|
|
|
from avwatch import coupling
|
|
from avwatch.config import PolicyConfig
|
|
from avwatch.models import Coupling, Finding, Item, VersionCheck, utcnow
|
|
from avwatch.risk import heuristics
|
|
from avwatch.versions import bump_kind, compare
|
|
|
|
|
|
def item(iid, installed=None, **kw):
|
|
base = dict(id=iid, name=iid.replace("-", " ").title(), vendor="TestCo",
|
|
category="video", installed=installed)
|
|
base.update(kw)
|
|
return Item(**base)
|
|
|
|
|
|
def finding(it, latest=None, age_days=90):
|
|
released = utcnow() - timedelta(days=age_days) if latest else None
|
|
chk = VersionCheck(item_id=it.id, latest=latest, released=released,
|
|
status="ok" if latest else "manual")
|
|
comp = compare(it.installed, latest) if latest else "unknown"
|
|
bump = bump_kind(it.installed, latest) if latest else "unknown"
|
|
f = Finding(item=it, check=chk, comparison=comp, bump=bump)
|
|
f.verdict = heuristics.score(it, chk, comp, bump, [], PolicyConfig())
|
|
return f
|
|
|
|
|
|
def run(items_list, findings_list):
|
|
by_id = {i.id: i for i in items_list}
|
|
return coupling.annotate(findings_list, by_id), by_id
|
|
|
|
|
|
class TestMustMatch(unittest.TestCase):
|
|
def test_existing_mismatch_is_blocking(self):
|
|
"""The switcher is already ahead of its panel -- that is broken now,
|
|
not a future risk."""
|
|
sw = item("atem", installed="10.4.1", criticality="showstopper",
|
|
coupled_with=[Coupling("panel", "must_match",
|
|
"panel drops offline")])
|
|
panel = item("panel", installed="10.2", criticality="showstopper")
|
|
fs, _ = run([sw, panel], [finding(sw, "10.4.1"), finding(panel, "10.4.1")])
|
|
|
|
sw_f = fs[0]
|
|
blocking = [c for c in sw_f.coupling if c.severity == "blocking"]
|
|
self.assertTrue(blocking, "a live mismatch must be blocking")
|
|
self.assertIn("MISMATCH NOW", blocking[0].message)
|
|
self.assertIn("10.2", blocking[0].message)
|
|
|
|
def test_matched_pair_with_update_pending_must_move_together(self):
|
|
a = item("rx", installed="1.0", coupled_with=[
|
|
Coupling("tx", "must_match", "they will not link")])
|
|
b = item("tx", installed="1.0")
|
|
fs, _ = run([a, b], [finding(a, "2.0"), finding(b, "2.0")])
|
|
|
|
blocking = [c for c in fs[0].coupling if c.severity == "blocking"]
|
|
self.assertTrue(blocking)
|
|
self.assertIn("same session", blocking[0].message)
|
|
|
|
def test_reverse_edge_is_mirrored(self):
|
|
"""Only one half of a pair declares the edge; both halves must warn."""
|
|
a = item("rx", installed="1.0", coupled_with=[
|
|
Coupling("tx", "must_match")])
|
|
b = item("tx", installed="2.0") # declares nothing
|
|
fs, _ = run([a, b], [finding(a, "2.0"), finding(b, "2.0")])
|
|
|
|
b_f = next(f for f in fs if f.item.id == "tx")
|
|
self.assertTrue(b_f.coupling,
|
|
"the item that declares no edge must still be warned")
|
|
|
|
def test_auto_suppresses_the_do_them_together_reminder(self):
|
|
"""ATEM Setup flashes switcher and panel in one pass, so reminding
|
|
the team to do both is noise about an unskippable step."""
|
|
sw = item("atem", installed="10.4", coupled_with=[
|
|
Coupling("panel", "must_match", auto=True)])
|
|
panel = item("panel", installed="10.4")
|
|
fs, _ = run([sw, panel], [finding(sw, "10.5"), finding(panel, "10.5")])
|
|
self.assertFalse(
|
|
[c for c in fs[0].coupling if c.severity == "blocking"],
|
|
"auto edges must not nag about doing both",
|
|
)
|
|
|
|
def test_auto_still_reports_a_real_divergence(self):
|
|
"""If an auto-flashed pair diverges, the automatic update FAILED --
|
|
that must still be loud."""
|
|
sw = item("atem", installed="10.4.1", coupled_with=[
|
|
Coupling("panel", "must_match", auto=True)])
|
|
panel = item("panel", installed="10.4")
|
|
fs, _ = run([sw, panel],
|
|
[finding(sw, "10.4.1"), finding(panel, "10.4.1")])
|
|
blocking = [c for c in fs[0].coupling if c.severity == "blocking"]
|
|
self.assertTrue(blocking, "a real divergence must still block")
|
|
self.assertIn("MISMATCH NOW", blocking[0].message)
|
|
|
|
def test_non_auto_still_nags(self):
|
|
sw = item("rx", installed="1.0", coupled_with=[
|
|
Coupling("tx", "must_match")])
|
|
tx = item("tx", installed="1.0")
|
|
fs, _ = run([sw, tx], [finding(sw, "2.0"), finding(tx, "2.0")])
|
|
self.assertTrue([c for c in fs[0].coupling if c.severity == "blocking"])
|
|
|
|
def test_different_numbering_schemes_flag_the_edge_not_the_gear(self):
|
|
"""DiGiCo number the console (Stealth Core 22.0.2242) and its rack
|
|
(1.11) on separate schemes. must_match there is a wrong edge, not a
|
|
real fault, and must not cry BROKEN NOW every scan."""
|
|
console = Item(id="console", name="Quantum 225", vendor="DiGiCo",
|
|
category="audio", installed="22.0.2242",
|
|
coupled_with=[Coupling("rack", "must_match")])
|
|
rack = Item(id="rack", name="DQ Rack", vendor="DiGiCo",
|
|
category="audio", installed="1.11")
|
|
fs, _ = run([console, rack],
|
|
[finding(console, "22.0.2242"), finding(rack, "1.11")])
|
|
|
|
sev = {c.severity for c in fs[0].coupling}
|
|
self.assertNotIn("blocking", sev,
|
|
"incomparable schemes must not report a real mismatch")
|
|
self.assertIn("info", sev)
|
|
self.assertTrue(any("numbering scheme" in c.message
|
|
for c in fs[0].coupling))
|
|
|
|
def test_close_versions_still_report_a_real_mismatch(self):
|
|
"""The guard must not swallow genuine mismatches."""
|
|
a = item("atem", installed="10.4.1",
|
|
coupled_with=[Coupling("panel", "must_match")])
|
|
b = item("panel", installed="10.2")
|
|
fs, _ = run([a, b], [finding(a, "10.4.1"), finding(b, "10.4.1")])
|
|
self.assertTrue([c for c in fs[0].coupling if c.severity == "blocking"])
|
|
|
|
def test_unknown_installed_is_warning_not_blocking(self):
|
|
a = item("rx", installed=None, coupled_with=[
|
|
Coupling("tx", "must_match")])
|
|
b = item("tx", installed="2.0")
|
|
fs, _ = run([a, b], [finding(a, "2.0"), finding(b, "2.0")])
|
|
sev = {c.severity for c in fs[0].coupling}
|
|
self.assertIn("warning", sev)
|
|
self.assertNotIn("blocking", sev)
|
|
|
|
def test_go_is_downgraded_when_coupled(self):
|
|
"""The headline rule: a coupled item is never plain 'go'."""
|
|
a = item("rx", installed="1.0", criticality="low", coupled_with=[
|
|
Coupling("tx", "must_match")])
|
|
b = item("tx", installed="1.0", criticality="low")
|
|
fa = finding(a, "1.0.1", age_days=400) # aged patch on low crit
|
|
self.assertEqual(fa.verdict.recommendation, "go")
|
|
|
|
fs, _ = run([a, b], [fa, finding(b, "1.0.1", age_days=400)])
|
|
self.assertNotEqual(fs[0].verdict.recommendation, "go")
|
|
self.assertTrue(any("Plan this as one job" in r
|
|
for r in fs[0].verdict.reasons))
|
|
|
|
|
|
class TestMustBeAtLeast(unittest.TestCase):
|
|
def test_tool_behind_device_is_blocking(self):
|
|
dev = item("qlxd4", installed="6.5", coupled_with=[
|
|
Coupling("wwb", "must_be_at_least", "WWB pushes device firmware")])
|
|
tool = item("wwb", installed="6.1")
|
|
fs, _ = run([dev, tool], [finding(dev, "6.5"), finding(tool, "6.5")])
|
|
|
|
blocking = [c for c in fs[0].coupling if c.severity == "blocking"]
|
|
self.assertTrue(blocking)
|
|
self.assertIn("must be at or above", blocking[0].message)
|
|
|
|
def test_tool_ahead_is_fine(self):
|
|
dev = item("qlxd4", installed="6.1", coupled_with=[
|
|
Coupling("wwb", "must_be_at_least")])
|
|
tool = item("wwb", installed="6.5")
|
|
fs, _ = run([dev, tool], [finding(dev, "6.5"), finding(tool, "6.5")])
|
|
self.assertFalse([c for c in fs[0].coupling if c.severity == "blocking"])
|
|
|
|
|
|
class TestHostOf(unittest.TestCase):
|
|
def test_major_platform_bump_blocks(self):
|
|
os_item = item("macos", installed="14.0", coupled_with=[
|
|
Coupling("protools", "host_of", "check Avid compatibility")])
|
|
app = item("protools", installed="2024.3")
|
|
fs, _ = run([os_item, app],
|
|
[finding(os_item, "15.0"), finding(app, "2024.3")])
|
|
blocking = [c for c in fs[0].coupling if c.severity == "blocking"]
|
|
self.assertTrue(blocking)
|
|
self.assertIn("platform change", blocking[0].message)
|
|
|
|
def test_patch_platform_bump_does_not_block(self):
|
|
os_item = item("macos", installed="14.0", coupled_with=[
|
|
Coupling("protools", "host_of")])
|
|
app = item("protools", installed="2024.3")
|
|
fs, _ = run([os_item, app],
|
|
[finding(os_item, "14.0.1"), finding(app, "2024.3")])
|
|
self.assertFalse([c for c in fs[0].coupling if c.severity == "blocking"])
|
|
|
|
def test_significant_any_blocks_on_a_patch(self):
|
|
"""NETGEAR's M4250 puts breaking changes in the third digit, so
|
|
13.0.4.x -> 13.0.5.x reads as a 'patch' but is anything but."""
|
|
sw = item("switch", installed="13.0.4.26", criticality="showstopper",
|
|
coupled_with=[Coupling("dante", "host_of",
|
|
note="the whole Dante network",
|
|
significant="any")])
|
|
dante = item("dante", installed="4.8")
|
|
fs, _ = run([sw, dante],
|
|
[finding(sw, "13.0.5.26"), finding(dante, "4.8")])
|
|
|
|
self.assertEqual(fs[0].bump, "patch", "sanity: reads as a patch")
|
|
blocking = [c for c in fs[0].coupling if c.severity == "blocking"]
|
|
self.assertTrue(blocking, "significant: any must still block")
|
|
self.assertIn("Any update", blocking[0].message)
|
|
|
|
def test_significant_major_stays_quiet_on_a_patch(self):
|
|
sw = item("switch", installed="13.0.4.26", coupled_with=[
|
|
Coupling("dante", "host_of")]) # default: major
|
|
dante = item("dante", installed="4.8")
|
|
fs, _ = run([sw, dante],
|
|
[finding(sw, "13.0.5.26"), finding(dante, "4.8")])
|
|
self.assertFalse([c for c in fs[0].coupling if c.severity == "blocking"])
|
|
|
|
|
|
class TestCarrierAndContent(unittest.TestCase):
|
|
def test_carrier_warns_when_device_behind(self):
|
|
tool = item("suu", installed="1.0", coupled_with=[
|
|
Coupling("ad4q", "carries_firmware_for")])
|
|
dev = item("ad4q", installed="1.0")
|
|
fs, _ = run([tool, dev], [finding(tool, "1.0"), finding(dev, "2.0")])
|
|
msgs = [c.message for c in fs[0].coupling]
|
|
self.assertTrue(any("Update this tool first" in m for m in msgs))
|
|
|
|
def test_content_warns_when_engine_behind(self):
|
|
show = item("vista-show", installed="1.0", coupled_with=[
|
|
Coupling("vista", "content_for")])
|
|
engine = item("vista", installed="3.1")
|
|
fs, _ = run([show, engine], [finding(show, "1.0"), finding(engine, "3.2")])
|
|
msgs = [c.message for c in fs[0].coupling]
|
|
self.assertTrue(any("Back this up first" in m for m in msgs))
|
|
|
|
|
|
class TestValidation(unittest.TestCase):
|
|
def test_dangling_id_reported(self):
|
|
a = item("a", coupled_with=[Coupling("nope", "must_match")])
|
|
problems = coupling.validate({"a": a})
|
|
self.assertTrue(any("not an inventory id" in p for p in problems))
|
|
|
|
def test_unknown_relation_reported(self):
|
|
a = item("a", coupled_with=[Coupling("b", "sometimes_maybe")])
|
|
b = item("b")
|
|
problems = coupling.validate({"a": a, "b": b})
|
|
self.assertTrue(any("unknown relation" in p for p in problems))
|
|
|
|
def test_self_coupling_reported(self):
|
|
a = item("a", coupled_with=[Coupling("a", "must_match")])
|
|
problems = coupling.validate({"a": a})
|
|
self.assertTrue(any("coupled to itself" in p for p in problems))
|
|
|
|
def test_dangling_id_is_info_not_blocking(self):
|
|
a = item("a", installed="1.0", coupled_with=[
|
|
Coupling("nope", "must_match")])
|
|
fs, _ = run([a], [finding(a, "2.0")])
|
|
self.assertEqual([c.severity for c in fs[0].coupling], ["info"])
|
|
|
|
def test_groups_only_returns_real_sets(self):
|
|
a = item("a", update_group="atem")
|
|
b = item("b", update_group="atem")
|
|
c = item("c", update_group="lonely")
|
|
grps = coupling.groups({"a": a, "b": b, "c": c})
|
|
self.assertIn("atem", grps)
|
|
self.assertNotIn("lonely", grps, "a group of one is not a set")
|
|
|
|
|
|
class TestUpdateClasses(unittest.TestCase):
|
|
"""Rows where 'should we install it?' is the wrong question."""
|
|
|
|
def setUp(self):
|
|
self.policy = PolicyConfig()
|
|
|
|
def _v(self, it, latest="2.0"):
|
|
chk = VersionCheck(item_id=it.id, latest=latest, status="ok")
|
|
return heuristics.score(it, chk, "behind", "major", [], self.policy)
|
|
|
|
def test_do_not_touch_never_recommended(self):
|
|
it = item("fire-alarm", installed="1.0", criticality="showstopper",
|
|
update_class="do_not_touch")
|
|
v = self._v(it)
|
|
self.assertEqual(v.recommendation, "avoid")
|
|
self.assertTrue(any("contractor" in r for r in v.reasons))
|
|
|
|
def test_vendor_pushed_gives_no_go_no_go(self):
|
|
it = item("youtube", installed="1.0", update_class="vendor_pushed")
|
|
v = self._v(it)
|
|
self.assertEqual(v.recommendation, "manual-check")
|
|
self.assertTrue(any("whether or not" in r or "no install decision" in r
|
|
for r in v.reasons))
|
|
|
|
def test_expiry_class_mentions_date(self):
|
|
it = item("ccli", installed="1.0", update_class="expiry",
|
|
expires="2027-03-01")
|
|
v = self._v(it)
|
|
self.assertTrue(any("2027-03-01" in r for r in v.reasons))
|
|
|
|
def test_unconfirmed_gives_no_advice(self):
|
|
it = item("main-pa", installed=None, criticality="high",
|
|
status="unconfirmed")
|
|
v = self._v(it)
|
|
self.assertEqual(v.recommendation, "manual-check")
|
|
self.assertTrue(any("guessed product" in r for r in v.reasons))
|
|
|
|
|
|
class TestShippedCouplingGraph(unittest.TestCase):
|
|
"""The real inventory files must have a sound coupling graph."""
|
|
|
|
def _items(self):
|
|
from pathlib import Path
|
|
|
|
from avwatch import inventory
|
|
|
|
path = Path(__file__).resolve().parent.parent / "inventory.yaml"
|
|
if not path.exists():
|
|
self.skipTest("inventory.yaml not present")
|
|
return inventory.by_id(inventory.load(path))
|
|
|
|
def test_graph_resolves(self):
|
|
problems = coupling.validate(self._items())
|
|
self.assertEqual(problems, [], "coupling graph has dangling edges")
|
|
|
|
def test_must_match_edges_are_reciprocal_or_mirrored(self):
|
|
"""Not a hard requirement (annotate mirrors them), but every
|
|
must_match target should at least exist and share the update_group."""
|
|
items = self._items()
|
|
for it in items.values():
|
|
for link in it.coupled_with:
|
|
if link.relation != "must_match":
|
|
continue
|
|
other = items[link.id]
|
|
if it.update_group and other.update_group:
|
|
self.assertEqual(
|
|
it.update_group, other.update_group,
|
|
f"{it.id} and {other.id} must_match but are in "
|
|
f"different update groups",
|
|
)
|
|
|
|
def test_unconfirmed_items_have_no_query(self):
|
|
"""An unconfirmed row must not carry a search query -- that is how a
|
|
guessed vendor's version reaches the report."""
|
|
from pathlib import Path
|
|
|
|
from avwatch import inventory
|
|
|
|
root = Path(__file__).resolve().parent.parent
|
|
for f in inventory.resolve_paths(root / "inventory.yaml"):
|
|
for it in inventory.load(f):
|
|
if it.is_confirmed:
|
|
continue
|
|
q = (it.source or {}).get("query", "")
|
|
self.assertFalse(
|
|
str(q).strip(),
|
|
f"{it.id} in {f.name} is unconfirmed but has query {q!r}",
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main(verbosity=2)
|