c3227c179c
감사에서 확정된 잔여 항목을 일괄 처리. 이전 커밋이 증상(스테일 번들)을 고쳤다면 이번에는 그것이 조용히 지나가게 만든 구조를 고친다. 보안 - CSP 도입. 인라인 2블록에 per-request nonce, unsafe-inline 없음. 빌드가 없다는 제약이 nonce를 막지는 않는다. 보안 헤더를 .htaccess 단독에서 PHP로 옮겨 개발/납품이 같아진다. - /logout에 CSRF. SameSite=Lax가 막아주지만, 사고 대응 중 운영자를 콘솔에서 튕겨내는 건 이 제품에선 실제 가용성 공격이라 예외로 둘 이유가 없다. - 잠긴 계정 423 → 401. 아무나 이름을 추측해 5회 실패로 잠근 뒤 423을 계정 실재 확인으로 쓸 수 있었다. 잠금 사실은 감사 로그에 남고, 익명 호출자는 못 읽는다. 엣지 - 큐 flush에 회당 상한. 정전 복구 직후 2000건을 비우는 동안 감지가 멈추던 구간을 없앤다. - 타이머 env 전수 검증. COOLDOWN_S=nan은 dedup을 무력화해 매 틱 발보시켰다. - T2 프레임 스킵(INFER_EVERY). 스킵 틱은 프레임을 읽어 버퍼만 비우고 링버퍼에 넣지 않는다 — 추론이 없으면 얼굴 박스도 없어 그대로 넣으면 증거 스냅샷에 미블러 얼굴이 실린다. T3(모션 게이트)는 의도적으로 미구현: 게이트가 틀리면 '움직이지 않는 쓰러진 작업자'에게 추론이 안 돈다. 실영상 없이 임계값을 정할 수 없다. - systemd 워치독(sd_notify) + 메모리 상한. Restart=always는 죽은 프로세스만 잡고 '살아 있는데 멈춘' 상태는 못 잡는다. UVC 해상도·버퍼 고정(링버퍼 RSS가 웹캠 기본값에 좌우). 테스트 — 감사의 최대 갭 - PDO 통합 테스트가 pdo_sqlite 부재로 통째로 스킵되고 CI도 없어 사실상 한 번도 안 돌았다. `-d extension=pdo_sqlite`로 되살리고 리포트·히트맵·익스포트 집계 SQL을 최초로 실행한다. - 얼굴 블러 테스트가 out is not None만 봤다 — blur()가 무동작이어도 통과. PIPA 대응의 유일한 실행 가능한 증거였는데 아무것도 증명하지 못했다. 디테일 파괴를 단언하도록 교체. - index.php 라우트 테이블 대조(프런트 호출 15개 교차검증), dist↔소스 패리티 테스트, CI 신설(스킵이 하나라도 있으면 실패). - ArraySessionStore.regenerate()가 데이터를 지워 실제 session_regenerate_id(true)와 달랐다. 더블이 실구현과 어긋난 자리 하나를 실제로 잡았다. 프런트 - 이벤트 목록 시그니처 재렌더 스킵 + 포커스 복원. aria-live 영역을 5초마다 통째로 다시 만들어 스크린리더가 200행을 재낭독하고 키보드 포커스가 body로 날아갔다. 같은 패턴이 device-tree.js에는 이미 있었다. - 폴링 중첩 가드(느린 공유호스팅에서 응답 역전), 콜드스타트 실패 표시. 납품 - VERSION 신설, 두 산출물에 version/git_commit/built_at 스탬프. 스테일 번들의 SHA256SUMS는 자기정합적이라 통과한다 — provenance 없이는 하류에서 탐지할 방법이 없다. - deploy/README·env.example을 Hailo 단독 번들 실제와 일치(템플릿이 없는 ONNX를 가리켰다). - 이벤트 보존 정책은 기본 0(영구). 산업안전 기록을 시스템이 임의로 지우지 않는다. 검증: pytest 251 / PHPUnit 162, 스킵 0. CSP nonce 결합은 실제 렌더 경로로 확인. 브라우저 실렌더는 MySQL이 필요해 이 환경에서 미수행.
87 lines
3.3 KiB
Python
87 lines
3.3 KiB
Python
import numpy as np
|
|
from edge.config import load_config
|
|
from edge.models import Person, BBox, Keypoint
|
|
from edge.detectors.zone import Zone
|
|
from edge.runtime import PipelineRuntime
|
|
|
|
|
|
def _lying_person():
|
|
kps = [Keypoint(150, 300, 0.9) for _ in range(17)]
|
|
kps[5] = Keypoint(100, 300, 0.9); kps[6] = Keypoint(100, 310, 0.9)
|
|
kps[11] = Keypoint(200, 300, 0.9); kps[12] = Keypoint(200, 310, 0.9)
|
|
kps[0] = Keypoint(90, 320, 0.9)
|
|
return Person(BBox(90, 290, 210, 320), kps)
|
|
|
|
|
|
class _Source:
|
|
def read(self): return True, np.full((360, 640, 3), 128, np.uint8)
|
|
|
|
|
|
class _Pose:
|
|
def infer(self, frame): return [_lying_person()]
|
|
|
|
|
|
class _Sink:
|
|
def __init__(self): self.events = []
|
|
def try_send(self, e, snaps): self.events.append(e); return True
|
|
|
|
|
|
def test_step_runs_pipeline_and_emits_fall():
|
|
cfg = load_config({"FALL_PRONE_S": "0.5"})
|
|
sink = _Sink()
|
|
rt = PipelineRuntime(cfg, _Source(), _Pose(), sink)
|
|
rt.step(now=0.0)
|
|
rt.step(now=0.6)
|
|
from edge.events import EventType
|
|
assert any(e.event_type == EventType.FALL for e in sink.events)
|
|
|
|
|
|
def test_apply_zones_hotswaps():
|
|
cfg = load_config({})
|
|
rt = PipelineRuntime(cfg, _Source(), _Pose(), _Sink())
|
|
rt.apply_zones([Zone("z1", "restricted", [(0, 0), (9, 0), (9, 9)])])
|
|
assert rt.zone_detector.zones[0].zone_id == "z1"
|
|
|
|
|
|
def test_infer_every_skips_inference_but_still_drains_the_camera():
|
|
# T2 저하 티어: 추론은 N틱마다. 스킵 틱에도 프레임은 읽어야 한다(v4l2 버퍼가 밀리면
|
|
# 몇 초 전 프레임이 올라와 모든 경보가 늦어진다).
|
|
class _CountingSource:
|
|
def __init__(self): self.reads = 0
|
|
def read(self):
|
|
self.reads += 1
|
|
return True, np.full((360, 640, 3), 128, np.uint8)
|
|
|
|
class _CountingPose:
|
|
def __init__(self): self.infers = 0
|
|
def infer(self, frame):
|
|
self.infers += 1
|
|
return [_lying_person()]
|
|
|
|
src, pose = _CountingSource(), _CountingPose()
|
|
cfg = load_config({"INFER_EVERY": "3"})
|
|
rt = PipelineRuntime(cfg, src, pose, _Sink())
|
|
for t in range(6):
|
|
rt.step(float(t))
|
|
assert src.reads == 6 # 매 틱 프레임을 뽑아 버퍼를 비운다
|
|
assert pose.infers == 2 # 추론은 3틱마다
|
|
|
|
|
|
def test_skipped_tick_never_puts_unanalyzed_frames_in_the_evidence_ring():
|
|
# 스킵 틱에는 얼굴 박스가 없다. 그 프레임이 링에 들어가면 증거 타임라인에 블러되지 않은
|
|
# 얼굴이 실려 나간다(PIPA). 링에는 추론된 프레임만 있어야 한다.
|
|
cfg = load_config({"INFER_EVERY": "4"})
|
|
rt = PipelineRuntime(cfg, _Source(), _Pose(), _Sink())
|
|
for t in range(8):
|
|
rt.step(float(t))
|
|
assert len(rt.ring.frames) == 2 # 8틱 중 추론된 2틱만
|
|
assert all(boxes for (_t, _f, boxes) in rt.ring.frames) # 전부 얼굴 박스를 동반
|
|
|
|
|
|
def test_dead_camera_still_reports_no_frame_on_a_skipped_tick():
|
|
# 스킵 틱이 죽은 카메라를 '정상'으로 보고하면 감시 공백이 초록으로 보인다.
|
|
class _DeadSource:
|
|
def read(self): return False, None
|
|
cfg = load_config({"INFER_EVERY": "2"})
|
|
rt = PipelineRuntime(cfg, _DeadSource(), _Pose(), _Sink())
|
|
assert rt.step(1.0) is None # 스킵 틱(1 % 2 != 0)에도 None
|