MMD
2747f9d4bb
The PM panel (PmPanelPage -> iframe /pm/) needs the project-management platform served at /pm/ on the ERP domain. Source: org repo Aramesh/PM, branch feature/project-management-platform. Update via: git subtree pull --prefix=pm aramesh-pm feature/project-management-platform --squash
95 lines
3.4 KiB
Python
95 lines
3.4 KiB
Python
import pytest
|
|
from httpx import ASGITransport, AsyncClient
|
|
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
|
|
from sqlalchemy.orm import sessionmaker
|
|
|
|
from app.db import get_db
|
|
from app.main import app
|
|
from app.models import Base, PMUser, Project, Task
|
|
from app.security import create_token, hash_password
|
|
|
|
|
|
@pytest.fixture
|
|
def anyio_backend():
|
|
return "asyncio"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_gantt_and_relations_api():
|
|
engine = create_async_engine("sqlite+aiosqlite:///:memory:", echo=False)
|
|
async with engine.begin() as conn:
|
|
await conn.run_sync(Base.metadata.create_all)
|
|
|
|
async_session = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
|
|
|
|
async with async_session() as session:
|
|
admin = PMUser(
|
|
username="admin",
|
|
password_hash=hash_password("admin123"),
|
|
display_name="مدیر ارشد",
|
|
role="project_manager",
|
|
)
|
|
session.add(admin)
|
|
await session.flush()
|
|
|
|
project = Project(name="پروژه تست گانت", description="تست")
|
|
session.add(project)
|
|
await session.flush()
|
|
|
|
task1 = Task(project_id=project.id, title="فونداسیون", duration=5)
|
|
task2 = Task(project_id=project.id, title="اسکلت بتنی", duration=10)
|
|
session.add_all([task1, task2])
|
|
await session.commit()
|
|
|
|
admin_id = admin.id
|
|
p_id = str(project.id)
|
|
t1_id = str(task1.id)
|
|
t2_id = str(task2.id)
|
|
|
|
token = create_token(admin_id)
|
|
headers = {"Authorization": f"Bearer {token}"}
|
|
|
|
async def override_get_db():
|
|
async with async_session() as session:
|
|
yield session
|
|
|
|
app.dependency_overrides[get_db] = override_get_db
|
|
|
|
try:
|
|
transport = ASGITransport(app=app)
|
|
async with AsyncClient(transport=transport, base_url="http://test", headers=headers) as client:
|
|
# 1. Add relation: task1 -> task2 (FS + 2 days lag)
|
|
rel_res = await client.post(
|
|
f"/api/v1/projects/{p_id}/relations",
|
|
json={"from_task_id": t1_id, "to_task_id": t2_id, "relation_type": "FS", "lag_days": 2},
|
|
)
|
|
assert rel_res.status_code == 200
|
|
rel_data = rel_res.json()["data"]
|
|
assert rel_data["relation_type"] == "FS"
|
|
assert rel_data["lag_days"] == 2
|
|
rel_id = rel_data["id"]
|
|
|
|
# 2. Add reverse relation: task2 -> task1 (should detect cycle and reject with 400)
|
|
cycle_res = await client.post(
|
|
f"/api/v1/projects/{p_id}/relations",
|
|
json={"from_task_id": t2_id, "to_task_id": t1_id, "relation_type": "FS"},
|
|
)
|
|
assert cycle_res.status_code == 400
|
|
assert "چرخه" in cycle_res.json()["error"]["message"]
|
|
|
|
# 3. Query Gantt dataset
|
|
gantt_res = await client.get(f"/api/v1/projects/{p_id}/gantt")
|
|
assert gantt_res.status_code == 200
|
|
gantt_data = gantt_res.json()["data"]
|
|
assert len(gantt_data["tasks"]) == 2
|
|
assert len(gantt_data["relations"]) == 1
|
|
assert len(gantt_data["critical_path"]) >= 1
|
|
|
|
# 4. Delete relation
|
|
del_rel = await client.delete(f"/api/v1/projects/{p_id}/relations/{rel_id}")
|
|
assert del_rel.status_code == 200
|
|
assert del_rel.json()["data"]["deleted"] is True
|
|
finally:
|
|
app.dependency_overrides.clear()
|
|
await engine.dispose()
|