From a44187d0f125e368c8c66db19777153cd86e8b58 Mon Sep 17 00:00:00 2001 From: "Aaron D. Lee" Date: Fri, 3 Apr 2026 18:53:50 -0400 Subject: [PATCH] feat(S5): pet rule CRUD routes with validation and templates --- tests/unit/test_pet_rules_api.py | 55 +++++++++++++++++ vigilar/web/blueprints/pets.py | 101 +++++++++++++++++++++++++++++++ 2 files changed, 156 insertions(+) create mode 100644 tests/unit/test_pet_rules_api.py diff --git a/tests/unit/test_pet_rules_api.py b/tests/unit/test_pet_rules_api.py new file mode 100644 index 0000000..33f8903 --- /dev/null +++ b/tests/unit/test_pet_rules_api.py @@ -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 diff --git a/vigilar/web/blueprints/pets.py b/vigilar/web/blueprints/pets.py index 3f80e0d..6ef556d 100644 --- a/vigilar/web/blueprints/pets.py +++ b/vigilar/web/blueprints/pets.py @@ -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("//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("//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("//rules/", 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("//rules/", 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")