336 lines
13 KiB
Python
336 lines
13 KiB
Python
"""Live end-to-end simulation for the policy-driven monitor workflow.
|
|
|
|
Drives the REAL HTTP routes through the Flask app against the LIVE database
|
|
(the same one the container serves), exercising the exact flow a user performs
|
|
in the browser:
|
|
|
|
1. Log in as an admin (mint a real JWT).
|
|
2. Confirm the monitor catalog lists the four shipped monitors.
|
|
3. Create a local process monitor (msedge.exe) via POST /api/endpoints/monitors.
|
|
4. Create a Configuration Policy scoped to a PC type, carrying monitor:<id>.
|
|
5. Wrap it in an active Business Policy.
|
|
6. Create a Computer of that PC type.
|
|
7. GET the services panel and assert the monitor surfaces as an Unknown row.
|
|
8. Negative + validation probes.
|
|
9. Clean up everything created (leaves the DB as found).
|
|
|
|
Run inside the container:
|
|
docker compose exec -T api python scripts/sim_policy_monitor_e2e.py
|
|
|
|
Exit code 0 = all simulations passed. Non-zero = a failure (printed).
|
|
"""
|
|
import sys
|
|
import uuid
|
|
from pathlib import Path
|
|
|
|
from dotenv import load_dotenv
|
|
|
|
load_dotenv(Path(__file__).resolve().parent.parent / '.env')
|
|
|
|
from shopdb import create_app
|
|
from shopdb.extensions import db
|
|
from shopdb.core.models import User, Role, Asset, AssetType, Setting
|
|
from flask_jwt_extended import create_access_token
|
|
|
|
|
|
PASS = "PASS"
|
|
FAIL = "FAIL"
|
|
|
|
|
|
class Sim:
|
|
def __init__(self):
|
|
self.results = []
|
|
self.created = {
|
|
"monitor_ids": [],
|
|
"configpolicy_ids": [],
|
|
"businesspolicy_ids": [],
|
|
"asset_ids": [],
|
|
"computer_ids": [],
|
|
"computertype_ids": [],
|
|
"setting_created": False,
|
|
}
|
|
|
|
def check(self, name, condition, detail=""):
|
|
status = PASS if condition else FAIL
|
|
self.results.append((status, name, detail))
|
|
marker = "[ OK ]" if condition else "[FAIL]"
|
|
line = f"{marker} {name}"
|
|
if detail:
|
|
line += f" -- {detail}"
|
|
print(line, flush=True)
|
|
return condition
|
|
|
|
def summary(self):
|
|
passed = sum(1 for r in self.results if r[0] == PASS)
|
|
failed = sum(1 for r in self.results if r[0] == FAIL)
|
|
print("\n" + "=" * 60, flush=True)
|
|
print(f"SIMULATION SUMMARY: {passed} passed, {failed} failed", flush=True)
|
|
print("=" * 60, flush=True)
|
|
return failed == 0
|
|
|
|
|
|
def admin_headers(app):
|
|
"""Mint a real admin JWT bound to an existing admin user, or a temp one."""
|
|
with app.app_context():
|
|
user = (
|
|
User.query.join(User.roles)
|
|
.filter(Role.rolename == "admin")
|
|
.first()
|
|
)
|
|
if user is None:
|
|
role = Role.query.filter_by(rolename="admin").first()
|
|
if role is None:
|
|
role = Role(rolename="admin", description="Administrator")
|
|
db.session.add(role)
|
|
db.session.flush()
|
|
user = User(
|
|
username=f"sim_admin_{uuid.uuid4().hex[:8]}",
|
|
email=f"sim_{uuid.uuid4().hex[:8]}@test.local",
|
|
isactive=True,
|
|
)
|
|
if hasattr(user, "set_password"):
|
|
user.set_password("simpass123!")
|
|
user.roles.append(role)
|
|
db.session.add(user)
|
|
db.session.commit()
|
|
token = create_access_token(
|
|
identity=str(user.userid),
|
|
additional_claims={
|
|
"username": user.username,
|
|
"roles": [r.rolename for r in user.roles],
|
|
},
|
|
)
|
|
return {"Authorization": f"Bearer {token}"}
|
|
|
|
|
|
def jdata(resp):
|
|
body = resp.get_json()
|
|
return (body or {}).get("data")
|
|
|
|
|
|
def run():
|
|
app = create_app()
|
|
headers = admin_headers(app)
|
|
sim = Sim()
|
|
client = app.test_client()
|
|
|
|
tag = uuid.uuid4().hex[:8]
|
|
pctype_name = f"SimPCType_{tag}"
|
|
|
|
# --- 1. Monitor catalog lists the shipped four ---------------------------
|
|
resp = client.get("/api/endpoints/monitors", headers=headers)
|
|
sim.check("GET /monitors returns 200", resp.status_code == 200,
|
|
f"status={resp.status_code}")
|
|
catalog = jdata(resp) or []
|
|
shipped = [m for m in catalog if m.get("source") == "shipped"]
|
|
shipped_names = sorted({m["displayname"] for m in shipped})
|
|
sim.check("Shipped monitors seeded (eDNC, UDC, MTConnect x2)",
|
|
len(shipped) >= 4,
|
|
f"shipped={len(shipped)} names={shipped_names}")
|
|
|
|
# --- 2. Create a local process monitor (msedge.exe) ----------------------
|
|
display = f"sim_msedge_{tag}.exe"
|
|
resp = client.post("/api/endpoints/monitors", headers=headers, json={
|
|
"kind": "process",
|
|
"displayname": display,
|
|
"processname": display,
|
|
"isactive": True,
|
|
})
|
|
sim.check("POST /monitors (process) returns 201",
|
|
resp.status_code == 201, f"status={resp.status_code}")
|
|
monitor = jdata(resp) or {}
|
|
monitor_id = monitor.get("endpointmonitorid")
|
|
if monitor_id:
|
|
sim.created["monitor_ids"].append(monitor_id)
|
|
sim.check("Created monitor has an id", bool(monitor_id),
|
|
f"id={monitor_id}")
|
|
|
|
# --- 2b. Validation probe: missing name is rejected ----------------------
|
|
resp = client.post("/api/endpoints/monitors", headers=headers, json={
|
|
"kind": "process", "displayname": "", "processname": "",
|
|
})
|
|
sim.check("POST /monitors with blank name rejected (400)",
|
|
resp.status_code == 400, f"status={resp.status_code}")
|
|
|
|
# --- 3. Create a PC type to scope to -------------------------------------
|
|
resp = client.post("/api/computers/types", headers=headers,
|
|
json={"computertype": pctype_name})
|
|
sim.check("POST /computers/types returns 2xx",
|
|
resp.status_code in (200, 201), f"status={resp.status_code}")
|
|
pctype = jdata(resp) or {}
|
|
pctype_id = pctype.get("computertypeid")
|
|
if pctype_id:
|
|
sim.created["computertype_ids"].append(pctype_id)
|
|
sim.check("PC type created with id", bool(pctype_id), f"id={pctype_id}")
|
|
|
|
# --- 4. Enable the feature flag ------------------------------------------
|
|
with app.app_context():
|
|
flag = Setting.query.filter_by(
|
|
key="computers_configpolicies_enabled").first()
|
|
if flag is None:
|
|
db.session.add(Setting(
|
|
key="computers_configpolicies_enabled", value="true"))
|
|
sim.created["setting_created"] = True
|
|
else:
|
|
sim._prev_flag = flag.value
|
|
flag.value = "true"
|
|
db.session.commit()
|
|
sim.check("Feature flag computers_configpolicies_enabled=true", True)
|
|
|
|
# --- 5. Create a Configuration Policy carrying monitor:<id> ---------------
|
|
resp = client.post("/api/computers/policies/configuration", headers=headers,
|
|
json={
|
|
"name": f"Sim Edge Coverage {tag}",
|
|
"description": "sim",
|
|
"scopejson": {"type": "pctype", "id": pctype_id},
|
|
"actions": [
|
|
{"action": f"monitor:{monitor_id}",
|
|
"mode": "gate"},
|
|
],
|
|
})
|
|
sim.check("POST /policies/configuration returns 2xx",
|
|
resp.status_code in (200, 201), f"status={resp.status_code}")
|
|
cpolicy = jdata(resp) or {}
|
|
cpolicy_id = cpolicy.get("configurationpolicyid")
|
|
if cpolicy_id:
|
|
sim.created["configpolicy_ids"].append(cpolicy_id)
|
|
sim.check("Configuration Policy created with id", bool(cpolicy_id),
|
|
f"id={cpolicy_id}")
|
|
|
|
# --- 5b. Round-trip: the monitor action persisted ------------------------
|
|
resp = client.get(
|
|
f"/api/computers/policies/configuration/{cpolicy_id}", headers=headers)
|
|
body = jdata(resp) or {}
|
|
actions = body.get("actions", [])
|
|
sim.check("Config policy round-trips the monitor action",
|
|
any(a.get("action") == f"monitor:{monitor_id}" for a in actions),
|
|
f"actions={[a.get('action') for a in actions]}")
|
|
|
|
# --- 6. Wrap it in an active Business Policy ------------------------------
|
|
resp = client.post("/api/computers/policies/business", headers=headers,
|
|
json={
|
|
"name": f"Sim Process Monitor Test {tag}",
|
|
"priority": 100,
|
|
"isactive": True,
|
|
"items": [
|
|
{"configurationpolicyid": cpolicy_id,
|
|
"sortorder": 0},
|
|
],
|
|
})
|
|
sim.check("POST /policies/business returns 2xx",
|
|
resp.status_code in (200, 201), f"status={resp.status_code}")
|
|
bpolicy = jdata(resp) or {}
|
|
bpolicy_id = bpolicy.get("businesspolicyid")
|
|
if bpolicy_id:
|
|
sim.created["businesspolicy_ids"].append(bpolicy_id)
|
|
sim.check("Business Policy created with id", bool(bpolicy_id),
|
|
f"id={bpolicy_id}")
|
|
|
|
# --- 7. Create a Computer of that PC type --------------------------------
|
|
hostname = f"SIMPC{tag.upper()}"
|
|
with app.app_context():
|
|
from plugins.computers.models import Computer
|
|
atype = AssetType.query.filter_by(assettype="computer").first()
|
|
if atype is None:
|
|
atype = AssetType(assettype="computer")
|
|
db.session.add(atype)
|
|
db.session.flush()
|
|
asset = Asset(assetnumber=hostname, assettypeid=atype.assettypeid,
|
|
isactive=True)
|
|
db.session.add(asset)
|
|
db.session.flush()
|
|
comp = Computer(assetid=asset.assetid, hostname=hostname,
|
|
computertypeid=pctype_id)
|
|
db.session.add(comp)
|
|
db.session.commit()
|
|
asset_id = asset.assetid
|
|
computer_id = comp.computerid
|
|
sim.created["asset_ids"].append(asset_id)
|
|
sim.created["computer_ids"].append(computer_id)
|
|
sim.check("Computer created for the scoped PC type", bool(asset_id),
|
|
f"assetid={asset_id}")
|
|
|
|
# --- 8. The monitor surfaces in the services panel -----------------------
|
|
resp = client.get(f"/api/endpoints/asset/{asset_id}/services",
|
|
headers=headers)
|
|
sim.check("GET /asset/<id>/services returns 200",
|
|
resp.status_code == 200, f"status={resp.status_code}")
|
|
rows = jdata(resp) or []
|
|
edge = next((r for r in rows if r.get("name") == display), None)
|
|
sim.check("Policy-attached monitor surfaces in panel", edge is not None,
|
|
f"rows={[r.get('name') for r in rows]}")
|
|
if edge:
|
|
sim.check("Surfaced row is a process", edge.get("kind") == "process",
|
|
f"kind={edge.get('kind')}")
|
|
sim.check("Surfaced row is Unknown (not yet reported)",
|
|
edge.get("statelabel") == "Unknown",
|
|
f"statelabel={edge.get('statelabel')}")
|
|
action = edge.get("action") or {}
|
|
sim.check("Surfaced row offers a Start action targeting the process",
|
|
action.get("target") == display
|
|
and action.get("targetkind") == "process",
|
|
f"action={action}")
|
|
|
|
# --- 9. Negative: flag off hides the row ---------------------------------
|
|
with app.app_context():
|
|
flag = Setting.query.filter_by(
|
|
key="computers_configpolicies_enabled").first()
|
|
flag.value = "false"
|
|
db.session.commit()
|
|
resp = client.get(f"/api/endpoints/asset/{asset_id}/services",
|
|
headers=headers)
|
|
rows = jdata(resp) or []
|
|
sim.check("Flag off hides the policy monitor from the panel",
|
|
not any(r.get("name") == display for r in rows),
|
|
f"rows={[r.get('name') for r in rows]}")
|
|
with app.app_context():
|
|
flag = Setting.query.filter_by(
|
|
key="computers_configpolicies_enabled").first()
|
|
flag.value = "true"
|
|
db.session.commit()
|
|
|
|
# --- Cleanup -------------------------------------------------------------
|
|
cleanup(app, client, headers, sim)
|
|
|
|
return sim.summary()
|
|
|
|
|
|
def cleanup(app, client, headers, sim):
|
|
print("\n--- cleanup ---", flush=True)
|
|
for bid in sim.created["businesspolicy_ids"]:
|
|
client.delete(f"/api/computers/policies/business/{bid}", headers=headers)
|
|
for cid in sim.created["configpolicy_ids"]:
|
|
client.delete(f"/api/computers/policies/configuration/{cid}",
|
|
headers=headers)
|
|
for mid in sim.created["monitor_ids"]:
|
|
client.delete(f"/api/endpoints/monitors/{mid}", headers=headers)
|
|
with app.app_context():
|
|
from plugins.computers.models import Computer, ComputerType
|
|
for cid in sim.created["computer_ids"]:
|
|
row = db.session.get(Computer, cid)
|
|
if row:
|
|
db.session.delete(row)
|
|
for aid in sim.created["asset_ids"]:
|
|
row = db.session.get(Asset, aid)
|
|
if row:
|
|
db.session.delete(row)
|
|
for tid in sim.created["computertype_ids"]:
|
|
row = db.session.get(ComputerType, tid)
|
|
if row:
|
|
db.session.delete(row)
|
|
# Restore or remove the feature flag as found.
|
|
flag = Setting.query.filter_by(
|
|
key="computers_configpolicies_enabled").first()
|
|
if flag is not None:
|
|
if sim.created["setting_created"]:
|
|
db.session.delete(flag)
|
|
elif hasattr(sim, "_prev_flag"):
|
|
flag.value = sim._prev_flag
|
|
db.session.commit()
|
|
print("cleanup done", flush=True)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
ok = run()
|
|
sys.exit(0 if ok else 1)
|