form_124_write: запись в формы 1 2 4 и тесты

api/src/api/v1/forms.py

@@ -5,9 +5,10 @@ from fastapi import APIRouter, Depends, HTTPException, Response
 
 from sqlalchemy.ext.asyncio import AsyncSession
 
+from src.services.budget_line_service import BudgetLineService
 from src.db.models.form_type import FormTypeEnum
 from src.services.sheet_service import SheetService
-from src.domain.schemas import BaseListResponse, BaseSingleResponse, BudgetFormResponse, DirectionSchemaEnum, SheetFormTypeResponse, SheetResponse
+from src.domain.schemas import AddLineSchema, BaseListResponse, BaseSingleResponse, BudgetFormResponse, CellPatch, CellsPatch, DirectionSchemaEnum, SheetFormTypeResponse, SheetResponse
 from src.db.models.app_user import AppUser
 from src.services.budget_form_service import BudgetFormService
 from src.api.v1.deps import get_current_active_user_with_set_db
@@ -16,6 +17,17 @@ from src.db.session import get_db
 router = APIRouter(prefix="/form", tags=["forms"])
 
 
+def _rows_to_sheet_response(result: list[tuple]) -> list[SheetResponse]:
+    return [
+        SheetResponse(
+            row_type=el[0],
+            depth=el[1],
+            sort_order=el[2],
+            data=el[3],
+        ) for el in result
+    ]
+
+
 @router.get("/")
 async def get_forms(
     offset: int = 0,
@@ -99,33 +111,201 @@ async def get_sheet(
     response.headers["X-DB-Time-Ms"] = f"{db_ms:.2f}"
     return BaseListResponse(
             count=len(result),
-            result=[
+            result=_rows_to_sheet_response(result),
-                SheetResponse(
-                    row_type=el[0],
-                    depth=el[1],
-                    sort_order=el[2],
-                    data=el[3],
-                ) for el in result
-            ]
         )
-    # try:
 
-    #     with db_cursor() as cur:
+
-    #         t0 = time.perf_counter()
+@router.patch("/{form_id}/sheet/{sheet}/cell")
-    #         cur.execute(
+async def update_cell(
-    #             "SELECT row_type, depth, sort_order, data "
+        form_id: int, 
-    #             "FROM v3.v_form_view(%s, %s, NULL, %s)",
+        sheet: str, 
-    #             (form_id, sheet, direction),
+        response: Response,
-    #         )
+        cell_body: CellPatch,
-    #         rows = cur.fetchall()
+        direction: Optional[DirectionSchemaEnum] = None,
-    #         db_ms = (time.perf_counter() - t0) * 1000
+        sections: Optional[str] = None,
-    #         payload = [
+        db: AsyncSession = Depends(get_db),
-    #             {"row_type": r[0], "depth": r[1], "sort_order": r[2], "data": r[3]}
+        current_user: AppUser = Depends(get_current_active_user_with_set_db),
-    #             for r in rows
+) -> BaseListResponse[SheetResponse]:
-    #         ]
+    if sections is not None:
-    #         return JSONResponse(
+        sections = sections.split(",")
-    #             content=payload,
+    t0 = time.perf_counter()
-    #             headers={"X-DB-Time-Ms": f"{db_ms:.2f}"},
+    bf_service = BudgetFormService(db)
-    #         )
+    sheet_service = SheetService(db)
-    # except psycopg2.Error as e:
+
-    #     raise HTTPException(400, str(e).strip())
+    form = await bf_service.get(
+        user=current_user,
+        budget_form_id=form_id,
+        load_form_type=True,
+    )
+    if not form:
+        raise HTTPException(404, f"Форма {form_id} не найдена")
+    if sheet not in form.form_type.sheet_list:
+        raise HTTPException(404, f"Лист {sheet} формы {form_id} не найден")
+    if sections and (rem_sects := set(sections) - set(form.form_type.section_list)):
+        raise HTTPException(404, f"Секций {', '.join(rem_sects)} формы {form_id} не найден")
+
+    budget_line_service = BudgetLineService(db)
+    budget_line = await budget_line_service.get(budget_line_id=cell_body.line_id, user=current_user)
+    if budget_line.budget_form_id != form_id:
+        raise HTTPException(404, f"Строка формы не найдена")
+    
+
+    result = await sheet_service.update_cell(
+        form_id=form_id, 
+        sheet=sheet,
+        direction=direction.value if direction else None,
+        sections=sections,
+        line_id=cell_body.line_id, 
+        column=cell_body.column, 
+        value=cell_body.value, 
+        user=current_user,
+    )
+    db_ms = (time.perf_counter() - t0) * 1000
+    response.headers["X-DB-Time-Ms"] = f"{db_ms:.2f}"
+    return BaseListResponse(
+            count=len(result),
+            result=_rows_to_sheet_response(result),
+        )
+
+
+
+
+@router.patch("/{form_id}/sheet/{sheet}/cells")
+async def update_cells(
+        form_id: int, 
+        sheet: str, 
+        response: Response,
+        cells_body: CellsPatch,
+        direction: Optional[DirectionSchemaEnum] = None,
+        sections: Optional[str] = None,
+        db: AsyncSession = Depends(get_db),
+        current_user: AppUser = Depends(get_current_active_user_with_set_db),
+) -> BaseListResponse[SheetResponse]:
+    if sections is not None:
+        sections = sections.split(",")
+    t0 = time.perf_counter()
+    bf_service = BudgetFormService(db)
+    sheet_service = SheetService(db)
+
+    form = await bf_service.get(
+        user=current_user,
+        budget_form_id=form_id,
+        load_form_type=True,
+    )
+    if not form:
+        raise HTTPException(404, f"Форма {form_id} не найдена")
+    if sheet not in form.form_type.sheet_list:
+        raise HTTPException(404, f"Лист {sheet} формы {form_id} не найден")
+    if sections and (rem_sects := set(sections) - set(form.form_type.section_list)):
+        raise HTTPException(404, f"Секций {', '.join(rem_sects)} формы {form_id} не найден")
+
+    budget_line_service = BudgetLineService(db)
+    budget_lines = await budget_line_service.get_list(
+        budget_line_ids=[cb.line_id for cb in cells_body.changes], user=current_user)
+    for budget_line in budget_lines:
+        if budget_line.budget_form_id != form_id:
+            raise HTTPException(404, f"Строка формы не найдена")
+    
+
+    result = await sheet_service.update_cells(
+        form_id=form_id, 
+        sheet=sheet,
+        direction=direction.value if direction else None,
+        sections=sections,
+        changes=[c.model_dump() for c in cells_body.changes],
+        user=current_user,
+    )
+    db_ms = (time.perf_counter() - t0) * 1000
+    response.headers["X-DB-Time-Ms"] = f"{db_ms:.2f}"
+    return BaseListResponse(
+            count=len(result),
+            result=_rows_to_sheet_response(result),
+    )
+
+
+
+@router.post("/{form_id}/sheet/{sheet}/line")
+async def add_line(
+        form_id: int, 
+        sheet: str, 
+        body: AddLineSchema,
+        response: Response,
+        db: AsyncSession = Depends(get_db),
+        current_user: AppUser = Depends(get_current_active_user_with_set_db),
+):
+    t0 = time.perf_counter()
+    bf_service = BudgetFormService(db)
+    sheet_service = SheetService(db)
+
+    form = await bf_service.get(
+        user=current_user,
+        budget_form_id=form_id,
+        load_form_type=True,
+    )
+    if not form:
+        raise HTTPException(404, f"Форма {form_id} не найдена")
+    if sheet not in form.form_type.sheet_list:
+        raise HTTPException(404, f"Лист {sheet} формы {form_id} не найден")
+
+    result = await sheet_service.add_line(
+        form_id=form_id,
+        sheet=sheet,
+        expense_item_id=body.expense_item_id,
+        item_id=body.item_id,
+        section_code=body.section_code,
+        direction=body.direction.value if body.direction else None,
+        name=body.name,
+        internal_order=body.internal_order,
+        vsp_id=body.vsp_id,
+        project_id=body.project_id,
+        justification=body.justification,
+        contract_number=body.contract_number,
+        contract_end_date=body.contract_end_date,
+        user=current_user,
+    )
+    db_ms = (time.perf_counter() - t0) * 1000
+    response.headers["X-DB-Time-Ms"] = f"{db_ms:.2f}"
+    return BaseListResponse(
+            count=len(result),
+            result=_rows_to_sheet_response(result),
+    )
+
+
+
+@router.delete("/{form_id}/sheet/{sheet}/line/{row_id}")
+async def delete_line(
+        form_id: int, 
+        sheet: str, 
+        row_id: int,
+        response: Response,
+        direction: str | None = None,
+        db: AsyncSession = Depends(get_db),
+        current_user: AppUser = Depends(get_current_active_user_with_set_db),
+):
+    t0 = time.perf_counter()
+    sheet_service = SheetService(db)
+    bf_service = BudgetFormService(db)
+
+    form = await bf_service.get(
+        user=current_user,
+        budget_form_id=form_id,
+        load_form_type=True,
+    )
+    if not form:
+        raise HTTPException(404, f"Форма {form_id} не найдена")
+    if sheet not in form.form_type.sheet_list:
+        raise HTTPException(404, f"Лист {sheet} формы {form_id} не найден")
+    
+    result = await sheet_service.delete_line(
+        form_id=form_id,
+        sheet=sheet,
+        row_id=row_id,
+        direction=direction,
+        user=current_user,
+    )
+    db_ms = (time.perf_counter() - t0) * 1000
+    response.headers["X-DB-Time-Ms"] = f"{db_ms:.2f}"
+    return BaseListResponse(
+        count=len(result),
+        result=_rows_to_sheet_response(result),
+    )

api/src/core/exception_handlers.py

@@ -1,5 +1,8 @@
-from fastapi import FastAPI, status
+# import psycopg2
+
+from fastapi import FastAPI, HTTPException, status
 from fastapi.responses import JSONResponse
+import sqlalchemy
 
 from src.core.errors import (
     AccessDeniedException,
@@ -10,6 +13,7 @@ from src.core.errors import (
 )
 
 
+
 def register_exception_handlers(app: FastAPI) -> None:
     @app.exception_handler(AccessDeniedException)
     async def access_denied_exception_handler(request, exc: AccessDeniedException):

api/src/db/session.py

@@ -14,7 +14,7 @@ async def get_db() -> AsyncGenerator:
     db = SessionLocal()
     try:
         yield db
-        # await db.commit()
+        await db.commit()
     finally:
         await db.close()
 

api/src/domain/schemas.py

@@ -2,6 +2,7 @@ from datetime import datetime
 import enum
 from typing import Any, Generic, Literal, Optional, TypeVar
 
+
 from pydantic import BaseModel, ConfigDict, EmailStr, Field
 
 T = TypeVar("T")
@@ -211,3 +212,17 @@ class AddProjectBody(BaseModel):
     object_address: Optional[str] = None
     staff_count: Optional[int] = None
     total_area: Optional[float] = None
+
+
+class AddLineSchema(BaseModel):
+    expense_item_id: Optional[int] = None
+    item_id: Optional[str] = None
+    section_code: Optional[str] = None
+    direction: Optional[DirectionSchemaEnum] = None
+    name: Optional[str] = None
+    internal_order: Optional[str] = None
+    vsp_id: Optional[int] = None
+    project_id: Optional[int] = None
+    justification: Optional[str] = None
+    contract_number: Optional[str] = None
+    contract_end_date: Optional[datetime] = None

api/src/repository/budget_line_repository.py

@@ -0,0 +1,17 @@
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from src.db.models.budget_line import BudgetLine
+
+
+class BudgetLineRepository:
+    def __init__(self, db: AsyncSession):
+        self.db = db
+
+    async def get(self, budget_line_id: int) -> BudgetLine | None:
+        query = select(BudgetLine).where(BudgetLine.id == budget_line_id).limit(1)
+        return (await self.db.execute(query)).scalar_one_or_none()
+
+    async def get_list(self, budget_line_ids: list[int]) -> BudgetLine | None:
+        query = select(BudgetLine).where(BudgetLine.id.in_(budget_line_ids))
+        return (await self.db.execute(query)).scalars().all()

api/src/repository/sheet_repository.py

@@ -1,5 +1,6 @@
 import enum
 import itertools
+import json
 
 from sqlalchemy import text
 from sqlalchemy.ext.asyncio import AsyncSession
@@ -44,26 +45,191 @@ class SheetRepository:
         ):
             raise SheetValidationError("Некорректные данные")
 
-        sections_str = 'NULL'
+
-        if sections:
-            sections_str = ",".join([f"'{section}'" for section in sections])
-            sections_str = f"ARRAY[{sections_str}]"
         if direction:
-            func_query = "v3.v_form_view(%s, '%s', %s, '%s')" % (
+            func_query = "v3.v_form_view(:form_id, :sheet, :sections, :direction)"
-                form_id, sheet, sections_str, direction,
-            )
         else:
-            func_query = "v3.v_form_view(%s, '%s', %s)" % (
+            func_query = "v3.v_form_view(:form_id, :sheet, :sections)"
-                form_id, sheet, sections_str,
+        
-            )
+        return (
-        result = (
             await self.db.execute(
                 text(
                     f"""
                         SELECT row_type, depth, sort_order, data 
                         FROM {func_query}
                     """
-                )
+                ),
+                {
+                    "form_id": form_id,
+                    "sheet": sheet,
+                    "direction": direction,
+                    "sections": sections,
+                }
+            )
+        ).all()
+    
+    async def update_cell(
+            self, 
+            form_id: int, 
+            sheet: str,
+            direction: str,
+            sections: list[str],
+            line_id: int, 
+            column: str, 
+            value: str | int | float | None = None, 
+            user_id: int = None,
+    ) -> list[tuple]:
+        if any(
+            [
+                not isinstance(form_id, int),
+                sheet not in ACCEPTABLE_SHEETS,
+                direction not in [s.value for s in DirectionEnum],
+                sections and (set(sections) - ACCEPTABLE_SECTIONS),
+            ]
+        ):
+            raise SheetValidationError("Некорректные данные")
+        if user_id:
+            func_query = "v3.upd_form_cell(:form_id, :sheet, :line_id, :column, :value, :user_id, :direction, :sections)"
+        else:
+            func_query = "v3.upd_form_cell(:form_id, :sheet, :line_id, :column, :value, :direction, :sections)"
+
+        return (
+            await self.db.execute(
+                text(
+                    f"""
+                        SELECT row_type, depth, sort_order, data 
+                        FROM {func_query}
+                    """
+                ),
+                {
+                    "form_id": form_id, 
+                    "sheet": sheet,
+                    "direction": direction,
+                    "sections": sections,
+                    "line_id": line_id, 
+                    "column": column, 
+                    "value": json.dumps(value), 
+                    "user_id": user_id,
+                }
+            )
+        ).all()
+
+    async def update_cells(
+            self, 
+            form_id: int, 
+            sheet: str,
+            direction: str,
+            sections: list[str],
+            changes: list[dict],
+            user_id: int = None,
+    ) -> list[tuple]:
+        if any(
+            [
+                not isinstance(form_id, int),
+                sheet not in ACCEPTABLE_SHEETS,
+                direction not in [s.value for s in DirectionEnum],
+                sections and (set(sections) - ACCEPTABLE_SECTIONS),
+            ]
+        ):
+            raise SheetValidationError("Некорректные данные")
+        changes = json.dumps(changes)
+        if user_id:
+            func_query = "v3.upd_form_cells(:form_id, :sheet, :changes, :user_id, :direction, :sections)"
+        else:
+            func_query = "v3.upd_form_cells(:form_id, :sheet, :changes, :direction, :sections)"
+
+        return (
+            await self.db.execute(
+                text(
+                    f"""
+                        SELECT row_type, depth, sort_order, data 
+                        FROM {func_query}
+                    """
+                ),
+                {
+                    "form_id": form_id, 
+                    "sheet": sheet,
+                    "direction": direction,
+                    "sections": sections,
+                    "changes": changes,
+                    "user_id": user_id,
+                }
+            )
+        ).all()
+
+    async def add_line(
+            self,
+            form_id: int,
+            sheet: str,
+            expense_item_id: int | None = None,
+            item_id: str | None = None,
+            section_code: str | None = None,
+            direction: str | None = None,
+            name: str | None = None,
+            internal_order: str | None = None,
+            vsp_id: int | None = None,
+            project_id: int | None = None,
+            justification: str | None = None,
+            contract_number: str | None = None,
+            contract_end_date: str | None = None,
+            user_id: int | None = None,
+    ) -> list[tuple]:
+        return (
+            await self.db.execute(
+                text(
+                    """
+                    SELECT row_type, depth, sort_order, data
+                    FROM v3.add_budget_line(
+                      :p_form_id,
+                      :p_expense_item_id,
+                      :p_sheet,
+                      :p_item_id,
+                      :p_section_code,
+                      :p_direction,
+                      :p_name,
+                      :p_internal_order,
+                      :p_vsp_id,
+                      :p_project_id,
+                      :p_justification,
+                      :p_contract_number,
+                      :p_contract_end_date
+                    )
+                    """
+                ),
+                {
+                    "p_form_id": form_id,
+                    "p_expense_item_id": expense_item_id,
+                    "p_sheet": sheet,
+                    "p_item_id": item_id,
+                    "p_section_code": section_code,
+                    "p_direction": direction,
+                    "p_name": name,
+                    "p_internal_order": internal_order,
+                    "p_vsp_id": vsp_id,
+                    "p_project_id": project_id,
+                    "p_justification": justification,
+                    "p_contract_number": contract_number,
+                    "p_contract_end_date": contract_end_date,
+                }
+            )
+        ).all()
+
+    async def delete_line(
+            self,
+            sheet: str,
+            row_id: int,
+            direction: str | None = None,
+    ) -> list[tuple]:
+        return (
+            await self.db.execute(
+                text(
+                    "SELECT row_type, depth, sort_order, data "
+                    "FROM v3.del_budget_line(:p_row_id, :p_direction, :p_sheet)"
+                ),
+                {
+                    "p_row_id": row_id,
+                    "p_direction": direction,
+                    "p_sheet": sheet,
+                }
             )
         ).all()
-        return result

api/src/services/budget_line_service.py

@@ -0,0 +1,28 @@
+
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from src.db.models.app_user import AppUser
+from src.db.models.budget_line import BudgetLine
+from src.repository.budget_line_repository import BudgetLineRepository
+
+
+
+
+class BudgetLineService:
+    def __init__(self, db: AsyncSession):
+        self.db = db
+        self.bl_repo = BudgetLineRepository(db)
+
+    async def get(
+            self, 
+            user: AppUser, 
+            budget_line_id: int,
+    ) -> BudgetLine | None:
+        return await self.bl_repo.get(budget_line_id=budget_line_id)
+
+    async def get_list(
+            self, 
+            user: AppUser, 
+            budget_line_ids: list[int],
+    ) -> list[BudgetLine]:
+        return await self.bl_repo.get_list(budget_line_ids=budget_line_ids)

api/src/services/sheet_service.py

@@ -1,6 +1,7 @@
 
 from sqlalchemy.ext.asyncio import AsyncSession
 
+from src.db.models.app_user import AppUser
 from src.db.models.budget_form import BudgetForm
 from src.repository.sheet_repository import SheetRepository
 
@@ -25,3 +26,114 @@ class SheetService:
             direction=direction,
             sections=sections,
         )
+    
+    async def update_cell(
+            self, 
+            form_id: int, 
+            sheet: str,
+            direction: str,
+            sections: list[str],
+            line_id: int, 
+            column: str, 
+            value: str | int | float | None = None, 
+            user: AppUser | None = None,
+    ) -> list[tuple]:
+        return await self.sheet_repo.update_cell(
+            form_id=form_id, 
+            sheet=sheet,
+            direction=direction,
+            sections=sections,
+            line_id=line_id, 
+            column=column, 
+            value=value, 
+            user_id=user.id if user else None,
+        )
+
+    async def update_cells(
+            self, 
+            form_id: int, 
+            sheet: str,
+            direction: str,
+            sections: list[str],
+            changes: list[dict],
+            user: AppUser | None = None,
+    ) -> list[tuple]:
+        return await self.sheet_repo.update_cells(
+            form_id=form_id, 
+            sheet=sheet,
+            direction=direction,
+            sections=sections,
+            changes=changes,
+            user_id=user.id if user else None,
+        )
+    
+    async def update_cell(
+            self, 
+            form_id: int, 
+            sheet: str,
+            direction: str,
+            sections: list[str],
+            line_id: int, 
+            column: str, 
+            value: str | int | float | None = None, 
+            user: AppUser | None = None,
+    ) -> list[tuple]:
+        return await self.sheet_repo.update_cell(
+            form_id=form_id, 
+            sheet=sheet,
+            direction=direction,
+            sections=sections,
+            line_id=line_id, 
+            column=column, 
+            value=value, 
+            user_id=user.id if user else None,
+        )
+
+
+    async def add_line(
+            self, 
+            form_id: int,
+            sheet: str,
+            expense_item_id: int | None = None,
+            item_id: str | None = None,
+            section_code: str | None = None,
+            direction: str | None = None,
+            name: str | None = None,
+            internal_order: str | None = None,
+            vsp_id: int | None = None,
+            project_id: int | None = None,
+            justification: str | None = None,
+            contract_number: str | None = None,
+            contract_end_date: str | None = None,
+            user: AppUser | None = None,
+    ) -> list[tuple]:
+        return await self.sheet_repo.add_line(
+            form_id=form_id,
+            sheet=sheet,
+            expense_item_id=expense_item_id,
+            item_id=item_id,
+            section_code=section_code,
+            direction=direction,
+            name=name,
+            internal_order=internal_order,
+            vsp_id=vsp_id,
+            project_id=project_id,
+            justification=justification,
+            contract_number=contract_number,
+            contract_end_date=contract_end_date,
+            user_id=user.id if user else None,
+        )
+
+    async def delete_line(
+            self,
+            form_id: int,
+            sheet: str,
+            row_id: int,
+            direction: str | None = None,
+            user: AppUser | None = None,
+    ) -> list[tuple]:
+        return await self.sheet_repo.delete_line(
+            sheet=sheet,
+            row_id=row_id,
+            direction=direction,
+        )

api/tests/integration/conftest.py

@@ -1,15 +1,36 @@
+import asyncio
 import os
 
 import pytest
 from fastapi.testclient import TestClient
 
+
+from src.db.base import SessionLocal
+from src.db.session import get_db
 from src.main import app
 
 
+
+
 @pytest.fixture
 def client():
+    db = SessionLocal()
+    loop = asyncio.get_event_loop()
+
+    try:
+        async def _get_db():
+            try:
+                yield db
+                await db.flush()
+            finally:
+                await db.close()
+        app.dependency_overrides[get_db] = _get_db
         with TestClient(app) as test_client:
             yield test_client
+        app.dependency_overrides.clear()
+    finally:
+        loop.run_until_complete(db.rollback())
+        loop.run_until_complete(db.close())
 
 
 @pytest.fixture

api/tests/integration/test_forms_api_smoke.py

@@ -0,0 +1,133 @@
+import pytest
+
+
+def _auth_headers(tokens: dict) -> dict:
+    return {"Authorization": f"Bearer {tokens['access_token']}"}
+
+
+def test_forms_list_smoke(client, admin_tokens):
+    response = client.get("/api/v1/form/", headers=_auth_headers(admin_tokens))
+    assert response.status_code == 200
+    payload = response.json()
+    assert "result" in payload
+    assert "count" in payload
+    assert isinstance(payload["result"], list)
+
+
+def test_forms_sheets_not_found_error_shape(client, admin_tokens):
+    response = client.get(
+        "/api/v1/form/999/sheets", headers=_auth_headers(admin_tokens)
+    )
+    assert response.status_code == 404
+    payload = response.json()
+    assert isinstance(payload, dict)
+    assert "detail" in payload
+    assert isinstance(payload["detail"], str)
+
+
+def test_forms_sheet_get_not_found_error_shape(client, admin_tokens):
+    response = client.get(
+        "/api/v1/form/999/sheet/AHR", headers=_auth_headers(admin_tokens)
+    )
+    assert response.status_code == 404
+    payload = response.json()
+    assert isinstance(payload, dict)
+    assert "detail" in payload
+
+
+def test_forms_cell_update_not_found_error_shape(client, admin_tokens):
+    response = client.patch(
+        "/api/v1/form/999/sheet/AHR/cell",
+        json={"line_id": 1, "column": "q1.m1", "value": 100},
+        headers=_auth_headers(admin_tokens),
+    )
+    assert response.status_code == 404
+    payload = response.json()
+    assert isinstance(payload, dict)
+    assert "detail" in payload
+
+
+def test_forms_cell_update(client, admin_tokens):
+    response = client.patch(
+        "/api/v1/form/1/sheet/AHR/cell?direction=Support&sections=plan,contract_summary",
+        json={
+            "line_id": 1,
+            "column": "plan.q1",
+            "value": 225
+        },
+        headers=_auth_headers(admin_tokens),
+    )
+    assert response.status_code == 200
+    payload = response.json()
+    assert isinstance(payload, dict)
+    assert "result" in payload
+
+
+def test_forms_cells_update_not_found_error_shape(client, admin_tokens):
+    response = client.patch(
+        "/api/v1/form/999/sheet/AHR/cells",
+        json={"changes": [{"line_id": 1, "column": "q1.m1", "value": 100}]},
+        headers=_auth_headers(admin_tokens),
+    )
+    assert response.status_code == 404
+    payload = response.json()
+    assert isinstance(payload, dict)
+    assert "detail" in payload
+
+
+def test_forms_cells_update(client, admin_tokens):
+    response = client.patch(
+        "/api/v1/form/1/sheet/AHR/cells?direction=Support&sections=plan,contract_summary",
+        json={
+            "changes": [
+                {
+                    "line_id": 1,
+                    "column": "plan.q1",
+                    "value": 226
+                }
+            ],
+        },
+        headers=_auth_headers(admin_tokens),
+    )
+    assert response.status_code == 200
+    payload = response.json()
+    assert isinstance(payload, dict)
+    assert "result" in payload
+
+
+def test_forms_add_line_not_found_error_shape(client, admin_tokens):
+    response = client.post(
+        "/api/v1/form/999/sheet/AHR/line",
+        json={},
+        headers=_auth_headers(admin_tokens),
+    )
+    assert response.status_code == 404
+    payload = response.json()
+    assert isinstance(payload, dict)
+    assert "detail" in payload
+
+
+def test_forms_add_line(client, admin_tokens):
+    response = client.post(
+        "/api/v1/form/1/sheet/AHR/line?direction=Support&sections=plan,contract_summary",
+        json={
+            "expense_item_id": 120,
+            "direction": "Support"
+        },
+        headers=_auth_headers(admin_tokens),
+    )
+    assert response.status_code == 200
+    payload = response.json()
+    assert isinstance(payload, dict)
+    assert "result" in payload
+
+
+def test_forms_delete_line_not_found_error_shape(client, admin_tokens):
+    response = client.delete(
+        "/api/v1/form/999/sheet/AHR/line/1",
+        headers=_auth_headers(admin_tokens),
+    )
+    assert response.status_code == 404
+    payload = response.json()
+    assert isinstance(payload, dict)
+    assert "detail" in payload