feat(S5): pet rule CRUD routes with validation and templates
This commit is contained in:
Aaron D. Lee
parent
931b453ba9
commit
a44187d0f1
55
tests/unit/test_pet_rules_api.py
Normal file
55
tests/unit/test_pet_rules_api.py
Normal file
@ -0,0 +1,55 @@
|
||||
import json
|
||||
import pytest
|
||||
from vigilar.config import VigilarConfig
|
||||
from vigilar.storage.queries import insert_pet, get_all_pets
|
||||
from vigilar.web.app import create_app
|
||||
|
||||
@pytest.fixture
|
||||
def rules_app(test_db):
|
||||
cfg = VigilarConfig()
|
||||
app = create_app(cfg)
|
||||
app.config["TESTING"] = True
|
||||
app.config["DB_ENGINE"] = test_db
|
||||
insert_pet(test_db, name="Milo", species="dog")
|
||||
return app
|
||||
|
||||
def _get_pet_id(rules_app):
|
||||
db = rules_app.config["DB_ENGINE"]
|
||||
return get_all_pets(db)[0]["id"]
|
||||
|
||||
def test_create_rule(rules_app):
|
||||
pet_id = _get_pet_id(rules_app)
|
||||
with rules_app.test_client() as c:
|
||||
rv = c.post(f"/pets/{pet_id}/rules", json={
|
||||
"name": "Outdoor timer",
|
||||
"conditions": [{"type": "in_zone_longer_than", "zone": "EXTERIOR", "minutes": 45}],
|
||||
"action": "push_notify", "action_message": "{pet_name} outside", "cooldown_minutes": 30})
|
||||
assert rv.status_code == 200
|
||||
assert rv.get_json()["ok"] is True
|
||||
|
||||
def test_list_rules(rules_app):
|
||||
pet_id = _get_pet_id(rules_app)
|
||||
with rules_app.test_client() as c:
|
||||
c.post(f"/pets/{pet_id}/rules", json={"name": "Test", "conditions": [], "action": "log_event", "cooldown_minutes": 30})
|
||||
rv = c.get(f"/pets/{pet_id}/rules")
|
||||
assert len(rv.get_json()["rules"]) == 1
|
||||
|
||||
def test_delete_rule(rules_app):
|
||||
pet_id = _get_pet_id(rules_app)
|
||||
with rules_app.test_client() as c:
|
||||
rv = c.post(f"/pets/{pet_id}/rules", json={"name": "Del", "conditions": [], "action": "log_event", "cooldown_minutes": 30})
|
||||
rule_id = rv.get_json()["id"]
|
||||
rv = c.delete(f"/pets/{pet_id}/rules/{rule_id}")
|
||||
assert rv.status_code == 200
|
||||
|
||||
def test_invalid_action(rules_app):
|
||||
pet_id = _get_pet_id(rules_app)
|
||||
with rules_app.test_client() as c:
|
||||
rv = c.post(f"/pets/{pet_id}/rules", json={"name": "Bad", "conditions": [], "action": "invalid", "cooldown_minutes": 30})
|
||||
assert rv.status_code == 400
|
||||
|
||||
def test_rule_templates(rules_app):
|
||||
with rules_app.test_client() as c:
|
||||
rv = c.get("/pets/api/rule-templates")
|
||||
assert rv.status_code == 200
|
||||
assert len(rv.get_json()) >= 5
|
||||
@ -223,6 +223,107 @@ def delete_pet(pet_id: str):
|
||||
return jsonify({"ok": True})
|
||||
|
||||
|
||||
VALID_ACTIONS = {"push_notify", "log_event", "start_recording"}
|
||||
VALID_CONDITION_TYPES = {
|
||||
"detected_in_zone", "not_seen_in_zone", "not_seen_anywhere",
|
||||
"detected_without_person", "in_zone_longer_than", "time_of_day",
|
||||
}
|
||||
|
||||
|
||||
@pets_bp.route("/<pet_id>/rules")
|
||||
def list_rules(pet_id):
|
||||
engine = _engine()
|
||||
if engine is None:
|
||||
return jsonify({"rules": []})
|
||||
from vigilar.storage.queries import get_pet_rules
|
||||
return jsonify({"rules": get_pet_rules(engine, pet_id)})
|
||||
|
||||
|
||||
@pets_bp.route("/<pet_id>/rules", methods=["POST"])
|
||||
def create_rule(pet_id):
|
||||
engine = _engine()
|
||||
if engine is None:
|
||||
return jsonify({"error": "database not available"}), 503
|
||||
data = request.get_json(silent=True) or {}
|
||||
name = data.get("name")
|
||||
conditions = data.get("conditions", [])
|
||||
action = data.get("action")
|
||||
action_message = data.get("action_message", "")
|
||||
cooldown = data.get("cooldown_minutes", 30)
|
||||
if not name:
|
||||
return jsonify({"error": "name required"}), 400
|
||||
if action not in VALID_ACTIONS:
|
||||
return jsonify({"error": f"Invalid action: {action}"}), 400
|
||||
if cooldown < 1:
|
||||
return jsonify({"error": "cooldown_minutes must be >= 1"}), 400
|
||||
for cond in conditions:
|
||||
if cond.get("type") not in VALID_CONDITION_TYPES:
|
||||
return jsonify({"error": f"Invalid condition type: {cond.get('type')}"}), 400
|
||||
from vigilar.storage.queries import count_pet_rules
|
||||
cfg = current_app.config.get("VIGILAR_CONFIG")
|
||||
max_rules = cfg.pets.max_rules_per_pet if cfg else 32
|
||||
if count_pet_rules(engine, pet_id) >= max_rules:
|
||||
return jsonify({"error": f"Max {max_rules} rules per pet"}), 400
|
||||
import json as json_mod
|
||||
from vigilar.storage.queries import insert_pet_rule
|
||||
rule_id = insert_pet_rule(engine, pet_id, name, json_mod.dumps(conditions),
|
||||
action, action_message, cooldown, data.get("priority", 0))
|
||||
return jsonify({"ok": True, "id": rule_id})
|
||||
|
||||
|
||||
@pets_bp.route("/<pet_id>/rules/<int:rule_id>", methods=["PUT"])
|
||||
def update_rule_route(pet_id, rule_id):
|
||||
engine = _engine()
|
||||
if engine is None:
|
||||
return jsonify({"error": "database not available"}), 503
|
||||
import json as json_mod
|
||||
data = request.get_json(silent=True) or {}
|
||||
updates = {}
|
||||
if "name" in data: updates["name"] = data["name"]
|
||||
if "conditions" in data: updates["conditions"] = json_mod.dumps(data["conditions"])
|
||||
if "action" in data:
|
||||
if data["action"] not in VALID_ACTIONS:
|
||||
return jsonify({"error": "Invalid action"}), 400
|
||||
updates["action"] = data["action"]
|
||||
if "action_message" in data: updates["action_message"] = data["action_message"]
|
||||
if "cooldown_minutes" in data: updates["cooldown_minutes"] = data["cooldown_minutes"]
|
||||
if "enabled" in data: updates["enabled"] = 1 if data["enabled"] else 0
|
||||
if "priority" in data: updates["priority"] = data["priority"]
|
||||
if not updates:
|
||||
return jsonify({"error": "no fields to update"}), 400
|
||||
from vigilar.storage.queries import update_pet_rule
|
||||
update_pet_rule(engine, rule_id, **updates)
|
||||
return jsonify({"ok": True})
|
||||
|
||||
|
||||
@pets_bp.route("/<pet_id>/rules/<int:rule_id>", methods=["DELETE"])
|
||||
def delete_rule_route(pet_id, rule_id):
|
||||
engine = _engine()
|
||||
if engine is None:
|
||||
return jsonify({"error": "database not available"}), 503
|
||||
from vigilar.storage.queries import delete_pet_rule
|
||||
delete_pet_rule(engine, rule_id)
|
||||
return jsonify({"ok": True})
|
||||
|
||||
|
||||
@pets_bp.route("/api/rule-templates")
|
||||
def rule_templates():
|
||||
return jsonify([
|
||||
{"name": "Outdoor timer", "conditions": [{"type": "in_zone_longer_than", "zone": "EXTERIOR", "minutes": 45}],
|
||||
"action": "push_notify", "action_message": "{pet_name} has been outside for {duration}", "cooldown_minutes": 30},
|
||||
{"name": "Needs to go out", "conditions": [{"type": "not_seen_in_zone", "zone": "EXTERIOR", "minutes": 240}],
|
||||
"action": "push_notify", "action_message": "{pet_name} hasn't been outside in {duration}", "cooldown_minutes": 60},
|
||||
{"name": "Missing pet", "conditions": [{"type": "not_seen_anywhere", "minutes": 480}],
|
||||
"action": "push_notify", "action_message": "{pet_name} hasn't been seen in {duration}", "cooldown_minutes": 120},
|
||||
{"name": "Wrong zone alert", "conditions": [{"type": "detected_in_zone", "zone": "EXTERIOR"}],
|
||||
"action": "push_notify", "action_message": "{pet_name} detected in {zone} — {camera}", "cooldown_minutes": 15},
|
||||
{"name": "On the loose", "conditions": [{"type": "detected_without_person"}],
|
||||
"action": "push_notify", "action_message": "{pet_name} spotted without supervision — {camera}", "cooldown_minutes": 5},
|
||||
{"name": "Night escape", "conditions": [{"type": "detected_in_zone", "zone": "EXTERIOR"}, {"type": "time_of_day", "start": "22:00", "end": "06:00"}],
|
||||
"action": "push_notify", "action_message": "{pet_name} is outside at night — {camera}", "cooldown_minutes": 30},
|
||||
])
|
||||
|
||||
|
||||
@pets_bp.route("/train", methods=["POST"])
|
||||
def train_model():
|
||||
pets_cfg = current_app.config.get("PETS_CONFIG")
|
||||
|
||||
Loading…
Reference in New Issue
Block a user