726 lines
24 KiB
Python
726 lines
24 KiB
Python
from __future__ import annotations
|
||
|
||
import hashlib
|
||
import hmac
|
||
import os
|
||
import json
|
||
from decimal import Decimal
|
||
|
||
import pytest
|
||
import pytest_asyncio
|
||
from sqlalchemy import JSON, select
|
||
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
|
||
|
||
os.environ["DATABASE_URL"] = "postgresql+asyncpg://tester:tester@localhost:5432/pmflow_test"
|
||
|
||
from app.models.ai_project_control import ( # noqa: E402
|
||
ComplianceCheckRun,
|
||
DevSession,
|
||
DevWorkspaceBinding,
|
||
EvidencePack,
|
||
FeedbackIngest,
|
||
GitHubAppInstallation,
|
||
GitHubAppRepository,
|
||
GitHubWebhookDelivery,
|
||
IterationRecommendation,
|
||
ModelCatalogItem,
|
||
ModelEvalRun,
|
||
ModelRecommendationSnapshot,
|
||
ModelUsageEvent,
|
||
ProgressEvent,
|
||
ProjectModelBinding,
|
||
PromptPackage,
|
||
ReleaseValidationRun,
|
||
RevenueEvent,
|
||
RepoBinding,
|
||
ROISnapshot,
|
||
TaskCodeLink,
|
||
)
|
||
from app.models.base import Base # noqa: E402
|
||
from app.models.feishu_item import FeishuItem # noqa: E402
|
||
from app.models.model_capability_matrix import Budget # noqa: E402
|
||
from app.models.model_requirement_binding import ModelRequirementBinding # noqa: E402
|
||
from app.models.model_version import ModelVersion # noqa: E402
|
||
from app.models.project import Project # noqa: E402
|
||
from app.models.signal import Signal # noqa: E402
|
||
from app.models.task import Task # noqa: E402
|
||
from app.models.truth_node import TruthNode # noqa: E402
|
||
from app.models.user import User # noqa: E402
|
||
from app.schemas import TaskBreakdownBusinessObject, TaskBreakdownPayload # noqa: E402
|
||
from app.services.ai_project_control_service import AIProjectControlService # noqa: E402
|
||
from app.core.config import settings # noqa: E402
|
||
from app.utils.timezone import now_cn_naive # noqa: E402
|
||
|
||
|
||
TABLES = [
|
||
User.__table__,
|
||
FeishuItem.__table__,
|
||
Signal.__table__,
|
||
Project.__table__,
|
||
TruthNode.__table__,
|
||
Task.__table__,
|
||
Budget.__table__,
|
||
ModelVersion.__table__,
|
||
ModelRequirementBinding.__table__,
|
||
ModelCatalogItem.__table__,
|
||
ProjectModelBinding.__table__,
|
||
ModelRecommendationSnapshot.__table__,
|
||
ModelUsageEvent.__table__,
|
||
ModelEvalRun.__table__,
|
||
PromptPackage.__table__,
|
||
DevWorkspaceBinding.__table__,
|
||
RepoBinding.__table__,
|
||
TaskCodeLink.__table__,
|
||
ProgressEvent.__table__,
|
||
DevSession.__table__,
|
||
ReleaseValidationRun.__table__,
|
||
FeedbackIngest.__table__,
|
||
IterationRecommendation.__table__,
|
||
RevenueEvent.__table__,
|
||
ROISnapshot.__table__,
|
||
ComplianceCheckRun.__table__,
|
||
EvidencePack.__table__,
|
||
GitHubAppInstallation.__table__,
|
||
GitHubAppRepository.__table__,
|
||
GitHubWebhookDelivery.__table__,
|
||
]
|
||
|
||
|
||
def _use_sqlite_json_types() -> None:
|
||
for table in TABLES:
|
||
for column in table.columns:
|
||
type_name = column.type.__class__.__name__.upper()
|
||
if type_name in {"JSONB", "ARRAY"}:
|
||
column.type = JSON()
|
||
|
||
|
||
@pytest_asyncio.fixture
|
||
async def db_env():
|
||
engine = create_async_engine("sqlite+aiosqlite:///:memory:")
|
||
_use_sqlite_json_types()
|
||
async with engine.begin() as conn:
|
||
await conn.run_sync(lambda sync_conn: Base.metadata.create_all(sync_conn, tables=TABLES))
|
||
maker = async_sessionmaker(engine, expire_on_commit=False)
|
||
async with maker() as session:
|
||
yield session
|
||
await engine.dispose()
|
||
|
||
|
||
async def _seed_project_graph(session):
|
||
user = User(
|
||
tenant_id=1,
|
||
username="owner",
|
||
email="owner@example.com",
|
||
password_hash="hashed",
|
||
name="Owner",
|
||
role="REVIEWER",
|
||
assigned_domains=[],
|
||
workload=0,
|
||
status="active",
|
||
is_active=True,
|
||
)
|
||
session.add(user)
|
||
await session.flush()
|
||
|
||
feishu_item = FeishuItem(
|
||
tenant_id=1,
|
||
raw_message="新建 AI 客服项目",
|
||
level="L1",
|
||
card_type="signal",
|
||
route_status="pending",
|
||
)
|
||
session.add(feishu_item)
|
||
await session.flush()
|
||
|
||
signal = Signal(
|
||
tenant_id=1,
|
||
source="manual",
|
||
content="需要一个支持知识库检索和客服回复的 AI 应用",
|
||
priority="high",
|
||
processed=False,
|
||
feishu_item_id=feishu_item.id,
|
||
title="AI 客服升级",
|
||
status="待处理",
|
||
planner_status="pending",
|
||
)
|
||
session.add(signal)
|
||
await session.flush()
|
||
|
||
project = Project(
|
||
tenant_id=1,
|
||
name="AI 客服升级",
|
||
title="AI 客服升级",
|
||
description="支持 RAG、客服对话和模型成本治理",
|
||
owner_id=user.id,
|
||
status="active",
|
||
assigned_role="BUILDER",
|
||
signal_count=1,
|
||
)
|
||
session.add(project)
|
||
await session.flush()
|
||
|
||
charter = TruthNode(
|
||
tenant_id=1,
|
||
node_type="charter",
|
||
parent_id=None,
|
||
project_id=project.id,
|
||
signal_id=signal.id,
|
||
created_by=user.id,
|
||
title="AI 客服升级宪章",
|
||
content={
|
||
"business_goal": "7 天内交付可验证 MVP",
|
||
"target_users": ["客服团队", "运营负责人"],
|
||
"success_criteria": ["首响低于 3 秒", "高频问题命中率达到 85%"],
|
||
"data_sources": ["客服 FAQ", "产品说明书"],
|
||
"content_labeling": "所有 AI 生成回复在前台显示 AI 标识",
|
||
"model_requirements": ["中文问答", "RAG", "代码集成"],
|
||
"delivery_boundary": ["仅做 MVP，不做 CRM 全量重构"],
|
||
},
|
||
version=1,
|
||
status="aligned",
|
||
)
|
||
session.add(charter)
|
||
await session.flush()
|
||
|
||
release = TruthNode(
|
||
tenant_id=1,
|
||
node_type="release",
|
||
parent_id=charter.id,
|
||
project_id=project.id,
|
||
signal_id=None,
|
||
created_by=user.id,
|
||
title="AI 客服 MVP 发布",
|
||
content={"charter_id": charter.id},
|
||
version=1,
|
||
status="pending",
|
||
)
|
||
session.add(release)
|
||
await session.flush()
|
||
|
||
task = Task(
|
||
tenant_id=1,
|
||
truth_node_id=charter.id,
|
||
project_id=project.id,
|
||
assignee_id=user.id,
|
||
ai_recommended_assignees=None,
|
||
ai_breakdown_payload=None,
|
||
manual_overrides=None,
|
||
title="接入 AI 模型与知识库",
|
||
description="实现问答主链、成本埋点和效果验证基线",
|
||
required_skill_tag_ids=[101],
|
||
status="todo",
|
||
priority="high",
|
||
estimated_hours=8,
|
||
actual_hours=None,
|
||
quality_score=None,
|
||
due_date=None,
|
||
completed_at=None,
|
||
)
|
||
session.add(task)
|
||
|
||
budget = Budget(
|
||
tenant_id=1,
|
||
budget_type="project",
|
||
project_id=project.id,
|
||
period="monthly",
|
||
budget_amount_usd=3000,
|
||
alert_threshold_percent=80,
|
||
hard_limit_enabled=True,
|
||
current_usage_usd=0,
|
||
current_period_start=None,
|
||
current_period_end=None,
|
||
status="active",
|
||
override_history={},
|
||
)
|
||
session.add(budget)
|
||
await session.commit()
|
||
return user, project, charter, release, task
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_model_recommendation_binding_and_prompt_package_flow(db_env) -> None:
|
||
session = db_env
|
||
user, project, charter, _release, task = await _seed_project_graph(session)
|
||
service = AIProjectControlService(session, tenant_id=1)
|
||
|
||
recommendations = await service.recommend_models(project_id=project.id, charter_id=charter.id)
|
||
assert len(recommendations["recommendations"]) == 3
|
||
top = recommendations["recommendations"][0]
|
||
|
||
binding = await service.bind_model(
|
||
project_id=project.id,
|
||
catalog_item_id=int(top["catalog_item_id"]),
|
||
selected_by_user_id=user.id,
|
||
charter_id=charter.id,
|
||
estimated_daily_tokens=120_000,
|
||
selection_reason={"score": 93, "note": "适合 MVP 先跑通"},
|
||
)
|
||
assert binding is not None
|
||
assert binding["catalog_item_id"] == top["catalog_item_id"]
|
||
assert float(binding["estimated_monthly_cost_usd"] or 0.0) > 0
|
||
|
||
legacy_binding = (
|
||
await session.execute(
|
||
select(ModelRequirementBinding).where(ModelRequirementBinding.truth_node_id == charter.id)
|
||
)
|
||
).scalar_one_or_none()
|
||
assert legacy_binding is not None
|
||
|
||
payload = TaskBreakdownPayload(
|
||
tasks=[
|
||
TaskBreakdownBusinessObject(
|
||
task_key="baseline",
|
||
title="先补业务基线",
|
||
description="整理 MVP 的功能边界",
|
||
type="planning",
|
||
estimated_hours=2,
|
||
required_skill_names=["产品"],
|
||
priority="high",
|
||
)
|
||
]
|
||
)
|
||
enriched = service.inject_model_tasks(payload=payload, active_binding=binding)
|
||
task_keys = {item.task_key for item in enriched.tasks}
|
||
assert "model-integration" in task_keys
|
||
assert "prompt-optimization" in task_keys
|
||
assert "model-eval" in task_keys
|
||
|
||
prompt_package = await service.create_prompt_package(
|
||
project_id=project.id,
|
||
charter_id=charter.id,
|
||
generated_by_user_id=user.id,
|
||
task_id=task.id,
|
||
)
|
||
assert prompt_package["task_id"] == str(task.id)
|
||
assert "模型绑定" in prompt_package["package_text"]
|
||
assert prompt_package["package_payload"]["model_binding"]["catalog_item_id"] == top["catalog_item_id"]
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_progress_events_and_github_ingest_follow_controlled_workflow(db_env) -> None:
|
||
session = db_env
|
||
user, project, charter, _release, task = await _seed_project_graph(session)
|
||
service = AIProjectControlService(session, tenant_id=1)
|
||
recommendations = await service.recommend_models(project_id=project.id, charter_id=charter.id)
|
||
selected_model = next(
|
||
(item for item in recommendations["recommendations"] if item["filing_status"] == "filed"),
|
||
recommendations["recommendations"][0],
|
||
)
|
||
await service.bind_model(
|
||
project_id=project.id,
|
||
catalog_item_id=int(selected_model["catalog_item_id"]),
|
||
selected_by_user_id=user.id,
|
||
charter_id=charter.id,
|
||
estimated_daily_tokens=80_000,
|
||
selection_reason={"score": 88},
|
||
)
|
||
|
||
repo_binding = await service.register_repo_binding(
|
||
project_id=project.id,
|
||
provider="github",
|
||
owner="pmflow",
|
||
repo_name="control-plane",
|
||
default_branch="main",
|
||
external_repo_id="repo-1",
|
||
)
|
||
|
||
commit_event = await service.create_or_apply_progress_event(
|
||
task_id=task.id,
|
||
project_id=project.id,
|
||
event_type="commit",
|
||
source="ide",
|
||
payload={"branch_name": "feature/model-control", "commit_sha": "abc123"},
|
||
user_id=user.id,
|
||
repo_binding_id=int(repo_binding["id"]),
|
||
)
|
||
assert commit_event["task_status"] == "in_progress"
|
||
assert commit_event["workflow_state"] == "in_progress"
|
||
|
||
link = (
|
||
await session.execute(
|
||
select(TaskCodeLink).where(
|
||
TaskCodeLink.task_id == task.id,
|
||
TaskCodeLink.repo_binding_id == int(repo_binding["id"]),
|
||
)
|
||
)
|
||
).scalar_one_or_none()
|
||
assert link is not None
|
||
assert link.branch_name == "feature/model-control"
|
||
|
||
pr_result = await service.ingest_github_event(
|
||
event_type="pull_request",
|
||
payload={
|
||
"action": "opened",
|
||
"repository": {"name": "control-plane", "owner": {"login": "pmflow"}},
|
||
"pull_request": {"number": 17, "head": {"ref": "feature/model-control"}},
|
||
},
|
||
)
|
||
assert pr_result["matched_links"] == 1
|
||
assert pr_result["events_created"][0]["task_status"] == "review"
|
||
assert pr_result["events_created"][0]["workflow_state"] == "in_review"
|
||
|
||
ignored = await service.ingest_github_event(
|
||
event_type="check_run",
|
||
payload={
|
||
"action": "completed",
|
||
"repository": {"name": "control-plane", "owner": {"login": "pmflow"}},
|
||
},
|
||
)
|
||
assert ignored["matched_links"] == 0
|
||
assert ignored["events_created"] == []
|
||
assert ignored["ignored_reason"] == "missing_branch_or_pr"
|
||
|
||
summary = await service.get_project_dev_sync_summary(project_id=project.id, current_user_id=user.id)
|
||
assert summary["summary"]["repo_binding_count"] == 1
|
||
assert summary["summary"]["active_task_count"] >= 1
|
||
assert summary["summary"]["recent_event_count"] >= 2
|
||
assert summary["summary"]["my_active_task_count"] == 1
|
||
assert summary["active_dev_tasks"][0]["code_link"]["pr_number"] == 17
|
||
assert summary["recent_progress_events"][0]["task"]["id"] == str(task.id)
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_validation_feedback_roi_and_compliance_chain(db_env) -> None:
|
||
session = db_env
|
||
user, project, charter, release, task = await _seed_project_graph(session)
|
||
service = AIProjectControlService(session, tenant_id=1)
|
||
recommendations = await service.recommend_models(project_id=project.id, charter_id=charter.id)
|
||
selected_model = next(
|
||
(item for item in recommendations["recommendations"] if item["filing_status"] == "filed"),
|
||
recommendations["recommendations"][0],
|
||
)
|
||
binding = await service.bind_model(
|
||
project_id=project.id,
|
||
catalog_item_id=int(selected_model["catalog_item_id"]),
|
||
selected_by_user_id=user.id,
|
||
charter_id=charter.id,
|
||
estimated_daily_tokens=100_000,
|
||
selection_reason={"score": 95},
|
||
)
|
||
prompt_package = await service.create_prompt_package(
|
||
project_id=project.id,
|
||
charter_id=charter.id,
|
||
generated_by_user_id=user.id,
|
||
task_id=task.id,
|
||
)
|
||
assert prompt_package["status"] == "active"
|
||
|
||
session.add(
|
||
ModelUsageEvent(
|
||
tenant_id=1,
|
||
project_id=project.id,
|
||
project_model_binding_id=int(binding["id"]),
|
||
source_type="manual",
|
||
event_key="usage-1",
|
||
input_tokens=120_000,
|
||
output_tokens=30_000,
|
||
cost_usd=Decimal("4.50"),
|
||
event_payload={"source": "qa"},
|
||
occurred_at=now_cn_naive(),
|
||
)
|
||
)
|
||
await session.commit()
|
||
|
||
validation = await service.run_release_validation(
|
||
project_id=project.id,
|
||
actor_user_id=user.id,
|
||
charter_id=charter.id,
|
||
release_node_id=release.id,
|
||
)
|
||
assert validation["status"] == "passed"
|
||
assert validation["score"] == 100.0
|
||
|
||
feedback = await service.ingest_feedback(
|
||
project_id=project.id,
|
||
actor_user_id=user.id,
|
||
charter_id=charter.id,
|
||
source_type="manual",
|
||
title="线上反馈",
|
||
content="回答有时偏慢，而且个别场景会出现幻觉。",
|
||
metrics={"csat": 3.8},
|
||
samples=[{"question": "退款规则", "issue": "答非所问"}],
|
||
)
|
||
focus_areas = feedback["recommendation"]["recommendation_summary"]["focus_areas"]
|
||
assert "performance" in focus_areas
|
||
assert "quality" in focus_areas
|
||
|
||
revenue = await service.record_revenue_event(
|
||
project_id=project.id,
|
||
amount_usd=180.0,
|
||
source_type="subscription",
|
||
currency="usd",
|
||
external_ref="sub-1",
|
||
event_payload={"plan": "pro"},
|
||
)
|
||
assert revenue["amount_usd"] == 180.0
|
||
|
||
roi = await service.get_roi_snapshots(project_id=project.id)
|
||
assert roi["latest"]["total_revenue_usd"] == 180.0
|
||
assert roi["latest"]["total_cost_usd"] == 4.5
|
||
assert roi["latest"]["roi_ratio"] > 0
|
||
|
||
compliance = await service.run_compliance_check(
|
||
project_id=project.id,
|
||
actor_user_id=user.id,
|
||
charter_id=charter.id,
|
||
release_node_id=release.id,
|
||
)
|
||
assert compliance["status"] == "passed"
|
||
assert compliance["score"] == 100.0
|
||
assert compliance["evidence_pack"]["package_kind"] == "compliance_release"
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_github_webhook_uses_raw_body_signature_and_delivery_dedup(db_env) -> None:
|
||
session = db_env
|
||
user, project, charter, _release, task = await _seed_project_graph(session)
|
||
service = AIProjectControlService(session, tenant_id=1)
|
||
recommendations = await service.recommend_models(project_id=project.id, charter_id=charter.id)
|
||
await service.bind_model(
|
||
project_id=project.id,
|
||
catalog_item_id=int(recommendations["recommendations"][0]["catalog_item_id"]),
|
||
selected_by_user_id=user.id,
|
||
charter_id=charter.id,
|
||
estimated_daily_tokens=60_000,
|
||
selection_reason={"score": 91},
|
||
)
|
||
|
||
repo_binding = await service.register_repo_binding(
|
||
project_id=project.id,
|
||
provider="github",
|
||
owner="pmflow",
|
||
repo_name="control-plane",
|
||
default_branch="main",
|
||
external_repo_id="repo-1",
|
||
)
|
||
await service.create_or_apply_progress_event(
|
||
task_id=task.id,
|
||
project_id=project.id,
|
||
event_type="commit",
|
||
source="ide",
|
||
payload={"branch_name": "feature/model-control", "commit_sha": "abc123"},
|
||
user_id=user.id,
|
||
repo_binding_id=int(repo_binding["id"]),
|
||
)
|
||
|
||
previous_secret = settings.github_app_webhook_secret
|
||
settings.github_app_webhook_secret = "test-secret"
|
||
try:
|
||
payload = {
|
||
"action": "opened",
|
||
"repository": {"name": "control-plane", "owner": {"login": "pmflow"}},
|
||
"pull_request": {"number": 17, "head": {"ref": "feature/model-control"}},
|
||
}
|
||
raw_body = json.dumps(payload, separators=(",", ":"), ensure_ascii=False).encode("utf-8")
|
||
signature = "sha256=" + hmac.new(
|
||
b"test-secret",
|
||
raw_body,
|
||
hashlib.sha256,
|
||
).hexdigest()
|
||
|
||
result = await service.ingest_github_webhook(
|
||
event_type="pull_request",
|
||
delivery_id="delivery-1",
|
||
signature_256=signature,
|
||
payload=payload,
|
||
raw_body=raw_body,
|
||
)
|
||
assert result["deduplicated"] is False
|
||
assert result["delivery"]["status"] == "processed"
|
||
|
||
dedup = await service.ingest_github_webhook(
|
||
event_type="pull_request",
|
||
delivery_id="delivery-1",
|
||
signature_256=signature,
|
||
payload=payload,
|
||
raw_body=raw_body,
|
||
)
|
||
assert dedup["deduplicated"] is True
|
||
|
||
with pytest.raises(ValueError, match="签名校验失败"):
|
||
await service.ingest_github_webhook(
|
||
event_type="pull_request",
|
||
delivery_id="delivery-2",
|
||
signature_256="sha256=bad-signature",
|
||
payload=payload,
|
||
raw_body=raw_body,
|
||
)
|
||
finally:
|
||
settings.github_app_webhook_secret = previous_secret
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_bind_github_repository_requires_installation_repo(db_env) -> None:
|
||
session = db_env
|
||
_user, project, _charter, _release, _task = await _seed_project_graph(session)
|
||
service = AIProjectControlService(session, tenant_id=1)
|
||
|
||
installation = GitHubAppInstallation(
|
||
tenant_id=1,
|
||
installation_id="inst-1",
|
||
account_login="pmflow",
|
||
account_type="Organization",
|
||
app_slug="pmflow-dev",
|
||
status="active",
|
||
permissions={"pull_requests": "write"},
|
||
events=["pull_request"],
|
||
app_metadata={},
|
||
)
|
||
session.add(installation)
|
||
await session.flush()
|
||
|
||
allowed_repo = GitHubAppRepository(
|
||
tenant_id=1,
|
||
installation_id="inst-1",
|
||
external_repo_id="1001",
|
||
owner="pmflow",
|
||
repo_name="allowed-repo",
|
||
full_name="pmflow/allowed-repo",
|
||
default_branch="main",
|
||
permissions={"pull_requests": "write"},
|
||
is_private=True,
|
||
status="active",
|
||
)
|
||
blocked_repo = GitHubAppRepository(
|
||
tenant_id=1,
|
||
installation_id="inst-1",
|
||
external_repo_id="1002",
|
||
owner="pmflow",
|
||
repo_name="blocked-repo",
|
||
full_name="pmflow/blocked-repo",
|
||
default_branch="main",
|
||
permissions={"pull_requests": "write"},
|
||
is_private=True,
|
||
status="archived",
|
||
)
|
||
session.add_all([allowed_repo, blocked_repo])
|
||
await session.commit()
|
||
|
||
bound = await service.bind_github_repository(
|
||
project_id=project.id,
|
||
github_repository_id=allowed_repo.id,
|
||
)
|
||
assert bound["full_name"] == "pmflow/allowed-repo"
|
||
|
||
with pytest.raises(ValueError, match="已授权仓库列表"):
|
||
await service.bind_github_repository(
|
||
project_id=project.id,
|
||
github_repository_id=blocked_repo.id,
|
||
)
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_sync_github_installations_from_platform_app(monkeypatch, db_env) -> None:
|
||
session = db_env
|
||
service = AIProjectControlService(session, tenant_id=1)
|
||
|
||
class FakeResponse:
|
||
def __init__(self, payload: dict):
|
||
self._payload = payload
|
||
self.headers = {}
|
||
|
||
def raise_for_status(self) -> None:
|
||
return None
|
||
|
||
def json(self) -> dict:
|
||
return self._payload
|
||
|
||
class FakeClient:
|
||
async def __aenter__(self):
|
||
return self
|
||
|
||
async def __aexit__(self, exc_type, exc, tb):
|
||
return None
|
||
|
||
async def get(self, url: str, *, headers: dict, params: dict | None = None):
|
||
if url == "https://api.github.com/app/installations":
|
||
assert headers["Authorization"] == "Bearer test-jwt"
|
||
return FakeResponse(
|
||
{
|
||
"installations": [
|
||
{
|
||
"id": 901,
|
||
"account": {"login": "pmflow-user", "type": "User"},
|
||
"target_type": "User",
|
||
"permissions": {"contents": "read", "pull_requests": "write"},
|
||
"events": ["push", "pull_request"],
|
||
"repository_selection": "selected",
|
||
"html_url": "https://github.com/settings/installations/901",
|
||
}
|
||
]
|
||
}
|
||
)
|
||
if url == "https://api.github.com/installation/repositories":
|
||
assert headers["Authorization"] == "Bearer installation-token"
|
||
return FakeResponse(
|
||
{
|
||
"repositories": [
|
||
{
|
||
"id": 10001,
|
||
"full_name": "pmflow-user/app",
|
||
"name": "app",
|
||
"owner": {"login": "pmflow-user"},
|
||
"default_branch": "main",
|
||
"private": True,
|
||
"permissions": {"pull": True, "push": False},
|
||
}
|
||
]
|
||
}
|
||
)
|
||
raise AssertionError(f"unexpected url {url}")
|
||
|
||
async def post(self, url: str, *, headers: dict):
|
||
assert headers["Authorization"] == "Bearer test-jwt"
|
||
if url == "https://api.github.com/app/installations/901/access_tokens":
|
||
return FakeResponse({"token": "installation-token"})
|
||
raise AssertionError(f"unexpected url {url}")
|
||
|
||
previous = (
|
||
settings.github_app_id,
|
||
settings.github_app_private_key,
|
||
settings.github_app_sync_endpoint,
|
||
settings.github_app_installations_fixture,
|
||
)
|
||
settings.github_app_id = "12345"
|
||
settings.github_app_private_key = "fake-key"
|
||
settings.github_app_sync_endpoint = None
|
||
settings.github_app_installations_fixture = None
|
||
monkeypatch.setattr(AIProjectControlService, "_github_app_jwt", lambda self: "test-jwt")
|
||
monkeypatch.setattr("app.services.ai_project_control_service.httpx.AsyncClient", lambda **kwargs: FakeClient())
|
||
try:
|
||
result = await service.sync_github_installations()
|
||
finally:
|
||
(
|
||
settings.github_app_id,
|
||
settings.github_app_private_key,
|
||
settings.github_app_sync_endpoint,
|
||
settings.github_app_installations_fixture,
|
||
) = previous
|
||
|
||
assert result["repository_count"] == 1
|
||
assert result["installations"][0]["installation_id"] == "901"
|
||
|
||
repos = await service.list_available_github_repositories()
|
||
assert repos[0]["full_name"] == "pmflow-user/app"
|
||
assert repos[0]["installation_account_login"] == "pmflow-user"
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_release_validation_blocked_generates_fix_task_and_prompt(db_env) -> None:
|
||
session = db_env
|
||
user, project, charter, release, _task = await _seed_project_graph(session)
|
||
service = AIProjectControlService(session, tenant_id=1)
|
||
|
||
validation = await service.run_release_validation(
|
||
project_id=project.id,
|
||
actor_user_id=user.id,
|
||
charter_id=charter.id,
|
||
release_node_id=release.id,
|
||
)
|
||
|
||
assert validation["allow_release"] is False
|
||
assert validation["status"] == "blocked"
|
||
assert validation["blocking_reason"]
|
||
assert validation["remediation_task"] is not None
|
||
assert validation["remediation_prompt_package"] is not None
|