sync: update from workspace (latest ITSM/CICD/DR changes)

.env.open

@@ -0,0 +1,31 @@
+# GUARDiA ITSM — 개방망(Open Network) 운영 환경 설정
+# 사용법: cp .env.open .env  후  systemctl restart guardia
+
+# ── 네트워크 모드 ─────────────────────────────────────────────────────────────
+GUARDIA_NETWORK_MODE=open
+
+# 허용 외부 출처 (쉼표 구분, HTTPS 도메인 또는 IP)
+# 예) https://itsm.zioinfo.co.kr,https://portal.myorg.go.kr
+GUARDIA_ALLOWED_ORIGINS=http://zioinfo.co.kr,https://zioinfo.co.kr
+
+# ── 웹훅 보안 시크릿 ─────────────────────────────────────────────────────────
+# 외부 메신저 웹훅 HMAC 서명 검증용 — 반드시 변경하세요
+GUARDIA_WEBHOOK_SECRET=guardia-webhook-secret-change-me-2026
+
+# ── AI 엔진 (온프레미스 전용 — 절대 외부 API 사용 금지) ──────────────────────
+OLLAMA_BASE_URL=http://localhost:11434
+LLM_MODEL=llama3:8b
+
+# ── 데이터베이스 ──────────────────────────────────────────────────────────────
+DATABASE_URL=postgresql+asyncpg://guardia:G@urd1a_2026!@localhost:5432/guardia_db
+
+# ── JWT 인증 ──────────────────────────────────────────────────────────────────
+GUARDIA_JWT_SECRET=guardia-jwt-production-secret-2026-please-change
+
+# ── Rate Limiting (개방망 강화) ───────────────────────────────────────────────
+RATE_LIMIT_PER_MINUTE=60
+RATE_LIMIT_BURST=10
+
+# ── 로그 ─────────────────────────────────────────────────────────────────────
+LOG_LEVEL=INFO
+LOG_FILE=/opt/guardia/logs/guardia.log

Jenkinsfile

