guardia-itsm/test_c1_cmdb.py

"""C-1 CMDB 확장 테스트"""
import sys, ast, os

os.environ.setdefault("GUARDIA_SECRET_KEY", "test-c1-secret-key-32bytes-padded!")
os.environ.setdefault("DATABASE_URL", "sqlite+aiosqlite:///./test_c1.db")

ok = True

print("=== 1. 구문 검사 ===")
files = ["routers/cmdb.py", "models.py", "main.py"]
for f in files:
    try:
        with open(f, encoding="utf-8") as fh:
            src = fh.read()
        ast.parse(src)
        print(f"  OK  {f}")
    except SyntaxError as e:
        print(f"  ERR {f}: {e}")
        ok = False

print("\n=== 2. models.py CI 모델 확인 ===")
with open("models.py", encoding="utf-8") as f:
    models_src = f.read()

model_checks = [
    ("class ConfigItem(Base):", "ConfigItem ORM 클래스"),
    ("class CIRelation(Base):", "CIRelation ORM 클래스"),
    ("class CIChangeLog(Base):", "CIChangeLog ORM 클래스"),
    ("class CIStatus(str, Enum):", "CIStatus Enum"),
    ("class CIType(str, Enum):", "CIType Enum"),
    ("class CIRelationType(str, Enum):", "CIRelationType Enum"),
    ("class CIChangeType(str, Enum):", "CIChangeType Enum"),
    ("ConfigItemOut", "ConfigItemOut Pydantic 스키마"),
    ("ConfigItemCreate", "ConfigItemCreate Pydantic 스키마"),
    ("ConfigItemUpdate", "ConfigItemUpdate Pydantic 스키마"),
    ("CIRelationOut", "CIRelationOut Pydantic 스키마"),
    ("CIChangeLogOut", "CIChangeLogOut Pydantic 스키마"),
    ("tb_ci", "tb_ci 테이블명"),
    ("tb_ci_relation", "tb_ci_relation 테이블명"),
    ("tb_ci_change_log", "tb_ci_change_log 테이블명"),
    ("DEPENDS_ON", "DEPENDS_ON 관계 타입"),
    ("HOSTED_ON", "HOSTED_ON 관계 타입"),
    ("linked_server_id", "서버 연결 컬럼"),
    ("attributes_json", "유연한 속성 JSON 컬럼"),
]
for sym, desc in model_checks:
    status = "OK" if sym in models_src else "ERR"
    if status == "ERR":
        ok = False
    print(f"  {status}  {desc}")

print("\n=== 3. routers/cmdb.py 엔드포인트 확인 ===")
with open("routers/cmdb.py", encoding="utf-8") as f:
    router_src = f.read()

endpoint_checks = [
    ('@router.post("/ci"', "POST /api/cmdb/ci (CI 생성)"),
    ('@router.get("/ci"', "GET /api/cmdb/ci (CI 목록)"),
    ('@router.get("/ci/stats"', "GET /api/cmdb/ci/stats"),
    ('@router.get("/ci/{ci_id}"', "GET /api/cmdb/ci/{ci_id}"),
    ('@router.patch("/ci/{ci_id}"', "PATCH /api/cmdb/ci/{ci_id}"),
    ('@router.delete("/ci/{ci_id}"', "DELETE /api/cmdb/ci/{ci_id} (폐기)"),
    ('@router.post("/ci/relations"', "POST CI 관계 추가"),
    ('@router.delete("/ci/relations/{relation_id}"', "DELETE CI 관계 삭제"),
    ('@router.get("/ci/{ci_id}/relations"', "GET CI 관계 조회"),
    ('@router.get("/ci/{ci_id}/history"', "GET CI 변경 이력"),
    ('@router.post("/ci/import-servers"', "POST 서버 CI 일괄 등록"),
    ("_next_ci_id", "CI ID 생성 함수"),
    ("_log_ci_change", "변경 이력 기록 함수"),
    ("CIChangeType.CREATE", "CREATE 변경 이력"),
    ("CIChangeType.RELATION_ADD", "RELATION_ADD 변경 이력"),
    ("RETIRED", "폐기 상태 처리"),
]
for sym, desc in endpoint_checks:
    status = "OK" if sym in router_src else "ERR"
    if status == "ERR":
        ok = False
    print(f"  {status}  {desc}")

print("\n=== 4. CI Enum 값 검증 ===")
try:
    import importlib.util
    spec = importlib.util.spec_from_file_location("models_mod", "models.py")
    models_mod = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(models_mod)

    # CIStatus
    ci_statuses = [e.value for e in models_mod.CIStatus]
    expected_statuses = ["PLANNED", "ACTIVE", "INACTIVE", "RETIRED", "DISPOSED"]
    for st in expected_statuses:
        status = "OK" if st in ci_statuses else "ERR"
        if status == "ERR":
            ok = False
        print(f"  {status}  CIStatus.{st}")

    # CIType
    ci_types = [e.value for e in models_mod.CIType]
    for t in ["SERVER", "NETWORK", "SOFTWARE", "SERVICE", "DATABASE"]:
        status = "OK" if t in ci_types else "ERR"
        if status == "ERR":
            ok = False
        print(f"  {status}  CIType.{t}")

    # CIRelationType
    rel_types = [e.value for e in models_mod.CIRelationType]
    for rt in ["DEPENDS_ON", "PART_OF", "HOSTED_ON", "CONNECTS_TO", "BACKS_UP"]:
        status = "OK" if rt in rel_types else "ERR"
        if status == "ERR":
            ok = False
        print(f"  {status}  CIRelationType.{rt}")

    # CIChangeType
    change_types = [e.value for e in models_mod.CIChangeType]
    for ct in ["CREATE", "UPDATE", "STATUS_CHANGE", "RETIRE", "RELATION_ADD", "RELATION_DEL"]:
        status = "OK" if ct in change_types else "ERR"
        if status == "ERR":
            ok = False
        print(f"  {status}  CIChangeType.{ct}")

