christianandClaude Fable 5.1
0eee778a92
A draft started before someone else's Apply used to carry that older config and would have put the newer changes back. Now every read of the draft replays only the draft's own changes onto the live settings (base config from history), so the review lists exactly what you changed. Values read in words (windows as days and hours), long ones fold open with was/now, commands are no longer red, and warnings the live config already had are not repeated. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
358 lines
22 KiB
Python
358 lines
22 KiB
Python
"""The settings page's server and probes, offline: routes against the fake Companion, the LAN
|
||
binding, the editing-code gate, the probe parsers on captured output, and the teaching pages."""
|
||
import json
|
||
import os
|
||
import socket
|
||
import sys
|
||
import tempfile
|
||
import threading
|
||
import time
|
||
import unittest
|
||
import urllib.error
|
||
import urllib.request
|
||
from http.server import ThreadingHTTPServer
|
||
|
||
ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||
sys.path.insert(0, ROOT)
|
||
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
||
import status_panel as sp # noqa: E402
|
||
from settings import keyguide, probes, server # noqa: E402
|
||
from test_status_panel import CONFIG, FakeCompanion, healthy_snapshot # noqa: E402
|
||
|
||
|
||
def get(port, path, json_body=None, origin=None, method=None, code=None, who=None):
|
||
data = json.dumps(json_body).encode() if json_body is not None else None
|
||
headers = {"Content-Type": "application/json", **({"Origin": origin} if origin else {}),
|
||
**({"X-Edit-Key": code} if code else {}), **({"X-Editor": who} if who else {})}
|
||
req = urllib.request.Request("http://127.0.0.1:%d%s" % (port, path), data=data, headers=headers, method=method)
|
||
try:
|
||
with urllib.request.urlopen(req, timeout=10) as r:
|
||
body = r.read()
|
||
return r.status, body
|
||
except urllib.error.HTTPError as e:
|
||
return e.code, e.read()
|
||
|
||
|
||
class TestServer(unittest.TestCase):
|
||
@classmethod
|
||
def setUpClass(cls):
|
||
snap = healthy_snapshot(time.time())
|
||
FakeCompanion.store = {"conns": snap.conns, "vars": dict(snap.vars),
|
||
"custom": {"sp_state": json.dumps({"all": {"x": "ALL OK", "b": 1, "f": 2, "b2": 1, "f2": 2, "s": 20}}),
|
||
"sp_hb": str(int(time.time())), "sp_cmd": "", "flash_tick": "0"}, "restarts": []}
|
||
cls.fake = ThreadingHTTPServer(("127.0.0.1", 0), FakeCompanion)
|
||
threading.Thread(target=cls.fake.serve_forever, daemon=True).start()
|
||
cls.data = tempfile.mkdtemp(prefix="sp-settings-")
|
||
cfg = json.loads(json.dumps(CONFIG))
|
||
cfg["companion"] = {"host": "127.0.0.1", "port": cls.fake.server_address[1]}
|
||
cfg["site"]["editing_code"] = "apple river stone"
|
||
json.dump(cfg, open(os.path.join(cls.data, "config.json"), "w"))
|
||
now = time.time()
|
||
json.dump({"pid": 1, "started_at": now - 100, "config_sha": "x", "version": "test"}, open(os.path.join(cls.data, "started.json"), "w"))
|
||
json.dump({"live": True, "reason": "a HyperDeck is recording", "time": now}, open(os.path.join(cls.data, "live-now.json"), "w"))
|
||
json.dump({"time": now, "conns": {}, "faults": {"atem": [{"check": "ATEM", "type": "conn", "sub": "conn", "level": "err",
|
||
"msg": "NO ANSWER", "since": now - 60, "acked": False}],
|
||
"midi": [{"check": "tracks/NA Tracks/Lighting PC", "type": "midi_peer", "sub": "peer",
|
||
"level": "err", "msg": "FX LIGHTS MIDI LOST", "since": now - 60, "acked": False}]},
|
||
"midi": {"NA Ingest": {"peers": [{"name": "Lighting PC", "state": "running"}]}}, "remote_midi": {},
|
||
"http": {"http://10.6.7.13:8080/": True}, "ping": {"10.1.3.227": 0}, "service_on": True, "dante_map": {}},
|
||
open(os.path.join(cls.data, "snapshot.json"), "w"))
|
||
sock = socket.socket(); sock.bind(("127.0.0.1", 0)); cls.port = sock.getsockname()[1]; sock.close()
|
||
cls.site, cls.servers = server.serve(cls.data, cls.port, ["127.0.0.1"])
|
||
|
||
@classmethod
|
||
def tearDownClass(cls):
|
||
for s in cls.servers:
|
||
s.shutdown()
|
||
cls.fake.shutdown()
|
||
|
||
def test_reads_are_open_and_secrets_are_masked(self):
|
||
status, body = get(self.port, "/api/config")
|
||
self.assertEqual(status, 200)
|
||
cfg = json.loads(body)["config"]
|
||
self.assertEqual(cfg["site"]["editing_code"], "set (17 characters)")
|
||
self.assertNotIn("apple river stone", body.decode())
|
||
|
||
def test_overview_shows_the_deck_the_problems_and_the_live_state(self):
|
||
status, body = get(self.port, "/api/overview")
|
||
o = json.loads(body)
|
||
self.assertEqual(o["deck"]["all"]["x"], "ALL OK")
|
||
self.assertTrue(o["companion_ok"])
|
||
self.assertEqual([p["text"] for p in o["problems"]], ["ATEM: NO ANSWER", "MIDI: FX LIGHTS MIDI LOST"])
|
||
words = {c["alias"]: c["word"] for c in o["computers"]}
|
||
self.assertEqual((words["tracks"], words["lighting"], words["prop"], words["ingest"]), ("Problem", "Problem", "Not looked yet", "OK"))
|
||
self.assertIn("hold the ATEM key", o["problems"][0]["action"])
|
||
self.assertTrue(o["live"]["live"])
|
||
self.assertEqual(len(o["computers"]), len(CONFIG["hosts"]))
|
||
self.assertEqual(o["watchdog"]["version"], "test")
|
||
|
||
def test_companion_view_lists_connections_with_their_watchers(self):
|
||
o = json.loads(get(self.port, "/api/companion")[1])
|
||
by = {c["label"]: c for c in o["connections"]}
|
||
self.assertIn("hd2", by["HyperDeck_2"]["watched_by"])
|
||
self.assertTrue(o["lights"]["variable sp_state"])
|
||
self.assertEqual(o["missing"], [])
|
||
|
||
def test_machines_use_what_the_watchdog_sees(self):
|
||
o = json.loads(get(self.port, "/api/machines")[1])
|
||
ingest = next(c for c in o["computers"] if c["self"])
|
||
self.assertEqual(ingest["word"], "OK")
|
||
bible = next(c for c in o["computers"] if c["alias"] == "bibletv")
|
||
web = next(i for i in bible["items"] if i["kind"] == "web")
|
||
self.assertEqual(web["light"], "ok") # from snapshot.json, not a probe
|
||
self.assertIn("mac", o["setup_steps"])
|
||
|
||
def test_gear_and_keys_and_pages(self):
|
||
gear = json.loads(get(self.port, "/api/gear")[1])
|
||
self.assertTrue(any(r["name"] == "HyperDeck_2" and "hd2" in r["watched_by"] for r in gear))
|
||
keys = json.loads(get(self.port, "/api/keys")[1])
|
||
hd2 = next(t for t in keys["tiles"] if t["id"] == "hd2")
|
||
self.assertIn("NO DISK", hd2["says"])
|
||
self.assertIn("off and on", hd2["holding"])
|
||
self.assertEqual(keys["hold"], "HOLD 2s")
|
||
for page in ("/key-guide.html", "/cheat.html", "/where.html", "/", "/app.js", "/style.css"):
|
||
status, body = get(self.port, page)
|
||
self.assertEqual(status, 200, page)
|
||
guide = get(self.port, "/key-guide.html")[1].decode()
|
||
for t in CONFIG["tiles"]:
|
||
self.assertIn(t["label"], guide)
|
||
self.assertNotIn("?", keyguide.cheat_sheet(CONFIG, "http://x/").split("<h2>Where to look")[0].split("</h1>")[1][:200])
|
||
|
||
def test_logs_are_whitelisted(self):
|
||
self.assertEqual(get(self.port, "/api/logs/nope")[0], 404)
|
||
self.assertEqual(get(self.port, "/api/logs/watchdog")[0], 200)
|
||
self.assertEqual(json.loads(get(self.port, "/api/logs/autofix")[1])["file"], None)
|
||
|
||
def test_editing_code_gate_and_lockout(self):
|
||
self.site.attempts.clear()
|
||
self.assertFalse(json.loads(get(self.port, "/api/editing/check", {"code": "nope"})[1])["ok"])
|
||
self.assertTrue(json.loads(get(self.port, "/api/editing/check", {"code": "Apple River Stone"})[1])["ok"])
|
||
for _ in range(5):
|
||
get(self.port, "/api/editing/check", {"code": "wrong"})
|
||
r = json.loads(get(self.port, "/api/editing/check", {"code": "apple river stone"})[1])
|
||
self.assertFalse(r["ok"])
|
||
self.assertIn("locked_for", r)
|
||
self.site.attempts.clear()
|
||
self.assertEqual(get(self.port, "/api/editing/check", {"code": "x"}, origin="http://evil.example")[0], 403)
|
||
|
||
# ---- step 2: the write routes ----
|
||
CODE = "apple river stone"
|
||
|
||
def put(self, path, body, method="PUT", who="sam", code=None):
|
||
st, raw = get(self.port, path, body, method=method, code=code or self.CODE, who=who)
|
||
return st, json.loads(raw)
|
||
|
||
def test_writes_need_the_code_and_a_name(self):
|
||
self.site.attempts.clear()
|
||
self.assertEqual(get(self.port, "/api/draft", {"config": {}}, method="PUT")[0], 401)
|
||
self.assertEqual(get(self.port, "/api/draft", {"config": {}}, method="PUT", code="wrong", who="sam")[0], 401)
|
||
self.assertEqual(get(self.port, "/api/draft", {"config": {}}, method="PUT", code=self.CODE)[0], 400) # no name
|
||
self.site.attempts.clear()
|
||
|
||
def test_draft_round_trip_conflict_and_discard(self):
|
||
self.site.attempts.clear()
|
||
live = self.site.load()
|
||
mine = json.loads(json.dumps(live)); mine["debounce_seconds"] = 4
|
||
st, r = self.put("/api/draft", {"config": mine})
|
||
self.assertEqual(st, 200)
|
||
v1 = r["draft"]["version"]
|
||
self.assertEqual(r["draft"]["config"]["debounce_seconds"], 4)
|
||
self.assertTrue(r["draft"]["config"]["site"]["editing_code"].startswith("set (")) # masked on the way out
|
||
self.assertEqual(json.loads(get(self.port, "/api/draft")[1])["draft"]["by"], "sam")
|
||
theirs = json.loads(json.dumps(live)); theirs["debounce_seconds"] = 5
|
||
st, r = self.put("/api/draft", {"config": theirs, "expect": v1}, who="chris")
|
||
self.assertEqual(st, 200)
|
||
st, r = self.put("/api/draft", {"config": mine, "expect": v1}) # sam's tab is behind
|
||
self.assertEqual(st, 409)
|
||
self.assertIn("chris", r["error"])
|
||
self.assertEqual(r["draft"]["config"]["debounce_seconds"], 5)
|
||
self.assertEqual(self.site.draft.read()["config"]["site"]["editing_code"], self.CODE) # the secret survived the masked round trip
|
||
st, r = self.put("/api/validate", {})
|
||
self.assertEqual(st, 200)
|
||
self.assertTrue(any("debounce seconds: %d → 5" % live["debounce_seconds"] in c["text"] for c in r["changes"]), r["changes"])
|
||
self.assertIn("purple", r["feel"].lower())
|
||
self.assertTrue(r["live"]["live"])
|
||
self.assertEqual(self.put("/api/draft", {}, method="DELETE")[0], 200)
|
||
self.assertIsNone(json.loads(get(self.port, "/api/draft")[1])["draft"])
|
||
|
||
def test_apply_waits_for_live_then_runs_restores_and_undoes(self):
|
||
from test_apply import FakeLaunchctl
|
||
self.site.attempts.clear()
|
||
fake = FakeLaunchctl(self.data)
|
||
self.site.applier.run, self.site.applier.sleep = fake, lambda s: None
|
||
live = self.site.load()
|
||
mine = json.loads(json.dumps(live)); mine["debounce_seconds"] = 6
|
||
self.put("/api/draft", {"config": mine})
|
||
st, r = self.put("/api/apply", {"note": "slower"}, method="POST")
|
||
self.assertEqual(st, 409)
|
||
self.assertTrue(r.get("live"))
|
||
self.assertIsNotNone(self.site.draft.read()) # the draft is kept
|
||
st, r = self.put("/api/apply", {"note": "slower", "emergency": {"word": "LIVE", "reason": "deck wrong"}}, method="POST")
|
||
self.assertEqual(st, 200, r)
|
||
self.site.applier.thread.join(30)
|
||
self.assertIsNone(self.site.draft.read())
|
||
status = json.loads(get(self.port, "/api/apply/status")[1])
|
||
self.assertEqual(status["state"], "done", status)
|
||
self.assertEqual(self.site.load()["debounce_seconds"], 6)
|
||
hist = json.loads(get(self.port, "/api/history")[1])["entries"]
|
||
self.assertEqual((hist[0]["by"], hist[0]["note"]), ("sam", "slower"))
|
||
detail = json.loads(get(self.port, "/api/history/" + hist[0]["id"])[1])
|
||
self.assertTrue(any("debounce seconds: %d → 6" % live["debounce_seconds"] in c["text"] for c in detail["changes_to_now"]), detail)
|
||
st, r = self.put("/api/restore/" + hist[0]["id"], {}, method="POST")
|
||
self.assertEqual(r["draft"]["config"]["debounce_seconds"], live["debounce_seconds"])
|
||
self.put("/api/draft", {}, method="DELETE")
|
||
st, r = self.put("/api/undo", {}, method="POST")
|
||
self.assertEqual((st, r["draft"]["config"]["debounce_seconds"], r["note"]), (200, live["debounce_seconds"], "slower"))
|
||
self.put("/api/draft", {}, method="DELETE")
|
||
# watchdog control waits for live too; a command goes to the watchdog through Companion
|
||
self.assertEqual(self.put("/api/watchdog/restart", {}, method="POST")[0], 409)
|
||
st, r = self.put("/api/command", {"text": "service off"}, method="POST")
|
||
self.assertEqual((st, r["sent"]), (200, "service off"))
|
||
self.assertIn("service off", FakeCompanion.store["custom"]["sp_cmd"])
|
||
self.assertEqual(self.put("/api/command", {"text": "rm -rf"}, method="POST")[0], 400)
|
||
self.assertEqual(self.put("/api/command", {"text": "test hd2 vista 0"}, method="POST")[0], 409) # no test while live
|
||
|
||
def test_setup_bundle_import_and_install_routes(self):
|
||
import io, zipfile
|
||
from test_apply import FakeLaunchctl # noqa: F401
|
||
self.site.attempts.clear()
|
||
st, raw = get(self.port, "/api/setup")
|
||
comps = json.loads(raw)["computers"]
|
||
self.assertEqual(st, 200)
|
||
self.assertEqual({c["alias"] for c in comps}, set(self.site.load()["hosts"]))
|
||
self.assertTrue(any(s["name"].startswith("Set-up script") for c in comps for s in c["scripts"]))
|
||
st, raw = get(self.port, "/api/site/bundle")
|
||
self.assertEqual(st, 200)
|
||
z = zipfile.ZipFile(io.BytesIO(raw))
|
||
self.assertIn("config.json", z.namelist())
|
||
bundle_cfg = json.loads(z.read("config.json"))
|
||
self.assertEqual(bundle_cfg["site"]["editing_code"], self.CODE) # the bundle is the real file
|
||
# a new campus from that bundle: this campus's own site block stays, computers are flagged
|
||
other = json.loads(json.dumps(bundle_cfg)); other["site"]["name"] = "Woonsocket"; other["site"]["editing_code"] = "theirs"
|
||
buf = io.BytesIO()
|
||
with zipfile.ZipFile(buf, "w") as z2:
|
||
z2.writestr("config.json", json.dumps(other))
|
||
req = urllib.request.Request("http://127.0.0.1:%d/api/site/import" % self.port, data=buf.getvalue(),
|
||
headers={"Content-Type": "application/octet-stream", "X-Edit-Key": self.CODE, "X-Editor": "sam"}, method="POST")
|
||
with urllib.request.urlopen(req, timeout=10) as r:
|
||
body = json.loads(r.read())
|
||
self.assertEqual(body["from"], "Woonsocket")
|
||
d = self.site.draft.read()
|
||
self.assertEqual(d["config"]["site"]["name"], self.site.load()["site"]["name"])
|
||
self.assertEqual(d["config"]["site"]["editing_code"], self.CODE)
|
||
self.assertTrue(all(h["notes"].startswith("copied from Woonsocket - check it") for h in d["config"]["hosts"].values()))
|
||
self.assertEqual(body["errors"], [])
|
||
self.put("/api/draft", {}, method="DELETE")
|
||
self.assertEqual(self.put("/api/site/import", {"nope": 1}, method="POST")[0], 400)
|
||
# the program: staged folder info is readable; an install is refused while live
|
||
st, raw = get(self.port, "/api/program/staged")
|
||
self.assertEqual(st, 200)
|
||
self.assertIn("running_version", json.loads(raw))
|
||
self.assertEqual(self.put("/api/program/install", {}, method="POST")[0], 409)
|
||
|
||
def test_names_check_lists_what_this_companion_lacks(self):
|
||
self.site.attempts.clear()
|
||
n = json.loads(get(self.port, "/api/site/names")[1])
|
||
self.assertTrue(n["companion_ok"])
|
||
live = self.site.load()
|
||
mine = json.loads(json.dumps(live))
|
||
for t in mine["tiles"]:
|
||
for c in t.get("checks", []):
|
||
if c.get("conn") == "HyperDeck_2":
|
||
c["conn"] = "HyperDeck_Z"
|
||
self.put("/api/draft", {"config": mine})
|
||
n = json.loads(get(self.port, "/api/site/names")[1])
|
||
gone = [c for c in n["connections"] if not c["live"]]
|
||
self.assertEqual([(c["name"], c["keys"]) for c in gone], [("HyperDeck_Z", ["HD2"])])
|
||
self.assertIn("HyperDeck_2", gone[0]["candidates"]) # same module, unused
|
||
self.assertNotIn("HyperDeck_1", gone[0]["candidates"]) # used by HD1
|
||
self.put("/api/draft", {}, method="DELETE")
|
||
|
||
def test_a_draft_is_rebased_when_the_live_settings_move_on(self):
|
||
from test_apply import FakeLaunchctl
|
||
self.site.attempts.clear()
|
||
fake = FakeLaunchctl(self.data); self.site.applier.run, self.site.applier.sleep = fake, lambda s: None
|
||
live = self.site.load()
|
||
mine = json.loads(json.dumps(live)); mine["hosts"]["lighting"]["notes"] = "sam's note"
|
||
self.put("/api/draft", {"config": mine}, who="sam") # sam's draft
|
||
d = self.site.draft.read()
|
||
self.site.draft.clear()
|
||
other = json.loads(json.dumps(live)); other["debounce_seconds"] = (live["debounce_seconds"] or 3) + 1 # chris applies something else
|
||
self.put("/api/draft", {"config": other}, who="chris")
|
||
self.put("/api/apply", {"note": "chris", "emergency": {"word": "LIVE", "reason": "test"}}, method="POST", who="chris")
|
||
self.site.applier.thread.join(30)
|
||
sp.write_json(self.site.draft.path, d) # sam's draft is back, based on the old settings
|
||
st, raw = get(self.port, "/api/draft")
|
||
r = json.loads(raw)
|
||
self.assertFalse(r["stale"])
|
||
self.assertEqual(r["draft"]["config"]["hosts"]["lighting"]["notes"], "sam's note") # sam's own change
|
||
self.assertEqual(r["draft"]["config"]["debounce_seconds"], other["debounce_seconds"]) # chris's change kept
|
||
st, v = self.put("/api/validate", {})
|
||
self.assertEqual(len(v["changes"]), 1); self.assertTrue(v["changes"][0]["text"].startswith("computer Lighting PC › notes"))
|
||
self.put("/api/draft", {}, method="DELETE")
|
||
|
||
def test_probe_is_refused_while_a_fix_runs(self):
|
||
json.dump({"fix-x": {"key": "hd2"}}, open(os.path.join(self.data, "fixes-running.json"), "w"))
|
||
try:
|
||
self.assertEqual(get(self.port, "/api/machines/probe", {})[0], 409)
|
||
finally:
|
||
os.remove(os.path.join(self.data, "fixes-running.json"))
|
||
|
||
|
||
class TestBindingAndProbes(unittest.TestCase):
|
||
def test_bind_addresses(self):
|
||
ifaces = [("en0", "10.1.1.1", True), ("en1", "10.6.3.222", True), ("en17", "169.254.1.2", True), ("en20", "10.66.1.2", True),
|
||
("en22", "10.104.40.6", True)]
|
||
self.assertEqual(server.bind_addresses(CONFIG, ifaces), ["127.0.0.1", "10.1.1.1", "10.6.3.222"]) # the LAN ports only
|
||
cfg = json.loads(json.dumps(CONFIG))
|
||
cfg["hosts"]["ingest"]["networks"] = []
|
||
self.assertEqual(server.bind_addresses(cfg, ifaces), ["127.0.0.1", "10.1.1.1", "10.6.3.222", "169.254.1.2", "10.66.1.2", "10.104.40.6"])
|
||
self.assertTrue(server.allowed_client("192.168.4.7"))
|
||
self.assertFalse(server.allowed_client("8.8.8.8"))
|
||
|
||
def test_mac_probe_parsing(self):
|
||
out = ("SP_NAME=NA-Tracks.localdomain\nSP_OSV=26.5.1\nSP_PY=Python 3.9.6\nSP_KEY=1\nSP_AGENTS\nPID\tStatus\tLabel\n"
|
||
"780\t0\tapplication.com.audinate.DanteVirtualSoundcard.1.2\n-\t0\tcom.google.keystone.agent\nSP_PROCS\n"
|
||
"/Applications/Ableton Live 11 Suite.app/Contents/MacOS/Live\n/usr/sbin/sshd\n/Library/Audio/dvsd\nSP_APPS\nAbleton Live 11 Suite.app\n"
|
||
"SP_HELPER\n-rwxr-xr-x 1 root wheel 375 Oct 8 12:05 /usr/local/sbin/status-panel-dvs-start\n/usr/local/sbin/status-panel-dvs-start\n"
|
||
"SP_SOLO\n\nSP_NOW\n1791900000\nSP_END\n")
|
||
parsed = probes.sections(out, ("SP_AGENTS", "SP_PROCS", "SP_APPS", "SP_HELPER", "SP_SOLO", "SP_NOW", "SP_END"))
|
||
items, found = probes.mac_items(CONFIG["hosts"]["tracks"], parsed, CONFIG)
|
||
by = {i["name"]: i for i in items}
|
||
self.assertEqual(by["Ableton Live 11 Suite"]["light"], "ok")
|
||
self.assertEqual(by["the DVS helper"]["light"], "ok")
|
||
self.assertEqual(by["Dante Virtual Soundcard"]["light"], "ok")
|
||
self.assertEqual(probes.values(out)["KEY"], "1")
|
||
stopped = out.replace("/Contents/MacOS/Live\n", "")
|
||
items, _ = probes.mac_items(CONFIG["hosts"]["tracks"], probes.sections(stopped, ("SP_AGENTS", "SP_PROCS", "SP_APPS", "SP_HELPER", "SP_SOLO", "SP_NOW", "SP_END")), CONFIG)
|
||
self.assertEqual({i["name"]: i for i in items}["Ableton Live 11 Suite"]["reason"], "not running")
|
||
|
||
def test_windows_probe_parsing(self):
|
||
out = ('SP_NAME=LIGHTING-PC\r\nSP_TASKS\r\n"\\StatusPanel\\Start Arena","N/A","Running"\r\n"\\StatusPanel\\Start rtpMIDI","N/A","Ready"\r\n'
|
||
'"\\StatusPanel\\Start Vista","N/A","Running"\r\nSP_PROCS\r\n"Vista.exe","1234","Console","1","100 K"\r\n"Arena.exe","1","Console","1","1 K"\r\n'
|
||
'SP_SVC\r\nSERVICE_NAME: sshd\r\n STATE : 4 RUNNING\r\nSERVICE_NAME: rtpMIDIService\r\n STATE : 4 RUNNING\r\n'
|
||
'SP_PS\r\nPublic\r\nEnabled\r\n10.1.1.1,10.6.3.222\r\nSP_END\r\n')
|
||
parsed = probes.sections(out, ("SP_TASKS", "SP_PROCS", "SP_SVC", "SP_PS", "SP_END"))
|
||
items, found, setup = probes.windows_items(CONFIG["hosts"]["lighting"], parsed)
|
||
by = {i["name"]: i for i in items}
|
||
self.assertEqual(by["Start Vista"]["light"], "ok")
|
||
self.assertEqual(by["Start rtpMIDI"]["reason"], "rtpMIDI.exe is not running")
|
||
self.assertEqual(by["rtpMIDIService"]["light"], "ok")
|
||
self.assertEqual(by["sshd"]["light"], "ok")
|
||
self.assertEqual((setup["profile"], setup["wake_on_lan"], setup["firewall_allow"]), ("Public", "Enabled", "10.1.1.1,10.6.3.222"))
|
||
self.assertIn("\\StatusPanel\\Start Arena", found["tasks"])
|
||
|
||
def test_probe_commands_never_carry_names_from_the_settings(self):
|
||
cfg = json.loads(json.dumps(CONFIG))
|
||
cfg["hosts"]["tracks"]["runs"][1]["process"] = "x; rm -rf ~" # a hostile name in the settings
|
||
cfg["site"]["dvs_helper"] = "/usr/local/sbin/x; rm -rf ~" # and a hostile path
|
||
line = probes.MAC_PROBE.format(**probes.quoted_paths(cfg))
|
||
self.assertNotIn("rm -rf ~", line.replace("'/usr/local/sbin/x; rm -rf ~'", "")) # only inside quotes
|
||
self.assertNotIn("x; rm -rf ~\n", line)
|
||
self.assertNotIn("Live", line) # names are matched here, never sent
|
||
self.assertIn('"$HOME"\'/Library/Application Support/status-panel\'', probes.quoted_paths(CONFIG)["solo"])
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|