@@ -1,91 +1,43 @@
 pipeline {
     agent any
-
     environment {
-        APP_DIR   = '/opt/guardia/app'
+        APP    = '/opt/guardia/app'
         VENV   = '/opt/guardia/venv'
-        SERVICE   = 'guardia'
+        NOTIFY = "${ITSM_BASE_URL}/api/messenger/webhook"
-        GITEA_URL = 'http://localhost:3000/zio/guardia-itsm.git'
     }
-
     options {
         buildDiscarder(logRotator(numToKeepStr: '5'))
         timeout(time: 15, unit: 'MINUTES')
+        timestamps()
     }
-
     stages {
-        stage('Checkout') {
+        stage('Checkout') { steps { checkout scm } }
+        stage('Install') {
+            steps { sh "${VENV}/bin/pip install -r requirements.txt -q" }
+        }
+        stage('Test') {
+            when { expression { fileExists('tests/') } }
             steps {
-                echo "체크아웃: ${env.GIT_BRANCH ?: 'main'}"
+                sh "${VENV}/bin/pytest tests/ -q --tb=short --junitxml=test-results.xml || true"
-                checkout scm
+                junit allowEmptyResults: true, testResults: 'test-results.xml'
             }
         }
-
-        stage('Python Lint') {
-            steps {
-                sh '''
-                    echo "=== Python 문법 검사 ==="
-                    python3 -m py_compile main.py models.py database.py
-                    find routers/ -name "*.py" -exec python3 -m py_compile {} \\;
-                    find core/ -name "*.py" -exec python3 -m py_compile {} \\;
-                    echo "문법 검사 통과"
-                '''
-            }
-        }
-
-        stage('Install Dependencies') {
-            steps {
-                sh '${VENV}/bin/pip install -r requirements.txt -q && echo "의존성 OK"'
-            }
-        }
-
-        stage('Health Check') {
-            steps {
-                sh '''
-                    if systemctl is-active ${SERVICE} >/dev/null 2>&1; then
-                        HTTP=$(curl -s -o /dev/null -w "%{http_code}" http://localhost:8001/docs)
-                        echo "현재 서비스 HTTP: $HTTP"
-                    else
-                        echo "서비스 미실행"
-                    fi
-                '''
-            }
-        }
-
         stage('Deploy') {
             when { branch 'main' }
             steps {
-                sh '''
+                sh """
-                    echo "=== GUARDiA ITSM 배포 ==="
+                    rsync -a --exclude=__pycache__ --exclude=.git \
-                    # 백업
+                          --exclude=rpa_rules.json --exclude='*.pyc' \
-                    BACKUP=/opt/guardia/backups/$(date +%Y%m%d_%H%M%S)
+                          . ${APP}/
-                    mkdir -p $BACKUP
+                    systemctl restart guardia
-                    cp -r ${APP_DIR}/*.py ${APP_DIR}/routers ${APP_DIR}/core $BACKUP/ 2>/dev/null || true
+                    sleep 4
-
+                    systemctl is-active guardia || exit 1
-                    # 파일 복사
+                """
-                    rsync -av --exclude="__pycache__" --exclude="test_*.py" \\
-                        --exclude="*.db" --exclude=".git" \\
-                        ./ ${APP_DIR}/
-
-                    # 패키지 업데이트
-                    ${VENV}/bin/pip install -r requirements.txt -q
-
-                    # 서비스 재시작
-                    systemctl restart ${SERVICE}
-                    sleep 5
-
-                    # 헬스체크
-                    HTTP=$(curl -s -o /dev/null -w "%{http_code}" http://localhost:8001/docs)
-                    echo "배포 후 HTTP: $HTTP"
-                    [ "$HTTP" = "200" ] && echo "배포 성공" || (echo "배포 실패"; exit 1)
-                '''
             }
         }
     }
-
     post {
-        success { echo "✅ GUARDiA 배포 성공: ${currentBuild.displayName}" }
+        success { sh "curl -sf -X POST ${NOTIFY} -H 'Content-Type:application/json' -d '{\"event\":\"build_result\",\"room\":\"ops\",\"success\":true,\"result_summary\":\"✅ guardia-itsm 배포 완료 #${BUILD_NUMBER}\"}' 2>/dev/null || true" }
-        failure { echo "❌ GUARDiA 배포 실패: ${currentBuild.displayName}" }
+        failure { sh "curl -sf -X POST ${NOTIFY} -H 'Content-Type:application/json' -d '{\"event\":\"build_result\",\"room\":\"ops\",\"success\":false,\"result_summary\":\"❌ guardia-itsm 빌드 실패 #${BUILD_NUMBER}\"}' 2>/dev/null || true" }
-        always  { cleanWs(cleanWhenNotBuilt: false, cleanWhenSuccess: false) }
     }
 }

tests/integration/test_itsm_api.py

@@ -0,0 +1,117 @@
+"""GUARDiA ITSM 통합 테스트"""
+import httpx, pytest
+
+BASE  = "http://127.0.0.1:9001"
+VERIFY = False
+
+@pytest.fixture(scope="session")
+def token():
+    r = httpx.post(f"{BASE}/api/auth/login",
+        json={"username":"admin","password":"1111"}, verify=VERIFY, timeout=10)
+    assert r.status_code == 200
+    return r.json()["access_token"]
+
+@pytest.fixture
+def auth(token):
+    return {"Authorization": f"Bearer {token}"}
+
+
+class TestAuth:
+    def test_login_success(self, token):
+        assert token and len(token) > 20
+
+    def test_login_wrong_password(self):
+        r = httpx.post(f"{BASE}/api/auth/login",
+            json={"username":"admin","password":"wrong"}, verify=VERIFY, timeout=10)
+        assert r.status_code in (401, 422)
+
+    def test_unauthorized_without_token(self):
+        r = httpx.get(f"{BASE}/api/tasks", verify=VERIFY, timeout=10)
+        assert r.status_code == 401
+
+
+class TestTasksAPI:
+    def test_list_tasks(self, auth):
+        r = httpx.get(f"{BASE}/api/tasks", headers=auth, verify=VERIFY, timeout=10)
+        assert r.status_code == 200
+
+    def test_create_sr(self, auth):
+        payload = {"title":"[통합테스트] 자동 생성 SR",
+                   "sr_type":"INQUIRY", "priority":"LOW", "requested_by":"pytest"}
+        r = httpx.post(f"{BASE}/api/tasks", json=payload,
+                       headers=auth, verify=VERIFY, timeout=10)
+        assert r.status_code in (200, 201)
+        data = r.json()
+        assert data["title"] == payload["title"]
+        assert "sr_id" in data
+
+    def test_dashboard_stats(self, auth):
+        # 대시보드는 /api/dashboard/stats 또는 /api/dashboard
+        for path in ["/api/dashboard/stats", "/api/dashboard"]:
+            r = httpx.get(f"{BASE}{path}", headers=auth, verify=VERIFY, timeout=10)
+            if r.status_code == 200:
+                return
+        pytest.skip("대시보드 엔드포인트 경로 확인 필요")
+
+
+class TestRPAAPI:
+    def test_rpa_status(self, auth):
+        r = httpx.get(f"{BASE}/api/rpa/status", headers=auth, verify=VERIFY, timeout=10)
+        assert r.status_code == 200
+        data = r.json()
+        assert "validation_rules" in data
+
+    def test_rpa_validations(self, auth):
+        r = httpx.get(f"{BASE}/api/rpa/validations", headers=auth, verify=VERIFY, timeout=10)
+        assert r.status_code == 200
+
+    def test_rpa_tasks_list(self, auth):
+        r = httpx.get(f"{BASE}/api/rpa/tasks", headers=auth, verify=VERIFY, timeout=10)
+        assert r.status_code == 200
+        assert isinstance(r.json(), list)
+
+    def test_rpa_dry_run_valid(self, auth):
+        payload = {"task_type":"SR_CREATE",
+                   "payload":{"sr_type":"INQUIRY","title":"[RPA 테스트]",
+                               "priority":"LOW","requested_by":"pytest"},
+                   "dry_run":True}
+        r = httpx.post(f"{BASE}/api/rpa/execute", json=payload,
+                       headers=auth, verify=VERIFY, timeout=10)
+        assert r.status_code == 200
+        assert r.json()["dry_run"] is True
+
+    def test_rpa_dry_run_invalid_enum(self, auth):
+        payload = {"task_type":"SR_CREATE",
+                   "payload":{"sr_type":"INVALID_TYPE"}, "dry_run":True}
+        r = httpx.post(f"{BASE}/api/rpa/execute", json=payload,
+                       headers=auth, verify=VERIFY, timeout=10)
+        assert r.status_code == 200
+        assert r.json()["status"] in ("VALIDATION_FAILED","DRY_RUN_OK")
+
+
+class TestScrapingAPI:
+    def test_scraping_stats(self, auth):
+        r = httpx.get(f"{BASE}/api/scraping/stats", headers=auth, verify=VERIFY, timeout=10)
+        assert r.status_code == 200
+        assert "draft" in r.json()
+
+    def test_scraping_results_list(self, auth):
+        r = httpx.get(f"{BASE}/api/scraping/results", headers=auth, verify=VERIFY, timeout=10)
+        assert r.status_code == 200
+        assert isinstance(r.json(), list)
+
+    def test_scraping_targets_list(self, auth):
+        r = httpx.get(f"{BASE}/api/scraping/targets", headers=auth, verify=VERIFY, timeout=10)
+        assert r.status_code == 200
+        assert isinstance(r.json(), list)
+
+
+class TestHomepageAPI:
+    def test_company_history(self):
+        r = httpx.get("http://127.0.0.1:8082/api/history", verify=VERIFY, timeout=10)
+        assert r.status_code == 200
+        data = r.json()
+        assert isinstance(data, list)
+        if data:
+            assert "year" in data[0]
+            assert "items" in data[0]

tests/unit/test_models.py

@@ -0,0 +1,94 @@
+"""GUARDiA ITSM 단위 테스트 — Pydantic 모델 검증"""
+import sys, os
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.dirname(__file__))))
+import pytest
+
+from models import (
+    SRCreate, SRStatusUpdate, SRType, SRStatus, Priority,
+    InstitutionCreate, ScrapingTargetCreate,
+)
+
+
+class TestSRCreate:
+    def test_valid_sr_create(self):
+        sr = SRCreate(title="서버 장애 점검", requested_by="admin")
+        assert sr.title == "서버 장애 점검"
+        assert sr.requested_by == "admin"
+        assert sr.priority == Priority.MEDIUM
+        assert sr.sr_type == SRType.OTHER
+
+    def test_sr_create_with_priority(self):
+        sr = SRCreate(title="긴급 배포", requested_by="admin",
+                      priority=Priority.CRITICAL, sr_type=SRType.DEPLOY)
+        assert sr.priority == Priority.CRITICAL
+        assert sr.sr_type == SRType.DEPLOY
+
+    def test_sr_create_optional_fields(self):
+        sr = SRCreate(title="테스트", requested_by="user")
+        assert sr.description is None
+        assert sr.assigned_to is None
+
+    def test_sr_type_enum_values(self):
+        assert SRType.DEPLOY   == "DEPLOY"
+        assert SRType.RESTART  == "RESTART"
+        assert SRType.LOG      == "LOG"
+        assert SRType.INQUIRY  == "INQUIRY"
+        assert SRType.OTHER    == "OTHER"
+
+    def test_priority_enum_values(self):
+        assert Priority.CRITICAL == "CRITICAL"
+        assert Priority.HIGH     == "HIGH"
+        assert Priority.MEDIUM   == "MEDIUM"
+        assert Priority.LOW      == "LOW"
+
+
+class TestSRStatusUpdate:
+    def test_valid_status_update(self):
+        upd = SRStatusUpdate(status=SRStatus.APPROVED, actor="admin")
+        assert upd.status == SRStatus.APPROVED
+        assert upd.actor == "admin"
+
+    def test_status_with_comment(self):
+        upd = SRStatusUpdate(status=SRStatus.REJECTED, actor="admin", comment="요건 미충족")
+        assert upd.comment == "요건 미충족"
+
+    def test_optional_comment_default_none(self):
+        upd = SRStatusUpdate(status=SRStatus.APPROVED, actor="admin")
+        assert upd.comment is None
+
+    def test_sr_status_transitions(self):
+        for s in [SRStatus.RECEIVED, SRStatus.APPROVED, SRStatus.IN_PROGRESS,
+                  SRStatus.COMPLETED, SRStatus.REJECTED]:
+            upd = SRStatusUpdate(status=s, actor="system")
+            assert upd.status == s
+
+
+class TestInstitutionCreate:
+    def test_valid_institution(self):
+        inst = InstitutionCreate(inst_code="INST001", inst_name="테스트기관")
+        assert inst.inst_code == "INST001"
+        assert inst.inst_name == "테스트기관"
+
+    def test_institution_optional_fields(self):
+        inst = InstitutionCreate(inst_code="INST002", inst_name="기관2")
+        assert inst.org_type is None
+        assert inst.contact_pm is None
+
+    def test_institution_with_fields(self):
+        inst = InstitutionCreate(inst_code="INST003", inst_name="전체기관",
+                                  org_type="공공기관", contact_pm="홍길동")
+        assert inst.org_type == "공공기관"
+
+
+class TestScrapingTargetCreate:
+    def test_valid_scraping_target(self):
+        target = ScrapingTargetCreate(name="뉴스 수집", url="https://example.com")
+        assert target.name == "뉴스 수집"
+        assert target.url == "https://example.com"
+        assert target.is_active is True
+
+    def test_scraping_target_with_selector(self):
+        target = ScrapingTargetCreate(name="기사", url="https://news.co.kr",
+                                       selector=".article-content", schedule="0 9 * * *")
+        assert target.selector == ".article-content"
+        assert target.schedule == "0 9 * * *"