except Exception as e:
    print(f"  ERR  Enum 로드 오류: {type(e).__name__}: {e}")
    ok = False

print("\n=== 5. CI 관계 타입 풍부성 검증 ===")
try:
    rel_types_set = set(e.value for e in models_mod.CIRelationType)
    assert len(rel_types_set) >= 5, f"관계 타입이 너무 적음: {len(rel_types_set)}"
    print(f"  OK  관계 타입 {len(rel_types_set)}개: {sorted(rel_types_set)}")

    lifespan_check = {
        "DEPENDS_ON": "A가 B에 의존",
        "PART_OF":    "A는 B의 구성요소",
        "HOSTED_ON":  "A는 B 위에서 실행",
        "CONNECTS_TO": "A↔B 네트워크",
        "BACKS_UP":   "A가 B를 백업",
    }
    for key, desc in lifespan_check.items():
        assert key in rel_types_set, f"{key} 없음"
        print(f"  OK  {key}: {desc}")

except AssertionError as e:
    print(f"  ERR  {e}")
    ok = False

print("\n=== 6. ConfigItemCreate Pydantic 모델 검증 ===")
try:
    from datetime import date

    # CI ID 형식 검증 (CI-YYYYMMDD-NNNN)
    from datetime import datetime
    today = datetime.utcnow().strftime("%Y%m%d")
    ci_id_example = f"CI-{today}-0001"
    assert ci_id_example.startswith("CI-"), "CI ID 형식 오류"
    # CI-YYYYMMDD-NNNN: 3+8+1+4 = 16자
    assert len(ci_id_example) == 16, f"CI ID 길이 오류: {len(ci_id_example)}"
    print(f"  OK  CI ID 형식: {ci_id_example}")

    # ConfigItemCreate 소스 구조 확인
    ci_create_start = models_src.find("class ConfigItemCreate(BaseModel):")
    ci_create_end   = models_src.find("\n\nclass ", ci_create_start + 1)
    ci_create_sec   = models_src[ci_create_start:ci_create_end]
    required_fields = ["name", "ci_type", "status", "owner", "location", "linked_server_id"]
    for f in required_fields:
        status = "OK" if f in ci_create_sec else "ERR"
        if status == "ERR":
            ok = False
        print(f"  {status}  ConfigItemCreate.{f}")

except Exception as e:
    print(f"  ERR  ConfigItemCreate 검증 오류: {type(e).__name__}: {e}")
    ok = False

print("\n=== 7. 변경 이력 구조 검증 ===")
try:
    # CIChangeLog 테이블 구조 확인
    change_start = models_src.find("class CIChangeLog(Base):")
    change_end   = models_src.find("\n\nclass ", change_start + 1)
    change_sec   = models_src[change_start:change_end]

    required_cols = [
        "ci_id_fk", "ci_id_str", "change_type",
        "field_name", "old_value", "new_value",
        "changed_by", "changed_at", "sr_id", "note"
    ]
    for col in required_cols:
        status = "OK" if col in change_sec else "ERR"
        if status == "ERR":
            ok = False
        print(f"  {status}  CIChangeLog.{col}")

    # ci_id_str 컬럼: CI 삭제 후에도 이력 조회 가능
    assert "ci_id_str" in change_sec, "ci_id_str 없음 (CI 삭제 후 조회 불가)"
    print(f"  OK  ci_id_str 컬럼: CI 삭제 후에도 이력 조회 가능")

except AssertionError as e:
    print(f"  ERR  {e}")
    ok = False

print("\n=== 8. import-servers 엔드포인트 검증 ===")
import_checks = [
    ("import-servers", "서버 CI 일괄 등록 엔드포인트"),
    ("type_map", "server_role→ci_type 매핑"),
    ("linked_server_id", "서버 연결 ID 저장"),
    ("WEB.*SERVER|SERVER.*WEB", "WEB 서버 타입 매핑"),
    ("MIDDLEWARE", "ESB → MIDDLEWARE 매핑"),
]
import re
for sym, desc in import_checks:
    # 정규식 검색 지원
    if "|" in sym or ".*" in sym:
        found = bool(re.search(sym, router_src))
    else:
        found = sym in router_src
    status = "OK" if found else "ERR"
    if status == "ERR":
        ok = False
    print(f"  {status}  {desc}")

print("\n=== C-1 CMDB 확장 테스트 완료 ===")
if ok:
    print("모든 검사 통과")
else:
    sys.exit(1)