"""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)