import type { BlockView } from '../blockView' import type { Inline, PropertyRow, RuleContext, Segment } from '../contract' import type { DefinitionStart } from '../nameResolution' import { describe, expect, it, vi } from 'vitest' import { toBlockViews } from '../blockView' import { createPythonAdapter } from '../languages/python' import { DocumentLookups } from '../ruleLookups' import { flatSegments, inlineOf, MemoryIndexHost, propertyLines, readFixture, WASM_DIR, withBlocks } from '../testSupport' import { pythonZhTranslator } from '../translators/pythonZh' import { WorkspaceIndexes } from '../workspaceIndex' import { fastapiRule } from './fastapi' const py = createPythonAdapter(WASM_DIR) const ROUTES = readFixture('routes.py') const SKILL_HUB = readFixture('skillHubRoutes.py') const CONSOLE = readFixture('consoleRoutes.py') /** 规则查名字的定义时查到的值(名字原文 → IndexEntry.value),别的没查到。 */ function known(values: Record): RuleContext['lookup'] { return (_index, node) => values[node.text] === undefined ? undefined : { key: node.text, name: { startIndex: 0, endIndex: 0 }, value: values[node.text] } } async function views(source: string, lookup?: RuleContext['lookup'], reports: string[] = []): Promise { return withBlocks(py, 'python', source, blocks => toBlockViews(blocks, { source, language: py, markers: [] }, pythonZhTranslator, { rules: [fastapiRule], report: message => reports.push(message), lookup })) } function flatten(list: readonly BlockView[]): BlockView[] { return list.flatMap(view => [view, ...flatten(view.children)]) } function properties(view: BlockView | undefined): PropertyRow[] { if (!view || !('properties' in view)) throw new Error(`块 ${view?.id} 不是属性表`) return view.properties } /** 函数名是 fn 的接口卡片。 */ async function route(source: string, fn: string, lookup?: RuleContext['lookup']): Promise { const found = flatten(await views(source, lookup)).find(view => view.kind === 'fastapi.route' && view.defines && source.slice(view.defines.startIndex, view.defines.endIndex) === fn) if (!found) throw new Error(`没有 ${fn} 的接口卡片`) return found } /** 属性表某一行(连子行格子)里所有的词。 */ function segmentsOf(row: PropertyRow | undefined): Segment[] { const pieces: Inline[] = [ ...(!row || 'markdown' in row.value ? [] : row.value), ...(row?.items ?? []).flatMap(item => item.cells.flatMap(inlineOf)), ] return flatSegments(pieces) } const HEADER = [ 'from typing import Annotated', '', 'from fastapi import APIRouter, Body, Cookie, Depends, FastAPI, File, Form, Header, HTTPException, Path, Query, Request, Security, UploadFile, status', '', 'router = APIRouter(prefix="/items", tags=["物品"], dependencies=[Depends(verify_token)], responses={401: {"description": "没登录"}})', 'app = FastAPI()', '', ].join('\n') describe('fastapi 库规则:接口卡片', () => { it('luffy 的 update_user:名称 / 说明 / 标签 / 异步 / 依赖,请求表、响应表;Forbidden 的状态码查到是 403,四条消息写在一起', async () => { const card = await route(ROUTES, 'update_user', known({ Forbidden: '403' })) expect(card.badge).toEqual({ label: '路由', color: { dark: '#61afef', light: '#1a6fb5' }, icon: 'globe' }) expect(propertyLines(properties(card))).toEqual([ '名称: update_user', '说明: 更新用户信息。\n\n管理员可更新任意用户,普通用户只能更新自己的基础资料。', '路由器: router', '标签: 用户管理', '异步: 是', '依赖: get_db(db)、get_current_user(current_user)', '请求: PUT /users/{user_id}', ' URL | /users/{user_id}', ' 方法 | PUT', ' 路径参数 | user_id: Int4Path', ' 请求体 | UserUpdateRequest', '响应: 200 / 403 / 422', ' 200 | UserDetailResponse | 成功', ' 403 | Forbidden | 无权更新此用户信息;无权修改角色;无权修改启用状态;无权修改密码', ' 422 | 参数校验失败 | FastAPI 自动返回', ]) expect(properties(card).map(row => row.columns)).toEqual([undefined, undefined, undefined, undefined, undefined, undefined, ['项', '值'], ['状态码', '内容', '说明']]) expect(card.attributes).toMatchObject({ kind: 'fastapi.route', access: 'public', async: 'true', httpMethod: 'PUT' }) expect(fastapiRule.attributeNames?.('zh')).toEqual({ attributes: { httpMethod: '请求方法' }, values: {} }) expect(ROUTES.slice(card.defines?.startIndex, card.defines?.endIndex)).toBe('update_user') // 函数体照常是子块,卡片的编辑范围是整个函数(连装饰器) expect(card.children.length).toBeGreaterThan(0) expect(card.code.split('\n')[0]).toBe('@router.put("/{user_id}", response_model=UserDetailResponse)') expect(card.code.split('\n').at(-1)).toBe(' )') }) it('着色:{参数} 按变量、别的路径按字符串,带源码下标;方法用 Swagger UI 的颜色;成功的状态码绿色,FastAPI 自动带的 422 淡色', async () => { const rows = properties(await route(ROUTES, 'update_user', known({ Forbidden: '403' }))) const request = rows.find(row => row.label === '请求') const [url, method] = request?.items ?? [] expect(flatSegments(inlineOf(url.cells[1])).map(piece => [piece.text, piece.role])).toEqual([['/users', 'string'], ['/', 'string'], ['{', 'variable'], ['user_id', 'variable'], ['}', 'variable']]) for (const piece of flatSegments(inlineOf(url.cells[1]))) expect(ROUTES.slice(piece.offset, (piece.offset ?? 0) + piece.text.length)).toBe(piece.text) // 路径里的 user_id 讲的是 update_user 的参数 user_id:高亮参数的各处出现时跟着亮,⌘ + 点击跳到参数 const param = ROUTES.indexOf('user_id: Int4Path', ROUTES.indexOf('def update_user')) expect(flatSegments(inlineOf(url.cells[1])).find(piece => piece.text === 'user_id')?.refers).toEqual({ startIndex: param, endIndex: param + 'user_id'.length }) expect(method.cells[1]).toEqual([{ text: 'PUT', role: 'keyword', color: { dark: '#fca130', light: '#b35f00' } }]) const [ok, forbidden, validation] = rows.find(row => row.label === '响应')?.items ?? [] expect(ok.cells[0]).toEqual([{ text: '200', role: 'number', color: { dark: '#2ea043', light: '#1a7f37' } }]) const model = ROUTES.indexOf('response_model=UserDetailResponse', ROUTES.indexOf('@router.put')) + 'response_model='.length expect(ok.cells[1]).toEqual([{ text: 'UserDetailResponse', role: 'type', offset: model }]) expect(forbidden.cells[0]).toEqual([{ text: '403', role: 'string' }]) expect(flatSegments(inlineOf(forbidden.cells[1]))).toEqual([{ text: 'Forbidden', role: 'type', offset: ROUTES.indexOf('Forbidden("无权更新') }]) expect(flatSegments([...inlineOf(validation.cells[1]), ...inlineOf(validation.cells[2])]).map(piece => piece.role)).toEqual(['hint', 'hint']) // 名字都带下标、指回源码原文 for (const piece of rows.flatMap(segmentsOf).filter(piece => piece.offset !== undefined && piece.role !== 'string' && piece.role !== 'variable')) expect(ROUTES.slice(piece.offset, (piece.offset ?? 0) + piece.text.length)).toBe(piece.text) }) it('没查到异常类的状态码(索引还没建好、语言扩展没答)时不列它', async () => { const lines = propertyLines(properties(await route(ROUTES, 'update_user'))) expect(lines.slice(-3)).toEqual(['响应: 200 / 422', ' 200 | UserDetailResponse | 成功', ' 422 | 参数校验失败 | FastAPI 自动返回']) }) it('luffy 的 list_skills:Annotated 里的 Query 是查询参数(类型取 Annotated 的第一项),Depends 是依赖,参数名是 _ 的只写依赖', async () => { expect(propertyLines(properties(await route(SKILL_HUB, 'list_skills')))).toEqual([ '名称: list_skills', '路由器: skill_hub_router', '标签: skill-hub', '异步: 是', '依赖: get_db(db)、get_current_user', '请求: GET /skill-hub/skills', ' URL | /skill-hub/skills', ' 方法 | GET', ' 查询参数 | query: 字符串 | 空, tag: 字符串 | 空, skip: 整数, limit: 整数', '响应: 200 / 422', ' 200 | list[HubSkillListItem] | 成功', ' 422 | 参数校验失败 | FastAPI 自动返回', ]) }) it('luffy 的 skill_hub:HTTPException 按 status_code 归,消息是 detail;Request 不算参数;-> None 没有内容', async () => { expect(propertyLines(properties(await route(SKILL_HUB, 'get_skill'))).slice(-3)).toEqual([ ' 200 | HubSkillDetail | 成功', ' 404 | HTTPException | f"技能不存在: {slug}"', ' 422 | 参数校验失败 | FastAPI 自动返回', ]) expect(propertyLines(properties(await route(SKILL_HUB, 'install'))).slice(5)).toEqual([ '请求: POST /skill-hub/apps/{app_id}/skills/{slug}', ' URL | /skill-hub/apps/{app_id}/skills/{slug}', ' 方法 | POST', ' 路径参数 | app_id: 字符串, slug: 字符串', '响应: 204 / 422', ' 204 | | 成功', ' 422 | 参数校验失败 | FastAPI 自动返回', ]) }) it('luffy 的 create_user:status_code=201 是成功那一行;Pydantic 模型是请求体;函数里调别的函数抛的异常不算', async () => { expect(propertyLines(properties(await route(CONSOLE, 'create_user', known({ Forbidden: '403' }))))).toEqual([ '名称: create_user', '说明: 创建新用户。\n\n需要管理员权限。密码会自动哈希。', '路由器: router', '标签: 用户管理-运维', '异步: 是', '依赖: get_db(db)、require_users_create', '请求: POST /admin/users', ' URL | /admin/users', ' 方法 | POST', ' 请求体 | UserCreateRequest', '响应: 201 / 422', ' 201 | UserDetailResponse | 成功', ' 422 | 参数校验失败 | FastAPI 自动返回', ]) }) it('luffy 的 list_users:别处写好的类型别名按索引查到的标记放(FilterQuery 是 Depends → 依赖),Literal 是简单类型', async () => { const lines = propertyLines(properties(await route(CONSOLE, 'list_users', known({ FilterQuery: 'Depends' })))) expect(lines).toContain('依赖: FilterQuery(filters)、get_db(db)、require_users_view') expect(lines).toContain(' 查询参数 | page: 整数, page_size: 整数, acquisition_channel_id: 整数 | 空, referred: Literal["yes", "no"] | 空') expect(lines.some(line => line.includes('请求体'))).toBe(false) }) it('参数放在哪儿照 FastAPI 的规则:路径里的名字、Query / Header / Cookie / Body / Form / File、上传文件、简单类型、枚举和类型别名(查到的)、Pydantic 模型;Request 不算', async () => { const source = `${HEADER} @router.post("/{item_id}/files/{path:path}") async def upload( item_id: int, path: str, request: Request, extra: Extra, q: str | None = None, ids: list[int] = Query(default=[]), color: Color = Color.RED, owner: Int4Query = None, token: Annotated[str, Header()] = "", session: str = Cookie(None), note: str = Form(...), upload: UploadFile = File(...), item: Item = Body(...), user: User = Security(get_user, scopes=["me"]), helper: Helper = Depends(), ): pass ` const lines = propertyLines(properties(await route(source, 'upload', known({ Color: 'Enum', Int4Query: 'Query' })))) expect(lines).toEqual(expect.arrayContaining([ '依赖: verify_token、get_user(user)、Helper(helper)', ' 路径参数 | item_id: 整数, path: 字符串', ' 查询参数 | q: 字符串 | 空, ids: 列表[整数], color: Color, owner: Int4Query', ' 请求头 | token: 字符串', ' Cookie | session: 字符串', ' 请求体 | extra: Extra, note: 字符串, upload: UploadFile, item: Item', ])) expect(lines.indexOf(' 请求头 | token: 字符串')).toBeLessThan(lines.indexOf(' Cookie | session: 字符串')) }) it('标签、依赖、响应叠上路由对象的;responses= 各占一行;status 常量、响应描述;嵌套函数里、光写 raise 的不算,raise ... from 算', async () => { const source = `${HEADER} @router.delete( "/{item_id}", status_code=status.HTTP_202_ACCEPTED, response_model=None, response_description="已受理", tags=["管理"], dependencies=[Depends(audit)], responses={404: {"model": ErrorOut, "description": "没找到"}, "5XX": {"description": "服务出错"}}, ) def remove(item_id: int): """删掉一个物品。""" try: drop(item_id) except KeyError as exc: raise HTTPException(status.HTTP_404_NOT_FOUND, "物品不存在") from exc except RuntimeError: raise if item_id < 0: raise HTTPException(status_code=409, detail="冲突") def inner(): raise HTTPException(status_code=500, detail="不算") ` expect(propertyLines(properties(await route(source, 'remove')))).toEqual([ '名称: remove', '说明: 删掉一个物品。', '路由器: router', '标签: 物品、管理', '异步: 否', '依赖: verify_token、audit', '请求: DELETE /items/{item_id}', ' URL | /items/{item_id}', ' 方法 | DELETE', ' 路径参数 | item_id: 整数', '响应: 202 / 401 / 404 / 409 / 422 / 5XX', ' 202 | | 已受理', ' 401 | | 没登录', ' 404 | ErrorOut、HTTPException | 没找到;物品不存在', ' 409 | HTTPException | 冲突', ' 422 | 参数校验失败 | FastAPI 自动返回', ' 5XX | | 服务出错', ]) }) it('直接挂在 FastAPI 应用上的接口:没有前缀;没有参数时没有 422;别的装饰器照写', async () => { const source = `${HEADER} @app.get("/health") @cache(60) async def health() -> dict: return {"ok": True} ` expect(propertyLines(properties(await route(source, 'health')))).toEqual([ '名称: health', '路由器: app', '异步: 是', '装饰器: @cache(60)', '请求: GET /health', ' URL | /health', ' 方法 | GET', '响应: 200', ' 200 | 字典 | 成功', ]) }) it('不认的:没导入 fastapi、路由对象不在这份文件里、不是路由对象的方法、不是挂接口的方法', async () => { const plain = 'router = APIRouter()\n\n@router.get("/")\ndef a():\n pass\n' expect(flatten(await views(plain)).some(view => view.kind === 'fastapi.route')).toBe(false) const source = `${HEADER} from .routes import other_router @other_router.get("/") def b(): pass @limiter.get("/") def c(): pass @router.websocket("/ws") async def d(ws): pass @router.get def e(): pass ` const reports: string[] = [] expect(flatten(await views(source, undefined, reports)).filter(view => view.kind === 'fastapi.route')).toEqual([]) expect(reports).toEqual([]) }) }) /** 名字是 name 的路由器卡片。 */ async function routerCard(source: string, name: string): Promise { const found = flatten(await views(source)).find(view => view.kind === 'fastapi.router' && view.defines && source.slice(view.defines.startIndex, view.defines.endIndex) === name) if (!found) throw new Error(`没有 ${name} 的路由器卡片`) return found } describe('fastapi 库规则:说明里的块标签', () => { const SOURCE = [ 'from fastapi import APIRouter, Depends, Query', '', 'router = APIRouter(prefix="/apps")', '', '@router.get("/{app_id}")', 'async def get_app(app_id: int, q: str | None = Query(None), db = Depends(get_db)) -> AppOut:', ' """取一个应用。', '', ' @param app_id 应用 id', ' @param q 按名字搜', ' @param db 数据库会话', ' @param qq 写错的名字', ' @return 应用详情', ' @throws [AppNotFound] 没有这个应用', ' """', ].join('\n') it('@param 对到请求表(一个参数一行,多一列说明),依赖参数不算对不上;@return 跟在响应后面;@throws 成异常表', async () => { const lines = propertyLines(properties(await route(SOURCE, 'get_app'))) expect(lines).toEqual([ '名称: get_app', '说明: 取一个应用。', '对不上的参数: 1 个', ' qq | 写错的名字', '路由器: router', '异步: 是', '依赖: get_db(db)', '请求: GET /apps/{app_id}', ' URL | /apps/{app_id} | ', ' 方法 | GET | ', ' 路径参数 | app_id: 整数 | 应用 id', ' 查询参数 | q: 字符串 | 空 | 按名字搜', '响应: 200 / 422 — 应用详情', ' 200 | AppOut | 成功', ' 422 | 参数校验失败 | FastAPI 自动返回', '异常: 1 个', ' [AppNotFound] | 没有这个应用', ]) }) }) describe('fastapi 库规则:路由器卡片', () => { it('luffy 的 router:名称 / 前缀 / 标签,接口表列这份文件里挂在它上面的接口(方法 / 完整路径 / 函数 / docstring 第一行)', async () => { const card = await routerCard(ROUTES, 'router') expect(card.badge).toEqual({ label: '路由器', color: { dark: '#7aa2f7', light: '#34548a' }, icon: 'network' }) expect(propertyLines(properties(card))).toEqual([ '名称: router', '前缀: /users', '标签: 用户管理', '接口: 3 个', ' GET | /users/{user_id} | get_user | 获取用户详情。', ' PUT | /users/{user_id} | update_user | 更新用户信息。', ' POST | /users/me/avatar | upload_current_user_avatar | 上传当前登录用户头像。', ]) expect(properties(card).at(-1)?.columns).toEqual(['方法', '路径', '函数', '说明']) expect(card.code).toBe('router = APIRouter(tags=["用户管理"], prefix="/users")') expect(ROUTES.slice(card.defines?.startIndex, card.defines?.endIndex)).toBe('router') }) it('函数名带下标(⌘ + 点击跳到接口卡片);路径里的 {user_id} 各讲各的函数的参数;方法用 Swagger UI 的颜色', async () => { const [getUser, updateUser] = properties(await routerCard(ROUTES, 'router')).at(-1)?.items ?? [] expect(getUser.cells[2]).toEqual([{ text: 'get_user', role: 'function', offset: ROUTES.indexOf('get_user(') }]) const paramOf = (fn: string): number => ROUTES.indexOf('user_id: Int4Path', ROUTES.indexOf(`def ${fn}`)) for (const [row, fn] of [[getUser, 'get_user'], [updateUser, 'update_user']] as const) expect(flatSegments(inlineOf(row.cells[1])).find(piece => piece.text === 'user_id')?.refers).toEqual({ startIndex: paramOf(fn), endIndex: paramOf(fn) + 'user_id'.length }) expect(updateUser.cells[0]).toEqual([{ text: 'PUT', role: 'keyword', color: { dark: '#fca130', light: '#b35f00' } }]) }) it('接口卡片写挂在哪个路由器上(名字带下标,⌘ + 点击跳到路由器卡片),排在说明后面', async () => { const rows = properties(await route(ROUTES, 'update_user')) expect(rows.map(row => row.label).slice(0, 4)).toEqual(['名称', '说明', '路由器', '标签']) expect(rows[2].value).toEqual([{ text: 'router', role: 'variable', offset: ROUTES.indexOf('router = APIRouter') }]) }) it('写了依赖、默认响应才有这两行;FastAPI 应用没有前缀;变量的说明是属性 docstring;没挂接口时「0 个」、不画表格', async () => { const source = `${HEADER}"""整个应用。""" empty = APIRouter() @router.get("/{item_id}") def show(item_id: int): """看一个物品。 带库存。 """ @app.post("/items") def create(): pass ` expect(propertyLines(properties(await routerCard(source, 'router')))).toEqual([ '名称: router', '前缀: /items', '标签: 物品', '依赖: verify_token', '默认响应: 401', ' 401 | | 没登录', '接口: 1 个', ' GET | /items/{item_id} | show | 看一个物品。', ]) expect(propertyLines(properties(await routerCard(source, 'app')))).toEqual([ '名称: app', '说明: 整个应用。', '接口: 1 个', ' POST | /items | create | ', ]) const empty = properties(await routerCard(source, 'empty')) expect(propertyLines(empty)).toEqual(['名称: empty', '接口: 0 个']) expect(empty.at(-1)?.columns).toBeUndefined() }) }) describe('fastapi 库规则:别的文件里的定义(工作区索引 + 语言扩展查定义)', () => { const SCHEMAS = 'backend/src/luffy_agent/core/exceptions/schemas.py' const PARAMS = 'backend/src/luffy_agent/core/params.py' const files: Record = { [SCHEMAS]: readFixture('appExceptions.py'), [PARAMS]: readFixture('coreParams.py'), 'backend/src/luffy_agent/user/routes.py': ROUTES, 'backend/src/luffy_agent/other/errors.py': 'class Forbidden(Exception):\n status_code = 418\n', 'backend/src/luffy_agent/other/models.py': 'import enum\n\nclass Color(str, enum.Enum):\n RED = "red"\n\nclass Plain:\n pass\n', } /** file 里 text 的起点(行、列 0 起),语言扩展答的定义处。 */ function at(file: string, text: string): DefinitionStart { const before = files[file].slice(0, files[file].indexOf(text)).split('\n') return { uri: `${MemoryIndexHost.FOLDER}/${file}`, line: before.length - 1, character: before.at(-1)?.length ?? 0 } } it('update_user 抛的 Forbidden:语言扩展查到定义在 core/exceptions/schemas.py,那个类上 status_code = 403;同名的别处的类不算', async () => { const indexes = new WorkspaceIndexes(new MemoryIndexHost(files, py)) const definitions = async (offset: number): Promise => ROUTES.startsWith('Forbidden', offset) ? [at(SCHEMAS, 'Forbidden(AppException)')] : [] let changed = 0 const lookups = new DocumentLookups(() => changed++, []) const first = lookups.session() expect(propertyLines(properties(await route(ROUTES, 'update_user', first.lookup)))).not.toContain(' 403 | Forbidden | 无权更新此用户信息;无权修改角色;无权修改启用状态;无权修改密码') lookups.resolve(1, first.requests(), { find: (index, key) => indexes.find(index, key), definitions }) await vi.waitFor(() => expect(changed).toBe(1)) expect(propertyLines(properties(await route(ROUTES, 'update_user', lookups.session().lookup)))).toContain(' 403 | Forbidden | 无权更新此用户信息;无权修改角色;无权修改启用状态;无权修改密码') indexes.dispose() }) it('索引认的:类上的 status_code(带不带注解、status 常量)、枚举类、Annotated 类型别名的标记;普通类不认', async () => { const host = new MemoryIndexHost({ ...files, 'backend/src/luffy_agent/other/more.py': 'from fastapi import status\n\nclass Gone(Exception):\n status_code: int = status.HTTP_410_GONE\n', }, py) const indexes = new WorkspaceIndexes(host) const lookups = new DocumentLookups(() => {}, []) const session = lookups.session() await route(`${HEADER} @router.get("/{item_id}") def show(item_id: Int4Path, color: Color, owner: Int4Query, plain: Plain): raise Gone("没了") `, 'show', session.lookup) const index = session.requests()[0].index const values = async (key: string): Promise => (await indexes.find(index, key)).map(({ entry }) => `${entry.value}`) expect(await values('Forbidden')).toEqual(['403', '418']) expect(await values('Gone')).toEqual(['410']) expect(await values('Color')).toEqual(['Enum']) expect(await values('Int4Path')).toEqual(['Path']) expect(await values('Int4Query')).toEqual(['Query']) expect(await values('Plain')).toEqual([]) expect(host.reports).toEqual([]) indexes.dispose() }) })