"""528-1 · 3 · 5 신규 프로젝트 그룹 공통(요구 345 · 347-8). - 산림청 판정 — *.go.kr(대소문자 · 하위 도메인) · 시스템관리자 · 아닌 메일 - D01 생성 — 일반 사용자 403 · 산림청 = PLAN · 설계속도 기본 · 연도 끝 기본 - E01 생성 — 원본 B06 미완료 400 · 남의 것 404 · 완료 = STABILITY + 원본 칸 - 공유 판정 — 지정 회사 GET 200 · PUT 403 · 미지정 403 · 같은 회사 그대로 - 가입 — go.kr = 회사 「산림청」 · ACTIVE · 화면이 보낸 회사 값 무시 · 다른 메일 = NO_COMPANY - 종류별 단계 시드 · storage 허용 목록 - 528-6 D · E 저장소 = 종류 단계만(B 양식 · B0x 빈 폴더 없음 · D 는 B04_PreProcess) · ROAD 그대로 """ import asyncio import pytest from fastapi import HTTPException from A07_Register import A07_Register_Router as reg_router from A07_Register.A07_Register_Schema import RegisterRequest from common_util import common_util_auth as auth from common_util import common_util_auth_repository as auth_repo from common_util.common_util_project_kind import KIND_STAGE_DIRS, init_kind_storage from common_util.common_util_workflow_state import initialize_project_stages from config.config_system import PROJECT_STORAGE_STAGE_DIRS from D01_PlanRegister import D01_PlanRegister_Router as d01 from E01_StabRegister import E01_StabRegister_Router as e01 class FakeCursor: """SQL 조각 → 돌려줄 한 줄. 실행한 SQL · 인자를 쌓음.""" def __init__(self, answers, log): self.answers, self.log, self.last = answers, log, None async def execute(self, sql, args=None): self.log.append((" ".join(sql.split()), args)) self.last = next((v for k, v in self.answers.items() if k in sql), None) async def fetchone(self): return self.last async def __aenter__(self): return self async def __aexit__(self, *a): return False class FakePool: def __init__(self, answers=None): self.answers, self.log = answers or {}, [] def acquire(self): return self def cursor(self, *a): return FakeCursor(self.answers, self.log) async def commit(self): self.log.append(("commit", None)) async def __aenter__(self): return self async def __aexit__(self, *a): return False class Req: def __init__(self, method="GET", project_id="p1"): self.method, self.path_params = method, {"project_id": project_id} USER = {"user_id": 9, "role": "USER", "company_id": 1, "email": "a@gmail.com"} FOREST = {"user_id": 8, "role": "USER", "company_id": 4, "email": "Kim@Forest.GO.KR"} ADMIN = {"user_id": 1, "role": "SYSTEM_ADMIN", "company_id": 1, "email": "x@aislo.kr"} @pytest.mark.parametrize( "email,expected", [ ("a@forest.go.kr", True), ("A@KFS.GO.KR", True), ("a@nifos.sub.forest.go.kr", True), ("a@gmail.com", False), ("a@notgo.kr", False), ("a@go.kr.evil.com", False), (None, False), ], ) def test_forest_email(email, expected): assert auth.is_forest_email(email) is expected def test_forest_service(): assert auth.is_forest_service(FOREST) assert auth.is_forest_service(ADMIN) assert not auth.is_forest_service(USER) def _plan(**kw): body = {"project_name": "예정", "project_year": 2026, "region": "홍천", "road_type": "fire"} return d01.CreatePlanRequest(**{**body, **kw}) def test_d01_plan_forbidden_for_user(): with pytest.raises(HTTPException) as e: asyncio.run(d01.post_plan_project(None, _plan(), USER)) assert e.value.status_code == 403 def test_d01_plan_created(monkeypatch): seen = {} async def fake_create(**kw): seen.update(kw) return "new-id" monkeypatch.setattr(d01, "create_kind_project", fake_create) out = asyncio.run(d01.post_plan_project(None, _plan(start_chainage_m=0), FOREST)) assert out == {"project_id": "new-id"} assert seen["kind"] == "PLAN" and seen["company_id"] == 4 assert seen["design_speed_kph"] == 20 and seen["project_year_end"] == 2026 asyncio.run(d01.post_plan_project(None, _plan(design_speed_kph=40), ADMIN)) assert seen["design_speed_kph"] == 40 asyncio.run(d01.post_plan_project(None, _plan(road_type="work", design_speed_kph=40), ADMIN)) assert seen["design_speed_kph"] == 20 # 작업임도 = 20 만 with pytest.raises(HTTPException) as e: asyncio.run(d01.post_plan_project(None, _plan(project_year_end=2025), FOREST)) assert e.value.status_code == 400 def _e01(monkeypatch, rows): seen = {} async def fake_sources(company_id, project_id=None): return rows async def fake_create(**kw): seen.update(kw) return "stab-id" monkeypatch.setattr(e01, "_sources", fake_sources) async def fake_copy(project_id, source_id): seen["copied"] = (project_id, source_id) return {"source_fingerprint": "f" * 40} monkeypatch.setattr(e01, "create_kind_project", fake_create) monkeypatch.setattr(e01, "_copy_source", fake_copy) async def fake_seed(project_id): seen["seeded"] = project_id monkeypatch.setattr(e01, "seed_initial", fake_seed) payload = e01.CreateStabRequest(source_project_id="s" * 36) return seen, lambda: asyncio.run(e01.post_stab_project(None, payload, USER)) def test_e01_requires_b06(monkeypatch): src = {"project_id": "s" * 36, "name": "임도", "region": "r", "road_type": "fire"} _, call = _e01(monkeypatch, [{**src, "project_year": 2026, "section_complete": 0}]) with pytest.raises(HTTPException) as e: call() assert e.value.status_code == 400 _, call = _e01(monkeypatch, []) with pytest.raises(HTTPException) as e: call() assert e.value.status_code == 404 seen, call = _e01(monkeypatch, [{**src, "project_year": 2026, "section_complete": 1}]) assert call() == {"project_id": "stab-id", "source_fingerprint": "f" * 40, "warnings": []} assert seen["copied"] == ("stab-id", "s" * 36) assert seen["kind"] == "STABILITY" and seen["source_project_id"] == "s" * 36 assert seen["name"] == "임도 안정성평가" assert seen["seeded"] == "stab-id" # E02 초기값(532-3) def _access(monkeypatch, method, assigned, owner_company=2): pool = FakePool( { "FROM projects": (owner_company,), "plan_route_assignees": (1,) if assigned else None, } ) monkeypatch.setattr(auth, "get_db_pool", lambda: pool) return asyncio.run(auth.require_project_access(Req(method), USER)) def test_shared_get_only(monkeypatch): assert _access(monkeypatch, "GET", True) is USER for method in ("PUT", "POST", "DELETE"): with pytest.raises(HTTPException) as e: _access(monkeypatch, method, True) assert e.value.status_code == 403 with pytest.raises(HTTPException) as e: _access(monkeypatch, "GET", False) assert e.value.status_code == 403 assert _access(monkeypatch, "PUT", False, owner_company=1) is USER def test_register_forest_company(monkeypatch): pool = FakePool({"FROM companies": (4,)}) monkeypatch.setattr(auth_repo, "get_db_pool", lambda: pool) assert asyncio.run(auth_repo.complete_registration(5, False, forest=True)) == "ACTIVE" update = [a for s, a in pool.log if s.startswith("UPDATE users")] assert update == [(4, 5)] pool = FakePool({"FROM companies": (4,)}) monkeypatch.setattr(auth_repo, "get_db_pool", lambda: pool) assert asyncio.run(auth_repo.complete_registration(5, False)) == "NO_COMPANY" assert not any("companies" in s for s, _ in pool.log) def test_register_verify_forces_forest(monkeypatch): seen = {} async def fake_user(email): return {"id": 5, "email": email, "status": "PENDING_EMAIL", "is_master": 0} async def fake_otp(user_id, kind): from datetime import datetime, timedelta return {"id": 1, "expires_at": datetime.utcnow() + timedelta(minutes=5), "otp_hash": "h"} async def fake_complete(user_id, is_master, forest=False): seen["forest"] = forest return "ACTIVE" if forest else "NO_COMPANY" async def noop(*a, **k): return None monkeypatch.setattr(reg_router, "get_user_by_email", fake_user) monkeypatch.setattr(reg_router, "get_active_otp", fake_otp) monkeypatch.setattr(reg_router, "verify_password", lambda a, b: True) monkeypatch.setattr(reg_router, "consume_otp", noop) monkeypatch.setattr(reg_router, "trust_device", noop) monkeypatch.setattr(reg_router, "complete_registration", fake_complete) class R: headers = {} body = reg_router.RegisterVerifyRequest(email="Lee@KFS.go.kr", otp_code="123456") resp = asyncio.run(reg_router.verify_registration(body, R())) assert seen["forest"] is True and b'"ACTIVE"' in resp.body body = reg_router.RegisterVerifyRequest(email="lee@gmail.com", otp_code="123456") resp = asyncio.run(reg_router.verify_registration(body, R())) assert seen["forest"] is False and b'"NO_COMPANY"' in resp.body def test_register_ignores_sent_company(): data = RegisterRequest( email="a@kfs.go.kr", password="12345678", password_confirm="12345678", name="이", terms_version="1.0", terms_agreed=True, privacy_agreed=True, company_name="다른회사", ).model_dump() assert "company_name" not in data @pytest.mark.parametrize("kind,count", [("ROAD", 7), ("PLAN", 3), ("STABILITY", 5)]) def test_stage_seed_by_kind(kind, count): log = [] asyncio.run(initialize_project_stages(FakeCursor({}, log), "p1", kind)) assert len(log) == count if kind == "PLAN": assert [a[2] for _, a in log] == ["PLAN_BASE", "PLAN_ANALYSIS", "PLAN_CONFIRM"] def test_storage_dirs_allowed(): for name in ("D02_PlanBase", "D04_PlanConfirm", "E02_StabGround", "E05_StabReport"): assert name in PROJECT_STORAGE_STAGE_DIRS @pytest.mark.parametrize("kind", ["PLAN", "STABILITY"]) def test_kind_storage_light(tmp_path, kind): init_kind_storage(tmp_path, kind) names = {p.name for p in tmp_path.iterdir()} assert names == set(KIND_STAGE_DIRS[kind]) assert ("B04_PreProcess" in names) == (kind == "PLAN") assert not [p for p in tmp_path.rglob("*") if p.is_file()] def test_road_storage_unchanged(tmp_path, monkeypatch): from B02_ProjRegister import B02_ProjRegister_Repository as b02 seeded = [] monkeypatch.setattr(b02, "seed_project", seeded.append) b02._initialize_project_storage(tmp_path, "p1") assert seeded == [tmp_path] assert (tmp_path / "B09_Estimation" / "v1").is_dir() assert (tmp_path / "project_manifest.json").is_file() @pytest.mark.parametrize("kind", ["PLAN", "STABILITY", None]) def test_resolve_layout_skips_kind(tmp_path, kind): from common_util.common_util_storage import ensure_project_storage_layout if kind: init_kind_storage(tmp_path, kind) before = {p.name for p in tmp_path.iterdir()} ensure_project_storage_layout(str(tmp_path)) after = {p.name for p in tmp_path.iterdir()} assert (after == before) == bool(kind) assert ("B09_Estimation" in after) == (kind is None)