"""فحص ذاتي لـ VDE — P1 (توزيع/الوضعان/عزل) + P2 (توقيع/تحقق/تحميل ساخن). تشغيل: python test_vde.py """ import base64 import os import sys import tempfile HERE = os.path.dirname(os.path.abspath(__file__)) sys.path.insert(0, os.path.dirname(HERE)) # edge/ ليُستورد `vde` import time # noqa: E402 from vde import loader, signing # noqa: E402 from vde.api import Detector # noqa: E402 from vde.builtins import (AudioAlert, CloudVLM, Compound, Inactivity, # noqa: E402 OnnxObject, VehicleCrossing, ZonePresence) from vde.engine import Engine, Services # noqa: E402 class FakeOnnx: # نموذج وهمي: 5 أسماك def __init__(self, path): pass def detect(self, jpeg, conf=0.35, iou=0.45): return [("fish", 0.9, (0, 0, 10, 10))] * 5 def make_engine(): emitted = [] svc = Services(frame_fn=lambda cam: b"JPEGDATA", vlm_fn=lambda key, frames, camera=None: {"flagged": True, "score": 8, "note": "عراك (اختبار)"}, log_fn=lambda m: None, onnx_cls=FakeOnnx) return Engine(svc, lambda k, c, d: emitted.append((k, c, d.flagged))), emitted def test_p1_modes_and_isolation(): eng, emitted = make_engine() # قاعدة منطقة (حدثي MQTT) eng.register("restricted_zone", ZonePresence(), "cam1", {"cooldown_s": 0}, None, 1, triggers=["zone:restricted_zone"]) eng.dispatch_event("cam1", "zone:restricted_zone") assert ("restricted_zone", "cam1", True) in emitted # نموذج ONNX (edge_model، حدثي) eng.register("fish_count", OnnxObject(), "cam2", {"detect_labels": ["fish"], "min_count": 3, "cooldown_s": 0}, "m.onnx", 1, triggers=["motion"]) eng.dispatch_event("cam2", "motion") assert ("fish_count", "cam2", True) in emitted # VLM سحابي (بثّي stream → on_frame) eng.register("fight_detection", CloudVLM(), "cam3", {"cooldown_s": 0}, None, 1, triggers=["stream"]) slot = eng.slots[("fight_detection", "cam3")] eng._run(slot, slot.detector.on_frame, b"JPEG") assert ("fight_detection", "cam3", True) in emitted # عزل + تعطيل تلقائي (P4) class Boom(Detector): triggers = ["motion"] def on_event(self, ev, ctx): raise RuntimeError("boom") eng.register("boom", Boom(), "cam4", {"cooldown_s": 0}, None, 1) for _ in range(5): eng.dispatch_event("cam4", "motion") assert eng.slots[("boom", "cam4")].enabled is False print("OK P1 — الوضعان (حدثي/بثّي) + كل الأنواع + عزل/تعطيل تلقائي") def test_p2_signing_and_hotload(): priv, pub = signing.generate_keypair() # توقيع سليم/مزيّف assert signing.verify(pub, b"x", signing.sign(priv, b"x")) assert not signing.verify(pub, b"y", signing.sign(priv, b"x")) src = open(os.path.join(HERE, "samples", "fish_count", "detector.py"), "rb").read() manifest = {"key": "fish_count", "version": 1, "kind": "edge_model", "entry": "detector.py", "triggers": ["motion"]} vmod = loader.pack(manifest, {"detector.py": src}, priv) # تحقق سليم assert loader.verify_package(vmod, pub)["manifest"]["key"] == "fish_count" # عبث بالبايتات → رفض bad = bytearray(vmod); bad[len(bad) // 2] ^= 0xFF try: loader.verify_package(bytes(bad), pub); assert False, "قبل حزمة معدّلة" except Exception: pass # مفتاح خاطئ → رفض _, pub2 = signing.generate_keypair() try: loader.verify_package(vmod, pub2); assert False, "قبل توقيعاً بمفتاح خاطئ" except ValueError: pass # تثبيت + تحميل ساخن + تسجيل + توزيع moddir = tempfile.mkdtemp(prefix="vde-mods-") man = loader.install(vmod, pub, moddir) det = loader.load_detector(man["_entry"], man["key"], man["version"]) eng, emitted = make_engine() eng.register(man["key"], det, "cam5", {"min_count": 20, "cooldown_s": 0}, man.get("_model"), man["version"], triggers=man["triggers"]) eng.dispatch_event("cam5", "motion") # 5 أسماك < 20 → مخالفة assert ("fish_count", "cam5", True) in emitted print("OK P2 — توقيع Ed25519 + رفض العبث/المفتاح الخاطئ + تحميل ساخن لموديول مرفوع") class _Resp: def __init__(self, ok=True, js=None, content=b""): self.ok, self._js, self.content = ok, js, content def json(self): return self._js class _FakeRequests: def __init__(self, listing, vmod): self.listing, self.vmod = listing, vmod def get(self, url, **k): return _Resp(js=self.listing) if url.endswith("/modules") \ else _Resp(content=self.vmod) def test_p2_ota(): from vde import registry_client priv, pub = signing.generate_keypair() src = open(os.path.join(HERE, "samples", "fish_count", "detector.py"), "rb").read() vmod = loader.pack({"key": "fish_count", "version": 1, "kind": "edge_model", "entry": "detector.py", "triggers": ["motion"], "model": "m.onnx"}, {"detector.py": src, "m.onnx": b"FAKEONNX"}, priv) listing = {"modules_version": 1, "modules": {"fish_count": {"latest": 1, "triggers": ["motion"]}}} registry_client.requests = _FakeRequests(listing, vmod) # نقل وهمي (بلا شبكة) eng, emitted = make_engine() moddir = tempfile.mkdtemp(prefix="vde-ota-") ota = registry_client.ModuleOTA( eng, "http://cloud", {}, base64.b64encode(pub).decode(), moddir, assignments_fn=lambda key: [("cam9", {"min_count": 20, "cooldown_s": 0})], log=lambda m: None) ota.sync() # يسحب + يتحقق + يحمّل ساخناً + يسجّل eng.dispatch_event("cam9", "motion") # 5 أسماك < 20 → مخالفة assert ("fish_count", "cam9", True) in emitted # لا إعادة تحميل عند نفس modules_version before = dict(ota._loaded) ota.sync() assert ota._loaded == before print("OK P2/OTA — استطلاع الإصدار + تنزيل + تحقق توقيع + تحميل ساخن + تسجيل لكل كاميرا") def test_2026_features(): eng, emitted = make_engine() # كشف الخمول: idle > threshold → تنبيه؛ نشاط يعيد الضبط eng.register("inactivity", Inactivity(), "cam1", {"inactive_s": 300, "check_s": 1, "cooldown_s": 0}, None, 1, triggers=["object:person", "motion", "schedule"]) eng.slots[("inactivity", "cam1")].detector.last_activity = time.time() - 400 eng.tick_schedule() assert ("inactivity", "cam1", True) in emitted before = sum(k == "inactivity" for k, *_ in emitted) eng.dispatch_event("cam1", "object:person") # نشاط → إعادة ضبط eng.tick_schedule() assert sum(k == "inactivity" for k, *_ in emitted) == before # لا تنبيه (نشِط) # عبور مركبة: مركبة داخل منطقة الميزة → تنبيه؛ خارجها → لا eng.register("vehicle_crossing", VehicleCrossing(), "cam1", {"cooldown_s": 0}, None, 1, triggers=["object:car", "object:truck"]) eng.dispatch_event("cam1", "object:car", {"zones": ["vehicle_crossing"]}) assert ("vehicle_crossing", "cam1", True) in emitted v0 = sum(k == "vehicle_crossing" for k, *_ in emitted) eng.dispatch_event("cam1", "object:car", {"zones": ["other"]}) assert sum(k == "vehicle_crossing" for k, *_ in emitted) == v0 # تنبيه مركّب: لا يُطلق حتى يتحقق الشرطان ضمن النافذة eng.register("compound_alert", Compound(), "cam1", {"conditions": ["weapon_detection", "crowd_density"], "window_s": 30, "cooldown_s": 0}, None, 1, triggers=["schedule"]) eng.register("weapon_detection", ZonePresence(), "cam1", {"cooldown_s": 0}, None, 1, triggers=["zone:weapon_detection"]) eng.register("crowd_density", ZonePresence(), "cam1", {"cooldown_s": 0}, None, 1, triggers=["zone:crowd_density"]) eng.dispatch_event("cam1", "zone:weapon_detection") # شرط أول فقط assert not any(k == "compound_alert" for k, *_ in emitted) eng.dispatch_event("cam1", "zone:crowd_density") # اكتمل → مركّب assert ("compound_alert", "cam1", True) in emitted # حدث صوتي: صراخ → حدث؛ صوت غير منبّه → لا شيء؛ ويعمل عبر MQTT audio eng.register("audio_event", AudioAlert(), "cam6", {"cooldown_s": 0}, None, 1, triggers=["audio"]) eng.dispatch_event("cam6", "audio", {"label": "scream"}) assert ("audio_event", "cam6", True) in emitted n = sum(k == "audio_event" for k, *_ in emitted) eng.dispatch_event("cam6", "audio", {"label": "dog"}) # ليس في alert_labels assert sum(k == "audio_event" for k, *_ in emitted) == n eng._on_mqtt("frigate/cam6/audio/glass_break", b"ON") # مسار MQTT الصوتي assert sum(k == "audio_event" for k, *_ in emitted) == n + 1 print("OK 2026 — خمول + مركبات + مركّب + صوت (MQTT frigate//audio/