diff --git a/backend/app/api/v1/endpoints/executions.py b/backend/app/api/v1/endpoints/executions.py index 6dd0f112..ccd45c91 100644 --- a/backend/app/api/v1/endpoints/executions.py +++ b/backend/app/api/v1/endpoints/executions.py @@ -71,12 +71,14 @@ async def list_executions( current_user: CurrentUser, page: int = Query(default=1, ge=1), page_size: int = Query(default=20, ge=1, le=100), + api_definition_id: UUID | None = None, ) -> Page[ExecutionResponse]: executions, total = await ExecutionService(session).list_executions( actor=current_user, project_id=project_id, page=page, page_size=page_size, + api_definition_id=api_definition_id, ) return Page( items=[ExecutionResponse.model_validate(execution) for execution in executions], diff --git a/backend/app/api/v1/endpoints/performance.py b/backend/app/api/v1/endpoints/performance.py index 6d45e722..2c22139c 100644 --- a/backend/app/api/v1/endpoints/performance.py +++ b/backend/app/api/v1/endpoints/performance.py @@ -60,6 +60,24 @@ async def create_performance_scenario( return PerformanceScenarioResponse.model_validate(scenario) +@router.get( + "/projects/{project_id}/performance-scenarios/{scenario_id}", + response_model=PerformanceScenarioResponse, +) +async def get_performance_scenario( + project_id: UUID, + scenario_id: UUID, + session: SessionDependency, + current_user: CurrentUser, +) -> PerformanceScenarioResponse: + scenario = await PerformanceScenarioService(session).get( + actor=current_user, + project_id=project_id, + scenario_id=scenario_id, + ) + return PerformanceScenarioResponse.model_validate(scenario) + + @router.post( "/projects/{project_id}/performance-scenarios/{scenario_id}/versions", response_model=PerformanceScenarioResponse, diff --git a/backend/app/api/v1/endpoints/test_assets.py b/backend/app/api/v1/endpoints/test_assets.py index c750c730..d6da25fa 100644 --- a/backend/app/api/v1/endpoints/test_assets.py +++ b/backend/app/api/v1/endpoints/test_assets.py @@ -1,22 +1,39 @@ from typing import Annotated from uuid import UUID -from fastapi import APIRouter, Header, Query, status +from fastapi import APIRouter, Header, Query, Request, Response, status from app.api.dependencies import CurrentUser, SessionDependency, WorkflowCoordinator from app.domain.test_assets import VersionChange +from app.http.test_asset_packages import package_request_contract, read_package_payload from app.schemas.common import Page +from app.schemas.tasking import TestPlanRunDetailResponse +from app.schemas.test_asset_packages import ( + PackageApplyRequest, + PackageApplyResponse, + PackageExportRequest, + PackagePreviewRequest, + PackagePreviewResponse, +) from app.schemas.test_assets import ( + AssetBulkDeleteRequest, AssetBulkMove, AssetBulkMoveResponse, AssetClone, + AssetDeleteTarget, + AssetDeletionPreviewRequest, + AssetDeletionPreviewResponse, + AssetDeletionResponse, + AssetDirectoryCountsResponse, TestCaseCreate, TestCaseResponse, + TestCaseRunHistoryResponse, TestCaseRunRequest, TestCaseRunResponse, TestCaseUpdate, TestCaseVersionResponse, TestSuiteCreate, + TestSuiteLatestRunResponse, TestSuiteResponse, TestSuiteUpdate, TestSuiteVersionResponse, @@ -24,12 +41,232 @@ VersionDiffResponse, VersionPublish, ) +from app.services.test_asset_deletion import TestAssetDeletionService +from app.services.test_asset_history import TestAssetHistoryService +from app.services.test_asset_package_imports import TestAssetPackageImportService +from app.services.test_asset_packages import TestAssetPackageExportService from app.services.test_assets import TestCaseService, TestSuiteService from app.services.test_case_runs import TestCaseRunService router = APIRouter(prefix="/projects/{project_id}") +@router.post("/test-assets/export") +async def export_test_asset_package( + project_id: UUID, + payload: PackageExportRequest, + session: SessionDependency, + current_user: CurrentUser, +) -> Response: + package = await TestAssetPackageExportService(session).export( + actor=current_user, project_id=project_id, selection=payload + ) + return Response( + content=package.model_dump_json(indent=2), + media_type="application/json", + headers={ + "Content-Disposition": f'attachment; filename="flowtest-test-assets-{project_id}.json"' + }, + ) + + +@router.post( + "/test-assets/import/preview", + response_model=PackagePreviewResponse, + openapi_extra=package_request_contract(PackagePreviewRequest), +) +async def preview_test_asset_package( + project_id: UUID, + request: Request, + session: SessionDependency, + current_user: CurrentUser, +) -> PackagePreviewResponse: + payload = await read_package_payload(request, PackagePreviewRequest) + return await TestAssetPackageImportService(session).preview( + actor=current_user, project_id=project_id, request=payload + ) + + +@router.post( + "/test-assets/import/apply", + response_model=PackageApplyResponse, + openapi_extra=package_request_contract(PackageApplyRequest), +) +async def apply_test_asset_package( + project_id: UUID, + request: Request, + session: SessionDependency, + current_user: CurrentUser, +) -> PackageApplyResponse: + payload = await read_package_payload(request, PackageApplyRequest) + return await TestAssetPackageImportService(session).apply( + actor=current_user, project_id=project_id, request=payload + ) + + +@router.get("/test-assets/directory-counts", response_model=AssetDirectoryCountsResponse) +async def test_asset_directory_counts( + project_id: UUID, + session: SessionDependency, + current_user: CurrentUser, + search: str | None = Query(default=None, max_length=200), + tag: str | None = Query(default=None, max_length=50), +) -> AssetDirectoryCountsResponse: + return await TestCaseService(session).directory_counts( + actor=current_user, project_id=project_id, search=search, tag=tag + ) + + +@router.post("/test-cases/deletion-preview", response_model=AssetDeletionPreviewResponse) +async def preview_test_case_deletion( + project_id: UUID, + payload: AssetDeletionPreviewRequest, + session: SessionDependency, + current_user: CurrentUser, +) -> AssetDeletionPreviewResponse: + return await TestAssetDeletionService(session).preview( + actor=current_user, + project_id=project_id, + kind="case", + asset_ids=payload.asset_ids, + ) + + +@router.post("/test-cases/bulk-delete", response_model=AssetDeletionResponse) +async def bulk_delete_test_cases( + project_id: UUID, + payload: AssetBulkDeleteRequest, + session: SessionDependency, + current_user: CurrentUser, +) -> AssetDeletionResponse: + return await TestAssetDeletionService(session).delete( + actor=current_user, + project_id=project_id, + kind="case", + targets=payload.assets, + ) + + +@router.delete("/test-cases/{case_id}", response_model=AssetDeletionResponse) +async def delete_test_case( + project_id: UUID, + case_id: UUID, + payload: AssetDeleteTarget, + session: SessionDependency, + current_user: CurrentUser, +) -> AssetDeletionResponse: + return await TestAssetDeletionService(session).delete_one( + actor=current_user, + project_id=project_id, + kind="case", + asset_id=case_id, + target=payload, + ) + + +@router.post("/test-suites/deletion-preview", response_model=AssetDeletionPreviewResponse) +async def preview_test_suite_deletion( + project_id: UUID, + payload: AssetDeletionPreviewRequest, + session: SessionDependency, + current_user: CurrentUser, +) -> AssetDeletionPreviewResponse: + return await TestAssetDeletionService(session).preview( + actor=current_user, + project_id=project_id, + kind="suite", + asset_ids=payload.asset_ids, + ) + + +@router.post("/test-suites/bulk-delete", response_model=AssetDeletionResponse) +async def bulk_delete_test_suites( + project_id: UUID, + payload: AssetBulkDeleteRequest, + session: SessionDependency, + current_user: CurrentUser, +) -> AssetDeletionResponse: + return await TestAssetDeletionService(session).delete( + actor=current_user, + project_id=project_id, + kind="suite", + targets=payload.assets, + ) + + +@router.delete("/test-suites/{suite_id}", response_model=AssetDeletionResponse) +async def delete_test_suite( + project_id: UUID, + suite_id: UUID, + payload: AssetDeleteTarget, + session: SessionDependency, + current_user: CurrentUser, +) -> AssetDeletionResponse: + return await TestAssetDeletionService(session).delete_one( + actor=current_user, + project_id=project_id, + kind="suite", + asset_id=suite_id, + target=payload, + ) + + +@router.get("/test-cases/{case_id}/runs", response_model=Page[TestCaseRunHistoryResponse]) +async def test_case_run_history( + project_id: UUID, + case_id: UUID, + session: SessionDependency, + current_user: CurrentUser, + version: int | None = Query(default=None, ge=1), + page: int = Query(default=1, ge=1), + page_size: int = Query(default=20, ge=1, le=100), +) -> Page[TestCaseRunHistoryResponse]: + items, total = await TestAssetHistoryService(session).case_runs( + actor=current_user, + project_id=project_id, + case_id=case_id, + version=version, + page=page, + page_size=page_size, + ) + return Page(items=items, total=total, page=page, page_size=page_size) + + +@router.get("/test-suites/runs/latest", response_model=list[TestSuiteLatestRunResponse]) +async def latest_test_suite_runs( + project_id: UUID, + session: SessionDependency, + current_user: CurrentUser, + suite_ids: Annotated[list[UUID] | None, Query(max_length=100)] = None, +) -> list[TestSuiteLatestRunResponse]: + return await TestAssetHistoryService(session).latest_suite_runs( + actor=current_user, + project_id=project_id, + suite_ids=suite_ids or [], + ) + + +@router.get("/test-suites/{suite_id}/runs", response_model=Page[TestPlanRunDetailResponse]) +async def test_suite_run_history( + project_id: UUID, + suite_id: UUID, + session: SessionDependency, + current_user: CurrentUser, + version: int | None = Query(default=None, ge=1), + page: int = Query(default=1, ge=1), + page_size: int = Query(default=20, ge=1, le=100), +) -> Page[TestPlanRunDetailResponse]: + items, total = await TestAssetHistoryService(session).suite_runs( + actor=current_user, + project_id=project_id, + suite_id=suite_id, + version=version, + page=page, + page_size=page_size, + ) + return Page(items=items, total=total, page=page, page_size=page_size) + + @router.get("/test-cases/runs/latest", response_model=list[TestCaseRunResponse]) async def latest_test_case_runs( project_id: UUID, @@ -85,6 +322,8 @@ async def list_test_cases( is_template: bool | None = None, page: int = Query(default=1, ge=1), page_size: int = Query(default=20, ge=1, le=100), + folder_id: UUID | None = None, + unfiled: bool = False, ) -> Page[TestCaseResponse]: items, total = await TestCaseService(session).list_cases( actor=current_user, @@ -94,6 +333,8 @@ async def list_test_cases( is_template=is_template, page=page, page_size=page_size, + folder_id=folder_id, + unfiled=unfiled, ) return Page( items=[TestCaseResponse.model_validate(item) for item in items], @@ -257,6 +498,8 @@ async def list_test_suites( tag: str | None = Query(default=None, max_length=50), page: int = Query(default=1, ge=1), page_size: int = Query(default=20, ge=1, le=100), + folder_id: UUID | None = None, + unfiled: bool = False, ) -> Page[TestSuiteResponse]: items, total = await TestSuiteService(session).list_suites( actor=current_user, @@ -265,6 +508,8 @@ async def list_test_suites( tag=tag, page=page, page_size=page_size, + folder_id=folder_id, + unfiled=unfiled, ) return Page( items=[TestSuiteResponse.model_validate(item) for item in items], diff --git a/backend/app/api/v1/endpoints/workflows.py b/backend/app/api/v1/endpoints/workflows.py index f0988be7..1fdcbcbe 100644 --- a/backend/app/api/v1/endpoints/workflows.py +++ b/backend/app/api/v1/endpoints/workflows.py @@ -59,12 +59,14 @@ async def list_workflows( current_user: CurrentUser, page: int = Query(default=1, ge=1), page_size: int = Query(default=20, ge=1, le=100), + search: str | None = Query(default=None, max_length=200), ) -> Page[WorkflowResponse]: items, total = await WorkflowService(session).list_workflows( actor=current_user, project_id=project_id, page=page, page_size=page_size, + search=search, ) return Page( items=[WorkflowResponse.model_validate(item) for item in items], diff --git a/backend/app/domain/test_asset_packages.py b/backend/app/domain/test_asset_packages.py new file mode 100644 index 00000000..a6b4d7c1 --- /dev/null +++ b/backend/app/domain/test_asset_packages.py @@ -0,0 +1,101 @@ +from typing import TYPE_CHECKING + +from app.domain.test_assets import definition_fingerprint + +if TYPE_CHECKING: + from app.schemas.test_asset_packages import ( + PackageCase, + PackageCaseVersion, + PackageSuite, + PackageSuiteVersion, + TestAssetPackage, + ) + + +def validate_package_structure(package: "TestAssetPackage") -> None: + groups = ( + package.folders, + package.environments, + package.workflows, + package.cases, + package.suites, + ) + for group in groups: + ids = [item.id for item in group] + if len(ids) != len(set(ids)): + raise ValueError("原生包中存在重复对象 ID") + if not package.cases and not package.suites: + raise ValueError("原生包未包含测试资产") + for workflow in package.workflows: + numbers = [version.version for version in workflow.versions] + if len(numbers) != len(set(numbers)): + raise ValueError("原生包存在重复流程版本引用") + _validate_folders(package) + for case in package.cases: + _validate_case(package, case) + for suite in package.suites: + _validate_suite(package, suite) + + +def _validate_folders(package: "TestAssetPackage") -> None: + parents = {folder.id: folder.parent_id for folder in package.folders} + for folder in package.folders: + seen = {folder.id} + parent = folder.parent_id + while parent is not None: + if parent not in parents or parent in seen: + raise ValueError("原生包目录存在缺失父目录或循环") + seen.add(parent) + parent = parents[parent] + assets: list[PackageCase | PackageSuite] = [*package.cases, *package.suites] + for asset in assets: + if asset.folder_id is not None and asset.folder_id not in parents: + raise ValueError("原生包测试资产引用了缺失目录") + + +def _validate_case(package: "TestAssetPackage", case: "PackageCase") -> None: + _validate_versions(case.versions) + workflows = {workflow.id: workflow for workflow in package.workflows} + environments = {environment.id for environment in package.environments} + for definition in [case.draft_definition, *(version.definition for version in case.versions)]: + workflow = workflows.get(definition.workflow_id) + if workflow is None or definition.environment_id not in environments: + raise ValueError("原生包用例缺失流程或环境引用") + if definition.workflow_version is not None and definition.workflow_version not in { + version.version for version in workflow.versions + }: + raise ValueError("原生包用例缺失固定流程版本引用") + + +def _validate_suite(package: "TestAssetPackage", suite: "PackageSuite") -> None: + _validate_versions(suite.versions) + cases = {case.id: case for case in package.cases} + for definition in [suite.draft_definition, *(version.definition for version in suite.versions)]: + ids = [item.test_case_id for item in definition.items] + if len(ids) != len(set(ids)): + raise ValueError("原生包套件存在重复成员") + for item in definition.items: + case = cases.get(item.test_case_id) + if case is None or ( + item.test_case_version is not None + and item.test_case_version not in {version.version for version in case.versions} + ): + raise ValueError("原生包套件缺失用例或固定用例版本") + if any( + item.test_case_version is None + for version in suite.versions + for item in version.definition.items + ): + raise ValueError("原生包已发布套件成员必须固定用例版本") + + +def _validate_versions(versions: "list[PackageCaseVersion] | list[PackageSuiteVersion]") -> None: + numbers = [version.version for version in versions] + if numbers != list(range(1, len(numbers) + 1)): + raise ValueError("原生包版本必须从 1 开始完整递增, 不能省略或重复") + for version in versions: + if ( + definition_fingerprint(version.definition.model_dump(mode="json")) + != version.fingerprint + ): + raise ValueError("原生包版本内容与指纹不一致") diff --git a/backend/app/http/test_asset_packages.py b/backend/app/http/test_asset_packages.py new file mode 100644 index 00000000..0a3f3adc --- /dev/null +++ b/backend/app/http/test_asset_packages.py @@ -0,0 +1,57 @@ +from typing import Any + +from fastapi import FastAPI, Request +from pydantic import BaseModel, JsonValue, ValidationError +from pydantic.json_schema import models_json_schema + +from app.core.errors import AppError +from app.schemas.test_asset_packages import PackageApplyRequest, PackagePreviewRequest +from app.services.test_asset_packages import PACKAGE_MAX_BYTES + + +def package_request_contract(schema: type[BaseModel]) -> dict[str, JsonValue]: + return { + "requestBody": { + "required": True, + "content": { + "application/json": {"schema": {"$ref": f"#/components/schemas/{schema.__name__}"}} + }, + } + } + + +def install_package_openapi(application: FastAPI) -> None: + original_openapi = application.openapi + _, definitions = models_json_schema( + [(PackagePreviewRequest, "validation"), (PackageApplyRequest, "validation")], + ref_template="#/components/schemas/{model}", + ) + + def package_openapi() -> dict[str, Any]: + document = original_openapi() + document.setdefault("components", {}).setdefault("schemas", {}).update(definitions["$defs"]) + return document + + application.openapi = package_openapi # type: ignore[method-assign] + + +async def read_package_payload[PackagePayload: BaseModel]( + request: Request, schema: type[PackagePayload] +) -> PackagePayload: + content = bytearray() + async for chunk in request.stream(): + if len(content) + len(chunk) > PACKAGE_MAX_BYTES: + raise AppError( + code="TEST_ASSET_PACKAGE_TOO_LARGE", + message="导入请求超过 10 MiB, 请缩小原生包", + status_code=413, + ) + content.extend(chunk) + try: + return schema.model_validate_json(content) + except ValidationError as error: + raise AppError( + code="TEST_ASSET_PACKAGE_INVALID", + message="原生包格式不合法, 请检查格式版本、完整引用、内容指纹和运行参数", + status_code=422, + ) from error diff --git a/backend/app/main.py b/backend/app/main.py index 8cebe4e5..00009f27 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -22,6 +22,7 @@ from app.core.storage import ensure_storage_bucket, object_storage from app.domain.runtime_profiles import RuntimeProfile from app.engine.events import ExecutionEventBus +from app.http.test_asset_packages import install_package_openapi from app.middleware.metrics import MetricsMiddleware from app.middleware.rate_limit import RateLimitMiddleware from app.middleware.trace import TraceIdMiddleware @@ -116,6 +117,7 @@ async def reset_request_redaction_policy( application.add_middleware(TraceIdMiddleware) register_exception_handlers(application) application.include_router(api_router, prefix=settings.api_v1_prefix) + install_package_openapi(application) _mount_frontend(application) return application diff --git a/backend/app/models/test_assets.py b/backend/app/models/test_assets.py index dc199b72..898834ed 100644 --- a/backend/app/models/test_assets.py +++ b/backend/app/models/test_assets.py @@ -1,3 +1,4 @@ +from datetime import datetime from typing import Any from uuid import UUID @@ -5,7 +6,9 @@ JSON, Boolean, CheckConstraint, + DateTime, ForeignKey, + Index, Integer, String, Text, @@ -21,6 +24,7 @@ class TestCase(UuidPrimaryKeyMixin, TimestampMixin, Base): __tablename__ = "test_cases" __table_args__ = ( UniqueConstraint("project_id", "name", name="uq_test_cases_project_name"), + Index("ix_test_cases_project_folder_active", "project_id", "archived_at", "folder_id"), CheckConstraint( "current_version IS NULL OR current_version >= 1", name="test_case_current_version", @@ -41,6 +45,7 @@ class TestCase(UuidPrimaryKeyMixin, TimestampMixin, Base): ) draft_definition: Mapped[dict[str, Any]] = mapped_column(JSON) current_version: Mapped[int | None] = mapped_column(Integer) + archived_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), index=True) created_by_id: Mapped[UUID] = mapped_column(ForeignKey("users.id", ondelete="RESTRICT")) @property @@ -69,6 +74,7 @@ class TestSuite(UuidPrimaryKeyMixin, TimestampMixin, Base): __tablename__ = "test_suites" __table_args__ = ( UniqueConstraint("project_id", "name", name="uq_test_suites_project_name"), + Index("ix_test_suites_project_folder_active", "project_id", "archived_at", "folder_id"), CheckConstraint( "current_version IS NULL OR current_version >= 1", name="test_suite_current_version", @@ -86,8 +92,13 @@ class TestSuite(UuidPrimaryKeyMixin, TimestampMixin, Base): tags: Mapped[list[str]] = mapped_column(JSON, default=list, server_default="[]") draft_definition: Mapped[dict[str, Any]] = mapped_column(JSON) current_version: Mapped[int | None] = mapped_column(Integer) + archived_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), index=True) created_by_id: Mapped[UUID] = mapped_column(ForeignKey("users.id", ondelete="RESTRICT")) + @property + def draft_fingerprint(self) -> str: + return definition_fingerprint(self.draft_definition) + class TestSuiteVersion(UuidPrimaryKeyMixin, TimestampMixin, Base): __tablename__ = "test_suite_versions" diff --git a/backend/app/repositories/executions.py b/backend/app/repositories/executions.py index 755759fd..dec9c63c 100644 --- a/backend/app/repositories/executions.py +++ b/backend/app/repositories/executions.py @@ -17,23 +17,29 @@ async def get(self, execution_id: UUID) -> APICallExecution | None: return await self._session.get(APICallExecution, execution_id) async def list_for_project( - self, *, project_id: UUID, offset: int, limit: int + self, + *, + project_id: UUID, + offset: int, + limit: int, + api_definition_id: UUID | None = None, ) -> tuple[list[APICallExecution], int]: + conditions = [APICallExecution.project_id == project_id] + if api_definition_id is not None: + conditions.append(APICallExecution.api_definition_id == api_definition_id) executions = list( ( await self._session.scalars( select(APICallExecution) - .where(APICallExecution.project_id == project_id) - .order_by(APICallExecution.started_at.desc()) + .where(*conditions) + .order_by(APICallExecution.started_at.desc(), APICallExecution.id.desc()) .offset(offset) .limit(limit) ) ).all() ) total = await self._session.scalar( - select(func.count()) - .select_from(APICallExecution) - .where(APICallExecution.project_id == project_id) + select(func.count()).select_from(APICallExecution).where(*conditions) ) return executions, int(total or 0) diff --git a/backend/app/repositories/reporting.py b/backend/app/repositories/reporting.py index 067c7e44..64edaed6 100644 --- a/backend/app/repositories/reporting.py +++ b/backend/app/repositories/reporting.py @@ -5,7 +5,12 @@ from sqlalchemy.ext.asyncio import AsyncSession from app.models.reporting import NotificationDelivery, NotificationWebhook -from app.models.workflows import Workflow, WorkflowExecution, WorkflowNodeExecution +from app.models.workflows import ( + Workflow, + WorkflowControlRecord, + WorkflowExecution, + WorkflowNodeExecution, +) class ReportingRepository: @@ -77,17 +82,37 @@ async def list_nodes(self, execution_id: UUID) -> list[WorkflowNodeExecution]: ).all() ) - async def list_children(self, execution_id: UUID) -> list[WorkflowExecution]: + async def list_children( + self, execution_id: UUID, *, project_id: UUID + ) -> list[WorkflowExecution]: return list( ( await self._session.scalars( select(WorkflowExecution) - .where(WorkflowExecution.parent_execution_id == execution_id) + .where( + WorkflowExecution.parent_execution_id == execution_id, + WorkflowExecution.project_id == project_id, + ) .order_by(WorkflowExecution.dataset_row_index) ) ).all() ) + async def list_control_records(self, execution_id: UUID) -> list[WorkflowControlRecord]: + return list( + ( + await self._session.scalars( + select(WorkflowControlRecord) + .where(WorkflowControlRecord.workflow_execution_id == execution_id) + .order_by( + WorkflowControlRecord.node_id, + WorkflowControlRecord.kind, + WorkflowControlRecord.ordinal, + ) + ) + ).all() + ) + async def get_webhook(self, webhook_id: UUID) -> NotificationWebhook | None: return await self._session.get(NotificationWebhook, webhook_id) diff --git a/backend/app/repositories/test_asset_deletion.py b/backend/app/repositories/test_asset_deletion.py new file mode 100644 index 00000000..3c99a20e --- /dev/null +++ b/backend/app/repositories/test_asset_deletion.py @@ -0,0 +1,174 @@ +from uuid import UUID + +from sqlalchemy import or_, select +from sqlalchemy.ext.asyncio import AsyncSession + +from app.models.tasking import TestPlan, TestPlanItem, TestPlanRun, TestPlanRunItem +from app.models.test_assets import TestCase, TestSuite, TestSuiteVersion, TestSuiteVersionItem +from app.models.workflows import WorkflowExecution +from app.schemas.test_assets import AssetReference, TestAssetKind, TestSuiteDefinitionInput + + +class TestAssetDeletionRepository: + def __init__(self, session: AsyncSession) -> None: + self._session = session + + async def selected_for_update( + self, *, project_id: UUID, kind: TestAssetKind, asset_ids: list[UUID] + ) -> list[TestCase | TestSuite]: + if kind == "case": + return list( + ( + await self._session.scalars( + select(TestCase) + .where(TestCase.project_id == project_id, TestCase.id.in_(asset_ids)) + .order_by(TestCase.id) + .with_for_update() + .execution_options(populate_existing=True) + ) + ).all() + ) + return list( + ( + await self._session.scalars( + select(TestSuite) + .where(TestSuite.project_id == project_id, TestSuite.id.in_(asset_ids)) + .order_by(TestSuite.id) + .with_for_update() + .execution_options(populate_existing=True) + ) + ).all() + ) + + async def references( + self, *, project_id: UUID, kind: TestAssetKind, asset_id: UUID + ) -> list[AssetReference]: + plan_rows = await self._session.execute( + select(TestPlan.id, TestPlan.name, TestPlanItem.target_version) + .join(TestPlanItem, TestPlanItem.test_plan_id == TestPlan.id) + .where( + TestPlan.project_id == project_id, + TestPlanItem.target_type == kind, + TestPlanItem.target_id == asset_id, + ) + .distinct() + ) + refs = [ + AssetReference(kind="test_plan", id=plan_id, name=name, version=version) + for plan_id, name, version in plan_rows + ] + refs.extend(await self._active_runs(project_id, kind, asset_id)) + if kind == "case": + refs.extend(await self._suite_plan_references(project_id, asset_id)) + refs.extend(await self._suite_references(project_id, asset_id)) + return refs + + async def _suite_plan_references(self, project_id: UUID, case_id: UUID) -> list[AssetReference]: + rows = await self._session.execute( + select(TestPlan.id, TestPlan.name, TestPlanItem.target_version) + .join(TestPlanItem, TestPlanItem.test_plan_id == TestPlan.id) + .join( + TestSuiteVersion, + (TestSuiteVersion.test_suite_id == TestPlanItem.target_id) + & (TestSuiteVersion.version == TestPlanItem.target_version), + ) + .join( + TestSuiteVersionItem, + TestSuiteVersionItem.test_suite_version_id == TestSuiteVersion.id, + ) + .where( + TestPlan.project_id == project_id, + TestPlanItem.target_type == "suite", + TestSuiteVersionItem.test_case_id == case_id, + ) + .distinct() + ) + return [ + AssetReference(kind="test_plan", id=plan_id, name=name, version=version) + for plan_id, name, version in rows + ] + + async def _active_runs( + self, project_id: UUID, kind: TestAssetKind, asset_id: UUID + ) -> list[AssetReference]: + item_filter = ( + TestPlanRunItem.target_id == asset_id + if kind == "case" + else TestPlanRunItem.target_snapshot["source_suite"]["id"].as_string() == str(asset_id) + ) + plan_rows = ( + await self._session.scalars( + select(TestPlanRun.id) + .join(TestPlanRunItem, TestPlanRunItem.test_plan_run_id == TestPlanRun.id) + .where( + TestPlanRun.project_id == project_id, + TestPlanRun.status.in_(("queued", "running")), + TestPlanRunItem.target_type == "case", + item_filter, + ) + .distinct() + ) + ).all() + refs = [ + AssetReference(kind="execution", id=run_id, name="活动计划运行") for run_id in plan_rows + ] + if kind == "suite": + return refs + legacy = ( + select(TestPlanRunItem.workflow_execution_id) + .join(TestPlanRun, TestPlanRun.id == TestPlanRunItem.test_plan_run_id) + .where( + TestPlanRun.project_id == project_id, + TestPlanRunItem.target_type == "case", + TestPlanRunItem.target_id == asset_id, + ) + ) + executions = ( + await self._session.scalars( + select(WorkflowExecution.id).where( + WorkflowExecution.project_id == project_id, + WorkflowExecution.status.in_(("queued", "running")), + or_( + WorkflowExecution.source_case_id == asset_id, + WorkflowExecution.id.in_(legacy), + ), + ) + ) + ).all() + refs.extend( + AssetReference(kind="execution", id=execution_id, name="活动用例运行") + for execution_id in executions + ) + return refs + + async def _suite_references(self, project_id: UUID, case_id: UUID) -> list[AssetReference]: + published = await self._session.execute( + select(TestSuite.id, TestSuite.name, TestSuiteVersion.version) + .join(TestSuiteVersion, TestSuiteVersion.test_suite_id == TestSuite.id) + .join( + TestSuiteVersionItem, + TestSuiteVersionItem.test_suite_version_id == TestSuiteVersion.id, + ) + .where( + TestSuite.project_id == project_id, + TestSuite.archived_at.is_(None), + TestSuiteVersionItem.test_case_id == case_id, + ) + .distinct() + ) + refs = [ + AssetReference(kind="test_suite", id=suite_id, name=name, version=version) + for suite_id, name, version in published + ] + suites = ( + await self._session.scalars( + select(TestSuite).where( + TestSuite.project_id == project_id, TestSuite.archived_at.is_(None) + ) + ) + ).all() + for suite in suites: + draft = TestSuiteDefinitionInput.model_validate(suite.draft_definition) + if any(item.test_case_id == case_id for item in draft.items): + refs.append(AssetReference(kind="test_suite", id=suite.id, name=suite.name)) + return refs diff --git a/backend/app/repositories/test_asset_history.py b/backend/app/repositories/test_asset_history.py new file mode 100644 index 00000000..559d211a --- /dev/null +++ b/backend/app/repositories/test_asset_history.py @@ -0,0 +1,243 @@ +"""Query asset-owned execution history without a project-wide recent-run window.""" + +from datetime import datetime +from uuid import UUID + +from sqlalchemy import Select, and_, func, literal, or_, select, union_all +from sqlalchemy.ext.asyncio import AsyncSession +from sqlalchemy.sql.elements import ColumnElement + +from app.models.tasking import TestPlanRun, TestPlanRunItem +from app.models.workflows import WorkflowExecution, WorkflowVersion +from app.schemas.test_assets import TestCaseRunHistoryResponse + + +class TestAssetHistoryRepository: + def __init__(self, session: AsyncSession) -> None: + self._session = session + + async def list_case_runs( + self, *, project_id: UUID, case_id: UUID, version: int | None, offset: int, limit: int + ) -> tuple[list[TestCaseRunHistoryResponse], int]: + history = union_all( + self._case_executions(project_id, case_id), + self._legacy_and_pending_case_runs(project_id, case_id), + ).subquery() + filters = [] if version is None else [history.c.case_version == version] + query = select(history).where(*filters) + total = await self._session.scalar(select(func.count()).select_from(query.subquery())) + rows = await self._session.execute( + query.order_by( + func.coalesce(history.c.started_at, history.c.created_at).desc(), + history.c.id.desc(), + ) + .offset(offset) + .limit(limit) + ) + return [TestCaseRunHistoryResponse.model_validate(row) for row in rows.mappings()], int( + total or 0 + ) + + @staticmethod + def _case_executions( + project_id: UUID, case_id: UUID + ) -> Select[ + tuple[ + UUID, + UUID, + int | None, + UUID | None, + int, + UUID, + str, + str | None, + datetime, + datetime, + UUID, + UUID, + ] + ]: + legacy_item = ( + select(TestPlanRunItem.id) + .join(TestPlanRun, TestPlanRun.id == TestPlanRunItem.test_plan_run_id) + .where( + TestPlanRun.project_id == project_id, + TestPlanRunItem.target_type == "case", + TestPlanRunItem.target_id == case_id, + TestPlanRunItem.workflow_execution_id == WorkflowExecution.id, + ) + .order_by(TestPlanRunItem.id) + .limit(1) + .correlate(WorkflowExecution) + .scalar_subquery() + ) + return ( + select( + WorkflowExecution.id.label("id"), + WorkflowExecution.id.label("execution_id"), + WorkflowExecution.source_case_version.label("case_version"), + WorkflowExecution.workflow_id.label("workflow_id"), + WorkflowVersion.version.label("workflow_version"), + WorkflowExecution.environment_id.label("environment_id"), + WorkflowExecution.status.label("status"), + WorkflowExecution.source_trigger.label("source"), + WorkflowExecution.started_at.label("started_at"), + WorkflowExecution.created_at.label("created_at"), + TestPlanRun.id.label("plan_run_id"), + TestPlanRun.test_plan_id.label("plan_id"), + ) + .join(WorkflowVersion, WorkflowVersion.id == WorkflowExecution.workflow_version_id) + .outerjoin( + TestPlanRunItem, + TestPlanRunItem.id + == func.coalesce(WorkflowExecution.source_plan_run_item_id, legacy_item), + ) + .outerjoin( + TestPlanRun, + and_( + TestPlanRun.id == TestPlanRunItem.test_plan_run_id, + TestPlanRun.project_id == project_id, + ), + ) + .where( + WorkflowExecution.project_id == project_id, + WorkflowExecution.source_case_id == case_id, + WorkflowExecution.parent_execution_id.is_(None), + ) + ) + + @staticmethod + def _legacy_and_pending_case_runs( + project_id: UUID, case_id: UUID + ) -> Select[tuple[UUID, UUID, int, UUID, int, UUID, str, str, datetime, datetime, UUID, UUID]]: + return ( + select( + func.coalesce(WorkflowExecution.id, TestPlanRunItem.id).label("id"), + WorkflowExecution.id.label("execution_id"), + TestPlanRunItem.target_version.label("case_version"), + TestPlanRunItem.workflow_id.label("workflow_id"), + TestPlanRunItem.workflow_version.label("workflow_version"), + TestPlanRunItem.environment_id.label("environment_id"), + func.coalesce(WorkflowExecution.status, TestPlanRunItem.status).label("status"), + literal("plan").label("source"), + WorkflowExecution.started_at.label("started_at"), + TestPlanRun.created_at.label("created_at"), + TestPlanRun.id.label("plan_run_id"), + TestPlanRun.test_plan_id.label("plan_id"), + ) + .select_from(TestPlanRunItem) + .join(TestPlanRun, TestPlanRun.id == TestPlanRunItem.test_plan_run_id) + .outerjoin( + WorkflowExecution, WorkflowExecution.id == TestPlanRunItem.workflow_execution_id + ) + .where( + TestPlanRun.project_id == project_id, + TestPlanRunItem.target_type == "case", + TestPlanRunItem.target_id == case_id, + or_( + WorkflowExecution.id.is_(None), + and_( + WorkflowExecution.project_id == project_id, + WorkflowExecution.source_case_id.is_(None), + WorkflowExecution.parent_execution_id.is_(None), + ), + ), + ) + ) + + async def list_suite_runs( + self, *, project_id: UUID, suite_id: UUID, version: int | None, offset: int, limit: int + ) -> tuple[list[tuple[TestPlanRun, list[TestPlanRunItem]]], int]: + filters = self._suite_filters(project_id, suite_id, version) + matches = ( + select(TestPlanRunItem.test_plan_run_id) + .join(TestPlanRun, TestPlanRun.id == TestPlanRunItem.test_plan_run_id) + .where(*filters) + .distinct() + ) + total = await self._session.scalar(select(func.count()).select_from(matches.subquery())) + runs = list( + ( + await self._session.scalars( + select(TestPlanRun) + .where(TestPlanRun.id.in_(matches)) + .order_by(TestPlanRun.created_at.desc(), TestPlanRun.id.desc()) + .offset(offset) + .limit(limit) + ) + ).all() + ) + if not runs: + return [], int(total or 0) + items = ( + await self._session.scalars( + select(TestPlanRunItem) + .join(TestPlanRun, TestPlanRun.id == TestPlanRunItem.test_plan_run_id) + .where(*filters, TestPlanRun.id.in_([run.id for run in runs])) + .order_by(TestPlanRunItem.position, TestPlanRunItem.id) + ) + ).all() + grouped: dict[UUID, list[TestPlanRunItem]] = {run.id: [] for run in runs} + for item in items: + grouped[item.test_plan_run_id].append(item) + return [(run, grouped[run.id]) for run in runs], int(total or 0) + + @staticmethod + def _suite_filters( + project_id: UUID, suite_id: UUID, version: int | None + ) -> list[ColumnElement[bool]]: + filters = [ + TestPlanRun.project_id == project_id, + TestPlanRunItem.target_type == "case", + TestPlanRunItem.target_snapshot["source_suite"]["id"].as_string() == str(suite_id), + ] + if version is not None: + filters.append( + TestPlanRunItem.target_snapshot["source_suite"]["version"].as_integer() == version + ) + return filters + + async def latest_suite_runs( + self, *, project_id: UUID, suite_ids: list[UUID] + ) -> list[tuple[UUID, TestPlanRun, list[TestPlanRunItem]]]: + suite = TestPlanRunItem.target_snapshot["source_suite"]["id"].as_string() + ranked = ( + select( + suite.label("suite_id"), + TestPlanRun.id.label("run_id"), + func.row_number() + .over( + partition_by=suite, + order_by=(TestPlanRun.created_at.desc(), TestPlanRun.id.desc()), + ) + .label("rank"), + ) + .join(TestPlanRun, TestPlanRun.id == TestPlanRunItem.test_plan_run_id) + .where( + TestPlanRun.project_id == project_id, + TestPlanRunItem.target_type == "case", + suite.in_([str(item) for item in suite_ids]), + ) + .subquery() + ) + rows = await self._session.execute( + select(ranked.c.suite_id, TestPlanRun, TestPlanRunItem) + .select_from(ranked) + .join(TestPlanRun, TestPlanRun.id == ranked.c.run_id) + .join(TestPlanRunItem, TestPlanRunItem.test_plan_run_id == TestPlanRun.id) + .where( + ranked.c.rank == 1, + suite == ranked.c.suite_id, + TestPlanRunItem.target_type == "case", + ) + .order_by( + TestPlanRun.created_at.desc(), TestPlanRun.id.desc(), TestPlanRunItem.position + ) + ) + grouped: dict[UUID, tuple[TestPlanRun, list[TestPlanRunItem]]] = {} + for suite_id, run, item in rows: + asset_id = UUID(suite_id) + if asset_id not in grouped: + grouped[asset_id] = (run, []) + grouped[asset_id][1].append(item) + return [(asset_id, run, items) for asset_id, (run, items) in grouped.items()] diff --git a/backend/app/repositories/test_asset_packages.py b/backend/app/repositories/test_asset_packages.py new file mode 100644 index 00000000..7bb8a9e3 --- /dev/null +++ b/backend/app/repositories/test_asset_packages.py @@ -0,0 +1,102 @@ +from typing import cast +from uuid import UUID + +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession + +from app.models.access import Folder +from app.models.api_assets import Environment +from app.models.test_assets import TestCase, TestCaseVersion, TestSuite, TestSuiteVersion +from app.models.workflows import Workflow, WorkflowVersion +from app.schemas.test_assets import TestAssetKind + + +class TestAssetPackageRepository: + def __init__(self, session: AsyncSession) -> None: + self._session = session + + async def asset( + self, project_id: UUID, kind: TestAssetKind, asset_id: UUID + ) -> TestCase | TestSuite | None: + model = TestCase if kind == "case" else TestSuite + result = await self._session.scalar( + select(model) + .where(model.project_id == project_id, model.id == asset_id) + .with_for_update() + .execution_options(populate_existing=True) + ) + return cast(TestCase | TestSuite | None, result) + + async def named( + self, project_id: UUID, kind: TestAssetKind, name: str + ) -> TestCase | TestSuite | None: + model = TestCase if kind == "case" else TestSuite + result = await self._session.scalar( + select(model) + .where(model.project_id == project_id, model.name == name) + .with_for_update() + .execution_options(populate_existing=True) + ) + return cast(TestCase | TestSuite | None, result) + + async def case_versions(self, case_id: UUID) -> list[TestCaseVersion]: + return list( + ( + await self._session.scalars( + select(TestCaseVersion) + .where(TestCaseVersion.test_case_id == case_id) + .order_by(TestCaseVersion.version) + ) + ).all() + ) + + async def suite_versions(self, suite_id: UUID) -> list[TestSuiteVersion]: + return list( + ( + await self._session.scalars( + select(TestSuiteVersion) + .where(TestSuiteVersion.test_suite_id == suite_id) + .order_by(TestSuiteVersion.version) + ) + ).all() + ) + + async def workflow(self, project_id: UUID, workflow_id: UUID) -> Workflow | None: + result = await self._session.scalar( + select(Workflow) + .where( + Workflow.project_id == project_id, + Workflow.id == workflow_id, + Workflow.archived_at.is_(None), + ) + .with_for_update() + ) + return result + + async def workflow_version(self, workflow_id: UUID, version: int) -> WorkflowVersion | None: + result = await self._session.scalar( + select(WorkflowVersion).where( + WorkflowVersion.workflow_id == workflow_id, WorkflowVersion.version == version + ) + ) + return result + + async def environment(self, project_id: UUID, environment_id: UUID) -> Environment | None: + result = await self._session.scalar( + select(Environment) + .where( + Environment.project_id == project_id, + Environment.id == environment_id, + Environment.archived_at.is_(None), + ) + .with_for_update() + ) + return result + + async def folder(self, project_id: UUID, folder_id: UUID) -> Folder | None: + result = await self._session.scalar( + select(Folder) + .where(Folder.project_id == project_id, Folder.id == folder_id) + .with_for_update() + ) + return result diff --git a/backend/app/repositories/test_assets.py b/backend/app/repositories/test_assets.py index 177eafe3..faf02db4 100644 --- a/backend/app/repositories/test_assets.py +++ b/backend/app/repositories/test_assets.py @@ -1,8 +1,11 @@ from collections.abc import Sequence from uuid import UUID -from sqlalchemy import func, or_, select +from sqlalchemy import case, cast, func, literal, or_, select +from sqlalchemy.dialects.postgresql import JSONB from sqlalchemy.ext.asyncio import AsyncSession +from sqlalchemy.orm import InstrumentedAttribute +from sqlalchemy.sql.elements import ColumnElement from app.models.test_assets import ( TestCase, @@ -11,6 +14,7 @@ TestSuiteVersion, TestSuiteVersionItem, ) +from app.schemas.test_assets import AssetDirectoryCountsResponse, AssetFolderCount TestAssetEntity = TestCase | TestCaseVersion | TestSuite | TestSuiteVersion | TestSuiteVersionItem @@ -65,23 +69,19 @@ async def list_cases( is_template: bool | None, offset: int, limit: int, + folder_id: UUID | None = None, + unfiled: bool = False, ) -> tuple[list[TestCase], int]: - filters = [TestCase.project_id == project_id] - if search: - pattern = f"%{search}%" - filters.append(or_(TestCase.name.ilike(pattern), TestCase.description.ilike(pattern))) + filters = self._catalog_filters(TestCase, project_id, search, tag, folder_id, unfiled) if is_template is not None: filters.append(TestCase.is_template == is_template) query = select(TestCase).where(*filters) - if tag: - matches = [ - item for item in (await self._session.scalars(query)).all() if tag in item.tags - ] - return matches[offset : offset + limit], len(matches) items = list( ( await self._session.scalars( - query.order_by(TestCase.updated_at.desc()).offset(offset).limit(limit) + query.order_by(TestCase.updated_at.desc(), TestCase.id.desc()) + .offset(offset) + .limit(limit) ) ).all() ) @@ -149,21 +149,17 @@ async def list_suites( tag: str | None, offset: int, limit: int, + folder_id: UUID | None = None, + unfiled: bool = False, ) -> tuple[list[TestSuite], int]: - filters = [TestSuite.project_id == project_id] - if search: - pattern = f"%{search}%" - filters.append(or_(TestSuite.name.ilike(pattern), TestSuite.description.ilike(pattern))) + filters = self._catalog_filters(TestSuite, project_id, search, tag, folder_id, unfiled) query = select(TestSuite).where(*filters) - if tag: - matches = [ - item for item in (await self._session.scalars(query)).all() if tag in item.tags - ] - return matches[offset : offset + limit], len(matches) items = list( ( await self._session.scalars( - query.order_by(TestSuite.updated_at.desc()).offset(offset).limit(limit) + query.order_by(TestSuite.updated_at.desc(), TestSuite.id.desc()) + .offset(offset) + .limit(limit) ) ).all() ) @@ -172,6 +168,78 @@ async def list_suites( ) return items, int(total or 0) + def _catalog_filters( + self, + model: type[TestCase] | type[TestSuite], + project_id: UUID, + search: str | None, + tag: str | None, + folder_id: UUID | None, + unfiled: bool, + ) -> list[ColumnElement[bool]]: + filters = [model.project_id == project_id, model.archived_at.is_(None)] + if search: + pattern = f"%{search}%" + filters.append(or_(model.name.ilike(pattern), model.description.ilike(pattern))) + if tag: + filters.append(self._tag_filter(model.tags, tag)) + if folder_id is not None: + filters.append(model.folder_id == folder_id) + if unfiled: + filters.append(model.folder_id.is_(None)) + return filters + + def _tag_filter( + self, column: InstrumentedAttribute[list[str]], tag: str + ) -> ColumnElement[bool]: + if self._session.get_bind().dialect.name == "sqlite": + values = func.json_each(column).table_valued("value") + return select(literal(1)).select_from(values).where(values.c.value == tag).exists() + return cast(column, JSONB).contains([tag]) + + async def directory_counts( + self, *, project_id: UUID, search: str | None, tag: str | None + ) -> AssetDirectoryCountsResponse: + case_rows = await self._session.execute( + select( + TestCase.folder_id, + func.count(), + func.sum(case((TestCase.current_version.is_not(None), 1), else_=0)), + ) + .where(*self._catalog_filters(TestCase, project_id, search, tag, None, False)) + .group_by(TestCase.folder_id) + ) + case_groups = case_rows.all() + cases = {folder_id: int(count) for folder_id, count, _published in case_groups} + # Aggregates retain one row per folder, rather than materializing assets. + published = sum(int(value or 0) for _folder, _count, value in case_groups) + suite_rows = await self._session.execute( + select(TestSuite.folder_id, func.count()) + .where( + *self._catalog_filters(TestSuite, project_id, search, tag, None, False), + ) + .group_by(TestSuite.folder_id) + ) + suites = {folder_id: int(count) for folder_id, count in suite_rows} + folder_ids = sorted( + {folder_id for folder_id in (*cases, *suites) if folder_id is not None}, key=str + ) + return AssetDirectoryCountsResponse( + case_total=sum(cases.values()), + suite_total=sum(suites.values()), + published_case_total=int(published or 0), + unfiled_cases=cases.get(None, 0), + unfiled_suites=suites.get(None, 0), + folders=[ + AssetFolderCount( + folder_id=folder_id, + cases=cases.get(folder_id, 0), + suites=suites.get(folder_id, 0), + ) + for folder_id in folder_ids + ], + ) + async def suite_name_exists( self, *, project_id: UUID, name: str, excluding_id: UUID | None = None ) -> bool: diff --git a/backend/app/repositories/workflows.py b/backend/app/repositories/workflows.py index 8b7c14d8..3d8f1131 100644 --- a/backend/app/repositories/workflows.py +++ b/backend/app/repositories/workflows.py @@ -4,7 +4,7 @@ from typing import Any from uuid import UUID -from sqlalchemy import case, delete, func, select, update +from sqlalchemy import case, delete, func, or_, select, update from sqlalchemy.ext.asyncio import AsyncSession from app.domain.control_reports import project_control_report @@ -128,26 +128,25 @@ async def list_versions(self, workflow_id: UUID) -> list[WorkflowVersion]: ) async def list_workflows( - self, *, project_id: UUID, offset: int, limit: int + self, *, project_id: UUID, offset: int, limit: int, search: str | None = None ) -> tuple[list[Workflow], int]: + filters = [Workflow.project_id == project_id, Workflow.archived_at.is_(None)] + if search: + pattern = f"%{search}%" + filters.append(or_(Workflow.name.ilike(pattern), Workflow.description.ilike(pattern))) items = list( ( await self._session.scalars( select(Workflow) - .where( - Workflow.project_id == project_id, - Workflow.archived_at.is_(None), - ) - .order_by(Workflow.updated_at.desc()) + .where(*filters) + .order_by(Workflow.updated_at.desc(), Workflow.id.desc()) .offset(offset) .limit(limit) ) ).all() ) total = await self._session.scalar( - select(func.count()) - .select_from(Workflow) - .where(Workflow.project_id == project_id, Workflow.archived_at.is_(None)) + select(func.count()).select_from(Workflow).where(*filters) ) return items, int(total or 0) diff --git a/backend/app/schemas/reporting.py b/backend/app/schemas/reporting.py index 13af552b..3af60e39 100644 --- a/backend/app/schemas/reporting.py +++ b/backend/app/schemas/reporting.py @@ -5,7 +5,7 @@ from pydantic import BaseModel, ConfigDict, Field, HttpUrl, JsonValue from app.domain.reporting import FailureCategory, NotificationEvent -from app.engine.results import NodeObservation +from app.engine.results import NodeObservation, NodeResult WebhookName = Annotated[str, Field(min_length=1, max_length=160)] @@ -40,6 +40,8 @@ class ReportNodeResponse(BaseModel): extraction: JsonValue = None assertion: JsonValue = None input_mappings: JsonValue = None + output: JsonValue = None + result: NodeResult | None = None error_code: str | None error_message: str | None diff --git a/backend/app/schemas/test_asset_packages.py b/backend/app/schemas/test_asset_packages.py new file mode 100644 index 00000000..e323c21e --- /dev/null +++ b/backend/app/schemas/test_asset_packages.py @@ -0,0 +1,200 @@ +from datetime import datetime +from typing import Annotated, Literal, Self +from uuid import UUID + +from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator + +from app.schemas.test_assets import ( + AssetName, + PublishedTestCaseDefinition, + TagName, + TestAssetKind, + TestCaseDefinitionInput, + TestSuiteDefinitionInput, +) + +Fingerprint = Annotated[str, Field(pattern=r"^[0-9a-f]{64}$")] +PositiveVersion = Annotated[int, Field(ge=1, strict=True)] +PackageAction = Literal["create", "clone", "update", "skip"] + + +class StrictPackageModel(BaseModel): + model_config = ConfigDict(extra="forbid") + + +class PackageFolder(StrictPackageModel): + id: UUID + name: Annotated[str, Field(min_length=1, max_length=160)] + parent_id: UUID | None + + +class PackageEnvironment(StrictPackageModel): + id: UUID + name: Annotated[str, Field(min_length=1, max_length=160)] + + +class PackageWorkflowVersion(StrictPackageModel): + version: PositiveVersion + fingerprint: Fingerprint + + +class PackageWorkflow(StrictPackageModel): + id: UUID + name: AssetName + versions: list[PackageWorkflowVersion] = Field(max_length=1000) + + +class PackageCaseVersion(StrictPackageModel): + version: PositiveVersion + definition: PublishedTestCaseDefinition + fingerprint: Fingerprint + change_note: str = Field(max_length=1000) + + +class PackageSuiteVersion(StrictPackageModel): + version: PositiveVersion + definition: TestSuiteDefinitionInput + fingerprint: Fingerprint + change_note: str = Field(max_length=1000) + + +class PackageCase(StrictPackageModel): + id: UUID + name: AssetName + description: str = Field(max_length=4000) + folder_id: UUID | None + tags: list[TagName] = Field(max_length=20) + is_template: bool + draft_definition: TestCaseDefinitionInput + versions: list[PackageCaseVersion] = Field(max_length=1000) + + +class PackageSuite(StrictPackageModel): + id: UUID + name: AssetName + description: str = Field(max_length=4000) + folder_id: UUID | None + tags: list[TagName] = Field(max_length=20) + draft_definition: TestSuiteDefinitionInput + versions: list[PackageSuiteVersion] = Field(max_length=1000) + + +class TestAssetPackage(StrictPackageModel): + format: Literal["flowtest-test-assets"] + format_version: Literal[1] + source_project_id: UUID + folders: list[PackageFolder] = Field(max_length=1000) + environments: list[PackageEnvironment] = Field(max_length=1000) + workflows: list[PackageWorkflow] = Field(max_length=1000) + cases: list[PackageCase] = Field(max_length=500) + suites: list[PackageSuite] = Field(max_length=100) + + @field_validator("format_version", mode="before") + @classmethod + def validate_format_version(cls, value: object) -> object: + if type(value) is not int: + raise ValueError("格式版本必须为整数") + return value + + @model_validator(mode="after") + def validate_structure(self) -> Self: + from app.domain.test_asset_packages import validate_package_structure + + validate_package_structure(self) + return self + + +class PackageExportRequest(StrictPackageModel): + case_ids: list[UUID] = Field(default_factory=list, max_length=100) + suite_ids: list[UUID] = Field(default_factory=list, max_length=100) + + @model_validator(mode="after") + def validate_selection(self) -> Self: + if not self.case_ids and not self.suite_ids: + raise ValueError("请明确选择需要导出的用例或套件") + if len(self.case_ids) != len(set(self.case_ids)) or len(self.suite_ids) != len( + set(self.suite_ids) + ): + raise ValueError("导出选择不能重复") + return self + + +class PackageAssetChoice(StrictPackageModel): + kind: TestAssetKind + source_id: UUID + action: PackageAction + name: AssetName | None = None + target_id: UUID | None = None + + +class PackageWorkflowBinding(StrictPackageModel): + target_id: UUID + versions: dict[PositiveVersion, PositiveVersion] = Field(default_factory=dict) + + +class PackageBindings(StrictPackageModel): + workflows: dict[UUID, PackageWorkflowBinding] = Field(default_factory=dict, max_length=1000) + environments: dict[UUID, UUID] = Field(default_factory=dict, max_length=1000) + folders: dict[UUID, UUID | None] = Field(default_factory=dict, max_length=1000) + + +class PackagePreviewRequest(StrictPackageModel): + package: TestAssetPackage + choices: list[PackageAssetChoice] = Field(default_factory=list, max_length=600) + bindings: PackageBindings = Field(default_factory=PackageBindings) + + +class PackageApplyRequest(PackagePreviewRequest): + expected_preview_fingerprint: Fingerprint + + +class PackageDependencyPreview(StrictPackageModel): + kind: Literal["workflow", "environment", "folder"] + source_id: UUID + source_name: str + target_id: UUID | None + target_name: str | None + source_version: int | None = None + target_version: int | None = None + fingerprint_changed: bool = False + problems: list[str] = Field(default_factory=list) + + +class PackageVersionMapping(StrictPackageModel): + source_version: int + target_version: int + creates_version: bool + + +class PackageAssetPreview(StrictPackageModel): + kind: TestAssetKind + source_id: UUID + source_name: str + action: PackageAction + target_id: UUID | None + target_name: str + existing_version: int | None + existing_updated_at: datetime | None + existing_draft_fingerprint: str | None + existing_content_fingerprint: str | None + versions: list[PackageVersionMapping] + problems: list[str] = Field(default_factory=list) + + +class PackagePreviewResponse(StrictPackageModel): + fingerprint: Fingerprint + can_apply: bool + assets: list[PackageAssetPreview] + dependencies: list[PackageDependencyPreview] + + +class PackageImportedAsset(StrictPackageModel): + kind: TestAssetKind + source_id: UUID + target_id: UUID + action: PackageAction + versions: list[PackageVersionMapping] + + +class PackageApplyResponse(StrictPackageModel): + assets: list[PackageImportedAsset] diff --git a/backend/app/schemas/test_assets.py b/backend/app/schemas/test_assets.py index c589d6fb..11505e7e 100644 --- a/backend/app/schemas/test_assets.py +++ b/backend/app/schemas/test_assets.py @@ -6,10 +6,12 @@ from pydantic import BaseModel, ConfigDict, Field, StringConstraints, field_validator from app.domain.api_assets import JsonValue +from app.schemas.tasking import TestPlanRunDetailResponse RuntimeName = Annotated[str, Field(pattern=r"^[A-Za-z_][A-Za-z0-9_.-]*$", max_length=160)] TagName = Annotated[str, Field(min_length=1, max_length=50)] AssetName = Annotated[str, StringConstraints(strip_whitespace=True, min_length=1, max_length=200)] +TestAssetKind = Literal["case", "suite"] def validate_runtime_headers(headers: dict[str, str]) -> dict[str, str]: @@ -82,6 +84,7 @@ class TestCaseResponse(BaseModel): draft_definition: StoredTestCaseDefinition draft_fingerprint: str current_version: int | None + archived_at: datetime | None created_by_id: UUID created_at: datetime updated_at: datetime @@ -120,6 +123,26 @@ class TestCaseRunResponse(BaseModel): created_new_version: bool = False +class TestCaseRunHistoryResponse(BaseModel): + id: UUID + execution_id: UUID | None + case_version: int + workflow_id: UUID + workflow_version: int + environment_id: UUID + status: Literal["queued", "running", "passed", "failed", "cancelled", "quarantined"] + source: Literal["direct", "plan"] + started_at: datetime | None + created_at: datetime + plan_run_id: UUID | None + plan_id: UUID | None + + +class TestSuiteLatestRunResponse(BaseModel): + suite_id: UUID + detail: TestPlanRunDetailResponse + + class TestCaseVersionResponse(BaseModel): model_config = ConfigDict(from_attributes=True) @@ -181,7 +204,9 @@ class TestSuiteResponse(BaseModel): description: str tags: list[str] draft_definition: TestSuiteDefinitionInput + draft_fingerprint: str current_version: int | None + archived_at: datetime | None created_by_id: UUID created_at: datetime updated_at: datetime @@ -213,6 +238,77 @@ class AssetBulkMoveResponse(BaseModel): updated: int +class AssetFolderCount(BaseModel): + folder_id: UUID + cases: int + suites: int + + +class AssetDirectoryCountsResponse(BaseModel): + case_total: int + suite_total: int + published_case_total: int + unfiled_cases: int + unfiled_suites: int + folders: list[AssetFolderCount] + + +class AssetDeletionPreviewRequest(BaseModel): + model_config = ConfigDict(extra="forbid") + asset_ids: list[UUID] = Field(min_length=1, max_length=100) + + @field_validator("asset_ids") + @classmethod + def unique_ids(cls, values: list[UUID]) -> list[UUID]: + if len(values) != len(set(values)): + raise ValueError("删除选择不能包含重复资产") + return values + + +class AssetDeleteTarget(BaseModel): + model_config = ConfigDict(extra="forbid") + id: UUID + expected_name: AssetName + expected_draft_fingerprint: str = Field(pattern=r"^[0-9a-f]{64}$") + expected_version: int | None = Field(ge=1) + expected_updated_at: datetime + + +class AssetReference(BaseModel): + kind: Literal["test_plan", "test_suite", "execution"] + id: UUID + name: str | None = None + version: int | None = None + + +class AssetDeletionPreviewTarget(BaseModel): + asset: AssetDeleteTarget + archived: bool + references: list[AssetReference] + + +class AssetDeletionPreviewResponse(BaseModel): + targets: list[AssetDeletionPreviewTarget] + + +class AssetBulkDeleteRequest(BaseModel): + model_config = ConfigDict(extra="forbid") + assets: list[AssetDeleteTarget] = Field(min_length=1, max_length=100) + + @field_validator("assets") + @classmethod + def unique_assets(cls, values: list[AssetDeleteTarget]) -> list[AssetDeleteTarget]: + ids = [item.id for item in values] + if len(ids) != len(set(ids)): + raise ValueError("删除选择不能包含重复资产") + return values + + +class AssetDeletionResponse(BaseModel): + archived_ids: list[UUID] + historical_data_retained: Literal[True] = True + + class VersionChangeResponse(BaseModel): path: str before: JsonValue diff --git a/backend/app/services/executions.py b/backend/app/services/executions.py index 7289251e..21012949 100644 --- a/backend/app/services/executions.py +++ b/backend/app/services/executions.py @@ -25,6 +25,7 @@ from app.domain.execution import ExecutionStatus from app.models.access import User from app.models.executions import APICallExecution, AssertionResult +from app.repositories.api_assets import APIAssetRepository from app.repositories.executions import ExecutionRepository from app.schemas.api_assets import MultipartBody from app.services.api_assets import APIAssetService, PreparedRequest @@ -68,6 +69,7 @@ def __init__( ) -> None: self._session = session self._repository = ExecutionRepository(session) + self._api_repository = APIAssetRepository(session) self._assets = APIAssetService(session) self._artifacts = ArtifactService(session) self._projects = ProjectService(session) @@ -227,13 +229,24 @@ async def execute( await client.aclose() async def list_executions( - self, *, actor: User, project_id: UUID, page: int, page_size: int + self, + *, + actor: User, + project_id: UUID, + page: int, + page_size: int, + api_definition_id: UUID | None = None, ) -> tuple[list[APICallExecution], int]: await self._projects.authorize(actor=actor, project_id=project_id, editing=False) + if api_definition_id is not None: + definition = await self._api_repository.get_definition(api_definition_id) + if definition is None or definition.project_id != project_id: + raise AppError(code="API_NOT_FOUND", message="接口不存在", status_code=404) return await self._repository.list_for_project( project_id=project_id, offset=(page - 1) * page_size, limit=page_size, + api_definition_id=api_definition_id, ) async def get_execution( diff --git a/backend/app/services/reporting.py b/backend/app/services/reporting.py index 57e155cc..1962ddd4 100644 --- a/backend/app/services/reporting.py +++ b/backend/app/services/reporting.py @@ -1,12 +1,12 @@ import html import json from collections import Counter, defaultdict -from dataclasses import dataclass +from dataclasses import asdict, dataclass from datetime import UTC, date, datetime, timedelta from typing import cast from uuid import UUID -from pydantic import JsonValue +from pydantic import BaseModel, JsonValue from sqlalchemy.ext.asyncio import AsyncSession from app.core.errors import AppError @@ -14,8 +14,10 @@ from app.engine.results import NodeObservation, NodeResult from app.models.access import User from app.models.artifacts import Artifact -from app.models.workflows import WorkflowExecution, WorkflowNodeExecution +from app.models.workflows import Workflow, WorkflowExecution, WorkflowNodeExecution +from app.repositories.durable_execution import DurableExecutionRepository from app.repositories.reporting import ReportingRepository +from app.schemas.durable_execution import ExecutionCheckpointResponse from app.services.artifacts import ArtifactService from app.services.audit import AuditService from app.services.projects import ProjectService @@ -55,6 +57,8 @@ class NodeReport: input_mappings: JsonValue error_code: str | None error_message: str | None + output: JsonValue = None + result: NodeResult | None = None @dataclass(frozen=True, slots=True) @@ -65,6 +69,24 @@ class ExecutionReportDetail: dataset_children: list[ExecutionReportSummary] +@dataclass(frozen=True, slots=True) +class ExportControlRecord: + node_id: str + kind: str + ordinal: int + status: str + test_verdict: str + payload: JsonValue + + +@dataclass(frozen=True, slots=True) +class ExecutionExportDetail(ExecutionReportDetail): + snapshot: dict[str, JsonValue] + control_records: list[ExportControlRecord] + checkpoints: list[ExecutionCheckpointResponse] + dataset_evidence: list["ExecutionExportDetail"] + + @dataclass(frozen=True, slots=True) class TrendPoint: date: date @@ -86,6 +108,7 @@ class ReportService: def __init__(self, session: AsyncSession) -> None: self._session = session self._reports = ReportingRepository(session) + self._durable = DurableExecutionRepository(session) self._projects = ProjectService(session) self._artifacts = ArtifactService(session) self._audit = AuditService(session) @@ -114,7 +137,7 @@ async def get_execution( await self._projects.authorize(actor=actor, project_id=project_id, editing=False) execution = await self._execution(project_id, execution_id) nodes = await self._reports.list_nodes(execution.id) - children = await self._reports.list_children(execution.id) + children = await self._reports.list_children(execution.id, project_id=project_id) return ExecutionReportDetail( summary=await self._summary(execution, nodes), nodes=[self._node(execution, item) for item in nodes], @@ -143,7 +166,7 @@ async def trend(self, *, actor: User, project_id: UUID, days: int) -> ReportTren ) async def export_html(self, *, actor: User, project_id: UUID, execution_id: UUID) -> Artifact: - detail = await self.get_execution( + detail = await self._export_detail( actor=actor, project_id=project_id, execution_id=execution_id, @@ -167,6 +190,51 @@ async def export_html(self, *, actor: User, project_id: UUID, execution_id: UUID await self._session.refresh(artifact) return artifact + async def _export_detail( + self, *, actor: User, project_id: UUID, execution_id: UUID + ) -> ExecutionExportDetail: + await self._projects.authorize(actor=actor, project_id=project_id, editing=False) + return await self._export_execution(project_id, execution_id, set(), 0) + + async def _export_execution( + self, project_id: UUID, execution_id: UUID, visited: set[UUID], depth: int + ) -> ExecutionExportDetail: + if execution_id in visited or len(visited) >= 1000 or depth > 16: + raise AppError( + code="REPORT_EXPORT_LIMIT_EXCEEDED", + message="报告执行树循环或超过导出边界; 请分别导出子运行", + status_code=422, + ) + visited.add(execution_id) + execution = await self._execution(project_id, execution_id) + nodes = await self._reports.list_nodes(execution_id) + children = await self._reports.list_children(execution_id, project_id=project_id) + records = await self._reports.list_control_records(execution_id) + checkpoints = await self._durable.list_checkpoints(execution_id) + return ExecutionExportDetail( + summary=await self._summary(execution, nodes), + nodes=[self._node(execution, item) for item in nodes], + context=cast(dict[str, JsonValue], execution.context), + dataset_children=[await self._summary(child) for child in children], + snapshot=cast(dict[str, JsonValue], execution.snapshot), + control_records=[ + ExportControlRecord( + node_id=item.node_id, + kind=item.kind, + ordinal=item.ordinal, + status=item.status, + test_verdict=item.test_verdict, + payload=cast(JsonValue, item.payload), + ) + for item in records + ], + checkpoints=[ExecutionCheckpointResponse.model_validate(item) for item in checkpoints], + dataset_evidence=[ + await self._export_execution(project_id, child.id, visited, depth + 1) + for child in children + ], + ) + async def _execution(self, project_id: UUID, execution_id: UUID) -> WorkflowExecution: execution = await self._reports.get_execution(execution_id) if execution is None or execution.project_id != project_id: @@ -191,13 +259,7 @@ async def _summary( return ExecutionReportSummary( id=execution.id, workflow_id=execution.workflow_id, - workflow_name=( - workflow.name - if workflow is not None - else "Sandbox Preview" - if execution.run_purpose == "preview" - else "已删除工作流" - ), + workflow_name=_report_workflow_name(execution, workflow), workflow_version=_workflow_version(execution.snapshot), status=execution.status, failure_category=classify_failure( @@ -215,8 +277,11 @@ async def _summary( @staticmethod def _node(execution: WorkflowExecution, node: WorkflowNodeExecution) -> NodeReport: - output = node.output if isinstance(node.output, dict) else {} - observations = _node_observations(node.result) + stored_output = node.output_summary if node.output_summary is not None else node.output + output = stored_output if isinstance(stored_output, dict) else {} + stored_result = node.result_summary if node.result_summary is not None else node.result + result = _node_result(stored_result) + observations = list(result.observations) if result is not None else [] latest = observations[-1] if observations else None request = ( cast(JsonValue, latest.request.model_dump(mode="json")) @@ -240,10 +305,12 @@ def _node(execution: WorkflowExecution, node: WorkflowNodeExecution) -> NodeRepo request=request, response=response, extraction=cast(JsonValue, output) if node.node_type == "extract" else None, - assertion=cast(JsonValue, output) if node.node_type == "assert" else None, + assertion=_node_assertion(node.node_type, output, result), input_mappings=cast(JsonValue, output.get("input_mappings")), error_code=node.error_code, error_message=node.error_message, + output=cast(JsonValue, stored_output), + result=result, ) @staticmethod @@ -276,13 +343,24 @@ def _prepared_request(snapshot: dict[str, object], node_id: str) -> JsonValue: return cast(JsonValue, api.get("prepared_request")) -def _node_observations(result: dict[str, object] | None) -> list[NodeObservation]: +def _node_result(result: dict[str, object] | None) -> NodeResult | None: if result is None: - return [] + return None try: - return list(NodeResult.model_validate(result).observations) + return NodeResult.model_validate(result) except ValueError: - return [] + # Legacy results can be incomplete; retain their separate stored output. + return None + + +def _node_assertion( + node_type: str, output: dict[str, object], result: NodeResult | None +) -> JsonValue: + if node_type == "assert": + return cast(JsonValue, output) + if result is not None and result.assertions: + return [cast(JsonValue, item.model_dump(mode="json")) for item in result.assertions] + return None def _response(node_type: str, output: dict[str, object]) -> JsonValue: @@ -333,24 +411,13 @@ def _render_html(detail: ExecutionReportDetail) -> str: ) payload = html.escape( json.dumps( - { - "context": detail.context, - "nodes": [ - { - "name": node.name, - "request": node.request, - "response": node.response, - "extraction": node.extraction, - "assertion": node.assertion, - } - for node in detail.nodes - ], - }, + asdict(detail), ensure_ascii=False, indent=2, - default=str, + default=_export_value, ) ) + duration = "未提供" if summary.duration_ms is None else f"{summary.duration_ms} ms" return ( '' "FlowTest 测试报告