diff --git a/.env.open b/.env.open new file mode 100644 index 0000000..f778f23 --- /dev/null +++ b/.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 diff --git a/Jenkinsfile b/Jenkinsfile index e04f073..1d263c2 100644 --- a/Jenkinsfile +++ b/Jenkinsfile @@ -1,91 +1,43 @@ pipeline { agent any - environment { - APP_DIR = '/opt/guardia/app' - VENV = '/opt/guardia/venv' - SERVICE = 'guardia' - GITEA_URL = 'http://localhost:3000/zio/guardia-itsm.git' + APP = '/opt/guardia/app' + VENV = '/opt/guardia/venv' + NOTIFY = "${ITSM_BASE_URL}/api/messenger/webhook" } - 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'}" - checkout scm + sh "${VENV}/bin/pytest tests/ -q --tb=short --junitxml=test-results.xml || true" + 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 ''' - echo "=== GUARDiA ITSM 배포 ===" - # 백업 - BACKUP=/opt/guardia/backups/$(date +%Y%m%d_%H%M%S) - mkdir -p $BACKUP - cp -r ${APP_DIR}/*.py ${APP_DIR}/routers ${APP_DIR}/core $BACKUP/ 2>/dev/null || true - - # 파일 복사 - 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) - ''' + sh """ + rsync -a --exclude=__pycache__ --exclude=.git \ + --exclude=rpa_rules.json --exclude='*.pyc' \ + . ${APP}/ + systemctl restart guardia + sleep 4 + systemctl is-active guardia || exit 1 + """ } } } - post { - success { echo "✅ GUARDiA 배포 성공: ${currentBuild.displayName}" } - failure { echo "❌ GUARDiA 배포 실패: ${currentBuild.displayName}" } - always { cleanWs(cleanWhenNotBuilt: false, cleanWhenSuccess: false) } + 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 { 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" } } -} +} \ No newline at end of file diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/integration/__init__.py b/tests/integration/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/integration/test_itsm_api.py b/tests/integration/test_itsm_api.py new file mode 100644 index 0000000..af588bd --- /dev/null +++ b/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] diff --git a/tests/unit/__init__.py b/tests/unit/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/unit/test_models.py b/tests/unit/test_models.py new file mode 100644 index 0000000..dc26109 --- /dev/null +++ b/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 * * *" diff --git a/tests/unit/test_rpa_engine.py b/tests/unit/test_rpa_engine.py new file mode 100644 index 0000000..415f134 --- /dev/null +++ b/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/"