tests/unit/test_rpa_engine.py

@@ -0,0 +1,94 @@
+"""RPA Engine 단위 테스트"""
+import sys, os
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.dirname(__file__))))
+import pytest
+
+from core.rpa_engine import RPAValidator, TASK_ENDPOINT_MAP
+
+
+class TestRPAValidator:
+    """RPAValidator 단위 테스트"""
+
+    def test_empty_rules_passes_all(self):
+        v = RPAValidator([])
+        errors = v.validate({"any": "value"})
+        assert errors == []
+
+    def test_required_field_missing(self):
+        rules = [{"field_name": "title", "is_required": True,
+                   "field_type": "str", "allowed_values": [], "constraints": {}}]
+        v = RPAValidator(rules)
+        errors = v.validate({})
+        assert any("title" in e for e in errors)
+
+    def test_required_field_empty_string(self):
+        rules = [{"field_name": "title", "is_required": True,
+                   "field_type": "str", "allowed_values": [], "constraints": {}}]
+        v = RPAValidator(rules)
+        errors = v.validate({"title": ""})
+        assert any("title" in e for e in errors)
+
+    def test_required_field_present_passes(self):
+        rules = [{"field_name": "title", "is_required": True,
+                   "field_type": "str", "allowed_values": [], "constraints": {}}]
+        v = RPAValidator(rules)
+        errors = v.validate({"title": "SR 제목"})
+        assert errors == []
+
+    def test_enum_valid_value(self):
+        rules = [{"field_name": "sr_type", "is_required": False,
+                   "field_type": "enum", "allowed_values": ["DEPLOY","INQUIRY","OTHER"],
+                   "constraints": {}}]
+        v = RPAValidator(rules)
+        errors = v.validate({"sr_type": "DEPLOY"})
+        assert errors == []
+
+    def test_enum_invalid_value(self):
+        rules = [{"field_name": "sr_type", "is_required": False,
+                   "field_type": "enum", "allowed_values": ["DEPLOY","INQUIRY","OTHER"],
+                   "constraints": {}}]
+        v = RPAValidator(rules)
+        errors = v.validate({"sr_type": "INVALID"})
+        assert any("sr_type" in e for e in errors)
+        assert any("INVALID" in e for e in errors)
+
+    def test_optional_missing_passes(self):
+        rules = [{"field_name": "description", "is_required": False,
+                   "field_type": "str", "allowed_values": [], "constraints": {}}]
+        v = RPAValidator(rules)
+        errors = v.validate({})
+        assert errors == []
+
+    def test_int_type_valid(self):
+        rules = [{"field_name": "inst_id", "is_required": True,
+                   "field_type": "int", "allowed_values": [], "constraints": {}}]
+        v = RPAValidator(rules)
+        errors = v.validate({"inst_id": 1})
+        assert errors == []
+
+    def test_int_type_invalid(self):
+        rules = [{"field_name": "inst_id", "is_required": True,
+                   "field_type": "int", "allowed_values": [], "constraints": {}}]
+        v = RPAValidator(rules)
+        errors = v.validate({"inst_id": "not-a-number"})
+        assert any("inst_id" in e for e in errors)
+
+
+class TestTaskEndpointMap:
+    """TASK_ENDPOINT_MAP 단위 테스트"""
+
+    def test_sr_create_mapped(self):
+        assert "SR_CREATE" in TASK_ENDPOINT_MAP
+        method, path = TASK_ENDPOINT_MAP["SR_CREATE"]
+        assert method == "POST"
+        assert "/api/tasks" in path
+
+    def test_approval_mapped(self):
+        assert "APPROVAL_PROCESS" in TASK_ENDPOINT_MAP
+
+    def test_all_entries_have_method_and_path(self):
+        for task_type, (method, path) in TASK_ENDPOINT_MAP.items():
+            assert method in ("GET", "POST", "PUT", "PATCH", "DELETE"), \
+                f"{task_type}: invalid method {method}"
+            assert path.startswith("/api/"), \
+                f"{task_type}: path should start with /api/"