Merge remote-tracking branch 'origin/dev' into sub_laptop_3

This commit is contained in:
2026-10-21 13:16:04 +09:00
30 changed files with 1370 additions and 106 deletions
+59
View File
@@ -0,0 +1,59 @@
/* =============================================================================
* design_calc_sites.ts
* 계산 자리 판(PLAN 504-2 · 요구 309-2) — B05 · B06 페이지 열 때 한 번 받아 페이지 수명 동안 씀.
* 판 = 서버 `resources/master_template/calc_sites.json`(`common_util_calc_sites.py`) · 쓰기 = 시스템 관리자
* 「계산 자리」 창(`B01_Dashboard_UI_CalcSites.ts`). 열린 페이지는 다음 열기부터 · 못 받으면 전부 both.
* ground_sample(G1) server = ground 단계가 격자 대신 서버 표본(`B05_Profile_Api_TerrainGrid.ts`)
* corridor_mesh(D3) server = 3D 미리보기 워커 안 띄움 · 서버 모형만(`B05_Profile_UI_Corridor_Worker_Client.ts`)
* ========================================================================== */
import { API_BASE_URL } from "@config/config_frontend";
export type CalcSiteValue = "both" | "server";
export interface CalcSiteGroup {
key: string;
nos: string[];
switchable: boolean;
fixed: "both" | "server" | "browser" | null;
why: string | null;
}
export interface CalcSitesBoard {
version: number;
sites: Record<string, CalcSiteValue>;
groups: CalcSiteGroup[];
}
async function request(init: RequestInit = {}): Promise<CalcSitesBoard> {
const path = init.method === "PUT" ? "/dashboard/admin/calc-sites" : "/dashboard/calc-sites";
const response = await fetch(`${API_BASE_URL}${path}`, {
credentials: "include",
headers: { "Content-Type": "application/json" },
...init,
});
const data = (await response.json()) as CalcSitesBoard & { detail?: string };
if (!response.ok) throw new Error(data.detail ?? `HTTP ${response.status}`);
return data;
}
export const fetchCalcSites = (): Promise<CalcSitesBoard> => request();
export const saveCalcSites = (sites: Record<string, CalcSiteValue>): Promise<CalcSitesBoard> =>
request({ method: "PUT", body: JSON.stringify({ sites }) });
let opened: Promise<Record<string, CalcSiteValue>> | null = null;
/** 페이지 열 때 — 새로 받음(앞 페이지 판 버림). */
export function openCalcSites(): Promise<Record<string, CalcSiteValue>> {
opened = fetchCalcSites().then(
(board) => board.sites,
() => ({}),
);
return opened;
}
/** 이 묶음을 조작 중에도 서버가 셈? — 받기 전이면 받기를 기다림. */
export async function calcOnServer(key: string): Promise<boolean> {
return (await (opened ?? openCalcSites()))[key] === "server";
}
+2
View File
@@ -40,6 +40,7 @@ import { usePipeIds } from "./design_pipe_ids";
import { createFacilityLane } from "./design_facility_lane";
import { pinDraftProject } from "./design_store";
import { pipelineDetail, reconcilePagePipeline } from "./design_pipeline_page";
import { openCalcSites } from "./design_calc_sites";
import { forgetPageData, onPageBusy } from "./design_page_data";
import { pageRouteId } from "./design_page_route";
import { setPipeSectionsSource } from "./design_structure_pipe_calc";
@@ -119,6 +120,7 @@ export function renderDesignPage(root: HTMLElement): MountedPage {
root.replaceChildren(page);
const projectId = localStorage.getItem(CURRENT_PROJECT_ID_KEY) ?? "";
pinDraftProject(projectId); // 연 프로젝트 고정(388-3) — 다른 탭이 바꿔도 이 탭 초안 칸 그대로
void openCalcSites(); // 계산 자리 판(504-2) — 열 때 한 번 · 열린 동안 바뀐 판은 다음 열기부터
usePipeIds(projectId || null); // 관 세션 id 장부 = 이 프로젝트(323-5 · 카드 · 알약 · 종단 · 3D 같은 id)
// 관 길이 · 흙막이 개수 = 공용 캐시(페이지 노선) 한 벌 — 마지막 연 보기의 사본이 아님(416 · B05 = B06) ·
// 노선 초안 동안 = 계산 줄 새 노선 상세(다시 센 횡단 · 428)
+9 -2
View File
@@ -16,7 +16,7 @@ interface SettingUser {
company_id?: number | null;
}
/** key = M01 은 「m01」 · M02 는 진입 인자 `open` */
/** key = M01 은 「m01」 · 계산 자리 창은 「calcSites」(대시보드 카드만 · 504-2) · M02 는 진입 인자 `open` */
export interface SettingLink {
key: string;
title: string;
@@ -57,6 +57,12 @@ export function settingLinks(user: SettingUser, companyId: number | string | nul
m02(L("B01_Dashboard_CardStructCombo"), "구조물조합", comboRoot("master"), desc),
m02(L("B01_Dashboard_CardStructOption"), "옵션", "master", desc),
m02(L("B01_Dashboard_CardDrawing"), "도면", "master", desc),
{
key: "calcSites",
title: L("B01_CalcSites_Title"),
desc: L("B01_CalcSites_Desc"),
route: ROUTES.B01_ACCOUNT,
},
];
}
// 회사 수정 = 회사 관리자 · 회사 마스터(M02 accessFor 와 같은 규칙)
@@ -76,7 +82,8 @@ export function mountSettingsDrawer(
activeKey: string,
companyId: number | string | null,
): void {
const links = settingLinks(user, companyId);
// 계산 자리 = 대시보드 창(504-2) — 페이지가 아니라 ☰ 에 안 둠
const links = settingLinks(user, companyId).filter((link) => link.key !== "calcSites");
const bar = createStepBar(
links.map((link) => link.title),
links.findIndex((link) => link.key === activeKey),
+5 -2
View File
@@ -29,8 +29,11 @@ async function post(path: string, body?: object): Promise<ApiResult> {
return result;
}
export function requestLogin(email: string, password: string): Promise<ApiResult> {
return post("/auth/login/request", { email, password });
export async function requestLogin(email: string, password: string): Promise<ApiResult> {
const result = await post("/auth/login/request", { email, password });
// 화면이 뜰 때 기억한 「로그인 안 됨」(5 s)을 응답 뒤에 버림 — 안 버리면 대시보드 가드가 되튕김(309-1)
clearSessionCache();
return result;
}
export function verifyLogin(email: string, otpCode: string): Promise<ApiResult> {
+9 -19
View File
@@ -1,5 +1,6 @@
"""로그인, 재인증, 세션 및 비밀번호 API."""
import asyncio
import logging
from datetime import datetime, timedelta
@@ -7,7 +8,6 @@ from fastapi import APIRouter, Depends, HTTPException, Request
from fastapi.responses import JSONResponse
from common_util.common_util_auth import (
create_session,
delete_session_cookie,
generate_device_token,
generate_otp,
@@ -22,16 +22,15 @@ from common_util.common_util_auth import (
)
from common_util.common_util_auth_repository import (
change_password,
clear_login_failures,
consume_otp,
delete_session,
finish_login,
get_active_otp,
get_user_by_email,
has_trusted_device,
record_failed_login,
record_login,
replace_otp,
trust_device,
)
from common_util.common_util_email import send_email_background
from common_util.common_util_email_templates import otp_email, security_alert_email
@@ -66,7 +65,8 @@ async def request_login(payload: LoginRequest, request: Request):
if user["account_locked_until"] and user["account_locked_until"] > datetime.utcnow():
await record_login(user["id"], email, "FAILURE", "ACCOUNT_LOCKED", agent)
raise HTTPException(status_code=423, detail="로그인 실패 누적으로 계정이 잠겼습니다.")
if not verify_password(payload.password, user["password_hash"]):
# bcrypt(cost 12 · 약 0.2 s)는 스레드에서 — 이벤트 루프(다른 요청)를 막지 않게
if not await asyncio.to_thread(verify_password, payload.password, user["password_hash"]):
failures = await record_failed_login(user["id"])
await record_login(user["id"], email, "FAILURE", "INVALID_PASSWORD", agent)
if failures >= 5 and ADMIN_EMAIL:
@@ -108,25 +108,15 @@ async def request_login(payload: LoginRequest, request: Request):
async def _finish_login(
user: dict, agent: str, previous_device_token_hash: str | None = None
) -> JSONResponse:
await clear_login_failures(user["id"])
device_token = generate_device_token()
await trust_device(
session_id = await finish_login(
user["id"],
hash_device_token(device_token),
user["email"],
agent,
hash_user_agent(agent),
hash_device_token(device_token),
previous_device_token_hash,
)
from config.config_db import get_db_pool
pool = get_db_pool()
async with pool.acquire() as connection, connection.cursor() as cursor:
await cursor.execute(
"UPDATE users SET last_login = CURRENT_TIMESTAMP WHERE id = %s",
(user["id"],),
)
await connection.commit()
session_id = await create_session(user["id"], agent)
await record_login(user["id"], user["email"], "SUCCESS", None, agent)
response = JSONResponse(
{
"status": "success",
@@ -146,7 +136,7 @@ async def verify_login(payload: OtpVerifyRequest, request: Request):
not user
or not otp
or otp["expires_at"] < datetime.utcnow()
or not verify_password(payload.otp_code, otp["otp_hash"])
or not await asyncio.to_thread(verify_password, payload.otp_code, otp["otp_hash"])
):
if user:
failures = await record_failed_login(user["id"])
+36
View File
@@ -7,9 +7,12 @@ from uuid import UUID
from fastapi import APIRouter, Depends, HTTPException, Path, Query, Request, Response
from pymysql.err import IntegrityError
from common_util import common_util_calc_sites as calc_sites
from common_util.common_util_audit import record_audit
from common_util.common_util_auth import require_company, require_system_admin, verify_session
from common_util.common_util_email import send_email
from common_util.common_util_project_delete import hard_delete_project
from config.config_db import get_db_pool
from config.config_system import APP_PUBLIC_BASE_URL, PROJECT_DELETE_HARD_ENABLED
from .B01_Dashboard_Knowledge_Chat import router as knowledge_chat_router
@@ -59,6 +62,7 @@ from .B01_Dashboard_Schema import (
AddMemberRequest,
AdminUpdateUserRequest,
AssignCompanyRequest,
CalcSitesRequest,
ChangeUserRoleRequest,
CreateCompanyRequest,
InviteMemberRequest,
@@ -579,5 +583,37 @@ async def system_projects(session: dict[str, Any] = Depends(require_system_admin
return {"status": "success", "projects": await list_all_projects()}
# ── 계산 자리 판(504-2) — 읽기 = 로그인 · 쓰기 = 시스템 관리자 · 바뀐 줄마다 시스템 로그 한 줄 ──
@router.get("/calc-sites")
async def read_calc_sites(session: dict[str, Any] = Depends(verify_session)):
_ = session
return {"status": "success", **calc_sites.read_sites(), "groups": calc_sites.GROUPS}
@router.put("/admin/calc-sites")
async def write_calc_sites(
payload: CalcSitesRequest,
request: Request,
session: dict[str, Any] = Depends(require_system_admin),
):
try:
board, changed = calc_sites.write_sites(payload.sites)
except ValueError as error:
raise HTTPException(status_code=400, detail=str(error)) from error
if changed:
async with get_db_pool().acquire() as connection, connection.cursor() as cursor:
for key in changed:
await record_audit(
cursor,
actor_id=int(session["user_id"]),
action="CALC_SITE_CHANGE",
resource_type="calc_site",
resource_ref=f"{key}={board['sites'][key]}",
request=request,
)
await connection.commit()
return {"status": "success", **board, "groups": calc_sites.GROUPS, "changed": changed}
router.include_router(assets_router)
router.include_router(knowledge_chat_router)
+6
View File
@@ -22,6 +22,12 @@ class UiPrefsRequest(BaseModel):
prefs: dict[str, str] = Field(default_factory=dict)
class CalcSitesRequest(BaseModel):
"""계산 자리 판 고침(504-2) — 바꿀 줄만 · 줄 · 값 검사는 `common_util_calc_sites`."""
sites: dict[str, str] = Field(default_factory=dict)
class CreateCompanyRequest(BaseModel):
name: str = Field(min_length=1, max_length=255)
business_registration_number: str = Field(min_length=1, max_length=20)
+110
View File
@@ -0,0 +1,110 @@
/* =============================================================================
* B01_Dashboard_UI_CalcSites.ts
* 「계산 자리」 창(PLAN 504-2 · 요구 309-2) — 시스템 관리자만(카드 = `settingLinks` · 쓰기 = 서버
* `require_system_admin`). 묶음 8 줄 · 칸 둘(「조작 중 브라우저 셈」 · 「[저장] 서버 정본」 늘 켬 · 잠금).
* 바꿀 수 있는 줄 = 서버 판 `groups[].switchable` · 나머지는 잠긴 체크 + 까닭 글.
* 고침은 창 안 사본에만 · [저장] 때만 바뀐 줄을 PUT(선례 `app_settings.ts`).
* ========================================================================== */
import { createButton, el, showToast } from "@ui/ui_template_elements";
import { table } from "@ui/ui_template_general_blocks";
import { askLeave } from "@ui/ui_template_leave_dialog";
import { openModal } from "@ui/ui_template_modal";
import {
fetchCalcSites,
saveCalcSites,
type CalcSiteGroup,
type CalcSiteValue,
type CalcSitesBoard,
} from "../A00_Common/design/design_calc_sites";
import { L } from "./B01_Dashboard_UI_Common";
type LocaleKey = Parameters<typeof L>[0];
const K = (key: string): string => L(`B01_CalcSites_${key}` as LocaleKey);
function check(checked: boolean, locked: boolean): HTMLInputElement {
const input = el("input", { attrs: { type: "checkbox" } });
input.checked = checked;
input.disabled = locked;
return input;
}
function groupRow(
group: CalcSiteGroup,
draft: Record<string, CalcSiteValue>,
changed: () => void,
): HTMLElement[] {
const name = el("span", {
children: [el("strong", { text: K(group.key) }), ` (${group.nos.join(" ")})`],
});
const browser = group.switchable
? check(draft[group.key] !== "server", false)
: check(group.fixed !== "server", true);
browser.dataset.calcSite = group.key;
browser.addEventListener("change", () => {
draft[group.key] = browser.checked ? "both" : "server";
changed();
});
const server = check(group.fixed !== "browser", true);
return [name, browser, server, el("span", { text: K(`Why_${group.why ?? group.key}`) })];
}
/** 시스템 설정 「계산 자리」 카드 — 판을 받아 창을 엶. */
export async function openCalcSitesDialog(): Promise<void> {
let board: CalcSitesBoard;
try {
board = await fetchCalcSites();
} catch (error) {
showToast(error instanceof Error ? error.message : K("LoadFailed"), "error");
return;
}
const draft = { ...board.sites };
const dirty = (): Record<string, CalcSiteValue> =>
Object.fromEntries(Object.entries(draft).filter(([key, value]) => board.sites[key] !== value));
const count = el("p", { className: "b01-dashboard__calc-count" });
const recount = (): void => {
const n = Object.values(draft).filter((value) => value === "server").length;
count.textContent = K("ServerCount").replace("{value}", String(n));
};
recount();
const save = async (): Promise<boolean> => {
const changes = dirty();
if (!Object.keys(changes).length) return true;
try {
board = await saveCalcSites(changes);
} catch (error) {
showToast(error instanceof Error ? error.message : String(error), "error");
return false;
}
showToast(K("Saved"), "success");
return true;
};
const saveButton = createButton({
label: L("Common_Btn_Save"),
variant: "filled",
onClick: () => void save().then((ok) => ok && modal.close()),
});
const modal = openModal({
title: K("Title"),
closeLabel: L("Common_Btn_Close"),
dialogClass: "b01-dashboard__calc-sites",
wide: true,
actions: [saveButton],
beforeClose: async () => {
if (!Object.keys(dirty()).length) return true;
const choice = await askLeave(K("Leave"));
return choice === "discard" || (choice === "save" && (await save()));
},
mount: (body) => {
body.append(
count,
el("p", { className: "b01-dashboard__calc-note", text: K("NextOpen") }),
table(
[K("Col_Group"), K("Col_Browser"), K("Col_Server"), K("Col_Why")],
board.groups.map((group) => groupRow(group, draft, recount)),
),
);
},
});
}
+5 -2
View File
@@ -33,6 +33,7 @@ import {
type ResourceData,
} from "./B01_Dashboard_Api_Fetch";
import { auditLogTable, userTable } from "./B01_Dashboard_UI_Admin";
import { openCalcSitesDialog } from "./B01_Dashboard_UI_CalcSites";
import { L, roleLabel } from "./B01_Dashboard_UI_Common";
import {
buildCompanyPanel,
@@ -165,13 +166,15 @@ function settingCard(title: string, desc: string, go: () => void): HTMLElement {
return card;
}
/** 시스템 설정 — 카드 규칙은 `settingLinks` 한 벌(☰ 바로가기와 같이 씀 · 334) */
/** 시스템 설정 — 카드 규칙은 `settingLinks` 한 벌(☰ 바로가기와 같이 씀 · 334) · 계산 자리 = 창(504-2) */
function buildSystemSettingsPanel(user: DashboardUser, companyId: number | null): HTMLElement {
const wrap = document.createElement("div");
wrap.className = "b01-dashboard__setting-grid";
wrap.append(
...settingLinks(user, companyId).map((link) =>
settingCard(link.title, link.desc, () => navigateTo(link.route, link.query)),
settingCard(link.title, link.desc, () =>
link.key === "calcSites" ? void openCalcSitesDialog() : navigateTo(link.route, link.query),
),
),
);
return wrap;
+13
View File
@@ -787,3 +787,16 @@
outline: 2px solid var(--color-royal-amethyst, #6843a7);
outline-offset: 2px;
}
/* 계산 자리 창(504-2) — 묶음 · 까닭 글은 줄바꿈 · 체크 칸은 가운데 */
.b01-dashboard__calc-sites .b01-dashboard__table td {
white-space: normal;
}
.b01-dashboard__calc-sites .b01-dashboard__table td:nth-child(2),
.b01-dashboard__calc-sites .b01-dashboard__table td:nth-child(3) {
text-align: center;
}
.b01-dashboard__calc-count,
.b01-dashboard__calc-note {
margin: 0 0 0.5rem;
}
+5 -1
View File
@@ -12,6 +12,7 @@
import { API_BASE_URL } from "@config/config_frontend";
import { fetchCachedBytes } from "../A00_Common/b_asset_cache";
import { calcOnServer } from "../A00_Common/design/design_calc_sites";
import { type DtmGrid, parseDtmGrid } from "../common_util/common_util_surface_sampler";
import type { GroundStageDeps } from "./logic/B05_Profile_Pipeline_RouteGround";
@@ -74,9 +75,12 @@ export function designTerrainGridNow(projectId: string): DesignTerrainGrid | nul
const ELEVATION_CHUNK = 4000;
/** 계산 묶음 ground 단계에 넣을 격자 · 서버 길(PLAN 346-1). 서버 길도 소수 6자리 — 격자 길과 같은 값. */
export function groundStageDeps(projectId: string): Pick<GroundStageDeps, "grid" | "remote"> {
export function groundStageDeps(
projectId: string,
): Pick<GroundStageDeps, "grid" | "remote" | "serverOnly"> {
return {
grid: () => designTerrainGridNow(projectId),
serverOnly: () => calcOnServer("ground_sample"),
async remote(points, terrain, signal) {
const chunks: Array<Array<[number, number]>> = [];
for (let i = 0; i < points.length; i += ELEVATION_CHUNK) {
@@ -5,6 +5,7 @@
* 워커는 처음 부를 때 하나 세워 페이지 수명 동안 다시 쓴다.
* ========================================================================== */
import { calcOnServer } from "../A00_Common/design/design_calc_sites";
import type { RoutePoint } from "./B05_Profile_Api_Fetch";
import type { CrossSection } from "../B06_Section/B06_Section_Api_Fetch";
import { buildCorridor, type CorridorBuildResult } from "./B05_Profile_UI_Corridor_Build";
@@ -48,12 +49,14 @@ function getWorker(): Worker | null {
return worker;
}
/** 초안 미리보기 짓기 — 결과는 `buildCorridor` 와 같음(같은 입력 · 같은 함수). */
/** 초안 미리보기 짓기 — 결과는 `buildCorridor` 와 같음(같은 입력 · 같은 함수).
* 계산 자리 판(504-2) D3 = server 면 안 지음(null) — 서버가 [저장] · [확정] 때 지은 모형만. */
export async function buildCorridorOffThread(
crossSections: CrossSection[],
routePoints: RoutePoint[],
designSamples?: Samples,
): Promise<CorridorBuildResult | null> {
if (await calcOnServer("corridor_mesh")) return null;
const onMain = (): CorridorBuildResult | null =>
buildCorridor(crossSections, routePoints, designSamples);
const target = getWorker();
@@ -102,6 +102,8 @@ export interface GroundStageDeps {
terrain: string | null,
signal: AbortSignal,
): Promise<Array<number | null>>;
/** 계산 자리 판(504-2) G1 = server — 격자가 있어도 서버 길 · 격자는 cross 도 쓰므로 여기서만 가름. */
serverOnly?(): boolean | Promise<boolean>;
/** 횡단 반폭(m) — 비면 서버 기본(20 m). */
crossHalfWidthM?(): number | null;
}
@@ -122,7 +124,7 @@ export function createGroundStage(deps: GroundStageDeps): StageFn<"ground"> {
const terrain = ctx.inputs.route?.terrain ?? null;
const halfWidth = deps.crossHalfWidthM?.() ?? SECTION_CROSS_HALF_WIDTH_M;
const offsets = crossOffsets(halfWidth, SECTION_CROSS_SAMPLE_INTERVAL_M);
const local = deps.grid();
const local = (await deps.serverOnly?.()) ? null : deps.grid();
const useGrid = local !== null && (terrain === null || terrain === local.design);
const tag = terrain ?? "design";
@@ -1,7 +1,7 @@
/**
* 저장소 tmp/tests에 있는 CAD 테스트 실행용 설정 (테스트 코드는 git 추적 제외).
* 저장소 resources/tester/cad 의 CAD 테스트 실행용 설정.
* npx vitest run --config vitest.tmp.config.ts
* tmp에는 node_modules가 없어 라이브러리를 openwebcad 쪽으로 이어 준다.
* 저장소 뿌리에는 node_modules가 없어 라이브러리를 openwebcad 쪽으로 이어 준다.
*/
import { resolve } from 'node:path';
import { defineConfig } from 'vitest/config';
@@ -13,9 +13,15 @@ export default defineConfig({
'@flatten-js/core': resolve('node_modules/@flatten-js/core'),
'es-toolkit/compat': resolve('node_modules/es-toolkit/dist/compat/index.mjs'),
'es-toolkit': resolve('node_modules/es-toolkit'),
xstate: resolve('node_modules/xstate'),
},
},
test: {
include: ['tmp/tests/cad/**/*.test.ts'],
include: ['resources/tester/cad/**/*.test.ts'],
// node 꼴(describe 없음 · `node --experimental-strip-types` 로 돎) 넷 + `@ui` 별칭 쓰는 하나(vitest.ui.config.ts)
exclude: [
'**/node_modules/**',
'resources/tester/cad/{dxf-clip,table-from-drawing,theme-auto-color,random-uuid-polyfill,ctrl-shortcut-sheet}.test.ts',
],
},
});
+1 -15
View File
@@ -3,7 +3,7 @@
import hashlib
import re
import secrets
from datetime import datetime, timedelta
from datetime import datetime
from typing import Any
import bcrypt
@@ -88,20 +88,6 @@ def delete_session_cookie(response: Response) -> None:
)
async def create_session(user_id: int, user_agent: str) -> str:
session_id = secrets.token_urlsafe(32)
expires_at = datetime.utcnow() + timedelta(seconds=SESSION_MAX_AGE_SECONDS)
pool = get_db_pool()
async with pool.acquire() as connection, connection.cursor() as cursor:
await cursor.execute(
"""INSERT INTO sessions (id, user_id, user_agent_hash, expires_at)
VALUES (%s, %s, %s, %s)""",
(session_id, user_id, hash_user_agent(user_agent), expires_at),
)
await connection.commit()
return session_id
async def verify_session(request: Request) -> dict[str, Any]:
session_id = request.cookies.get(SESSION_COOKIE_NAME)
if not session_id:
+87 -42
View File
@@ -9,7 +9,7 @@ from typing import Any
import aiomysql
from config.config_db import get_db_pool
from config.config_system import EMAIL_REVERIFY_DAYS, OTP_VALID_MINUTES
from config.config_system import EMAIL_REVERIFY_DAYS, OTP_VALID_MINUTES, SESSION_MAX_AGE_SECONDS
def _company_code() -> str:
@@ -188,44 +188,88 @@ async def record_login(
) -> None:
pool = get_db_pool()
async with pool.acquire() as connection, connection.cursor() as cursor:
await cursor.execute(
"""INSERT INTO login_logs (user_id, email, status, failure_reason, user_agent)
VALUES (%s, %s, %s, %s, %s)""",
(user_id, email, status, reason, user_agent),
)
await _insert_login_log(cursor, user_id, email, status, reason, user_agent)
await connection.commit()
async def _insert_login_log(
cursor: Any,
user_id: int | None,
email: str,
status: str,
reason: str | None,
user_agent: str,
) -> None:
await cursor.execute(
"""INSERT INTO login_logs (user_id, email, status, failure_reason, user_agent)
VALUES (%s, %s, %s, %s, %s)""",
(user_id, email, status, reason, user_agent),
)
async def record_failed_login(user_id: int) -> int:
pool = get_db_pool()
async with pool.acquire() as connection, connection.cursor() as cursor:
await cursor.execute(
"""UPDATE users SET login_failures = login_failures + 1,
last_failed_at = CURRENT_TIMESTAMP,
account_locked_until = CASE WHEN login_failures + 1 >= 5
THEN DATE_ADD(CURRENT_TIMESTAMP, INTERVAL 15 MINUTE)
ELSE account_locked_until END
last_failed_at = CURRENT_TIMESTAMP
WHERE id = %s""",
(user_id,),
)
# 잠금은 따로 — 한 UPDATE 안에서는 MariaDB 가 앞 대입(+1)을 본 값으로 판정해 4 번째에 잠김
await cursor.execute(
"""UPDATE users
SET account_locked_until = DATE_ADD(CURRENT_TIMESTAMP, INTERVAL 15 MINUTE)
WHERE id = %s AND login_failures >= 5""",
(user_id,),
)
await cursor.execute("SELECT login_failures FROM users WHERE id = %s", (user_id,))
failures = (await cursor.fetchone())[0]
await connection.commit()
return failures
async def clear_login_failures(user_id: int) -> None:
async def finish_login(
user_id: int,
email: str,
user_agent: str,
user_agent_hash: str,
device_token_hash: str,
previous_device_token_hash: str | None,
) -> str:
"""로그인 마무리 쓰기(실패 지움·last_login · 신뢰 기기 · 세션 · login_logs) = 한 트랜잭션.
하나라도 실패하면 모두 되돌리고 예외를 올린다 — 반쪽 로그인 없음."""
session_id = secrets.token_urlsafe(32)
expires_at = datetime.utcnow() + timedelta(seconds=SESSION_MAX_AGE_SECONDS)
pool = get_db_pool()
async with pool.acquire() as connection, connection.cursor() as cursor:
await cursor.execute(
"""UPDATE users SET login_failures = 0, last_failed_at = NULL,
account_locked_until = NULL WHERE id = %s""",
(user_id,),
)
await connection.commit()
await connection.begin()
try:
await cursor.execute(
"""UPDATE users SET login_failures = 0, last_failed_at = NULL,
account_locked_until = NULL, last_login = CURRENT_TIMESTAMP
WHERE id = %s""",
(user_id,),
)
await _trust_device(
cursor, user_id, device_token_hash, user_agent_hash, previous_device_token_hash
)
await cursor.execute(
"""INSERT INTO sessions (id, user_id, user_agent_hash, expires_at)
VALUES (%s, %s, %s, %s)""",
(session_id, user_id, user_agent_hash, expires_at),
)
await _insert_login_log(cursor, user_id, email, "SUCCESS", None, user_agent)
await connection.commit()
except Exception:
await connection.rollback()
raise
return session_id
async def has_trusted_device(user_id: int, token_hash: str) -> bool:
"""읽기만 — last_used_at 갱신은 finish_login 트랜잭션의 _trust_device 가 함."""
valid_after = datetime.utcnow() - timedelta(days=EMAIL_REVERIFY_DAYS)
pool = get_db_pool()
async with pool.acquire() as connection, connection.cursor() as cursor:
@@ -234,15 +278,7 @@ async def has_trusted_device(user_id: int, token_hash: str) -> bool:
WHERE user_id = %s AND token_hash = %s AND last_used_at >= %s LIMIT 1""",
(user_id, token_hash, valid_after),
)
row = await cursor.fetchone()
if not row:
return False
await cursor.execute(
"UPDATE trusted_devices SET last_used_at = CURRENT_TIMESTAMP WHERE id = %s",
(row[0],),
)
await connection.commit()
return True
return bool(await cursor.fetchone())
async def trust_device(
@@ -253,25 +289,34 @@ async def trust_device(
) -> None:
pool = get_db_pool()
async with pool.acquire() as connection, connection.cursor() as cursor:
if previous_token_hash:
await cursor.execute(
"""UPDATE trusted_devices
SET token_hash = %s, user_agent_hash = %s,
verified_at = CURRENT_TIMESTAMP, last_used_at = CURRENT_TIMESTAMP
WHERE user_id = %s AND token_hash = %s""",
(token_hash, user_agent_hash, user_id, previous_token_hash),
)
if cursor.rowcount:
await connection.commit()
return
await cursor.execute(
"""INSERT INTO trusted_devices (user_id, user_agent_hash, token_hash)
VALUES (%s, %s, %s)""",
(user_id, user_agent_hash, token_hash),
)
await _trust_device(cursor, user_id, token_hash, user_agent_hash, previous_token_hash)
await connection.commit()
async def _trust_device(
cursor: Any,
user_id: int,
token_hash: str,
user_agent_hash: str,
previous_token_hash: str | None,
) -> None:
if previous_token_hash:
await cursor.execute(
"""UPDATE trusted_devices
SET token_hash = %s, user_agent_hash = %s,
verified_at = CURRENT_TIMESTAMP, last_used_at = CURRENT_TIMESTAMP
WHERE user_id = %s AND token_hash = %s""",
(token_hash, user_agent_hash, user_id, previous_token_hash),
)
if cursor.rowcount:
return
await cursor.execute(
"""INSERT INTO trusted_devices (user_id, user_agent_hash, token_hash)
VALUES (%s, %s, %s)""",
(user_id, user_agent_hash, token_hash),
)
async def delete_session(session_id: str) -> None:
pool = get_db_pool()
async with pool.acquire() as connection, connection.cursor() as cursor:
+101
View File
@@ -0,0 +1,101 @@
"""계산 자리 판(PLAN 504-2 · 요구 309-2) — 시스템 관리자 「계산 자리」 창이 읽고 쓰는 판 하나.
판 = `resources/master_template/calc_sites.json` `{version, sites: {묶음: "both" | "server"}}`.
both = 조작 중 브라우저 셈 + [저장] 서버 정본(지금 등록부 자리) · server = 조작 중에도 서버 길.
판이 없거나 깨지면 기본값(전부 both = 등록부 그대로). [저장] · [확정] 서버 정본 길은 판을 안 읽음.
묶음 8 줄 = 등록부(`design_calc_registry*.ts`) # 전부 · 같은지 시험 `test_309_2_calc_sites.py`.
"""
from __future__ import annotations
import json
import os
from pathlib import Path
from typing import Any
ROOT = Path(__file__).resolve().parents[1]
PATH = ROOT / "resources" / "master_template" / "calc_sites.json"
VALUES = ("both", "server")
# 묶음 · 등록부 # · 바꿀 수 있음 · 잠금 줄의 고정 자리(both · server · browser) · 까닭 열쇠
GROUPS: tuple[dict[str, Any], ...] = (
{"key": "ground_sample", "nos": ["G1"], "switchable": True, "fixed": None, "why": None},
{"key": "corridor_mesh", "nos": ["D3"], "switchable": True, "fixed": None, "why": None},
{
"key": "quantity",
"nos": ["E1", "F7", "F8", "C4"],
"switchable": False,
"fixed": "both",
"why": "later",
},
{"key": "carry", "nos": ["R9", "F10"], "switchable": False, "fixed": "both", "why": "later"},
{
"key": "stages",
"nos": ["R1", "R3", "R5", "R6", "R7", "G2", "G4", "P1", "P2", "P3", "P4", "P5", "P6"]
+ ["P8", "C1", "C2", "C3", "C5", "C6", "F1", "F2", "F3", "F4"],
"switchable": False,
"fixed": "both",
"why": "pair_later",
},
{
"key": "structure_pair",
"nos": ["F5", "F6", "F9"],
"switchable": False,
"fixed": "both",
"why": "pair",
},
{
"key": "fixed_server",
"nos": ["E2", "G3", "D1", "D2", "N1"],
"switchable": False,
"fixed": "server",
"why": "server",
},
{
"key": "fixed_browser",
"nos": ["R2", "R4", "R8", "P7", "C7", "D4"],
"switchable": False,
"fixed": "browser",
"why": "browser",
},
)
SWITCHABLE = frozenset(g["key"] for g in GROUPS if g["switchable"])
LOCKED = frozenset(g["key"] for g in GROUPS if not g["switchable"])
def read_sites() -> dict[str, Any]:
"""판 — 없음 · 깨짐 · 모르는 값 = 그 줄 기본값(both)."""
try:
raw = json.loads(PATH.read_text(encoding="utf-8"))
except (OSError, ValueError):
raw = {}
stored = raw.get("sites") if isinstance(raw, dict) else None
stored = stored if isinstance(stored, dict) else {}
sites = {key: stored.get(key) if stored.get(key) in VALUES else "both" for key in SWITCHABLE}
version = raw.get("version") if isinstance(raw, dict) else None
return {"version": version if isinstance(version, int) else 0, "sites": sites}
def check_changes(changes: dict[str, str]) -> None:
"""잠금 줄 · 모르는 줄 · 모르는 값 = ValueError(창구가 400)."""
for key, value in changes.items():
if key in LOCKED:
raise ValueError(f"잠긴 줄 — {key}")
if key not in SWITCHABLE:
raise ValueError(f"모르는 줄 — {key}")
if value not in VALUES:
raise ValueError(f"모르는 값 — {key}={value}")
def write_sites(changes: dict[str, str]) -> tuple[dict[str, Any], list[str]]:
"""검사 → 바뀐 줄만 얹어 판 번호 +1 · 통째 바꿔 씀(반쯤 쓴 판 없음). 돌려줌 = 새 판 · 바뀐 줄."""
check_changes(changes)
current = read_sites()
changed = sorted(k for k, v in changes.items() if current["sites"][k] != v)
if not changed:
return current, []
board = {"version": current["version"] + 1, "sites": {**current["sites"], **changes}}
temp = PATH.with_suffix(".json.tmp")
temp.write_text(json.dumps(board, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
os.replace(temp, PATH)
return board, changed
@@ -0,0 +1,7 @@
{
"version": 1,
"sites": {
"ground_sample": "both",
"corridor_mesh": "both"
}
}
+1 -1
View File
@@ -10,7 +10,7 @@ import {
prepareClip,
} from '../../../B07_DesignDetail/openwebcad/src/integration/dxf-clip.ts';
const sample = readFileSync('resources/tester/fixtures/b07_webclip_sample.dxf', 'utf-8');
const sample = readFileSync(new URL('../fixtures/b07_webclip_sample.dxf', import.meta.url), 'utf-8');
assert.equal(isDxfText(sample), true, '견본 DXF 알아봄');
assert.equal(isDxfText(sample.replace(/\n/g, '\r\n')), true, 'CRLF DXF 도 알아봄');
assert.equal(isDxfText('0\nSECTION\n'), true, '앞 공백 없는 DXF');
+114
View File
@@ -0,0 +1,114 @@
"""503-4 로그인 잠금 횟수 · 쿠키 재로그인 — 살아 있는 서버(8001)에서 손으로 돌림.
./venv/Scripts/python.exe resources/tester/helper_503_4_lock_live.py [time|lock]
time = 쿠키 있는 재로그인 다섯 번 시간(중앙값) · last_used_at 갱신 확인
lock = 일반 계정 4 번 틀림(안 잠김) → 5 번(잠김) → 6 번(423) · 끝나면 users 행 되돌림
계정 = tmp/verify_accounts.json 「일반」(값은 출력 안 함).
"""
import asyncio
import json
import statistics
import sys
import time
from pathlib import Path
import aiomysql
import httpx
ROOT = Path(__file__).resolve().parents[2]
sys.path.insert(0, str(ROOT))
from config.config_system import DB_HOST, DB_NAME, DB_PASSWORD, DB_PORT, DB_USER # noqa: E402
BASE = "http://localhost:8001/api"
ACC = json.loads((ROOT / "tmp/verify_accounts.json").read_text(encoding="utf-8"))["일반"]
COLS = "login_failures, last_failed_at, account_locked_until"
async def _q(sql, args=()):
conn = await aiomysql.connect(
host=DB_HOST,
port=DB_PORT,
db=DB_NAME,
user=DB_USER,
password=DB_PASSWORD,
autocommit=True,
init_command="SET time_zone='+00:00'",
)
try:
async with conn.cursor() as cur:
await cur.execute(sql, args)
return await cur.fetchall()
finally:
conn.close()
def q(sql, args=()):
return asyncio.run(_q(sql, args))
def row():
return q(f"SELECT id, {COLS} FROM users WHERE email = %s", (ACC["email"],))[0]
def login(client, password):
t = time.perf_counter()
r = client.post("/auth/login/request", json={"email": ACC["email"], "password": password})
return r.status_code, time.perf_counter() - t
def run_time():
client = httpx.Client(base_url=BASE, timeout=60)
assert login(client, ACC["password"])[0] == 200
uid = row()[0]
times = []
for _ in range(5):
before = q("SELECT MAX(last_used_at) FROM trusted_devices WHERE user_id = %s", (uid,))
time.sleep(1.1)
code, dt = login(client, ACC["password"])
after = q("SELECT MAX(last_used_at) FROM trusted_devices WHERE user_id = %s", (uid,))
assert code == 200 and after[0][0] > before[0][0], (code, before, after)
times.append(dt)
print(
"cookie relogin s:",
[round(t, 3) for t in times],
"median",
round(statistics.median(times), 3),
)
def run_lock():
orig = row()
uid = orig[0]
print("orig failures", orig[1], "locked", orig[3] is not None)
try:
q(
"UPDATE users SET login_failures = 0, last_failed_at = NULL,"
" account_locked_until = NULL WHERE id = %s",
(uid,),
)
client = httpx.Client(base_url=BASE, timeout=60)
for n in range(1, 7):
code, _ = login(client, "wrong-password-503-4")
r = row()
print(f"try {n}: http {code} failures {r[1]} locked {r[3] is not None}")
if n <= 4:
assert code == 401 and r[1] == n and r[3] is None
elif n == 5:
assert code == 401 and r[1] == 5 and r[3] is not None
else:
assert code == 423
print("LOCK OK")
finally:
q(
"UPDATE users SET login_failures = %s, last_failed_at = %s,"
" account_locked_until = %s WHERE id = %s",
(orig[1], orig[2], orig[3], uid),
)
print("restored", row()[1:] == orig[1:])
if __name__ == "__main__":
{"time": run_time, "lock": run_lock}[sys.argv[1] if len(sys.argv) > 1 else "lock"]()
@@ -90,7 +90,7 @@ def _settings(user: dict, active: str, company) -> dict:
icons = re.search(r"^const ICONS.*$", src, re.M).group(0)
script = (
"const ROUTES = { M01_MASTER_DATA: 'm01-master-data',"
" M02_MASTER_TEMPLATE: 'm02-master-template' };\n"
" M02_MASTER_TEMPLATE: 'm02-master-template', B01_ACCOUNT: 'b01-account' };\n"
"const L = (k: string) => k;\nlet bar: any = null;\n"
"const createStepBar = (titles: string[], active: number, o: any) =>"
" ({ titles, active, icons: o.icons, home: o.homeButton, routes: o.routes });\n"
@@ -118,8 +118,10 @@ def test_설정_바로가기_역할_셋_지금_칸():
["구조물조합", M02, {"open": "구조물조합", "root": "personal", "company": "3"}],
["옵션", M02, {"open": "옵션", "root": "master", "company": "3"}],
["도면", M02, {"open": "도면", "root": "master", "company": "3"}],
["calcSites", "b01-account", None], # 계산 자리(504-2) = 대시보드 카드만 · ☰ 엔 없음
]
bar = admin["bar"]
assert len(bar["titles"]) == 4
assert bar["active"] == 2 and bar["home"] is True
assert bar["icons"] == ["🗄️", "🧱", "🎛️", "📑"]
assert _settings({"role": "SYSTEM_ADMIN"}, "m01", None)["bar"]["active"] == 0
+1 -1
View File
@@ -57,7 +57,7 @@ def test_공용_양식_칸_라벨_단추_규칙():
label = _block(OVERLAY, ".ui-panel-form .ui-field,\n.ui-panel-form .b05-route__field")
assert "gap: var(--panel-label-gap);" in label
assert "display: none;" in _block(OVERLAY, ".ui-panel-form .ui-field__error:not(.is-visible)")
button = _block(OVERLAY, ".ui-panel-form .ui-btn:not(.ui-help-btn)")
button = _block(OVERLAY, ".ui-panel-form .ui-btn:not(.ui-help-btn, .ui-step-nav-btn)")
# 「구조물 배치」 단추(b05-route__irregular-btn)와 같은 크기
assert "padding: var(--spacing-8);" in button
assert "font-size: var(--text-caption);" in button
+295
View File
@@ -0,0 +1,295 @@
"""503-1 · 503-2 로그인 흠(요구 309-1 · 310).
- 되튕김 — requestLogin 응답 뒤 세션 확인 캐시를 비움
- 마무리 쓰기 = 한 커넥션 · 한 트랜잭션(begin 1 · commit 1) · 실패 = rollback · 예외
- bcrypt 확인 = 이벤트 루프 밖 스레드
- 보안 그대로 — 5 번 틀림 15 분 잠금 · 잠긴 계정 423 · 실패 기록
- 503-4 — 4 번 틀림 = 안 잠김 · 5 번 = 잠김 · 경고 메일 · 쿠키 재로그인 커밋 1
"""
import asyncio
import re
import threading
from pathlib import Path
import pytest
from fastapi import HTTPException
from A06_Login import A06_Login_Router as router
from common_util import common_util_auth_repository as repo
from common_util.common_util_auth import DEVICE_TOKEN_COOKIE_NAME
ROOT = Path(__file__).resolve().parents[2]
class FakeCursor:
def __init__(self, conn, fail_on=None):
self.conn, self.fail_on, self.rowcount = conn, fail_on, 0
async def execute(self, sql, args=None):
self.conn.log.append(("execute", " ".join(sql.split())))
if self.fail_on and self.fail_on in sql:
raise RuntimeError("db down")
self.rowcount = 1
async def __aenter__(self):
return self
async def __aexit__(self, *a):
return False
class FakeConn:
def __init__(self, fail_on=None):
self.log, self.fail_on = [], fail_on
def cursor(self, *a):
return FakeCursor(self, self.fail_on)
async def begin(self):
self.log.append(("begin",))
async def commit(self):
self.log.append(("commit",))
async def rollback(self):
self.log.append(("rollback",))
async def __aenter__(self):
return self
async def __aexit__(self, *a):
return False
class FakePool:
def __init__(self, fail_on=None):
self.conns = []
self.fail_on = fail_on
def acquire(self):
conn = FakeConn(self.fail_on)
self.conns.append(conn)
return conn
def _finish(monkeypatch, fail_on=None, previous=None):
pool = FakePool(fail_on)
monkeypatch.setattr(repo, "get_db_pool", lambda: pool)
result = asyncio.run(repo.finish_login(7, "a@b.c", "ua", "uah", "dth", previous))
return pool, result
def test_되튕김_캐시_비움():
src = (ROOT / "A06_Login/A06_Login_Api_Fetch.ts").read_text(encoding="utf-8")
body = re.search(r"export async function requestLogin\(.*?\n}\n", src, re.S).group(0)
assert body.index('await post("/auth/login/request"') < body.index("clearSessionCache();")
def test_마무리_한_트랜잭션(monkeypatch):
pool, session_id = _finish(monkeypatch)
assert len(pool.conns) == 1
log = pool.conns[0].log
kinds = [e[0] for e in log]
assert kinds.count("begin") == 1 and kinds.count("commit") == 1 and "rollback" not in kinds
assert kinds[0] == "begin" and kinds[-1] == "commit"
sqls = [e[1] for e in log if e[0] == "execute"]
assert "login_failures = 0" in sqls[0] and "last_login = CURRENT_TIMESTAMP" in sqls[0]
assert sqls[1].startswith("INSERT INTO trusted_devices")
assert sqls[2].startswith("INSERT INTO sessions")
assert sqls[3].startswith("INSERT INTO login_logs")
assert len(sqls) == 4 and len(session_id) >= 40
def test_마무리_신뢰기기_갱신(monkeypatch):
pool, _ = _finish(monkeypatch, previous="old")
sqls = [e[1] for e in pool.conns[0].log if e[0] == "execute"]
assert sqls[1].startswith("UPDATE trusted_devices")
assert not any(s.startswith("INSERT INTO trusted_devices") for s in sqls)
@pytest.mark.parametrize("fail_on", ["INSERT INTO sessions", "INSERT INTO login_logs"])
def test_마무리_실패_되돌림(monkeypatch, fail_on):
with pytest.raises(RuntimeError):
_finish(monkeypatch, fail_on=fail_on)
log = repo.get_db_pool().conns[0].log
kinds = [e[0] for e in log]
assert "commit" not in kinds and kinds[-1] == "rollback"
def _user(**kw):
base = {
"id": 7,
"email": "a@b.c",
"role": "USER",
"status": "ACTIVE",
"password_hash": "h",
"account_locked_until": None,
"last_email_verified_at": None,
}
return base | kw
class _Req:
headers = {"user-agent": "t"}
cookies = {}
def _payload():
return router.LoginRequest(email="a@example.com", password="pw-12345678")
def test_bcrypt_스레드(monkeypatch):
seen = {}
def fake_verify(value, hashed):
seen["thread"] = threading.current_thread() is threading.main_thread()
return False
async def noop(*a, **k):
return 1
async def get_user(email):
return _user()
monkeypatch.setattr(router, "verify_password", fake_verify)
monkeypatch.setattr(router, "get_user_by_email", get_user)
monkeypatch.setattr(router, "record_failed_login", noop)
monkeypatch.setattr(router, "record_login", noop)
with pytest.raises(HTTPException) as e:
asyncio.run(router.request_login(_payload(), _Req()))
assert e.value.status_code == 401 and seen["thread"] is False
def test_잠금_그대로(monkeypatch):
src = (ROOT / "common_util/common_util_auth_repository.py").read_text(encoding="utf-8")
assert "login_failures >= 5" in src and "INTERVAL 15 MINUTE" in src
logged = []
async def rec(*a):
logged.append(a[3])
async def get_user(email):
from datetime import datetime, timedelta
return _user(account_locked_until=datetime.utcnow() + timedelta(minutes=5))
monkeypatch.setattr(router, "get_user_by_email", get_user)
monkeypatch.setattr(router, "record_login", rec)
with pytest.raises(HTTPException) as e:
asyncio.run(router.request_login(_payload(), _Req()))
assert e.value.status_code == 423 and logged == ["ACCOUNT_LOCKED"]
class _UsersRow:
"""users 한 행 흉내 — record_failed_login 의 SQL 을 실행 순서대로 반영."""
def __init__(self):
self.failures, self.locked = 0, False
class _RowCursor(FakeCursor):
def __init__(self, conn, row):
super().__init__(conn)
self.row = row
async def execute(self, sql, args=None):
await super().execute(sql, args)
flat = " ".join(sql.split())
if "login_failures = login_failures + 1" in flat:
self.row.failures += 1
if "account_locked_until = DATE_ADD" in flat:
assert "login_failures >= 5" in flat and "+ 1" not in flat
if self.row.failures >= 5:
self.row.locked = True
async def fetchone(self):
return (self.row.failures,)
def _fail_n(monkeypatch, n):
row, pool = _UsersRow(), FakePool()
pool.acquire = lambda: _with_cursor(pool, row)
monkeypatch.setattr(repo, "get_db_pool", lambda: pool)
for _ in range(n):
asyncio.run(repo.record_failed_login(7))
return row
def _with_cursor(pool, row):
conn = FakeConn()
conn.cursor = lambda *a: _RowCursor(conn, row)
pool.conns.append(conn)
return conn
def test_잠금_4번_안잠김(monkeypatch):
row = _fail_n(monkeypatch, 4)
assert row.failures == 4 and row.locked is False
def test_잠금_5번_잠김(monkeypatch):
row = _fail_n(monkeypatch, 5)
assert row.failures == 5 and row.locked is True
@pytest.mark.parametrize("failures,sent", [(4, 0), (5, 1)])
def test_경고_메일(monkeypatch, failures, sent):
mails = []
async def failed(user_id):
return failures
async def noop(*a, **k):
return None
async def get_user(email):
return _user()
monkeypatch.setattr(router, "verify_password", lambda *a: False)
monkeypatch.setattr(router, "get_user_by_email", get_user)
monkeypatch.setattr(router, "record_failed_login", failed)
monkeypatch.setattr(router, "record_login", noop)
monkeypatch.setattr(router, "ADMIN_EMAIL", "admin@example.com")
monkeypatch.setattr(router, "send_email_background", lambda *a: mails.append(a))
with pytest.raises(HTTPException) as e:
asyncio.run(router.request_login(_payload(), _Req()))
assert e.value.status_code == 401 and len(mails) == sent
if sent:
assert mails[0][0] == "admin@example.com"
class _TrustedCursor(FakeCursor):
async def fetchone(self):
return (11,)
def test_쿠키_재로그인_커밋_1(monkeypatch):
from datetime import datetime
pool = FakePool()
orig = pool.acquire
def acquire():
conn = orig()
conn.cursor = lambda *a: _TrustedCursor(conn)
return conn
pool.acquire = acquire
async def get_user(email):
return _user(last_email_verified_at=datetime.utcnow())
class Req(_Req):
cookies = {DEVICE_TOKEN_COOKIE_NAME: "tok"}
monkeypatch.setattr(repo, "get_db_pool", lambda: pool)
monkeypatch.setattr(router, "verify_password", lambda *a: True)
monkeypatch.setattr(router, "get_user_by_email", get_user)
monkeypatch.setattr(router, "AUTH_OTP_DISABLED", False)
res = asyncio.run(router.request_login(_payload(), Req()))
assert res.status_code == 200
kinds = [e[0] for c in pool.conns for e in c.log]
sqls = [e[1] for c in pool.conns for e in c.log if e[0] == "execute"]
assert kinds.count("commit") == 1 and kinds.count("begin") == 1
assert sum(s.startswith("UPDATE trusted_devices") for s in sqls) == 1
+260
View File
@@ -0,0 +1,260 @@
"""PLAN 504-2 (요구 309-2 · 310) — 계산 자리 판 · 시스템 관리자 창구 · G1 · D3 연결.
못박는 것
- 묶음 8 줄 = 등록부 # 전부(45 · 겹침 없음) · 바꿀 수 있는 줄(G1 · D3) = browser_now · 고정 줄 = 그 자리.
- 커밋한 판 = 기본값(전부 both = 등록부 그대로) · 판 없음 · 깨짐 = 기본값.
- GET = 로그인 · PUT = 시스템 관리자만(ADMIN · USER 403) · 잠금 · 모르는 줄 · 값 = 400 · 바뀐 줄마다
시스템 로그 CALC_SITE_CHANGE 한 줄 · 같은 값 = 판 · 로그 그대로.
- ground 단계 = 판(serverOnly) 따라 격자 · 서버 길 · cross 의 격자는 그대로 · D3 = 워커 안 띄움.
- 카드 = 시스템 관리자 갈래에만.
./venv/Scripts/python.exe -m pytest resources/tester/test_309_2_calc_sites.py -q
"""
from __future__ import annotations
import json
import re
import shutil
import subprocess
import sys
from pathlib import Path
import pytest
from fastapi import FastAPI
from fastapi.testclient import TestClient
ROOT = Path(__file__).resolve().parents[2]
if str(ROOT) not in sys.path:
sys.path.insert(0, str(ROOT))
from B01_Dashboard import B01_Dashboard_Router as dash # noqa: E402
from common_util import common_util_calc_sites as calc_sites # noqa: E402
from common_util.common_util_auth import verify_session # noqa: E402
REGISTRY = (
ROOT / "A00_Common/design/logic/design_calc_registry.ts",
ROOT / "A00_Common/design/logic/design_calc_registry_facility.ts",
)
TSC = ROOT / "config" / "node_modules" / "typescript" / "bin" / "tsc"
def _text(rel: str) -> str:
return (ROOT / rel).read_text(encoding="utf-8")
def _registry_sites() -> dict[str, str]:
pairs: dict[str, str] = {}
for path in REGISTRY:
src = path.read_text(encoding="utf-8")
for no, site in re.findall(r'no: "(\w+)",[\s\S]*?site: "(\w+)"', src):
pairs[no] = site
return pairs
def test_묶음_여덟_줄_등록부_전부():
sites = _registry_sites()
assert len(sites) == 45
nos = [no for group in calc_sites.GROUPS for no in group["nos"]]
assert len(calc_sites.GROUPS) == 8
assert len(nos) == len(set(nos)) and set(nos) == set(sites)
assert calc_sites.SWITCHABLE == {"ground_sample", "corridor_mesh"}
for group in calc_sites.GROUPS:
got = {sites[no] for no in group["nos"]}
if group["switchable"] or group["fixed"] == "both":
assert got == {"browser_now"}, group["key"]
elif group["fixed"] == "server":
assert got <= {"server_canon", "server_only"}, group["key"]
else:
assert got == {"browser_only"}, group["key"]
by_key = {g["key"]: g["nos"] for g in calc_sites.GROUPS}
assert by_key["ground_sample"] == ["G1"] and by_key["corridor_mesh"] == ["D3"]
def test_커밋한_판_기본값_없음_깨짐(tmp_path, monkeypatch):
committed = json.loads(calc_sites.PATH.read_text(encoding="utf-8"))
assert committed["sites"] == {"ground_sample": "both", "corridor_mesh": "both"}
assert set(calc_sites.read_sites()["sites"].values()) == {"both"}
monkeypatch.setattr(calc_sites, "PATH", tmp_path / "none.json")
assert calc_sites.read_sites() == {
"version": 0,
"sites": {"ground_sample": "both", "corridor_mesh": "both"},
}
broken = tmp_path / "broken.json"
broken.write_text('{"sites": {"ground_sample": "server", "corridor_mesh": "x"', "utf-8")
monkeypatch.setattr(calc_sites, "PATH", broken)
assert set(calc_sites.read_sites()["sites"].values()) == {"both"}
class _Cursor:
def __init__(self, log: list):
self.log = log
async def execute(self, sql, args=None):
self.log.append((sql, args))
async def __aenter__(self):
return self
async def __aexit__(self, *exc):
return False
class _Conn(_Cursor):
def cursor(self):
return _Cursor(self.log)
async def commit(self):
self.log.append(("COMMIT", None))
class _Pool:
def __init__(self, log: list):
self.log = log
def acquire(self):
return _Conn(self.log)
@pytest.fixture
def client(tmp_path, monkeypatch):
board = tmp_path / "calc_sites.json"
board.write_text(calc_sites.PATH.read_text(encoding="utf-8"), "utf-8")
monkeypatch.setattr(calc_sites, "PATH", board)
log: list = []
monkeypatch.setattr(dash, "get_db_pool", lambda: _Pool(log))
app = FastAPI()
app.include_router(dash.router)
role = {"role": "SYSTEM_ADMIN"}
app.dependency_overrides[verify_session] = lambda: {"role": role["role"], "user_id": 7}
return TestClient(app), role, board, log
def test_읽기_로그인_쓰기_시스템_관리자만(client):
http, role, board, log = client
before = board.read_text("utf-8")
for name in ("USER", "ADMIN"):
role["role"] = name
got = http.get("/api/dashboard/calc-sites")
assert got.status_code == 200
assert got.json()["sites"]["ground_sample"] == "both"
assert [g["key"] for g in got.json()["groups"]][:2] == ["ground_sample", "corridor_mesh"]
put = http.put(
"/api/dashboard/admin/calc-sites", json={"sites": {"ground_sample": "server"}}
)
assert put.status_code == 403, name
assert board.read_text("utf-8") == before and log == []
def test_잠금_모르는_줄_400_감사_한_줄(client):
http, _role, board, log = client
url = "/api/dashboard/admin/calc-sites"
for bad in ({"stages": "server"}, {"fixed_server": "both"}, {"nope": "server"}):
assert http.put(url, json={"sites": bad}).status_code == 400, bad
assert http.put(url, json={"sites": {"ground_sample": "browser"}}).status_code == 400
assert log == []
done = http.put(url, json={"sites": {"ground_sample": "server"}})
assert done.status_code == 200 and done.json()["changed"] == ["ground_sample"]
saved = json.loads(board.read_text("utf-8"))
assert saved == {"version": 2, "sites": {"ground_sample": "server", "corridor_mesh": "both"}}
inserts = [args for sql, args in log if "system_audit_logs" in sql]
assert len(inserts) == 1
assert inserts[0][:5] == (7, "CALC_SITE_CHANGE", "calc_site", None, "ground_sample=server")
assert log[-1][0] == "COMMIT"
# 같은 값 = 판 번호 · 로그 그대로
again = http.put(url, json={"sites": {"ground_sample": "server"}})
assert again.json()["changed"] == [] and json.loads(board.read_text("utf-8"))["version"] == 2
assert len([1 for sql, _ in log if "system_audit_logs" in sql]) == 1
# 되돌림
http.put(url, json={"sites": {"ground_sample": "both"}})
assert json.loads(board.read_text("utf-8"))["sites"]["ground_sample"] == "both"
_RUNNER = """
import { createGroundStage } from "./B05_Profile/logic/B05_Profile_Pipeline_RouteGround.js";
const axis = Float64Array.from({ length: 61 }, (_, i) => i);
const grid = { x: axis, y: axis, z: new Float32Array(61 * 61).fill(5),
validBits: new Uint8Array(Math.ceil((61 * 61) / 8)).fill(255) };
const run = async (serverOnly) => {
const calls = [];
const stage = createGroundStage({
grid: () => ({ grid, design: "las" }),
remote: async (points) => { calls.push(points.length); return points.map(() => 9); },
...(serverOnly === undefined ? {} : { serverOnly }),
});
const station = { key: "k0", chainageM: 0, x: 30, y: 30, azimuthDeg: 0, left: [0, 1],
kind: "grid", id: null, ground: null };
const out = await stage({
generation: 1, range: null, signal: new AbortController().signal,
inputs: { route: null }, out: { stations: [station], route: null }, before: {}, prev: null,
});
return { calls: calls.length, center: out.stations[0].ground.centerZ };
};
console.log(JSON.stringify([
await run(undefined), await run(() => false), await run(async () => true), await run(() => true),
]));
"""
@pytest.mark.skipif(shutil.which("node") is None or not TSC.exists(), reason="node · tsc 없음")
def test_ground_판_따라_길_고름(tmp_path):
out = tmp_path / "js"
subprocess.run( # noqa: S603 — 고정 실행 파일
["node", str(TSC), str(ROOT / "B05_Profile/logic/B05_Profile_Pipeline_RouteGround.ts"),
"--outDir", str(out), "--rootDir", str(ROOT), "--module", "esnext", "--target", "es2022",
"--moduleResolution", "bundler", "--ignoreConfig"],
cwd=str(ROOT), check=True, capture_output=True,
) # fmt: skip
for emitted in out.rglob("*.js"):
emitted.write_text(
re.sub(r'(from\s+"\.\.?/[^"]+)"', r'\1.js"', emitted.read_text(encoding="utf-8")),
encoding="utf-8",
)
(out / "runner.mjs").write_text(_RUNNER, encoding="utf-8")
result = subprocess.run( # noqa: S603
["node", str(out / "runner.mjs")], capture_output=True, text=True, timeout=60
)
assert result.returncode == 0, result.stderr
grid, off, server_async, server = json.loads(result.stdout)
assert grid == off == {"calls": 0, "center": 5} # 판 없음 · both = 격자
assert server_async == server and server["calls"] == 1 and server["center"] == 9
def test_연결_자리_cross_격자_그대로():
deps = _text("B05_Profile/B05_Profile_Api_TerrainGrid.ts")
assert "grid: () => designTerrainGridNow(projectId)," in deps # cross 가 같이 쓰는 격자
assert 'serverOnly: () => calcOnServer("ground_sample"),' in deps
stage = _text("B05_Profile/logic/B05_Profile_Pipeline_RouteGround.ts")
assert "const local = (await deps.serverOnly?.()) ? null : deps.grid();" in stage
page = _text("A00_Common/design/design_pipeline_page.ts")
assert len(page.splitlines()) <= 699
assert "createCrossStage({ grid: groundDeps.grid })" in page
assert "calc_sites" not in page and "calcOnServer" not in page
worker = _text("B05_Profile/B05_Profile_UI_Corridor_Worker_Client.ts")
body = worker.split("export async function buildCorridorOffThread(")[1]
assert body.index('if (await calcOnServer("corridor_mesh")) return null;') < body.index(
"getWorker()"
)
assert "void openCalcSites();" in _text("A00_Common/design/design_page.ts")
# 저장 길은 판을 안 읽음
for rel in ("A00_Common/design/design_save.ts", "A00_Common/design/design_save_api.ts"):
assert "calc_sites" not in _text(rel) and "calcOnServer" not in _text(rel)
def test_카드_시스템_관리자만_글_다_있음():
nav = _text("A00_Common/m_settings_nav.ts")
links = nav.split("export function mountSettingsDrawer")[0]
admin, rest = links.split('if (user.role === "SYSTEM_ADMIN") {')[1].split("// 회사 수정 =")
assert 'key: "calcSites"' in admin and "calcSites" not in rest
assert 'settingLinks(user, companyId).filter((link) => link.key !== "calcSites")' in nav
page = _text("B01_Dashboard/B01_Dashboard_UI_Page.ts")
assert 'link.key === "calcSites" ? void openCalcSitesDialog()' in page
locale = _text("ui_template/ui_template_locale_m1.ts")
for group in calc_sites.GROUPS:
assert f"B01_CalcSites_{group['key']}:" in locale, group["key"]
assert f"B01_CalcSites_Why_{group['why'] or group['key']}:" in locale, group["key"]
ui = _text("B01_Dashboard/B01_Dashboard_UI_CalcSites.ts")
assert 'check(group.fixed !== "browser", true)' in ui # 서버 정본 칸 = 늘 잠금
assert "saveCalcSites(changes)" in ui and "dirty()" in ui # 바뀐 줄만 PUT
@@ -0,0 +1,151 @@
"""PLAN 504-2 화면 — 자기 포트(5177) Playwright 머리 없는 브라우저 · 서버가 떠 있을 때만.
- 시스템 관리자 = 「계산 자리」 카드 · 창(밝음 · 어두움 한 장씩 · tmp/shots_504_2)
- 회사 관리자 · 일반 = 카드 없음 · PUT 403
- G1 체크 끔 → [저장] → B05 열기(검증_sub2) → 지반 표본이 서버 길(`/route/elevations`) · 판 기본값으로 되돌림
./venv/Scripts/python.exe -m pytest resources/tester/test_309_2_calc_sites_browser.py -q
"""
from __future__ import annotations
import json
import subprocess
import urllib.request
from pathlib import Path
import pytest
ROOT = Path(__file__).resolve().parents[2]
BASE = "http://localhost:5177"
BOARD = ROOT / "resources" / "master_template" / "calc_sites.json"
PROJECT = "cb15892a-5788-4a26-9316-20a15421f68b" # 검증_sub2(OWNERS.md)
SHOTS = ROOT / "tmp" / "shots_504_2"
CARD = "계산 자리"
def _up() -> bool:
try:
urllib.request.urlopen(BASE, timeout=3)
return True
except Exception:
return False
pytestmark = pytest.mark.skipif(not _up(), reason="5177 서버 없음")
def _cred(who: str) -> dict:
if who == "시스템 관리자":
return json.loads((ROOT / "tmp" / "aislo_cred.json").read_text(encoding="utf-8"))
accounts = json.loads((ROOT / "tmp" / "verify_accounts.json").read_text(encoding="utf-8"))
return {k: accounts[who][k] for k in ("email", "password")}
@pytest.fixture(scope="module")
def browser():
sync = pytest.importorskip("playwright.sync_api")
committed = subprocess.run(
["git", "show", "HEAD:resources/master_template/calc_sites.json"],
cwd=ROOT, capture_output=True, check=True,
).stdout # fmt: skip
with sync.sync_playwright() as play:
chrome = play.chromium.launch(channel="msedge", headless=True)
yield chrome
chrome.close()
BOARD.write_bytes(committed) # 판 = 커밋한 기본값 그대로
def _context(browser, who: str):
ctx = browser.new_context(viewport={"width": 1400, "height": 900})
r = ctx.request.post(f"{BASE}/api/auth/login/request", data=_cred(who), timeout=60000)
assert r.ok, f"로그인 실패 {r.status}" # 비밀번호는 안 찍음
return ctx
def _dashboard(ctx):
page = ctx.new_page()
page.goto(f"{BASE}/#/b01-account", wait_until="domcontentloaded")
page.wait_for_selector(".b01-dashboard__setting-card", timeout=60000)
return page
def _cards(page) -> list[str]:
return page.eval_on_selector_all(
".b01-dashboard__setting-title", "(els) => els.map((e) => e.textContent.trim())"
)
@pytest.mark.parametrize("who", ["회사관리자", "일반"])
def test_다른_역할_카드_없음_403(browser, who):
ctx = _context(browser, who)
try:
assert CARD not in _cards(_dashboard(ctx))
put = ctx.request.put(
f"{BASE}/api/dashboard/admin/calc-sites",
data={"sites": {"ground_sample": "server"}},
)
assert put.status == 403
assert ctx.request.get(f"{BASE}/api/dashboard/calc-sites").ok # 읽기 = 로그인
finally:
ctx.close()
def _elevation_calls(ctx) -> int:
page = ctx.new_page()
# 대시보드가 지금 프로젝트 칸을 비움 — 페이지 스크립트보다 먼저 넣음
page.add_init_script(f"localStorage.setItem('frd_current_project_id', '{PROJECT}')")
calls: list[str] = []
page.on("request", lambda r: "/route/elevations" in r.url and calls.append(r.url))
page.goto(f"{BASE}/#/b05-profile", wait_until="domcontentloaded")
page.wait_for_function("() => !!window.__aisloPipeline", timeout=90000)
page.wait_for_timeout(8000)
page.close()
return len(calls)
def test_시스템_관리자_카드_창_G1_서버_길(browser):
ctx = _context(browser, "시스템 관리자")
try:
before = _elevation_calls(ctx) # 기본값 = 격자(브라우저)
page = _dashboard(ctx)
assert CARD in _cards(page)
page.get_by_text(CARD, exact=True).click()
dialog = page.wait_for_selector(".b01-dashboard__calc-sites", timeout=30000)
rows = page.eval_on_selector_all(
".b01-dashboard__calc-sites tbody tr",
"(trs) => trs.map((tr) => [...tr.querySelectorAll('input')]"
".map((i) => [i.checked, i.disabled]))",
)
assert len(rows) == 8
assert rows[0] == [[True, False], [True, True]] # G1 바꿀 수 있음 · 서버 정본 잠금
assert all(row[0][1] and row[1][1] for row in rows[2:]) # 나머지 잠금
assert "지금 서버 0 줄" in dialog.inner_text()
SHOTS.mkdir(parents=True, exist_ok=True)
dialog.screenshot(path=str(SHOTS / "light.png"))
page.evaluate("document.documentElement.setAttribute('data-theme', 'dark')")
dialog.screenshot(path=str(SHOTS / "dark.png"))
page.evaluate("document.documentElement.setAttribute('data-theme', 'light')")
page.click("input[data-calc-site='ground_sample']")
assert "지금 서버 1 줄" in dialog.inner_text()
page.locator(".b01-dashboard__calc-sites .ui-modal__actions button").click()
page.wait_for_selector(".b01-dashboard__calc-sites", state="detached", timeout=30000)
assert json.loads(BOARD.read_text("utf-8"))["sites"]["ground_sample"] == "server"
logs = ctx.request.get(f"{BASE}/api/dashboard/admin/audit-logs").json()["items"]
assert any(
log.get("action") == "CALC_SITE_CHANGE"
and log.get("resource_ref") == "ground_sample=server"
for log in logs
)
after = _elevation_calls(ctx)
print(f"elevations 요청: 기본값 {before} · 서버 {after}")
assert after > before
back = ctx.request.put(
f"{BASE}/api/dashboard/admin/calc-sites", data={"sites": {"ground_sample": "both"}}
)
assert back.ok and back.json()["sites"]["ground_sample"] == "both"
finally:
ctx.close()
+8 -6
View File
@@ -3,7 +3,7 @@
① storage 떠 두기 · 되돌림 — 바뀐 · 새로 생긴 · 지운 파일만 되씀(안 바뀐 파일 그대로).
② 실제 창구(검증_sub8 · DB · storage) — 중간 차례 실패 → DB 7 표 행 · storage 해시 시작 그대로 ·
차례 이름 · 까닭 · restored. 같은 본문 정상 = 옛 차례(안쪽 창구 차례로 부름)와 행 · 파일 같음.
검증 프로젝트가 없는 자리(다른 PC · DB 없음)는 ② 건너뜀.
실DB(②)는 `AISLO_LIVE_DB=1` 일 때만 · 검증 프로젝트가 없는 자리(다른 PC · DB 없음)도 건너뜀.
② 는 살아 있는 검증_sub8 DB · storage 는 임시 사본(492 · `helper_temp_storage`) · 노선 바뀌어도 됨
(최신 노선 id · 추가 측점 자리를 그때 읽음).
@@ -15,6 +15,7 @@ from __future__ import annotations
import asyncio
import hashlib
import json
import os
import sys
from pathlib import Path
from uuid import UUID
@@ -79,7 +80,8 @@ def _db_ready() -> Path | None:
return None
SHARED_ROOT = _db_ready()
#: 켤 때만 — `AISLO_LIVE_DB=1`(공유 검증 DB 행을 실제로 씀 · 기본 건너뜀 · DB 도 안 열음)
SHARED_ROOT = _db_ready() if os.environ.get("AISLO_LIVE_DB") == "1" else None
#: 시험이 쓰는 뿌리 = 공유 storage 의 임시 사본(492 · `sandbox` 가 세움 · Synology 로 안 번지게)
ROOT_DIR: Path | None = None
@@ -96,10 +98,10 @@ def sandbox():
def live(test):
"""실DB · 사본 storage 시험 — 검증 프로젝트 없으면 건너뜀."""
return pytest.mark.skipif(SHARED_ROOT is None, reason="검증_sub8 · DB 없음")(
pytest.mark.usefixtures("sandbox")(test)
)
"""실DB · 사본 storage 시험 — `AISLO_LIVE_DB=1` 이고 검증 프로젝트가 있을 때만."""
return pytest.mark.skipif(
SHARED_ROOT is None, reason="AISLO_LIVE_DB=1 일 때만(검증_sub8 · DB 필요)"
)(pytest.mark.usefixtures("sandbox")(test))
def _state(root: Path) -> tuple[dict, dict]:
+8 -6
View File
@@ -9,7 +9,7 @@
④ 실제 창구(검증_sub8 DB · storage 는 임시 사본) — 노선 차례 뒤 실패(시험 갈래 · 넘김 실패)
= 노선 행 · 자식 표 · storage(초기값 사본 포함) 그대로 · 정상 한 본문(교점 끌기) = 옛 차례
(노선 저장 → 다시 얹기 → 횡단 저장)와 행 · 파일 같음 · 시설 수 그대로 · 초기값 그대로(493).
검증 프로젝트가 없는 자리(다른 PC · DB 없음)는 ④ 건너뜀.
실DB(④)는 `AISLO_LIVE_DB=1` 일 때만 · 검증 프로젝트가 없는 자리(다른 PC · DB 없음)도 건너뜀.
./venv/Scripts/python.exe -m pytest resources/tester/test_472_6_route_save.py -q
"""
@@ -18,6 +18,7 @@ from __future__ import annotations
import asyncio
import json
import os
import re
import shutil
import subprocess
@@ -176,7 +177,8 @@ def _db_ready() -> Path | None:
return None
SHARED_ROOT = _db_ready()
#: 켤 때만 — `AISLO_LIVE_DB=1`(공유 검증 DB 행을 실제로 씀 · 기본 건너뜀 · DB 도 안 열음)
SHARED_ROOT = _db_ready() if os.environ.get("AISLO_LIVE_DB") == "1" else None
#: 시험이 쓰는 뿌리 = 공유 storage 의 임시 사본(492 · `sandbox` 가 세움 · Synology 로 안 번지게)
ROOT_DIR: Path | None = None
@@ -193,10 +195,10 @@ def sandbox():
def live(test):
"""실DB · 사본 storage 시험 — 검증 프로젝트 없으면 건너뜀."""
return pytest.mark.skipif(SHARED_ROOT is None, reason="검증_sub8 · DB 없음")(
pytest.mark.usefixtures("sandbox")(test)
)
"""실DB · 사본 storage 시험 — `AISLO_LIVE_DB=1` 이고 검증 프로젝트가 있을 때만."""
return pytest.mark.skipif(
SHARED_ROOT is None, reason="AISLO_LIVE_DB=1 일 때만(검증_sub8 · DB 필요)"
)(pytest.mark.usefixtures("sandbox")(test))
def _rows() -> dict:
+7 -2
View File
@@ -35,11 +35,12 @@ def _cards(user: dict, company_id: int | None = None) -> list:
script = (
"interface DashboardUser { role: string; is_master: boolean }\n"
"const ROUTES = { M01_MASTER_DATA: 'm01-master-data',"
" M02_MASTER_TEMPLATE: 'm02-master-template' };\n"
" M02_MASTER_TEMPLATE: 'm02-master-template', B01_ACCOUNT: 'b01-account' };\n"
"const L = (k: string) => k;\n"
"let went = '';\n"
"const navigateTo = (r: string, q?: Record<string, string>) =>"
" { went = r + (q ? '?' + new URLSearchParams(q) : ''); };\n"
"const openCalcSitesDialog = async () => { went = 'calc-sites-dialog'; };\n"
"const mk = (): any => ({ children: [] as any[], textContent: '', className: '',"
" on: null as any,"
" append(...c: any[]) { this.children.push(...c); },"
@@ -66,6 +67,8 @@ def _cards(user: dict, company_id: int | None = None) -> list:
M02 = "m02-master-template"
# 계산 자리(504-2) — 시스템 관리자만 · 이동 대신 창
CALC_SITES = ["B01_CalcSites_Title", "B01_CalcSites_Desc", "calc-sites-dialog"]
READ_CO = (
"B01_Dashboard_PermMasterRead · B01_Dashboard_PermCompanyRead · B01_Dashboard_PermPersonalWrite"
)
@@ -73,13 +76,14 @@ WRITE_CO = READ_CO.replace("CompanyRead", "CompanyWrite")
@pytest.mark.skipif(shutil.which("node") is None, reason="node 가 없음")
def test_시스템_관리자_카드_넷_고른_회사_해시():
def test_시스템_관리자_카드_다섯_고른_회사_해시():
m = "B01_Dashboard_PermMasterWrite"
assert _cards({"role": "SYSTEM_ADMIN", "is_master": False}, 3) == [
["B01_Dashboard_MasterElements", "B01_Dashboard_PermSourceRead", "m01-master-data"],
["B01_Dashboard_CardStructCombo", m, f"{M02}?open=구조물조합&root=master&company=3"],
["B01_Dashboard_CardStructOption", m, f"{M02}?open=옵션&root=master&company=3"],
["B01_Dashboard_CardDrawing", m, f"{M02}?open=도면&root=master&company=3"],
CALC_SITES,
]
# 회사가 없으면 해시에 안 실음
assert _cards({"role": "SYSTEM_ADMIN", "is_master": False})[1][2] == (
@@ -229,6 +233,7 @@ def test_구조물_조합_기본_층_개인():
["B01_Dashboard_CardStructCombo", m, f"{combo}&company=3"],
["B01_Dashboard_CardStructOption", m, f"{M02}?open=옵션&root=master&company=3"],
["B01_Dashboard_CardDrawing", m, f"{M02}?open=도면&root=master&company=3"],
CALC_SITES,
]
assert _cards({"role": "ADMIN", "is_master": False, "company_id": 1}) == [
["B01_Dashboard_CardStructCombo", WRITE_CO, combo],
@@ -42,7 +42,7 @@ def test_draft_pipe_options_are_replayed_onto_the_cache():
cross = _read("A00_Common", "design", "design_view_cross.ts")
assert "replayDraftPipeOptions" not in cross
store = _read("B06_Section", "B06_Section_Section_Store.ts")
assert "await withDraftShapes(detail, projectId);" in store
assert "await withDraftShapes(detail, projectId, false, frame);" in store
shared = _read("A00_Common", "design", "design_shared.ts")
assert "shared.lastPipes = pipes" in shared
profile = _read("A00_Common", "design", "design_view_profile.ts")
+50
View File
@@ -76,4 +76,54 @@ export const ui_locales_m1 = {
M01_GroupExchange: ["환율", "Exchange rates"],
M01_GroupRate: ["요율", "Rates"],
M01_GroupLogic: ["단가산출 로직", "Unit-price logic"],
B01_CalcSites_Title: ["계산 자리", "Calc sites"],
B01_CalcSites_Desc: ["조작 중 계산 = 브라우저 · 서버 고름", "Browser or server while editing"],
B01_CalcSites_ServerCount: ["지금 서버 {value} 줄", "{value} rows on server now"],
B01_CalcSites_NextOpen: [
"열린 B05 · B06 페이지는 다음 열기(새로고침)부터 바뀜",
"Open B05 · B06 pages change from the next open (reload)",
],
B01_CalcSites_Col_Group: ["묶음", "Group"],
B01_CalcSites_Col_Browser: ["조작 중 브라우저 셈", "Browser while editing"],
B01_CalcSites_Col_Server: ["[저장] 서버 정본", "Server canon on [Save]"],
B01_CalcSites_Col_Why: ["까닭", "Why"],
B01_CalcSites_ground_sample: ["지반 표본", "Ground sampling"],
B01_CalcSites_corridor_mesh: ["3D 길 모형", "3D corridor mesh"],
B01_CalcSites_quantity: ["토량 · 구조물 셈", "Earthwork · structure calc"],
B01_CalcSites_carry: ["이월 · 다시 얹기", "Carry · rebase"],
B01_CalcSites_stages: [
"노선 · 측점 · 종단 · 따라감 · 횡단 단계",
"Route · station · profile · follow · cross stages",
],
B01_CalcSites_structure_pair: ["구조물 짝", "Structure pairs"],
B01_CalcSites_fixed_server: ["고정 서버", "Fixed server"],
B01_CalcSites_fixed_browser: ["고정 브라우저", "Fixed browser"],
B01_CalcSites_Why_ground_sample: [
"끄면 격자 대신 서버 표본(조작마다 왕복)",
"Off = server sampling instead of the grid (round trip per edit)",
],
B01_CalcSites_Why_corridor_mesh: [
"끄면 미리보기 안 지음 · [저장] · [확정] 때 서버가 지은 모형만",
"Off = no preview · only the model the server built on [Save] · [Confirm]",
],
B01_CalcSites_Why_later: ["서버 계산 창구 뒤(2단계)", "After the server calc endpoint (step 2)"],
B01_CalcSites_Why_pair_later: [
"파이썬 짝 섞임 · 짝 모으기(472-4) 뒤",
"Mixed Python pairs · after pair merge (472-4)",
],
B01_CalcSites_Why_pair: [
"파이썬 정본 짝 유지 · TS 는 미리보기",
"Python canon pair · TS = preview",
],
B01_CalcSites_Why_server: [
"서버 단독(GIS · 래스터 · 값어치)",
"Server only (GIS · raster · value)",
],
B01_CalcSites_Why_browser: [
"입력 도구 · 그림 — 서버 셈 없음",
"Input tools · drawing — no server calc",
],
B01_CalcSites_Saved: ["계산 자리 저장함", "Calc sites saved"],
B01_CalcSites_Leave: ["저장 안 한 계산 자리가 있음", "You have unsaved calc sites"],
B01_CalcSites_LoadFailed: ["계산 자리를 못 읽음", "Failed to load calc sites"],
} as const;