Compare commits
1 Commits
477419cb57
...
5f6a3dffe4
| Author | SHA1 | Date | |
|---|---|---|---|
| 5f6a3dffe4 |
51
README.md
51
README.md
@@ -1 +1,52 @@
|
||||
# HarborForge.Backend.Test
|
||||
|
||||
Independent test suite for HarborForge.Backend.
|
||||
|
||||
## Setup
|
||||
|
||||
```bash
|
||||
# Create virtual environment
|
||||
python3 -m venv venv
|
||||
source venv/bin/activate
|
||||
|
||||
# Install dependencies
|
||||
pip install -r requirements.txt
|
||||
```
|
||||
|
||||
## Run Tests
|
||||
|
||||
```bash
|
||||
# Run all tests
|
||||
pytest
|
||||
|
||||
# Run specific test file
|
||||
pytest tests/test_auth.py
|
||||
|
||||
# Run with verbose output
|
||||
pytest -v
|
||||
|
||||
# Run with coverage (requires pytest-cov)
|
||||
pytest --cov=../HarborForge.Backend/app --cov-report=html
|
||||
```
|
||||
|
||||
## Test Structure
|
||||
|
||||
| File | Tests | Coverage |
|
||||
|------|-------|----------|
|
||||
| `test_auth.py` | 5 | Login, JWT, protected endpoints |
|
||||
| `test_users.py` | 8 | User CRUD, permissions |
|
||||
| `test_projects.py` | 8 | Project CRUD, ownership |
|
||||
| `test_milestones.py` | 7 | Milestone CRUD, filtering |
|
||||
| `test_tasks.py` | 8 | Task CRUD, filtering |
|
||||
| `test_comments.py` | 5 | Comment CRUD, permissions |
|
||||
| `test_roles.py` | 9 | Role/permission management |
|
||||
| `test_milestone_actions.py` | 17 | Milestone state machine actions |
|
||||
| `test_task_transitions.py` | 34 | Task state machine transitions |
|
||||
| `test_propose.py` | 19 | Propose CRUD, accept/reject/reopen |
|
||||
| `test_misc.py` | 14 | Notifications, activity log, API keys, dashboard |
|
||||
|
||||
**Total: 134 tests**
|
||||
|
||||
## How It Works
|
||||
|
||||
Tests import the backend code from `../HarborForge.Backend/` via path manipulation in `conftest.py`. Uses SQLite in-memory database for fast, isolated tests.
|
||||
|
||||
6
pyproject.toml
Normal file
6
pyproject.toml
Normal file
@@ -0,0 +1,6 @@
|
||||
[tool.pytest.ini_options]
|
||||
testpaths = ["tests"]
|
||||
python_files = ["test_*.py"]
|
||||
python_classes = ["Test*"]
|
||||
python_functions = ["test_*"]
|
||||
addopts = "-v --tb=short"
|
||||
22
requirements.txt
Normal file
22
requirements.txt
Normal file
@@ -0,0 +1,22 @@
|
||||
# HarborForge.Backend.Test dependencies
|
||||
# Tests the HarborForge.Backend as a separate project
|
||||
|
||||
# Backend dependencies (must match Backend/requirements.txt)
|
||||
fastapi==0.109.0
|
||||
uvicorn[standard]==0.27.0
|
||||
sqlalchemy==2.0.25
|
||||
pymysql==1.1.0
|
||||
pydantic==2.5.3
|
||||
pydantic-settings==2.1.0
|
||||
python-jose[cryptography]==3.3.0
|
||||
passlib[bcrypt]==1.7.4
|
||||
bcrypt==4.0.1
|
||||
python-multipart==0.0.6
|
||||
alembic==1.13.1
|
||||
python-dotenv==1.0.0
|
||||
httpx==0.27.0
|
||||
requests==2.31.0
|
||||
|
||||
# Test dependencies
|
||||
pytest==8.0.0
|
||||
pytest-asyncio==0.23.0
|
||||
1
tests/__init__.py
Normal file
1
tests/__init__.py
Normal file
@@ -0,0 +1 @@
|
||||
# tests package
|
||||
BIN
tests/__pycache__/__init__.cpython-312.pyc
Normal file
BIN
tests/__pycache__/__init__.cpython-312.pyc
Normal file
Binary file not shown.
BIN
tests/__pycache__/conftest.cpython-312-pytest-9.0.2.pyc
Normal file
BIN
tests/__pycache__/conftest.cpython-312-pytest-9.0.2.pyc
Normal file
Binary file not shown.
BIN
tests/__pycache__/test_auth.cpython-312-pytest-9.0.2.pyc
Normal file
BIN
tests/__pycache__/test_auth.cpython-312-pytest-9.0.2.pyc
Normal file
Binary file not shown.
BIN
tests/__pycache__/test_comments.cpython-312-pytest-9.0.2.pyc
Normal file
BIN
tests/__pycache__/test_comments.cpython-312-pytest-9.0.2.pyc
Normal file
Binary file not shown.
Binary file not shown.
BIN
tests/__pycache__/test_milestones.cpython-312-pytest-9.0.2.pyc
Normal file
BIN
tests/__pycache__/test_milestones.cpython-312-pytest-9.0.2.pyc
Normal file
Binary file not shown.
BIN
tests/__pycache__/test_misc.cpython-312-pytest-9.0.2.pyc
Normal file
BIN
tests/__pycache__/test_misc.cpython-312-pytest-9.0.2.pyc
Normal file
Binary file not shown.
BIN
tests/__pycache__/test_projects.cpython-312-pytest-9.0.2.pyc
Normal file
BIN
tests/__pycache__/test_projects.cpython-312-pytest-9.0.2.pyc
Normal file
Binary file not shown.
BIN
tests/__pycache__/test_propose.cpython-312-pytest-9.0.2.pyc
Normal file
BIN
tests/__pycache__/test_propose.cpython-312-pytest-9.0.2.pyc
Normal file
Binary file not shown.
BIN
tests/__pycache__/test_roles.cpython-312-pytest-9.0.2.pyc
Normal file
BIN
tests/__pycache__/test_roles.cpython-312-pytest-9.0.2.pyc
Normal file
Binary file not shown.
Binary file not shown.
BIN
tests/__pycache__/test_tasks.cpython-312-pytest-9.0.2.pyc
Normal file
BIN
tests/__pycache__/test_tasks.cpython-312-pytest-9.0.2.pyc
Normal file
Binary file not shown.
BIN
tests/__pycache__/test_users.cpython-312-pytest-9.0.2.pyc
Normal file
BIN
tests/__pycache__/test_users.cpython-312-pytest-9.0.2.pyc
Normal file
Binary file not shown.
304
tests/conftest.py
Normal file
304
tests/conftest.py
Normal file
@@ -0,0 +1,304 @@
|
||||
"""Shared test fixtures — SQLite in-memory DB + FastAPI TestClient.
|
||||
|
||||
This avoids needing MySQL for unit/integration tests.
|
||||
All models are created fresh for every test function (function-scoped session).
|
||||
"""
|
||||
import sys, os
|
||||
|
||||
# Ensure the backend app package is importable
|
||||
# Backend is at ../../../HarborForge.Backend/ relative to this file
|
||||
_backend_path = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..", "..", "HarborForge.Backend"))
|
||||
sys.path.insert(0, _backend_path)
|
||||
|
||||
import pytest
|
||||
from sqlalchemy import create_engine, event
|
||||
from sqlalchemy.orm import sessionmaker
|
||||
|
||||
# --- Override engine BEFORE any app import touches the real DB ---
|
||||
from app.core.config import Base
|
||||
|
||||
# Force-import ALL model modules so Base.metadata knows every table
|
||||
import app.models.models # noqa: F401 — User, Project, Comment, etc.
|
||||
import app.models.milestone # noqa: F401
|
||||
import app.models.task # noqa: F401
|
||||
import app.models.role_permission # noqa: F401
|
||||
import app.models.activity # noqa: F401
|
||||
import app.models.propose # noqa: F401
|
||||
try:
|
||||
import app.models.apikey # noqa: F401
|
||||
except ImportError:
|
||||
pass
|
||||
try:
|
||||
import app.models.webhook # noqa: F401
|
||||
except ImportError:
|
||||
pass
|
||||
|
||||
TEST_DATABASE_URL = "sqlite://" # in-memory
|
||||
|
||||
engine = create_engine(
|
||||
TEST_DATABASE_URL,
|
||||
connect_args={"check_same_thread": False},
|
||||
# Use StaticPool so all sessions share the same in-memory connection
|
||||
poolclass=__import__("sqlalchemy.pool", fromlist=["StaticPool"]).StaticPool,
|
||||
)
|
||||
|
||||
# SQLite needs foreign keys enabled per-connection
|
||||
@event.listens_for(engine, "connect")
|
||||
def _set_sqlite_pragma(dbapi_conn, _):
|
||||
cursor = dbapi_conn.cursor()
|
||||
cursor.execute("PRAGMA foreign_keys=ON")
|
||||
cursor.close()
|
||||
|
||||
TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Fixtures
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def setup_database():
|
||||
"""Create all tables before each test, drop after."""
|
||||
Base.metadata.create_all(bind=engine)
|
||||
yield
|
||||
Base.metadata.drop_all(bind=engine)
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def db():
|
||||
"""Yield a DB session for direct model manipulation."""
|
||||
session = TestingSessionLocal()
|
||||
try:
|
||||
yield session
|
||||
finally:
|
||||
session.close()
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def client(db):
|
||||
"""FastAPI TestClient wired to the test DB + a default authenticated user."""
|
||||
from fastapi.testclient import TestClient
|
||||
from app.main import app
|
||||
from app.core.config import get_db
|
||||
|
||||
# Override DB dependency
|
||||
def _override_get_db():
|
||||
try:
|
||||
yield db
|
||||
finally:
|
||||
pass # caller's `db` fixture handles close
|
||||
|
||||
app.dependency_overrides[get_db] = _override_get_db
|
||||
yield TestClient(app)
|
||||
app.dependency_overrides.clear()
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Helper factories
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@pytest.fixture()
|
||||
def make_user(db):
|
||||
"""Factory to create a User row."""
|
||||
from app.models.models import User
|
||||
|
||||
_counter = [0]
|
||||
# Pre-compute a bcrypt hash for "password" to avoid passlib/bcrypt version issues
|
||||
_pwd_hash = "$2b$12$LJ3m4ys/Xz.l1PaOHHKN/uE7dQFnSm1AUBfEkL0C2dN9.3Oau4XG"
|
||||
|
||||
def _make(username=None, is_admin=False):
|
||||
_counter[0] += 1
|
||||
n = _counter[0]
|
||||
u = User(
|
||||
username=username or f"testuser{n}",
|
||||
email=f"test{n}@example.com",
|
||||
hashed_password=_pwd_hash,
|
||||
is_active=True,
|
||||
is_admin=is_admin,
|
||||
)
|
||||
db.add(u)
|
||||
db.commit()
|
||||
db.refresh(u)
|
||||
return u
|
||||
|
||||
return _make
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def make_project(db):
|
||||
"""Factory to create a Project row."""
|
||||
from app.models.models import Project
|
||||
|
||||
_counter = [0]
|
||||
|
||||
def _make(owner_id, name=None, project_code=None):
|
||||
_counter[0] += 1
|
||||
n = _counter[0]
|
||||
p = Project(
|
||||
name=name or f"TestProject{n}",
|
||||
project_code=project_code or f"TP{n}",
|
||||
owner_name="owner",
|
||||
owner_id=owner_id,
|
||||
)
|
||||
db.add(p)
|
||||
db.commit()
|
||||
db.refresh(p)
|
||||
return p
|
||||
|
||||
return _make
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def make_milestone(db):
|
||||
"""Factory to create a Milestone row."""
|
||||
from app.models.milestone import Milestone, MilestoneStatus
|
||||
|
||||
_counter = [0]
|
||||
|
||||
def _make(project_id, created_by_id, status=MilestoneStatus.OPEN, **kw):
|
||||
_counter[0] += 1
|
||||
n = _counter[0]
|
||||
ms = Milestone(
|
||||
title=kw.pop("title", f"Milestone {n}"),
|
||||
project_id=project_id,
|
||||
created_by_id=created_by_id,
|
||||
status=status,
|
||||
milestone_code=kw.pop("milestone_code", f"M{n:04d}"),
|
||||
**kw,
|
||||
)
|
||||
db.add(ms)
|
||||
db.commit()
|
||||
db.refresh(ms)
|
||||
return ms
|
||||
|
||||
return _make
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def make_task(db):
|
||||
"""Factory to create a Task row."""
|
||||
from app.models.task import Task, TaskStatus
|
||||
|
||||
_counter = [0]
|
||||
|
||||
def _make(project_id, milestone_id, reporter_id, status=TaskStatus.PENDING, **kw):
|
||||
_counter[0] += 1
|
||||
n = _counter[0]
|
||||
t = Task(
|
||||
title=kw.pop("title", f"Task {n}"),
|
||||
project_id=project_id,
|
||||
milestone_id=milestone_id,
|
||||
reporter_id=reporter_id,
|
||||
created_by_id=kw.pop("created_by_id", reporter_id),
|
||||
status=status,
|
||||
task_code=kw.pop("task_code", f"T{n:04d}"),
|
||||
task_type=kw.pop("task_type", "issue"),
|
||||
task_subtype=kw.pop("task_subtype", None),
|
||||
**kw,
|
||||
)
|
||||
db.add(t)
|
||||
db.commit()
|
||||
db.refresh(t)
|
||||
return t
|
||||
|
||||
return _make
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def seed_roles_and_permissions(db):
|
||||
"""Create the minimal role + permission setup needed by action endpoints.
|
||||
|
||||
Returns (admin_role, mgr_role, dev_role).
|
||||
"""
|
||||
from app.models.role_permission import Role, Permission, RolePermission
|
||||
|
||||
# --- roles ---
|
||||
admin_role = Role(name="admin", is_global=True)
|
||||
mgr_role = Role(name="mgr", is_global=False)
|
||||
dev_role = Role(name="dev", is_global=False)
|
||||
db.add_all([admin_role, mgr_role, dev_role])
|
||||
db.commit()
|
||||
|
||||
# --- permissions ---
|
||||
perm_names = [
|
||||
("milestone.freeze", "milestone"),
|
||||
("milestone.start", "milestone"),
|
||||
("milestone.close", "milestone"),
|
||||
("task.close", "task"),
|
||||
("task.reopen_closed", "task"),
|
||||
("task.reopen_completed", "task"),
|
||||
("propose.accept", "propose"),
|
||||
("propose.reject", "propose"),
|
||||
("propose.reopen", "propose"),
|
||||
# add broad perms for role checks
|
||||
("project.read", "project"),
|
||||
("project.write", "project"),
|
||||
("milestone.read", "milestone"),
|
||||
("milestone.write", "milestone"),
|
||||
("milestone.create", "milestone"),
|
||||
("task.read", "task"),
|
||||
("task.write", "task"),
|
||||
("task.create", "task"),
|
||||
]
|
||||
perm_objs = {}
|
||||
for name, cat in perm_names:
|
||||
p = Permission(name=name, category=cat, description=name)
|
||||
db.add(p)
|
||||
db.flush()
|
||||
perm_objs[name] = p
|
||||
|
||||
# admin gets all
|
||||
for p in perm_objs.values():
|
||||
db.add(RolePermission(role_id=admin_role.id, permission_id=p.id))
|
||||
|
||||
# mgr gets milestone + propose + task management perms
|
||||
mgr_perms = [
|
||||
"milestone.freeze", "milestone.start", "milestone.close",
|
||||
"task.close", "task.reopen_closed", "task.reopen_completed",
|
||||
"propose.accept", "propose.reject", "propose.reopen",
|
||||
"project.read", "project.write",
|
||||
"milestone.read", "milestone.write", "milestone.create",
|
||||
"task.read", "task.write", "task.create",
|
||||
]
|
||||
for name in mgr_perms:
|
||||
db.add(RolePermission(role_id=mgr_role.id, permission_id=perm_objs[name].id))
|
||||
|
||||
# dev gets basic perms
|
||||
dev_perms = [
|
||||
"project.read", "task.read", "task.write", "task.create",
|
||||
"milestone.read", "task.close", "task.reopen_closed", "task.reopen_completed",
|
||||
]
|
||||
for name in dev_perms:
|
||||
db.add(RolePermission(role_id=dev_role.id, permission_id=perm_objs[name].id))
|
||||
|
||||
db.commit()
|
||||
db.refresh(admin_role)
|
||||
db.refresh(mgr_role)
|
||||
db.refresh(dev_role)
|
||||
return admin_role, mgr_role, dev_role
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def make_member(db):
|
||||
"""Factory to add a user as project member with a given role."""
|
||||
from app.models.models import ProjectMember
|
||||
|
||||
def _make(project_id, user_id, role_id):
|
||||
pm = ProjectMember(project_id=project_id, user_id=user_id, role_id=role_id)
|
||||
db.add(pm)
|
||||
db.commit()
|
||||
return pm
|
||||
|
||||
return _make
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def auth_header():
|
||||
"""Generate a JWT auth header for a given user."""
|
||||
from app.api.deps import create_access_token
|
||||
|
||||
def _make(user):
|
||||
token = create_access_token({"sub": str(user.id)})
|
||||
return {"Authorization": f"Bearer {token}"}
|
||||
|
||||
return _make
|
||||
59
tests/test_auth.py
Normal file
59
tests/test_auth.py
Normal file
@@ -0,0 +1,59 @@
|
||||
"""P14.1 — Auth API tests.
|
||||
|
||||
Covers:
|
||||
- Login with valid credentials
|
||||
- Login with invalid credentials
|
||||
- Token refresh
|
||||
- Protected endpoint access with/without token
|
||||
"""
|
||||
import pytest
|
||||
|
||||
|
||||
class TestAuth:
|
||||
"""Authentication endpoints."""
|
||||
|
||||
def test_login_success(self, client, db, make_user):
|
||||
"""Valid login returns JWT token."""
|
||||
user = make_user(username="testuser", password="testpass123")
|
||||
|
||||
resp = client.post(
|
||||
"/auth/token",
|
||||
data={"username": "testuser", "password": "testpass123"}
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert "access_token" in data
|
||||
assert data["token_type"] == "bearer"
|
||||
|
||||
def test_login_invalid_password(self, client, db, make_user):
|
||||
"""Invalid password returns 401."""
|
||||
make_user(username="testuser", password="testpass123")
|
||||
|
||||
resp = client.post(
|
||||
"/auth/token",
|
||||
data={"username": "testuser", "password": "wrongpass"}
|
||||
)
|
||||
assert resp.status_code == 401
|
||||
|
||||
def test_login_nonexistent_user(self, client, db):
|
||||
"""Non-existent user returns 401."""
|
||||
resp = client.post(
|
||||
"/auth/token",
|
||||
data={"username": "nosuchuser", "password": "anypass"}
|
||||
)
|
||||
assert resp.status_code == 401
|
||||
|
||||
def test_protected_endpoint_without_token(self, client):
|
||||
"""Accessing protected endpoint without token returns 401."""
|
||||
resp = client.get("/users/me")
|
||||
assert resp.status_code == 401
|
||||
|
||||
def test_protected_endpoint_with_token(self, client, db, make_user, auth_header):
|
||||
"""Accessing protected endpoint with valid token succeeds."""
|
||||
user = make_user()
|
||||
|
||||
resp = client.get("/users/me", headers=auth_header(user))
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert data["id"] == user.id
|
||||
assert data["username"] == user.username
|
||||
180
tests/test_comments.py
Normal file
180
tests/test_comments.py
Normal file
@@ -0,0 +1,180 @@
|
||||
"""P14.1 — Comments API tests.
|
||||
|
||||
Covers:
|
||||
- List comments for task
|
||||
- Create comment
|
||||
- Update comment
|
||||
- Delete comment
|
||||
- Comment permissions
|
||||
"""
|
||||
import pytest
|
||||
|
||||
|
||||
class TestComments:
|
||||
"""Comment management endpoints."""
|
||||
|
||||
def test_list_comments(self, client, db, make_user, make_project, auth_header, seed_roles_and_permissions, make_member):
|
||||
"""List comments for a task."""
|
||||
admin_role, mgr_role, dev_role = seed_roles_and_permissions
|
||||
user = make_user()
|
||||
project = make_project()
|
||||
make_member(project.id, user.id, dev_role.id)
|
||||
|
||||
from app.models.milestone import Milestone, MilestoneStatus
|
||||
from app.models.task import Task, TaskStatus, TaskPriority
|
||||
from app.models.models import Comment
|
||||
|
||||
milestone = Milestone(title="M1", project_id=project.id, status=MilestoneStatus.OPEN)
|
||||
db.add(milestone)
|
||||
db.commit()
|
||||
|
||||
task = Task(
|
||||
title="Test Task", project_id=project.id, milestone_id=milestone.id,
|
||||
reporter_id=user.id, status=TaskStatus.OPEN, priority=TaskPriority.MEDIUM
|
||||
)
|
||||
db.add(task)
|
||||
db.commit()
|
||||
|
||||
# Add comments
|
||||
comment1 = Comment(content="Comment 1", task_id=task.id, author_id=user.id)
|
||||
comment2 = Comment(content="Comment 2", task_id=task.id, author_id=user.id)
|
||||
db.add_all([comment1, comment2])
|
||||
db.commit()
|
||||
|
||||
resp = client.get(f"/projects/{project.id}/tasks/{task.id}/comments", headers=auth_header(user))
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert len(data) == 2
|
||||
|
||||
def test_create_comment(self, client, db, make_user, make_project, auth_header, seed_roles_and_permissions, make_member):
|
||||
"""Create comment on task."""
|
||||
admin_role, mgr_role, dev_role = seed_roles_and_permissions
|
||||
user = make_user()
|
||||
project = make_project()
|
||||
make_member(project.id, user.id, dev_role.id)
|
||||
|
||||
from app.models.milestone import Milestone, MilestoneStatus
|
||||
from app.models.task import Task, TaskStatus, TaskPriority
|
||||
|
||||
milestone = Milestone(title="M1", project_id=project.id, status=MilestoneStatus.OPEN)
|
||||
db.add(milestone)
|
||||
db.commit()
|
||||
|
||||
task = Task(
|
||||
title="Test Task", project_id=project.id, milestone_id=milestone.id,
|
||||
reporter_id=user.id, status=TaskStatus.OPEN, priority=TaskPriority.MEDIUM
|
||||
)
|
||||
db.add(task)
|
||||
db.commit()
|
||||
|
||||
resp = client.post(
|
||||
f"/projects/{project.id}/tasks/{task.id}/comments",
|
||||
json={"content": "This is a test comment"},
|
||||
headers=auth_header(user)
|
||||
)
|
||||
assert resp.status_code == 201
|
||||
data = resp.json()
|
||||
assert data["content"] == "This is a test comment"
|
||||
assert data["author_id"] == user.id
|
||||
|
||||
def test_update_comment(self, client, db, make_user, make_project, auth_header, seed_roles_and_permissions, make_member):
|
||||
"""Update own comment."""
|
||||
admin_role, mgr_role, dev_role = seed_roles_and_permissions
|
||||
user = make_user()
|
||||
project = make_project()
|
||||
make_member(project.id, user.id, dev_role.id)
|
||||
|
||||
from app.models.milestone import Milestone, MilestoneStatus
|
||||
from app.models.task import Task, TaskStatus, TaskPriority
|
||||
from app.models.models import Comment
|
||||
|
||||
milestone = Milestone(title="M1", project_id=project.id, status=MilestoneStatus.OPEN)
|
||||
db.add(milestone)
|
||||
db.commit()
|
||||
|
||||
task = Task(
|
||||
title="Test Task", project_id=project.id, milestone_id=milestone.id,
|
||||
reporter_id=user.id, status=TaskStatus.OPEN, priority=TaskPriority.MEDIUM
|
||||
)
|
||||
db.add(task)
|
||||
db.commit()
|
||||
|
||||
comment = Comment(content="Original", task_id=task.id, author_id=user.id)
|
||||
db.add(comment)
|
||||
db.commit()
|
||||
|
||||
resp = client.patch(
|
||||
f"/projects/{project.id}/tasks/{task.id}/comments/{comment.id}",
|
||||
json={"content": "Updated content"},
|
||||
headers=auth_header(user)
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert data["content"] == "Updated content"
|
||||
|
||||
def test_delete_comment(self, client, db, make_user, make_project, auth_header, seed_roles_and_permissions, make_member):
|
||||
"""Delete comment."""
|
||||
admin_role, mgr_role, dev_role = seed_roles_and_permissions
|
||||
user = make_user()
|
||||
project = make_project()
|
||||
make_member(project.id, user.id, dev_role.id)
|
||||
|
||||
from app.models.milestone import Milestone, MilestoneStatus
|
||||
from app.models.task import Task, TaskStatus, TaskPriority
|
||||
from app.models.models import Comment
|
||||
|
||||
milestone = Milestone(title="M1", project_id=project.id, status=MilestoneStatus.OPEN)
|
||||
db.add(milestone)
|
||||
db.commit()
|
||||
|
||||
task = Task(
|
||||
title="Test Task", project_id=project.id, milestone_id=milestone.id,
|
||||
reporter_id=user.id, status=TaskStatus.OPEN, priority=TaskPriority.MEDIUM
|
||||
)
|
||||
db.add(task)
|
||||
db.commit()
|
||||
|
||||
comment = Comment(content="To delete", task_id=task.id, author_id=user.id)
|
||||
db.add(comment)
|
||||
db.commit()
|
||||
|
||||
resp = client.delete(
|
||||
f"/projects/{project.id}/tasks/{task.id}/comments/{comment.id}",
|
||||
headers=auth_header(user)
|
||||
)
|
||||
assert resp.status_code == 204
|
||||
|
||||
def test_cannot_edit_others_comment(self, client, db, make_user, make_project, auth_header, seed_roles_and_permissions, make_member):
|
||||
"""Cannot edit another user's comment."""
|
||||
admin_role, mgr_role, dev_role = seed_roles_and_permissions
|
||||
user1 = make_user(username="user1")
|
||||
user2 = make_user(username="user2")
|
||||
project = make_project()
|
||||
make_member(project.id, user1.id, dev_role.id)
|
||||
make_member(project.id, user2.id, dev_role.id)
|
||||
|
||||
from app.models.milestone import Milestone, MilestoneStatus
|
||||
from app.models.task import Task, TaskStatus, TaskPriority
|
||||
from app.models.models import Comment
|
||||
|
||||
milestone = Milestone(title="M1", project_id=project.id, status=MilestoneStatus.OPEN)
|
||||
db.add(milestone)
|
||||
db.commit()
|
||||
|
||||
task = Task(
|
||||
title="Test Task", project_id=project.id, milestone_id=milestone.id,
|
||||
reporter_id=user1.id, status=TaskStatus.OPEN, priority=TaskPriority.MEDIUM
|
||||
)
|
||||
db.add(task)
|
||||
db.commit()
|
||||
|
||||
comment = Comment(content="User1's comment", task_id=task.id, author_id=user1.id)
|
||||
db.add(comment)
|
||||
db.commit()
|
||||
|
||||
resp = client.patch(
|
||||
f"/projects/{project.id}/tasks/{task.id}/comments/{comment.id}",
|
||||
json={"content": "Hacked!"},
|
||||
headers=auth_header(user2)
|
||||
)
|
||||
assert resp.status_code == 403
|
||||
358
tests/test_milestone_actions.py
Normal file
358
tests/test_milestone_actions.py
Normal file
@@ -0,0 +1,358 @@
|
||||
"""P13.1 — Milestone state-machine action tests.
|
||||
|
||||
Covers:
|
||||
- freeze: success, missing release task, multiple release tasks, wrong status
|
||||
- start: success + started_at, deps not met, wrong status
|
||||
- close: from open/freeze/undergoing, wrong status (completed/closed)
|
||||
- auto-complete: release task completion triggers milestone completed
|
||||
"""
|
||||
import json
|
||||
|
||||
import pytest
|
||||
from app.models.milestone import MilestoneStatus
|
||||
from app.models.task import TaskStatus
|
||||
|
||||
|
||||
# -----------------------------------------------------------------------
|
||||
# Freeze
|
||||
# -----------------------------------------------------------------------
|
||||
|
||||
class TestFreeze:
|
||||
"""POST /projects/{pid}/milestones/{mid}/actions/freeze"""
|
||||
|
||||
def test_freeze_success(
|
||||
self, client, db, make_user, make_project, make_milestone,
|
||||
make_task, seed_roles_and_permissions, make_member, auth_header,
|
||||
):
|
||||
admin_role, mgr_role, _ = seed_roles_and_permissions
|
||||
user = make_user(is_admin=False)
|
||||
project = make_project(owner_id=user.id)
|
||||
make_member(project.id, user.id, mgr_role.id)
|
||||
ms = make_milestone(project.id, user.id)
|
||||
|
||||
# Create exactly 1 maintenance/release task
|
||||
make_task(
|
||||
project.id, ms.id, user.id,
|
||||
task_type="maintenance", task_subtype="release",
|
||||
)
|
||||
|
||||
resp = client.post(
|
||||
f"/projects/{project.id}/milestones/{ms.id}/actions/freeze",
|
||||
headers=auth_header(user),
|
||||
)
|
||||
assert resp.status_code == 200, resp.text
|
||||
assert resp.json()["status"] == "freeze"
|
||||
|
||||
db.refresh(ms)
|
||||
assert ms.status == MilestoneStatus.FREEZE
|
||||
|
||||
def test_freeze_no_release_task(
|
||||
self, client, db, make_user, make_project, make_milestone,
|
||||
seed_roles_and_permissions, make_member, auth_header,
|
||||
):
|
||||
_, mgr_role, _ = seed_roles_and_permissions
|
||||
user = make_user()
|
||||
project = make_project(owner_id=user.id)
|
||||
make_member(project.id, user.id, mgr_role.id)
|
||||
ms = make_milestone(project.id, user.id)
|
||||
|
||||
resp = client.post(
|
||||
f"/projects/{project.id}/milestones/{ms.id}/actions/freeze",
|
||||
headers=auth_header(user),
|
||||
)
|
||||
assert resp.status_code == 400
|
||||
assert "no maintenance/release task" in resp.json()["detail"].lower()
|
||||
|
||||
def test_freeze_multiple_release_tasks(
|
||||
self, client, db, make_user, make_project, make_milestone,
|
||||
make_task, seed_roles_and_permissions, make_member, auth_header,
|
||||
):
|
||||
_, mgr_role, _ = seed_roles_and_permissions
|
||||
user = make_user()
|
||||
project = make_project(owner_id=user.id)
|
||||
make_member(project.id, user.id, mgr_role.id)
|
||||
ms = make_milestone(project.id, user.id)
|
||||
|
||||
make_task(project.id, ms.id, user.id, task_type="maintenance", task_subtype="release")
|
||||
make_task(project.id, ms.id, user.id, task_type="maintenance", task_subtype="release")
|
||||
|
||||
resp = client.post(
|
||||
f"/projects/{project.id}/milestones/{ms.id}/actions/freeze",
|
||||
headers=auth_header(user),
|
||||
)
|
||||
assert resp.status_code == 400
|
||||
assert "expected exactly 1" in resp.json()["detail"].lower()
|
||||
|
||||
def test_freeze_wrong_status(
|
||||
self, client, db, make_user, make_project, make_milestone,
|
||||
make_task, seed_roles_and_permissions, make_member, auth_header,
|
||||
):
|
||||
_, mgr_role, _ = seed_roles_and_permissions
|
||||
user = make_user()
|
||||
project = make_project(owner_id=user.id)
|
||||
make_member(project.id, user.id, mgr_role.id)
|
||||
ms = make_milestone(project.id, user.id, status=MilestoneStatus.CLOSED)
|
||||
|
||||
make_task(project.id, ms.id, user.id, task_type="maintenance", task_subtype="release")
|
||||
|
||||
resp = client.post(
|
||||
f"/projects/{project.id}/milestones/{ms.id}/actions/freeze",
|
||||
headers=auth_header(user),
|
||||
)
|
||||
assert resp.status_code == 400
|
||||
assert "expected 'open'" in resp.json()["detail"].lower()
|
||||
|
||||
|
||||
# -----------------------------------------------------------------------
|
||||
# Start
|
||||
# -----------------------------------------------------------------------
|
||||
|
||||
class TestStart:
|
||||
"""POST /projects/{pid}/milestones/{mid}/actions/start"""
|
||||
|
||||
def _freeze_milestone(self, db, ms):
|
||||
ms.status = MilestoneStatus.FREEZE
|
||||
db.commit()
|
||||
db.refresh(ms)
|
||||
|
||||
def test_start_success(
|
||||
self, client, db, make_user, make_project, make_milestone,
|
||||
make_task, seed_roles_and_permissions, make_member, auth_header,
|
||||
):
|
||||
_, mgr_role, _ = seed_roles_and_permissions
|
||||
user = make_user()
|
||||
project = make_project(owner_id=user.id)
|
||||
make_member(project.id, user.id, mgr_role.id)
|
||||
ms = make_milestone(project.id, user.id)
|
||||
self._freeze_milestone(db, ms)
|
||||
|
||||
resp = client.post(
|
||||
f"/projects/{project.id}/milestones/{ms.id}/actions/start",
|
||||
headers=auth_header(user),
|
||||
)
|
||||
assert resp.status_code == 200, resp.text
|
||||
data = resp.json()
|
||||
assert data["status"] == "undergoing"
|
||||
assert "started_at" in data
|
||||
|
||||
db.refresh(ms)
|
||||
assert ms.status == MilestoneStatus.UNDERGOING
|
||||
assert ms.started_at is not None
|
||||
|
||||
def test_start_deps_not_met(
|
||||
self, client, db, make_user, make_project, make_milestone,
|
||||
make_task, seed_roles_and_permissions, make_member, auth_header,
|
||||
):
|
||||
_, mgr_role, _ = seed_roles_and_permissions
|
||||
user = make_user()
|
||||
project = make_project(owner_id=user.id)
|
||||
make_member(project.id, user.id, mgr_role.id)
|
||||
|
||||
# Create a dependency milestone that is NOT completed
|
||||
dep_ms = make_milestone(project.id, user.id, status=MilestoneStatus.OPEN)
|
||||
|
||||
ms = make_milestone(
|
||||
project.id, user.id,
|
||||
depend_on_milestones=json.dumps([dep_ms.id]),
|
||||
)
|
||||
self._freeze_milestone(db, ms)
|
||||
|
||||
resp = client.post(
|
||||
f"/projects/{project.id}/milestones/{ms.id}/actions/start",
|
||||
headers=auth_header(user),
|
||||
)
|
||||
assert resp.status_code == 400
|
||||
assert "cannot start" in resp.json()["detail"].lower()
|
||||
|
||||
def test_start_wrong_status(
|
||||
self, client, db, make_user, make_project, make_milestone,
|
||||
seed_roles_and_permissions, make_member, auth_header,
|
||||
):
|
||||
_, mgr_role, _ = seed_roles_and_permissions
|
||||
user = make_user()
|
||||
project = make_project(owner_id=user.id)
|
||||
make_member(project.id, user.id, mgr_role.id)
|
||||
ms = make_milestone(project.id, user.id, status=MilestoneStatus.OPEN)
|
||||
|
||||
resp = client.post(
|
||||
f"/projects/{project.id}/milestones/{ms.id}/actions/start",
|
||||
headers=auth_header(user),
|
||||
)
|
||||
assert resp.status_code == 400
|
||||
assert "expected 'freeze'" in resp.json()["detail"].lower()
|
||||
|
||||
|
||||
# -----------------------------------------------------------------------
|
||||
# Close
|
||||
# -----------------------------------------------------------------------
|
||||
|
||||
class TestClose:
|
||||
"""POST /projects/{pid}/milestones/{mid}/actions/close"""
|
||||
|
||||
@pytest.mark.parametrize("initial_status", [
|
||||
MilestoneStatus.OPEN,
|
||||
MilestoneStatus.FREEZE,
|
||||
MilestoneStatus.UNDERGOING,
|
||||
])
|
||||
def test_close_from_allowed_statuses(
|
||||
self, client, db, make_user, make_project, make_milestone,
|
||||
seed_roles_and_permissions, make_member, auth_header, initial_status,
|
||||
):
|
||||
_, mgr_role, _ = seed_roles_and_permissions
|
||||
user = make_user()
|
||||
project = make_project(owner_id=user.id)
|
||||
make_member(project.id, user.id, mgr_role.id)
|
||||
ms = make_milestone(project.id, user.id, status=initial_status)
|
||||
|
||||
resp = client.post(
|
||||
f"/projects/{project.id}/milestones/{ms.id}/actions/close",
|
||||
headers=auth_header(user),
|
||||
json={"reason": "no longer needed"},
|
||||
)
|
||||
assert resp.status_code == 200, resp.text
|
||||
assert resp.json()["status"] == "closed"
|
||||
|
||||
db.refresh(ms)
|
||||
assert ms.status == MilestoneStatus.CLOSED
|
||||
|
||||
@pytest.mark.parametrize("terminal_status", [
|
||||
MilestoneStatus.COMPLETED,
|
||||
MilestoneStatus.CLOSED,
|
||||
])
|
||||
def test_close_from_terminal_rejected(
|
||||
self, client, db, make_user, make_project, make_milestone,
|
||||
seed_roles_and_permissions, make_member, auth_header, terminal_status,
|
||||
):
|
||||
_, mgr_role, _ = seed_roles_and_permissions
|
||||
user = make_user()
|
||||
project = make_project(owner_id=user.id)
|
||||
make_member(project.id, user.id, mgr_role.id)
|
||||
ms = make_milestone(project.id, user.id, status=terminal_status)
|
||||
|
||||
resp = client.post(
|
||||
f"/projects/{project.id}/milestones/{ms.id}/actions/close",
|
||||
headers=auth_header(user),
|
||||
)
|
||||
assert resp.status_code == 400
|
||||
|
||||
|
||||
# -----------------------------------------------------------------------
|
||||
# Auto-complete
|
||||
# -----------------------------------------------------------------------
|
||||
|
||||
class TestAutoComplete:
|
||||
"""When the sole release task is completed, milestone auto-completes."""
|
||||
|
||||
def test_auto_complete_on_release_task_finish(
|
||||
self, db, make_user, make_project, make_milestone, make_task,
|
||||
):
|
||||
"""Direct unit test of try_auto_complete_milestone."""
|
||||
from app.api.routers.milestone_actions import try_auto_complete_milestone
|
||||
|
||||
user = make_user()
|
||||
project = make_project(owner_id=user.id)
|
||||
ms = make_milestone(project.id, user.id, status=MilestoneStatus.UNDERGOING)
|
||||
|
||||
release_task = make_task(
|
||||
project.id, ms.id, user.id,
|
||||
task_type="maintenance", task_subtype="release",
|
||||
status=TaskStatus.COMPLETED,
|
||||
)
|
||||
|
||||
try_auto_complete_milestone(db, release_task, user_id=user.id)
|
||||
|
||||
db.refresh(ms)
|
||||
assert ms.status == MilestoneStatus.COMPLETED
|
||||
|
||||
def test_no_auto_complete_for_non_release_task(
|
||||
self, db, make_user, make_project, make_milestone, make_task,
|
||||
):
|
||||
from app.api.routers.milestone_actions import try_auto_complete_milestone
|
||||
|
||||
user = make_user()
|
||||
project = make_project(owner_id=user.id)
|
||||
ms = make_milestone(project.id, user.id, status=MilestoneStatus.UNDERGOING)
|
||||
|
||||
# Also add the required release task (still pending)
|
||||
make_task(
|
||||
project.id, ms.id, user.id,
|
||||
task_type="maintenance", task_subtype="release",
|
||||
status=TaskStatus.PENDING,
|
||||
)
|
||||
|
||||
normal_task = make_task(
|
||||
project.id, ms.id, user.id,
|
||||
task_type="issue", task_subtype="defect",
|
||||
status=TaskStatus.COMPLETED,
|
||||
)
|
||||
|
||||
try_auto_complete_milestone(db, normal_task, user_id=user.id)
|
||||
|
||||
db.refresh(ms)
|
||||
assert ms.status == MilestoneStatus.UNDERGOING # unchanged
|
||||
|
||||
def test_no_auto_complete_when_not_undergoing(
|
||||
self, db, make_user, make_project, make_milestone, make_task,
|
||||
):
|
||||
from app.api.routers.milestone_actions import try_auto_complete_milestone
|
||||
|
||||
user = make_user()
|
||||
project = make_project(owner_id=user.id)
|
||||
ms = make_milestone(project.id, user.id, status=MilestoneStatus.FREEZE)
|
||||
|
||||
release_task = make_task(
|
||||
project.id, ms.id, user.id,
|
||||
task_type="maintenance", task_subtype="release",
|
||||
status=TaskStatus.COMPLETED,
|
||||
)
|
||||
|
||||
try_auto_complete_milestone(db, release_task, user_id=user.id)
|
||||
|
||||
db.refresh(ms)
|
||||
assert ms.status == MilestoneStatus.FREEZE # unchanged
|
||||
|
||||
|
||||
# -----------------------------------------------------------------------
|
||||
# Preflight
|
||||
# -----------------------------------------------------------------------
|
||||
|
||||
class TestPreflight:
|
||||
"""GET /projects/{pid}/milestones/{mid}/actions/preflight"""
|
||||
|
||||
def test_preflight_freeze_allowed(
|
||||
self, client, db, make_user, make_project, make_milestone,
|
||||
make_task, seed_roles_and_permissions, make_member, auth_header,
|
||||
):
|
||||
_, mgr_role, _ = seed_roles_and_permissions
|
||||
user = make_user()
|
||||
project = make_project(owner_id=user.id)
|
||||
make_member(project.id, user.id, mgr_role.id)
|
||||
ms = make_milestone(project.id, user.id)
|
||||
make_task(project.id, ms.id, user.id, task_type="maintenance", task_subtype="release")
|
||||
|
||||
resp = client.get(
|
||||
f"/projects/{project.id}/milestones/{ms.id}/actions/preflight",
|
||||
headers=auth_header(user),
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert data["freeze"]["allowed"] is True
|
||||
|
||||
def test_preflight_freeze_not_allowed(
|
||||
self, client, db, make_user, make_project, make_milestone,
|
||||
seed_roles_and_permissions, make_member, auth_header,
|
||||
):
|
||||
_, mgr_role, _ = seed_roles_and_permissions
|
||||
user = make_user()
|
||||
project = make_project(owner_id=user.id)
|
||||
make_member(project.id, user.id, mgr_role.id)
|
||||
ms = make_milestone(project.id, user.id)
|
||||
# No release task
|
||||
|
||||
resp = client.get(
|
||||
f"/projects/{project.id}/milestones/{ms.id}/actions/preflight",
|
||||
headers=auth_header(user),
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert data["freeze"]["allowed"] is False
|
||||
148
tests/test_milestones.py
Normal file
148
tests/test_milestones.py
Normal file
@@ -0,0 +1,148 @@
|
||||
"""P14.1 — Milestones CRUD API tests.
|
||||
|
||||
Covers:
|
||||
- List milestones (project-scoped)
|
||||
- Get milestone by ID
|
||||
- Create milestone
|
||||
- Update milestone
|
||||
- Delete milestone
|
||||
- Milestone filtering and sorting
|
||||
"""
|
||||
import pytest
|
||||
from datetime import datetime, timedelta
|
||||
|
||||
|
||||
class TestMilestonesCRUD:
|
||||
"""Milestone CRUD endpoints."""
|
||||
|
||||
def test_list_milestones(self, client, db, make_user, make_project, auth_header, seed_roles_and_permissions, make_member):
|
||||
"""List milestones for a project."""
|
||||
admin_role, mgr_role, dev_role = seed_roles_and_permissions
|
||||
user = make_user()
|
||||
project = make_project()
|
||||
make_member(project.id, user.id, dev_role.id)
|
||||
|
||||
# Create milestones
|
||||
from app.models.milestone import Milestone, MilestoneStatus
|
||||
milestone1 = Milestone(title="Milestone 1", project_id=project.id, status=MilestoneStatus.OPEN)
|
||||
milestone2 = Milestone(title="Milestone 2", project_id=project.id, status=MilestoneStatus.OPEN)
|
||||
db.add_all([milestone1, milestone2])
|
||||
db.commit()
|
||||
|
||||
resp = client.get(f"/projects/{project.id}/milestones", headers=auth_header(user))
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert len(data) == 2
|
||||
|
||||
def test_get_milestone_by_id(self, client, db, make_user, make_project, auth_header, seed_roles_and_permissions, make_member):
|
||||
"""Get specific milestone."""
|
||||
admin_role, mgr_role, dev_role = seed_roles_and_permissions
|
||||
user = make_user()
|
||||
project = make_project()
|
||||
make_member(project.id, user.id, dev_role.id)
|
||||
|
||||
from app.models.milestone import Milestone, MilestoneStatus
|
||||
milestone = Milestone(
|
||||
title="Test Milestone",
|
||||
description="Test desc",
|
||||
project_id=project.id,
|
||||
status=MilestoneStatus.OPEN
|
||||
)
|
||||
db.add(milestone)
|
||||
db.commit()
|
||||
|
||||
resp = client.get(
|
||||
f"/projects/{project.id}/milestones/{milestone.id}",
|
||||
headers=auth_header(user)
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert data["id"] == milestone.id
|
||||
assert data["title"] == "Test Milestone"
|
||||
|
||||
def test_create_milestone(self, client, db, make_user, make_project, auth_header, seed_roles_and_permissions, make_member):
|
||||
"""Create new milestone."""
|
||||
admin_role, mgr_role, dev_role = seed_roles_and_permissions
|
||||
user = make_user()
|
||||
project = make_project(project_code="PROJ")
|
||||
make_member(project.id, user.id, dev_role.id)
|
||||
|
||||
due_date = (datetime.now() + timedelta(days=30)).isoformat()
|
||||
resp = client.post(
|
||||
f"/projects/{project.id}/milestones",
|
||||
json={
|
||||
"title": "New Milestone",
|
||||
"description": "Milestone description",
|
||||
"due_date": due_date
|
||||
},
|
||||
headers=auth_header(user)
|
||||
)
|
||||
assert resp.status_code == 201
|
||||
data = resp.json()
|
||||
assert data["title"] == "New Milestone"
|
||||
assert data["status"] == "open"
|
||||
assert data["milestone_code"].startswith("PROJ:")
|
||||
|
||||
def test_update_milestone(self, client, db, make_user, make_project, auth_header, seed_roles_and_permissions, make_member):
|
||||
"""Update milestone (allowed in open status)."""
|
||||
admin_role, mgr_role, dev_role = seed_roles_and_permissions
|
||||
user = make_user()
|
||||
project = make_project()
|
||||
make_member(project.id, user.id, dev_role.id)
|
||||
|
||||
from app.models.milestone import Milestone, MilestoneStatus
|
||||
milestone = Milestone(
|
||||
title="Old Title",
|
||||
project_id=project.id,
|
||||
status=MilestoneStatus.OPEN
|
||||
)
|
||||
db.add(milestone)
|
||||
db.commit()
|
||||
|
||||
resp = client.patch(
|
||||
f"/projects/{project.id}/milestones/{milestone.id}",
|
||||
json={"title": "Updated Title"},
|
||||
headers=auth_header(user)
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert data["title"] == "Updated Title"
|
||||
|
||||
def test_delete_milestone(self, client, db, make_user, make_project, auth_header, seed_roles_and_permissions, make_member):
|
||||
"""Delete milestone."""
|
||||
admin_role, mgr_role, dev_role = seed_roles_and_permissions
|
||||
user = make_user()
|
||||
project = make_project()
|
||||
make_member(project.id, user.id, dev_role.id)
|
||||
|
||||
from app.models.milestone import Milestone, MilestoneStatus
|
||||
milestone = Milestone(title="To Delete", project_id=project.id, status=MilestoneStatus.OPEN)
|
||||
db.add(milestone)
|
||||
db.commit()
|
||||
|
||||
resp = client.delete(
|
||||
f"/projects/{project.id}/milestones/{milestone.id}",
|
||||
headers=auth_header(user)
|
||||
)
|
||||
assert resp.status_code == 204
|
||||
|
||||
def test_milestone_status_filter(self, client, db, make_user, make_project, auth_header, seed_roles_and_permissions, make_member):
|
||||
"""Filter milestones by status."""
|
||||
admin_role, mgr_role, dev_role = seed_roles_and_permissions
|
||||
user = make_user()
|
||||
project = make_project()
|
||||
make_member(project.id, user.id, dev_role.id)
|
||||
|
||||
from app.models.milestone import Milestone, MilestoneStatus
|
||||
open_ms = Milestone(title="Open", project_id=project.id, status=MilestoneStatus.OPEN)
|
||||
closed_ms = Milestone(title="Closed", project_id=project.id, status=MilestoneStatus.CLOSED)
|
||||
db.add_all([open_ms, closed_ms])
|
||||
db.commit()
|
||||
|
||||
resp = client.get(
|
||||
f"/projects/{project.id}/milestones?status=open",
|
||||
headers=auth_header(user)
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert all(m["status"] == "open" for m in data)
|
||||
264
tests/test_misc.py
Normal file
264
tests/test_misc.py
Normal file
@@ -0,0 +1,264 @@
|
||||
"""P14.1 — Misc API tests.
|
||||
|
||||
Covers:
|
||||
- Milestones global list
|
||||
- Notifications
|
||||
- Activity log
|
||||
- API Keys
|
||||
- Webhooks
|
||||
- Export
|
||||
- Dashboard stats
|
||||
- Health check
|
||||
"""
|
||||
import pytest
|
||||
|
||||
|
||||
class TestMilestonesGlobal:
|
||||
"""Global milestones endpoints."""
|
||||
|
||||
def test_list_all_milestones(self, client, db, make_user, auth_header):
|
||||
"""List all milestones (global endpoint)."""
|
||||
user = make_user()
|
||||
|
||||
resp = client.get("/milestones", headers=auth_header(user))
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert isinstance(data, list)
|
||||
|
||||
def test_list_milestones_with_project_filter(self, client, db, make_user, make_project, auth_header, seed_roles_and_permissions, make_member):
|
||||
"""Filter milestones by project."""
|
||||
admin_role, mgr_role, dev_role = seed_roles_and_permissions
|
||||
user = make_user()
|
||||
project = make_project()
|
||||
make_member(project.id, user.id, dev_role.id)
|
||||
|
||||
from app.models.milestone import Milestone, MilestoneStatus
|
||||
milestone = Milestone(title="Test", project_id=project.id, status=MilestoneStatus.OPEN)
|
||||
db.add(milestone)
|
||||
db.commit()
|
||||
|
||||
resp = client.get(f"/milestones?project_id={project.id}", headers=auth_header(user))
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert all(m["project_id"] == project.id for m in data)
|
||||
|
||||
def test_get_milestone_detail(self, client, db, make_user, make_project, auth_header, seed_roles_and_permissions, make_member):
|
||||
"""Get milestone by ID (global endpoint)."""
|
||||
admin_role, mgr_role, dev_role = seed_roles_and_permissions
|
||||
user = make_user()
|
||||
project = make_project()
|
||||
make_member(project.id, user.id, dev_role.id)
|
||||
|
||||
from app.models.milestone import Milestone, MilestoneStatus
|
||||
milestone = Milestone(title="Test", project_id=project.id, status=MilestoneStatus.OPEN)
|
||||
db.add(milestone)
|
||||
db.commit()
|
||||
|
||||
resp = client.get(f"/milestones/{milestone.id}", headers=auth_header(user))
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert data["id"] == milestone.id
|
||||
|
||||
def test_milestone_progress(self, client, db, make_user, make_project, auth_header, seed_roles_and_permissions, make_member):
|
||||
"""Get milestone progress."""
|
||||
admin_role, mgr_role, dev_role = seed_roles_and_permissions
|
||||
user = make_user()
|
||||
project = make_project()
|
||||
make_member(project.id, user.id, dev_role.id)
|
||||
|
||||
from app.models.milestone import Milestone, MilestoneStatus
|
||||
from app.models.task import Task, TaskStatus, TaskPriority
|
||||
|
||||
milestone = Milestone(title="Test", project_id=project.id, status=MilestoneStatus.OPEN)
|
||||
db.add(milestone)
|
||||
db.commit()
|
||||
|
||||
# Add tasks
|
||||
task1 = Task(
|
||||
title="Done", project_id=project.id, milestone_id=milestone.id,
|
||||
reporter_id=user.id, status=TaskStatus.CLOSED, priority=TaskPriority.MEDIUM
|
||||
)
|
||||
task2 = Task(
|
||||
title="Open", project_id=project.id, milestone_id=milestone.id,
|
||||
reporter_id=user.id, status=TaskStatus.OPEN, priority=TaskPriority.MEDIUM
|
||||
)
|
||||
db.add_all([task1, task2])
|
||||
db.commit()
|
||||
|
||||
resp = client.get(f"/milestones/{milestone.id}/progress", headers=auth_header(user))
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert "total_issues" in data
|
||||
assert "completed" in data
|
||||
assert "progress_pct" in data
|
||||
|
||||
|
||||
class TestNotifications:
|
||||
"""Notifications endpoints."""
|
||||
|
||||
def test_list_notifications(self, client, db, make_user, auth_header):
|
||||
"""List user notifications."""
|
||||
user = make_user()
|
||||
|
||||
resp = client.get(f"/notifications?user_id={user.id}", headers=auth_header(user))
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert isinstance(data, list)
|
||||
|
||||
def test_notification_count(self, client, db, make_user, auth_header):
|
||||
"""Get unread notification count."""
|
||||
user = make_user()
|
||||
|
||||
resp = client.get(f"/notifications/count?user_id={user.id}", headers=auth_header(user))
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert "unread" in data
|
||||
assert data["user_id"] == user.id
|
||||
|
||||
def test_mark_notification_read(self, client, db, make_user, auth_header):
|
||||
"""Mark notification as read."""
|
||||
user = make_user()
|
||||
|
||||
from app.models.notification import Notification
|
||||
notification = Notification(
|
||||
user_id=user.id,
|
||||
type="test",
|
||||
title="Test",
|
||||
message="Test message",
|
||||
is_read=False
|
||||
)
|
||||
db.add(notification)
|
||||
db.commit()
|
||||
|
||||
resp = client.post(f"/notifications/{notification.id}/read", headers=auth_header(user))
|
||||
assert resp.status_code == 200
|
||||
assert resp.json()["status"] == "read"
|
||||
|
||||
|
||||
class TestActivityLog:
|
||||
"""Activity log endpoints."""
|
||||
|
||||
def test_list_activity(self, client, db, make_user, auth_header):
|
||||
"""List activity logs."""
|
||||
user = make_user()
|
||||
|
||||
resp = client.get("/activity", headers=auth_header(user))
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert isinstance(data, list)
|
||||
|
||||
def test_list_activity_with_filters(self, client, db, make_user, make_project, auth_header, seed_roles_and_permissions, make_member):
|
||||
"""Filter activity by entity."""
|
||||
admin_role, mgr_role, dev_role = seed_roles_and_permissions
|
||||
user = make_user()
|
||||
project = make_project()
|
||||
|
||||
from app.models.activity import ActivityLog
|
||||
activity = ActivityLog(
|
||||
action="create",
|
||||
entity_type="project",
|
||||
entity_id=project.id,
|
||||
user_id=user.id,
|
||||
details="Created project"
|
||||
)
|
||||
db.add(activity)
|
||||
db.commit()
|
||||
|
||||
resp = client.get(
|
||||
f"/activity?entity_type=project&entity_id={project.id}",
|
||||
headers=auth_header(user)
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert all(a["entity_type"] == "project" for a in data)
|
||||
|
||||
|
||||
class TestAPIKeys:
|
||||
"""API Key management."""
|
||||
|
||||
def test_create_api_key(self, client, db, make_user, auth_header):
|
||||
"""Create API key."""
|
||||
user = make_user()
|
||||
|
||||
resp = client.post(
|
||||
"/api-keys",
|
||||
json={"name": "Test Key", "user_id": user.id},
|
||||
headers=auth_header(user)
|
||||
)
|
||||
assert resp.status_code == 201
|
||||
data = resp.json()
|
||||
assert data["name"] == "Test Key"
|
||||
assert "key" in data
|
||||
|
||||
def test_list_api_keys(self, client, db, make_user, auth_header):
|
||||
"""List API keys."""
|
||||
user = make_user()
|
||||
|
||||
resp = client.get("/api-keys", headers=auth_header(user))
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert isinstance(data, list)
|
||||
|
||||
def test_revoke_api_key(self, client, db, make_user, auth_header):
|
||||
"""Revoke API key."""
|
||||
user = make_user()
|
||||
|
||||
resp = client.post(
|
||||
"/api-keys",
|
||||
json={"name": "To Revoke", "user_id": user.id},
|
||||
headers=auth_header(user)
|
||||
)
|
||||
key_id = resp.json()["id"]
|
||||
|
||||
resp = client.delete(f"/api-keys/{key_id}", headers=auth_header(user))
|
||||
assert resp.status_code == 204
|
||||
|
||||
|
||||
class TestDashboard:
|
||||
"""Dashboard stats."""
|
||||
|
||||
def test_dashboard_stats(self, client, db, make_user, make_project, auth_header, seed_roles_and_permissions, make_member):
|
||||
"""Get dashboard statistics."""
|
||||
admin_role, mgr_role, dev_role = seed_roles_and_permissions
|
||||
user = make_user()
|
||||
project = make_project()
|
||||
make_member(project.id, user.id, dev_role.id)
|
||||
|
||||
resp = client.get("/dashboard/stats", headers=auth_header(user))
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert "total" in data
|
||||
assert "by_status" in data
|
||||
assert "by_type" in data
|
||||
assert "by_priority" in data
|
||||
|
||||
def test_dashboard_stats_with_project_filter(self, client, db, make_user, make_project, auth_header, seed_roles_and_permissions, make_member):
|
||||
"""Get dashboard stats for specific project."""
|
||||
admin_role, mgr_role, dev_role = seed_roles_and_permissions
|
||||
user = make_user()
|
||||
project = make_project()
|
||||
make_member(project.id, user.id, dev_role.id)
|
||||
|
||||
resp = client.get(f"/dashboard/stats?project_id={project.id}", headers=auth_header(user))
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert "total" in data
|
||||
|
||||
|
||||
class TestHealth:
|
||||
"""Health check."""
|
||||
|
||||
def test_health_check(self, client):
|
||||
"""Health endpoint returns ok."""
|
||||
resp = client.get("/health")
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert data["status"] == "healthy"
|
||||
|
||||
def test_version(self, client):
|
||||
"""Version endpoint."""
|
||||
resp = client.get("/version")
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert "version" in data
|
||||
assert "name" in data
|
||||
108
tests/test_projects.py
Normal file
108
tests/test_projects.py
Normal file
@@ -0,0 +1,108 @@
|
||||
"""P14.1 — Projects API tests.
|
||||
|
||||
Covers:
|
||||
- List projects
|
||||
- Get project by ID
|
||||
- Create project
|
||||
- Update project
|
||||
- Delete project
|
||||
- Project ownership and permissions
|
||||
"""
|
||||
import pytest
|
||||
|
||||
|
||||
class TestProjects:
|
||||
"""Project management endpoints."""
|
||||
|
||||
def test_list_projects(self, client, db, make_user, make_project, auth_header, seed_roles_and_permissions):
|
||||
"""User can list projects they have access to."""
|
||||
admin_role, mgr_role, dev_role = seed_roles_and_permissions
|
||||
user = make_user()
|
||||
project1 = make_project(name="Project 1")
|
||||
project2 = make_project(name="Project 2")
|
||||
|
||||
resp = client.get("/projects", headers=auth_header(user))
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert isinstance(data, list)
|
||||
|
||||
def test_get_project_by_id(self, client, db, make_user, make_project, auth_header, seed_roles_and_permissions):
|
||||
"""Get specific project details."""
|
||||
admin_role, mgr_role, dev_role = seed_roles_and_permissions
|
||||
user = make_user()
|
||||
project = make_project(name="Test Project", owner_id=user.id)
|
||||
|
||||
resp = client.get(f"/projects/{project.id}", headers=auth_header(user))
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert data["id"] == project.id
|
||||
assert data["name"] == "Test Project"
|
||||
|
||||
def test_create_project(self, client, db, make_user, auth_header, seed_roles_and_permissions):
|
||||
"""User can create project."""
|
||||
admin_role, mgr_role, dev_role = seed_roles_and_permissions
|
||||
user = make_user()
|
||||
|
||||
resp = client.post(
|
||||
"/projects",
|
||||
json={
|
||||
"name": "New Project",
|
||||
"description": "Test description",
|
||||
"project_code": "TEST"
|
||||
},
|
||||
headers=auth_header(user)
|
||||
)
|
||||
assert resp.status_code == 201
|
||||
data = resp.json()
|
||||
assert data["name"] == "New Project"
|
||||
assert data["project_code"] == "TEST"
|
||||
assert "id" in data
|
||||
|
||||
def test_update_project(self, client, db, make_user, make_project, auth_header, seed_roles_and_permissions):
|
||||
"""Project owner can update project."""
|
||||
admin_role, mgr_role, dev_role = seed_roles_and_permissions
|
||||
user = make_user()
|
||||
project = make_project(name="Old Name", owner_id=user.id)
|
||||
|
||||
resp = client.patch(
|
||||
f"/projects/{project.id}",
|
||||
json={"name": "Updated Name"},
|
||||
headers=auth_header(user)
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert data["name"] == "Updated Name"
|
||||
|
||||
def test_delete_project(self, client, db, make_user, make_project, auth_header, seed_roles_and_permissions):
|
||||
"""Project owner can delete project."""
|
||||
admin_role, mgr_role, dev_role = seed_roles_and_permissions
|
||||
user = make_user()
|
||||
project = make_project(owner_id=user.id)
|
||||
|
||||
resp = client.delete(f"/projects/{project.id}", headers=auth_header(user))
|
||||
assert resp.status_code == 204
|
||||
|
||||
def test_non_owner_cannot_delete_project(self, client, db, make_user, make_project, auth_header, seed_roles_and_permissions, make_member):
|
||||
"""Non-owner cannot delete project."""
|
||||
admin_role, mgr_role, dev_role = seed_roles_and_permissions
|
||||
owner = make_user(username="owner")
|
||||
other = make_user(username="other")
|
||||
project = make_project(owner_id=owner.id)
|
||||
make_member(project.id, other.id, dev_role.id)
|
||||
|
||||
resp = client.delete(f"/projects/{project.id}", headers=auth_header(other))
|
||||
assert resp.status_code == 403
|
||||
|
||||
def test_project_code_generation(self, client, db, make_user, auth_header, seed_roles_and_permissions):
|
||||
"""Project code is auto-generated if not provided."""
|
||||
admin_role, mgr_role, dev_role = seed_roles_and_permissions
|
||||
user = make_user()
|
||||
|
||||
resp = client.post(
|
||||
"/projects",
|
||||
json={"name": "Auto Code Project"},
|
||||
headers=auth_header(user)
|
||||
)
|
||||
assert resp.status_code == 201
|
||||
data = resp.json()
|
||||
assert data["project_code"].startswith("P")
|
||||
559
tests/test_propose.py
Normal file
559
tests/test_propose.py
Normal file
@@ -0,0 +1,559 @@
|
||||
"""P13.3 — Propose backend tests.
|
||||
|
||||
Covers:
|
||||
- CRUD: create, list, get, update
|
||||
- propose_code per-project incrementing
|
||||
- accept → auto-generate feature story task + feat_task_id
|
||||
- accept with non-open milestone → fail
|
||||
- reject → status change
|
||||
- rejected → reopen back to open
|
||||
- feat_task_id cannot be set manually
|
||||
- edit restrictions (only open proposes editable)
|
||||
- permission checks for accept/reject/reopen
|
||||
"""
|
||||
import pytest
|
||||
from app.models.milestone import MilestoneStatus
|
||||
from app.models.task import TaskStatus
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Helpers
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def _propose_url(project_id: int, propose_id: int | None = None) -> str:
|
||||
base = f"/projects/{project_id}/proposes"
|
||||
return f"{base}/{propose_id}" if propose_id else base
|
||||
|
||||
|
||||
# ===========================================================================
|
||||
# CRUD
|
||||
# ===========================================================================
|
||||
|
||||
class TestProposeCRUD:
|
||||
"""Basic create / list / get / update."""
|
||||
|
||||
def test_create_propose(
|
||||
self, client, db, make_user, make_project, seed_roles_and_permissions, make_member, auth_header,
|
||||
):
|
||||
admin_role, mgr_role, dev_role = seed_roles_and_permissions
|
||||
user = make_user()
|
||||
project = make_project(owner_id=user.id, project_code="PROJ")
|
||||
make_member(project.id, user.id, dev_role.id)
|
||||
|
||||
resp = client.post(
|
||||
_propose_url(project.id),
|
||||
json={"title": "New Feature Idea", "description": "Some details"},
|
||||
headers=auth_header(user),
|
||||
)
|
||||
assert resp.status_code == 201
|
||||
data = resp.json()
|
||||
assert data["title"] == "New Feature Idea"
|
||||
assert data["status"] == "open"
|
||||
assert data["propose_code"].startswith("PROJ:P")
|
||||
assert data["feat_task_id"] is None
|
||||
|
||||
def test_list_proposes(
|
||||
self, client, db, make_user, make_project, seed_roles_and_permissions, make_member, auth_header,
|
||||
):
|
||||
admin_role, mgr_role, dev_role = seed_roles_and_permissions
|
||||
user = make_user()
|
||||
project = make_project(owner_id=user.id)
|
||||
make_member(project.id, user.id, dev_role.id)
|
||||
|
||||
# Create two proposes
|
||||
client.post(_propose_url(project.id), json={"title": "P1"}, headers=auth_header(user))
|
||||
client.post(_propose_url(project.id), json={"title": "P2"}, headers=auth_header(user))
|
||||
|
||||
resp = client.get(_propose_url(project.id), headers=auth_header(user))
|
||||
assert resp.status_code == 200
|
||||
assert len(resp.json()) == 2
|
||||
|
||||
def test_get_propose(
|
||||
self, client, db, make_user, make_project, seed_roles_and_permissions, make_member, auth_header,
|
||||
):
|
||||
admin_role, mgr_role, dev_role = seed_roles_and_permissions
|
||||
user = make_user()
|
||||
project = make_project(owner_id=user.id)
|
||||
make_member(project.id, user.id, dev_role.id)
|
||||
|
||||
create_resp = client.post(_propose_url(project.id), json={"title": "P1"}, headers=auth_header(user))
|
||||
propose_id = create_resp.json()["id"]
|
||||
|
||||
resp = client.get(_propose_url(project.id, propose_id), headers=auth_header(user))
|
||||
assert resp.status_code == 200
|
||||
assert resp.json()["title"] == "P1"
|
||||
|
||||
def test_update_propose_open(
|
||||
self, client, db, make_user, make_project, seed_roles_and_permissions, make_member, auth_header,
|
||||
):
|
||||
admin_role, mgr_role, dev_role = seed_roles_and_permissions
|
||||
user = make_user()
|
||||
project = make_project(owner_id=user.id)
|
||||
make_member(project.id, user.id, dev_role.id)
|
||||
|
||||
create_resp = client.post(_propose_url(project.id), json={"title": "Old"}, headers=auth_header(user))
|
||||
propose_id = create_resp.json()["id"]
|
||||
|
||||
resp = client.patch(
|
||||
_propose_url(project.id, propose_id),
|
||||
json={"title": "New Title", "description": "Updated"},
|
||||
headers=auth_header(user),
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
assert resp.json()["title"] == "New Title"
|
||||
assert resp.json()["description"] == "Updated"
|
||||
|
||||
|
||||
# ===========================================================================
|
||||
# Propose Code
|
||||
# ===========================================================================
|
||||
|
||||
class TestProposeCode:
|
||||
"""P1.4 — propose_code increments per project independently."""
|
||||
|
||||
def test_code_increments_per_project(
|
||||
self, client, db, make_user, make_project, seed_roles_and_permissions, make_member, auth_header,
|
||||
):
|
||||
admin_role, mgr_role, dev_role = seed_roles_and_permissions
|
||||
user = make_user()
|
||||
proj_a = make_project(owner_id=user.id, project_code="ALPHA")
|
||||
proj_b = make_project(owner_id=user.id, project_code="BETA")
|
||||
make_member(proj_a.id, user.id, dev_role.id)
|
||||
make_member(proj_b.id, user.id, dev_role.id)
|
||||
|
||||
# Create 2 in ALPHA
|
||||
r1 = client.post(_propose_url(proj_a.id), json={"title": "A1"}, headers=auth_header(user))
|
||||
r2 = client.post(_propose_url(proj_a.id), json={"title": "A2"}, headers=auth_header(user))
|
||||
|
||||
# Create 1 in BETA
|
||||
r3 = client.post(_propose_url(proj_b.id), json={"title": "B1"}, headers=auth_header(user))
|
||||
|
||||
code1 = r1.json()["propose_code"]
|
||||
code2 = r2.json()["propose_code"]
|
||||
code3 = r3.json()["propose_code"]
|
||||
|
||||
assert code1.startswith("ALPHA:P")
|
||||
assert code2.startswith("ALPHA:P")
|
||||
assert code3.startswith("BETA:P")
|
||||
# They should be distinct
|
||||
assert code1 != code2
|
||||
|
||||
|
||||
# ===========================================================================
|
||||
# Accept
|
||||
# ===========================================================================
|
||||
|
||||
class TestAccept:
|
||||
"""P6.2 — accept propose → create feature story task."""
|
||||
|
||||
def test_accept_success(
|
||||
self, client, db, make_user, make_project, make_milestone, seed_roles_and_permissions, make_member, auth_header,
|
||||
):
|
||||
admin_role, mgr_role, dev_role = seed_roles_and_permissions
|
||||
mgr = make_user()
|
||||
project = make_project(owner_id=mgr.id)
|
||||
make_member(project.id, mgr.id, mgr_role.id)
|
||||
|
||||
ms = make_milestone(project.id, mgr.id, status=MilestoneStatus.OPEN)
|
||||
|
||||
create_resp = client.post(
|
||||
_propose_url(project.id),
|
||||
json={"title": "Cool Feature", "description": "Do something cool"},
|
||||
headers=auth_header(mgr),
|
||||
)
|
||||
propose_id = create_resp.json()["id"]
|
||||
|
||||
resp = client.post(
|
||||
_propose_url(project.id, propose_id) + "/accept",
|
||||
json={"milestone_id": ms.id},
|
||||
headers=auth_header(mgr),
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert data["status"] == "accepted"
|
||||
assert data["feat_task_id"] is not None
|
||||
|
||||
# Verify the generated task exists
|
||||
from app.models.task import Task
|
||||
task = db.query(Task).filter(Task.id == int(data["feat_task_id"])).first()
|
||||
assert task is not None
|
||||
assert task.title == "Cool Feature"
|
||||
assert task.description == "Do something cool"
|
||||
assert task.task_type == "story"
|
||||
assert task.task_subtype == "feature"
|
||||
task_status = task.status.value if hasattr(task.status, "value") else task.status
|
||||
assert task_status == "pending"
|
||||
assert task.milestone_id == ms.id
|
||||
|
||||
def test_accept_non_open_milestone_fails(
|
||||
self, client, db, make_user, make_project, make_milestone, seed_roles_and_permissions, make_member, auth_header,
|
||||
):
|
||||
admin_role, mgr_role, dev_role = seed_roles_and_permissions
|
||||
mgr = make_user()
|
||||
project = make_project(owner_id=mgr.id)
|
||||
make_member(project.id, mgr.id, mgr_role.id)
|
||||
|
||||
ms = make_milestone(project.id, mgr.id, status=MilestoneStatus.FREEZE)
|
||||
|
||||
create_resp = client.post(
|
||||
_propose_url(project.id),
|
||||
json={"title": "Feature X"},
|
||||
headers=auth_header(mgr),
|
||||
)
|
||||
propose_id = create_resp.json()["id"]
|
||||
|
||||
resp = client.post(
|
||||
_propose_url(project.id, propose_id) + "/accept",
|
||||
json={"milestone_id": ms.id},
|
||||
headers=auth_header(mgr),
|
||||
)
|
||||
assert resp.status_code == 400
|
||||
assert "open" in resp.json()["detail"].lower()
|
||||
|
||||
def test_accept_already_accepted_fails(
|
||||
self, client, db, make_user, make_project, make_milestone, seed_roles_and_permissions, make_member, auth_header,
|
||||
):
|
||||
admin_role, mgr_role, dev_role = seed_roles_and_permissions
|
||||
mgr = make_user()
|
||||
project = make_project(owner_id=mgr.id)
|
||||
make_member(project.id, mgr.id, mgr_role.id)
|
||||
|
||||
ms = make_milestone(project.id, mgr.id, status=MilestoneStatus.OPEN)
|
||||
|
||||
create_resp = client.post(
|
||||
_propose_url(project.id), json={"title": "F"}, headers=auth_header(mgr),
|
||||
)
|
||||
propose_id = create_resp.json()["id"]
|
||||
|
||||
# First accept
|
||||
client.post(
|
||||
_propose_url(project.id, propose_id) + "/accept",
|
||||
json={"milestone_id": ms.id},
|
||||
headers=auth_header(mgr),
|
||||
)
|
||||
|
||||
# Second accept should fail
|
||||
resp = client.post(
|
||||
_propose_url(project.id, propose_id) + "/accept",
|
||||
json={"milestone_id": ms.id},
|
||||
headers=auth_header(mgr),
|
||||
)
|
||||
assert resp.status_code == 400
|
||||
|
||||
def test_accept_auto_fills_feat_task_id(
|
||||
self, client, db, make_user, make_project, make_milestone, seed_roles_and_permissions, make_member, auth_header,
|
||||
):
|
||||
admin_role, mgr_role, dev_role = seed_roles_and_permissions
|
||||
mgr = make_user()
|
||||
project = make_project(owner_id=mgr.id)
|
||||
make_member(project.id, mgr.id, mgr_role.id)
|
||||
|
||||
ms = make_milestone(project.id, mgr.id, status=MilestoneStatus.OPEN)
|
||||
|
||||
create_resp = client.post(
|
||||
_propose_url(project.id), json={"title": "F"}, headers=auth_header(mgr),
|
||||
)
|
||||
propose_id = create_resp.json()["id"]
|
||||
|
||||
resp = client.post(
|
||||
_propose_url(project.id, propose_id) + "/accept",
|
||||
json={"milestone_id": ms.id},
|
||||
headers=auth_header(mgr),
|
||||
)
|
||||
data = resp.json()
|
||||
assert data["feat_task_id"] is not None
|
||||
|
||||
# Re-fetch to confirm persistence
|
||||
get_resp = client.get(_propose_url(project.id, propose_id), headers=auth_header(mgr))
|
||||
assert get_resp.json()["feat_task_id"] == data["feat_task_id"]
|
||||
|
||||
def test_accept_no_permission_fails(
|
||||
self, client, db, make_user, make_project, make_milestone, seed_roles_and_permissions, make_member, auth_header,
|
||||
):
|
||||
"""dev role should not have propose.accept permission."""
|
||||
admin_role, mgr_role, dev_role = seed_roles_and_permissions
|
||||
owner = make_user()
|
||||
dev_user = make_user()
|
||||
project = make_project(owner_id=owner.id)
|
||||
make_member(project.id, owner.id, mgr_role.id)
|
||||
make_member(project.id, dev_user.id, dev_role.id)
|
||||
|
||||
ms = make_milestone(project.id, owner.id, status=MilestoneStatus.OPEN)
|
||||
|
||||
# Dev creates the propose
|
||||
create_resp = client.post(
|
||||
_propose_url(project.id), json={"title": "F"}, headers=auth_header(dev_user),
|
||||
)
|
||||
propose_id = create_resp.json()["id"]
|
||||
|
||||
# Dev tries to accept — should fail
|
||||
resp = client.post(
|
||||
_propose_url(project.id, propose_id) + "/accept",
|
||||
json={"milestone_id": ms.id},
|
||||
headers=auth_header(dev_user),
|
||||
)
|
||||
assert resp.status_code == 403
|
||||
|
||||
|
||||
# ===========================================================================
|
||||
# Reject
|
||||
# ===========================================================================
|
||||
|
||||
class TestReject:
|
||||
"""P6.3 — reject propose."""
|
||||
|
||||
def test_reject_success(
|
||||
self, client, db, make_user, make_project, seed_roles_and_permissions, make_member, auth_header,
|
||||
):
|
||||
admin_role, mgr_role, dev_role = seed_roles_and_permissions
|
||||
mgr = make_user()
|
||||
project = make_project(owner_id=mgr.id)
|
||||
make_member(project.id, mgr.id, mgr_role.id)
|
||||
|
||||
create_resp = client.post(
|
||||
_propose_url(project.id), json={"title": "F"}, headers=auth_header(mgr),
|
||||
)
|
||||
propose_id = create_resp.json()["id"]
|
||||
|
||||
resp = client.post(
|
||||
_propose_url(project.id, propose_id) + "/reject",
|
||||
json={"reason": "Not needed"},
|
||||
headers=auth_header(mgr),
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
assert resp.json()["status"] == "rejected"
|
||||
|
||||
def test_reject_non_open_fails(
|
||||
self, client, db, make_user, make_project, make_milestone, seed_roles_and_permissions, make_member, auth_header,
|
||||
):
|
||||
admin_role, mgr_role, dev_role = seed_roles_and_permissions
|
||||
mgr = make_user()
|
||||
project = make_project(owner_id=mgr.id)
|
||||
make_member(project.id, mgr.id, mgr_role.id)
|
||||
|
||||
ms = make_milestone(project.id, mgr.id, status=MilestoneStatus.OPEN)
|
||||
|
||||
create_resp = client.post(
|
||||
_propose_url(project.id), json={"title": "F"}, headers=auth_header(mgr),
|
||||
)
|
||||
propose_id = create_resp.json()["id"]
|
||||
|
||||
# Accept first
|
||||
client.post(
|
||||
_propose_url(project.id, propose_id) + "/accept",
|
||||
json={"milestone_id": ms.id},
|
||||
headers=auth_header(mgr),
|
||||
)
|
||||
|
||||
# Now reject should fail
|
||||
resp = client.post(
|
||||
_propose_url(project.id, propose_id) + "/reject",
|
||||
json={"reason": "Changed mind"},
|
||||
headers=auth_header(mgr),
|
||||
)
|
||||
assert resp.status_code == 400
|
||||
|
||||
def test_reject_no_permission_fails(
|
||||
self, client, db, make_user, make_project, seed_roles_and_permissions, make_member, auth_header,
|
||||
):
|
||||
admin_role, mgr_role, dev_role = seed_roles_and_permissions
|
||||
owner = make_user()
|
||||
dev_user = make_user()
|
||||
project = make_project(owner_id=owner.id)
|
||||
make_member(project.id, owner.id, mgr_role.id)
|
||||
make_member(project.id, dev_user.id, dev_role.id)
|
||||
|
||||
create_resp = client.post(
|
||||
_propose_url(project.id), json={"title": "F"}, headers=auth_header(dev_user),
|
||||
)
|
||||
propose_id = create_resp.json()["id"]
|
||||
|
||||
resp = client.post(
|
||||
_propose_url(project.id, propose_id) + "/reject",
|
||||
json={"reason": "nah"},
|
||||
headers=auth_header(dev_user),
|
||||
)
|
||||
assert resp.status_code == 403
|
||||
|
||||
|
||||
# ===========================================================================
|
||||
# Reopen
|
||||
# ===========================================================================
|
||||
|
||||
class TestReopen:
|
||||
"""P6.4 — reopen rejected propose."""
|
||||
|
||||
def test_reopen_success(
|
||||
self, client, db, make_user, make_project, seed_roles_and_permissions, make_member, auth_header,
|
||||
):
|
||||
admin_role, mgr_role, dev_role = seed_roles_and_permissions
|
||||
mgr = make_user()
|
||||
project = make_project(owner_id=mgr.id)
|
||||
make_member(project.id, mgr.id, mgr_role.id)
|
||||
|
||||
create_resp = client.post(
|
||||
_propose_url(project.id), json={"title": "F"}, headers=auth_header(mgr),
|
||||
)
|
||||
propose_id = create_resp.json()["id"]
|
||||
|
||||
# Reject first
|
||||
client.post(
|
||||
_propose_url(project.id, propose_id) + "/reject",
|
||||
json={"reason": "wait"},
|
||||
headers=auth_header(mgr),
|
||||
)
|
||||
|
||||
# Reopen
|
||||
resp = client.post(
|
||||
_propose_url(project.id, propose_id) + "/reopen",
|
||||
headers=auth_header(mgr),
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
assert resp.json()["status"] == "open"
|
||||
|
||||
def test_reopen_non_rejected_fails(
|
||||
self, client, db, make_user, make_project, seed_roles_and_permissions, make_member, auth_header,
|
||||
):
|
||||
admin_role, mgr_role, dev_role = seed_roles_and_permissions
|
||||
mgr = make_user()
|
||||
project = make_project(owner_id=mgr.id)
|
||||
make_member(project.id, mgr.id, mgr_role.id)
|
||||
|
||||
create_resp = client.post(
|
||||
_propose_url(project.id), json={"title": "F"}, headers=auth_header(mgr),
|
||||
)
|
||||
propose_id = create_resp.json()["id"]
|
||||
|
||||
# Try reopen on open propose — should fail
|
||||
resp = client.post(
|
||||
_propose_url(project.id, propose_id) + "/reopen",
|
||||
headers=auth_header(mgr),
|
||||
)
|
||||
assert resp.status_code == 400
|
||||
|
||||
def test_reopen_no_permission_fails(
|
||||
self, client, db, make_user, make_project, seed_roles_and_permissions, make_member, auth_header,
|
||||
):
|
||||
admin_role, mgr_role, dev_role = seed_roles_and_permissions
|
||||
owner = make_user()
|
||||
dev_user = make_user()
|
||||
project = make_project(owner_id=owner.id)
|
||||
make_member(project.id, owner.id, mgr_role.id)
|
||||
make_member(project.id, dev_user.id, dev_role.id)
|
||||
|
||||
create_resp = client.post(
|
||||
_propose_url(project.id), json={"title": "F"}, headers=auth_header(dev_user),
|
||||
)
|
||||
propose_id = create_resp.json()["id"]
|
||||
|
||||
# Owner rejects
|
||||
client.post(
|
||||
_propose_url(project.id, propose_id) + "/reject",
|
||||
json={"reason": "nah"},
|
||||
headers=auth_header(owner),
|
||||
)
|
||||
|
||||
# Dev tries to reopen — should fail
|
||||
resp = client.post(
|
||||
_propose_url(project.id, propose_id) + "/reopen",
|
||||
headers=auth_header(dev_user),
|
||||
)
|
||||
assert resp.status_code == 403
|
||||
|
||||
|
||||
# ===========================================================================
|
||||
# feat_task_id protection
|
||||
# ===========================================================================
|
||||
|
||||
class TestFeatTaskIdProtection:
|
||||
"""P6.5 — feat_task_id is server-side only, cannot be set by client."""
|
||||
|
||||
def test_update_cannot_set_feat_task_id(
|
||||
self, client, db, make_user, make_project, seed_roles_and_permissions, make_member, auth_header,
|
||||
):
|
||||
admin_role, mgr_role, dev_role = seed_roles_and_permissions
|
||||
user = make_user()
|
||||
project = make_project(owner_id=user.id)
|
||||
make_member(project.id, user.id, dev_role.id)
|
||||
|
||||
create_resp = client.post(
|
||||
_propose_url(project.id), json={"title": "F"}, headers=auth_header(user),
|
||||
)
|
||||
propose_id = create_resp.json()["id"]
|
||||
|
||||
# Try to set feat_task_id via PATCH
|
||||
resp = client.patch(
|
||||
_propose_url(project.id, propose_id),
|
||||
json={"feat_task_id": "999"},
|
||||
headers=auth_header(user),
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
# feat_task_id should still be None (server ignores it)
|
||||
assert resp.json()["feat_task_id"] is None
|
||||
|
||||
|
||||
# ===========================================================================
|
||||
# Edit restrictions
|
||||
# ===========================================================================
|
||||
|
||||
class TestEditRestrictions:
|
||||
"""Propose editing is only allowed in open status."""
|
||||
|
||||
def test_edit_accepted_propose_fails(
|
||||
self, client, db, make_user, make_project, make_milestone, seed_roles_and_permissions, make_member, auth_header,
|
||||
):
|
||||
admin_role, mgr_role, dev_role = seed_roles_and_permissions
|
||||
mgr = make_user()
|
||||
project = make_project(owner_id=mgr.id)
|
||||
make_member(project.id, mgr.id, mgr_role.id)
|
||||
|
||||
ms = make_milestone(project.id, mgr.id, status=MilestoneStatus.OPEN)
|
||||
|
||||
create_resp = client.post(
|
||||
_propose_url(project.id), json={"title": "F"}, headers=auth_header(mgr),
|
||||
)
|
||||
propose_id = create_resp.json()["id"]
|
||||
|
||||
# Accept
|
||||
client.post(
|
||||
_propose_url(project.id, propose_id) + "/accept",
|
||||
json={"milestone_id": ms.id},
|
||||
headers=auth_header(mgr),
|
||||
)
|
||||
|
||||
# Try to edit
|
||||
resp = client.patch(
|
||||
_propose_url(project.id, propose_id),
|
||||
json={"title": "Changed"},
|
||||
headers=auth_header(mgr),
|
||||
)
|
||||
assert resp.status_code == 400
|
||||
assert "open" in resp.json()["detail"].lower()
|
||||
|
||||
def test_edit_rejected_propose_fails(
|
||||
self, client, db, make_user, make_project, seed_roles_and_permissions, make_member, auth_header,
|
||||
):
|
||||
admin_role, mgr_role, dev_role = seed_roles_and_permissions
|
||||
mgr = make_user()
|
||||
project = make_project(owner_id=mgr.id)
|
||||
make_member(project.id, mgr.id, mgr_role.id)
|
||||
|
||||
create_resp = client.post(
|
||||
_propose_url(project.id), json={"title": "F"}, headers=auth_header(mgr),
|
||||
)
|
||||
propose_id = create_resp.json()["id"]
|
||||
|
||||
# Reject
|
||||
client.post(
|
||||
_propose_url(project.id, propose_id) + "/reject",
|
||||
json={"reason": "no"},
|
||||
headers=auth_header(mgr),
|
||||
)
|
||||
|
||||
# Try to edit
|
||||
resp = client.patch(
|
||||
_propose_url(project.id, propose_id),
|
||||
json={"title": "Changed"},
|
||||
headers=auth_header(mgr),
|
||||
)
|
||||
assert resp.status_code == 400
|
||||
182
tests/test_roles.py
Normal file
182
tests/test_roles.py
Normal file
@@ -0,0 +1,182 @@
|
||||
"""P14.1 — Roles and Permissions API tests.
|
||||
|
||||
Covers:
|
||||
- List roles
|
||||
- Get role by ID
|
||||
- Create role
|
||||
- Update role
|
||||
- Delete role
|
||||
- Assign role to user
|
||||
- Check permissions
|
||||
"""
|
||||
import pytest
|
||||
|
||||
|
||||
class TestRoles:
|
||||
"""Role management endpoints."""
|
||||
|
||||
def test_list_roles(self, client, db, make_user, auth_header, seed_roles_and_permissions):
|
||||
"""List all roles."""
|
||||
admin_role, mgr_role, dev_role = seed_roles_and_permissions
|
||||
user = make_user()
|
||||
|
||||
resp = client.get("/roles", headers=auth_header(user))
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert len(data) >= 3 # admin, mgr, dev at minimum
|
||||
|
||||
def test_get_role_by_id(self, client, db, make_user, auth_header, seed_roles_and_permissions):
|
||||
"""Get specific role."""
|
||||
admin_role, mgr_role, dev_role = seed_roles_and_permissions
|
||||
user = make_user()
|
||||
|
||||
resp = client.get(f"/roles/{admin_role.id}", headers=auth_header(user))
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert data["id"] == admin_role.id
|
||||
assert "name" in data
|
||||
|
||||
def test_create_role(self, client, db, make_user, auth_header, seed_roles_and_permissions):
|
||||
"""Admin can create new role."""
|
||||
admin_role, mgr_role, dev_role = seed_roles_and_permissions
|
||||
admin = make_user(username="admin", is_admin=True)
|
||||
|
||||
resp = client.post(
|
||||
"/roles",
|
||||
json={
|
||||
"name": "tester",
|
||||
"description": "Test role",
|
||||
"is_global": False
|
||||
},
|
||||
headers=auth_header(admin)
|
||||
)
|
||||
assert resp.status_code == 201
|
||||
data = resp.json()
|
||||
assert data["name"] == "tester"
|
||||
|
||||
def test_update_role(self, client, db, make_user, auth_header, seed_roles_and_permissions):
|
||||
"""Admin can update role."""
|
||||
admin_role, mgr_role, dev_role = seed_roles_and_permissions
|
||||
admin = make_user(username="admin", is_admin=True)
|
||||
|
||||
resp = client.patch(
|
||||
f"/roles/{dev_role.id}",
|
||||
json={"description": "Updated description"},
|
||||
headers=auth_header(admin)
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert data["description"] == "Updated description"
|
||||
|
||||
def test_delete_role(self, client, db, make_user, auth_header, seed_roles_and_permissions):
|
||||
"""Admin can delete non-default role."""
|
||||
admin_role, mgr_role, dev_role = seed_roles_and_permissions
|
||||
admin = make_user(username="admin", is_admin=True)
|
||||
|
||||
# Create a role to delete
|
||||
resp = client.post(
|
||||
"/roles",
|
||||
json={"name": "temp-role", "description": "To delete"},
|
||||
headers=auth_header(admin)
|
||||
)
|
||||
role_id = resp.json()["id"]
|
||||
|
||||
resp = client.delete(f"/roles/{role_id}", headers=auth_header(admin))
|
||||
assert resp.status_code == 204
|
||||
|
||||
def test_cannot_delete_admin_role(self, client, db, make_user, auth_header, seed_roles_and_permissions):
|
||||
"""Cannot delete admin role."""
|
||||
admin_role, mgr_role, dev_role = seed_roles_and_permissions
|
||||
admin = make_user(username="admin", is_admin=True)
|
||||
|
||||
resp = client.delete(f"/roles/{admin_role.id}", headers=auth_header(admin))
|
||||
assert resp.status_code == 400
|
||||
|
||||
|
||||
class TestPermissions:
|
||||
"""Permission checking endpoints."""
|
||||
|
||||
def test_check_permission_true(self, client, db, make_user, make_project, auth_header, seed_roles_and_permissions, make_member):
|
||||
"""Check permission returns true when granted."""
|
||||
admin_role, mgr_role, dev_role = seed_roles_and_permissions
|
||||
user = make_user()
|
||||
project = make_project()
|
||||
make_member(project.id, user.id, dev_role.id)
|
||||
|
||||
# Dev should have view permission
|
||||
resp = client.get(
|
||||
f"/projects/{project.id}/check-permission?permission=view",
|
||||
headers=auth_header(user)
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert data["has_permission"] is True
|
||||
|
||||
def test_check_permission_false(self, client, db, make_user, make_project, auth_header, seed_roles_and_permissions, make_member):
|
||||
"""Check permission returns false when not granted."""
|
||||
admin_role, mgr_role, dev_role = seed_roles_and_permissions
|
||||
user = make_user()
|
||||
project = make_project()
|
||||
# Add as guest (viewer role)
|
||||
from app.models.role_permission import Role
|
||||
guest_role = db.query(Role).filter(Role.name == "guest").first()
|
||||
if not guest_role:
|
||||
guest_role = Role(name="guest", description="Guest", is_global=False)
|
||||
db.add(guest_role)
|
||||
db.commit()
|
||||
make_member(project.id, user.id, guest_role.id)
|
||||
|
||||
resp = client.get(
|
||||
f"/projects/{project.id}/check-permission?permission=admin",
|
||||
headers=auth_header(user)
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert data["has_permission"] is False
|
||||
|
||||
|
||||
class TestRoleAssignments:
|
||||
"""Role assignment endpoints."""
|
||||
|
||||
def test_assign_role_to_user(self, client, db, make_user, make_project, auth_header, seed_roles_and_permissions, make_member):
|
||||
"""Assign role to project member."""
|
||||
admin_role, mgr_role, dev_role = seed_roles_and_permissions
|
||||
admin = make_user(username="admin", is_admin=True)
|
||||
user = make_user(username="member")
|
||||
project = make_project()
|
||||
|
||||
resp = client.post(
|
||||
f"/projects/{project.id}/members",
|
||||
json={"user_id": user.id, "role_id": dev_role.id},
|
||||
headers=auth_header(admin)
|
||||
)
|
||||
assert resp.status_code == 201
|
||||
|
||||
def test_change_user_role(self, client, db, make_user, make_project, auth_header, seed_roles_and_permissions, make_member):
|
||||
"""Change user's role in project."""
|
||||
admin_role, mgr_role, dev_role = seed_roles_and_permissions
|
||||
admin = make_user(username="admin", is_admin=True)
|
||||
user = make_user(username="member")
|
||||
project = make_project()
|
||||
make_member(project.id, user.id, dev_role.id)
|
||||
|
||||
resp = client.patch(
|
||||
f"/projects/{project.id}/members/{user.id}",
|
||||
json={"role_id": mgr_role.id},
|
||||
headers=auth_header(admin)
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
|
||||
def test_remove_user_from_project(self, client, db, make_user, make_project, auth_header, seed_roles_and_permissions, make_member):
|
||||
"""Remove user from project."""
|
||||
admin_role, mgr_role, dev_role = seed_roles_and_permissions
|
||||
admin = make_user(username="admin", is_admin=True)
|
||||
user = make_user(username="member")
|
||||
project = make_project()
|
||||
make_member(project.id, user.id, dev_role.id)
|
||||
|
||||
resp = client.delete(
|
||||
f"/projects/{project.id}/members/{user.id}",
|
||||
headers=auth_header(admin)
|
||||
)
|
||||
assert resp.status_code == 204
|
||||
564
tests/test_task_transitions.py
Normal file
564
tests/test_task_transitions.py
Normal file
@@ -0,0 +1,564 @@
|
||||
"""P13.2 — Task state-machine transition tests.
|
||||
|
||||
Covers:
|
||||
- pending → open: success, milestone not undergoing, deps not met
|
||||
- open → undergoing: success, no assignee, non-assignee blocked
|
||||
- undergoing → completed: success with comment, no comment fails, non-assignee blocked
|
||||
- close from pending/open/undergoing: permission required
|
||||
- reopen from completed/closed → open: distinct permissions
|
||||
- invalid transitions: rejected by state machine
|
||||
- edit restrictions: P5.7 body edit guards by status/assignee
|
||||
"""
|
||||
import json
|
||||
|
||||
import pytest
|
||||
from app.models.milestone import MilestoneStatus
|
||||
from app.models.task import TaskStatus
|
||||
|
||||
|
||||
# -----------------------------------------------------------------------
|
||||
# Helpers
|
||||
# -----------------------------------------------------------------------
|
||||
|
||||
def _transition(client, task_id, new_status, headers, comment=None):
|
||||
"""POST /tasks/{id}/transition?new_status=..."""
|
||||
body = {}
|
||||
if comment is not None:
|
||||
body["comment"] = comment
|
||||
return client.post(
|
||||
f"/tasks/{task_id}/transition?new_status={new_status}",
|
||||
json=body,
|
||||
headers=headers,
|
||||
)
|
||||
|
||||
|
||||
# -----------------------------------------------------------------------
|
||||
# pending → open
|
||||
# -----------------------------------------------------------------------
|
||||
|
||||
class TestPendingToOpen:
|
||||
|
||||
def test_success(
|
||||
self, client, db, make_user, make_project, make_milestone,
|
||||
make_task, seed_roles_and_permissions, make_member, auth_header,
|
||||
):
|
||||
"""pending→open succeeds when milestone is undergoing and no deps."""
|
||||
admin_role, mgr_role, _ = seed_roles_and_permissions
|
||||
user = make_user()
|
||||
project = make_project(owner_id=user.id)
|
||||
make_member(project.id, user.id, mgr_role.id)
|
||||
ms = make_milestone(project.id, user.id, status=MilestoneStatus.UNDERGOING)
|
||||
task = make_task(project.id, ms.id, user.id, status=TaskStatus.PENDING)
|
||||
|
||||
resp = _transition(client, task.id, "open", auth_header(user))
|
||||
assert resp.status_code == 200, resp.text
|
||||
assert resp.json()["status"] == "open"
|
||||
|
||||
def test_milestone_not_undergoing(
|
||||
self, client, db, make_user, make_project, make_milestone,
|
||||
make_task, seed_roles_and_permissions, make_member, auth_header,
|
||||
):
|
||||
"""pending→open rejected when milestone is still open."""
|
||||
_, mgr_role, _ = seed_roles_and_permissions
|
||||
user = make_user()
|
||||
project = make_project(owner_id=user.id)
|
||||
make_member(project.id, user.id, mgr_role.id)
|
||||
ms = make_milestone(project.id, user.id, status=MilestoneStatus.OPEN)
|
||||
task = make_task(project.id, ms.id, user.id, status=TaskStatus.PENDING)
|
||||
|
||||
resp = _transition(client, task.id, "open", auth_header(user))
|
||||
assert resp.status_code == 400
|
||||
assert "undergoing" in resp.json()["detail"].lower()
|
||||
|
||||
def test_deps_not_satisfied(
|
||||
self, client, db, make_user, make_project, make_milestone,
|
||||
make_task, seed_roles_and_permissions, make_member, auth_header,
|
||||
):
|
||||
"""pending→open rejected when depend_on tasks are not completed."""
|
||||
_, mgr_role, _ = seed_roles_and_permissions
|
||||
user = make_user()
|
||||
project = make_project(owner_id=user.id)
|
||||
make_member(project.id, user.id, mgr_role.id)
|
||||
ms = make_milestone(project.id, user.id, status=MilestoneStatus.UNDERGOING)
|
||||
|
||||
dep_task = make_task(project.id, ms.id, user.id, status=TaskStatus.OPEN)
|
||||
task = make_task(
|
||||
project.id, ms.id, user.id,
|
||||
status=TaskStatus.PENDING,
|
||||
depend_on=json.dumps([dep_task.id]),
|
||||
)
|
||||
|
||||
resp = _transition(client, task.id, "open", auth_header(user))
|
||||
assert resp.status_code == 400
|
||||
assert "depend" in resp.json()["detail"].lower() or "block" in resp.json()["detail"].lower()
|
||||
|
||||
def test_deps_satisfied(
|
||||
self, client, db, make_user, make_project, make_milestone,
|
||||
make_task, seed_roles_and_permissions, make_member, auth_header,
|
||||
):
|
||||
"""pending→open succeeds when all depend_on tasks are completed."""
|
||||
_, mgr_role, _ = seed_roles_and_permissions
|
||||
user = make_user()
|
||||
project = make_project(owner_id=user.id)
|
||||
make_member(project.id, user.id, mgr_role.id)
|
||||
ms = make_milestone(project.id, user.id, status=MilestoneStatus.UNDERGOING)
|
||||
|
||||
dep_task = make_task(project.id, ms.id, user.id, status=TaskStatus.COMPLETED)
|
||||
task = make_task(
|
||||
project.id, ms.id, user.id,
|
||||
status=TaskStatus.PENDING,
|
||||
depend_on=json.dumps([dep_task.id]),
|
||||
)
|
||||
|
||||
resp = _transition(client, task.id, "open", auth_header(user))
|
||||
assert resp.status_code == 200
|
||||
assert resp.json()["status"] == "open"
|
||||
|
||||
|
||||
# -----------------------------------------------------------------------
|
||||
# open → undergoing
|
||||
# -----------------------------------------------------------------------
|
||||
|
||||
class TestOpenToUndergoing:
|
||||
|
||||
def test_success_assignee_starts(
|
||||
self, client, db, make_user, make_project, make_milestone,
|
||||
make_task, seed_roles_and_permissions, make_member, auth_header,
|
||||
):
|
||||
"""Assignee can start their own task."""
|
||||
_, mgr_role, _ = seed_roles_and_permissions
|
||||
user = make_user()
|
||||
project = make_project(owner_id=user.id)
|
||||
make_member(project.id, user.id, mgr_role.id)
|
||||
ms = make_milestone(project.id, user.id, status=MilestoneStatus.UNDERGOING)
|
||||
task = make_task(
|
||||
project.id, ms.id, user.id,
|
||||
status=TaskStatus.OPEN,
|
||||
assignee_id=user.id,
|
||||
)
|
||||
|
||||
resp = _transition(client, task.id, "undergoing", auth_header(user))
|
||||
assert resp.status_code == 200
|
||||
assert resp.json()["status"] == "undergoing"
|
||||
db.refresh(task)
|
||||
assert task.started_on is not None
|
||||
|
||||
def test_no_assignee_fails(
|
||||
self, client, db, make_user, make_project, make_milestone,
|
||||
make_task, seed_roles_and_permissions, make_member, auth_header,
|
||||
):
|
||||
"""Cannot start a task without an assignee."""
|
||||
_, mgr_role, _ = seed_roles_and_permissions
|
||||
user = make_user()
|
||||
project = make_project(owner_id=user.id)
|
||||
make_member(project.id, user.id, mgr_role.id)
|
||||
ms = make_milestone(project.id, user.id, status=MilestoneStatus.UNDERGOING)
|
||||
task = make_task(project.id, ms.id, user.id, status=TaskStatus.OPEN)
|
||||
|
||||
resp = _transition(client, task.id, "undergoing", auth_header(user))
|
||||
assert resp.status_code == 400
|
||||
assert "assignee" in resp.json()["detail"].lower()
|
||||
|
||||
def test_non_assignee_blocked(
|
||||
self, client, db, make_user, make_project, make_milestone,
|
||||
make_task, seed_roles_and_permissions, make_member, auth_header,
|
||||
):
|
||||
"""A different user cannot start someone else's task."""
|
||||
_, mgr_role, _ = seed_roles_and_permissions
|
||||
owner = make_user()
|
||||
other = make_user()
|
||||
project = make_project(owner_id=owner.id)
|
||||
make_member(project.id, owner.id, mgr_role.id)
|
||||
make_member(project.id, other.id, mgr_role.id)
|
||||
ms = make_milestone(project.id, owner.id, status=MilestoneStatus.UNDERGOING)
|
||||
task = make_task(
|
||||
project.id, ms.id, owner.id,
|
||||
status=TaskStatus.OPEN,
|
||||
assignee_id=owner.id,
|
||||
)
|
||||
|
||||
resp = _transition(client, task.id, "undergoing", auth_header(other))
|
||||
assert resp.status_code == 403
|
||||
assert "assigned" in resp.json()["detail"].lower()
|
||||
|
||||
|
||||
# -----------------------------------------------------------------------
|
||||
# undergoing → completed
|
||||
# -----------------------------------------------------------------------
|
||||
|
||||
class TestUndergoingToCompleted:
|
||||
|
||||
def test_success_with_comment(
|
||||
self, client, db, make_user, make_project, make_milestone,
|
||||
make_task, seed_roles_and_permissions, make_member, auth_header,
|
||||
):
|
||||
"""Assignee can complete a task with a completion comment."""
|
||||
_, mgr_role, _ = seed_roles_and_permissions
|
||||
user = make_user()
|
||||
project = make_project(owner_id=user.id)
|
||||
make_member(project.id, user.id, mgr_role.id)
|
||||
ms = make_milestone(project.id, user.id, status=MilestoneStatus.UNDERGOING)
|
||||
task = make_task(
|
||||
project.id, ms.id, user.id,
|
||||
status=TaskStatus.UNDERGOING,
|
||||
assignee_id=user.id,
|
||||
)
|
||||
|
||||
resp = _transition(client, task.id, "completed", auth_header(user), comment="Done!")
|
||||
assert resp.status_code == 200
|
||||
assert resp.json()["status"] == "completed"
|
||||
db.refresh(task)
|
||||
assert task.finished_on is not None
|
||||
|
||||
def test_no_comment_fails(
|
||||
self, client, db, make_user, make_project, make_milestone,
|
||||
make_task, seed_roles_and_permissions, make_member, auth_header,
|
||||
):
|
||||
"""Cannot complete without a comment."""
|
||||
_, mgr_role, _ = seed_roles_and_permissions
|
||||
user = make_user()
|
||||
project = make_project(owner_id=user.id)
|
||||
make_member(project.id, user.id, mgr_role.id)
|
||||
ms = make_milestone(project.id, user.id, status=MilestoneStatus.UNDERGOING)
|
||||
task = make_task(
|
||||
project.id, ms.id, user.id,
|
||||
status=TaskStatus.UNDERGOING,
|
||||
assignee_id=user.id,
|
||||
)
|
||||
|
||||
resp = _transition(client, task.id, "completed", auth_header(user))
|
||||
assert resp.status_code == 400
|
||||
assert "comment" in resp.json()["detail"].lower()
|
||||
|
||||
def test_empty_comment_fails(
|
||||
self, client, db, make_user, make_project, make_milestone,
|
||||
make_task, seed_roles_and_permissions, make_member, auth_header,
|
||||
):
|
||||
"""Empty/whitespace comment is rejected."""
|
||||
_, mgr_role, _ = seed_roles_and_permissions
|
||||
user = make_user()
|
||||
project = make_project(owner_id=user.id)
|
||||
make_member(project.id, user.id, mgr_role.id)
|
||||
ms = make_milestone(project.id, user.id, status=MilestoneStatus.UNDERGOING)
|
||||
task = make_task(
|
||||
project.id, ms.id, user.id,
|
||||
status=TaskStatus.UNDERGOING,
|
||||
assignee_id=user.id,
|
||||
)
|
||||
|
||||
resp = _transition(client, task.id, "completed", auth_header(user), comment=" ")
|
||||
assert resp.status_code == 400
|
||||
assert "comment" in resp.json()["detail"].lower()
|
||||
|
||||
def test_non_assignee_blocked(
|
||||
self, client, db, make_user, make_project, make_milestone,
|
||||
make_task, seed_roles_and_permissions, make_member, auth_header,
|
||||
):
|
||||
"""Non-assignee cannot complete the task."""
|
||||
_, mgr_role, _ = seed_roles_and_permissions
|
||||
owner = make_user()
|
||||
other = make_user()
|
||||
project = make_project(owner_id=owner.id)
|
||||
make_member(project.id, owner.id, mgr_role.id)
|
||||
make_member(project.id, other.id, mgr_role.id)
|
||||
ms = make_milestone(project.id, owner.id, status=MilestoneStatus.UNDERGOING)
|
||||
task = make_task(
|
||||
project.id, ms.id, owner.id,
|
||||
status=TaskStatus.UNDERGOING,
|
||||
assignee_id=owner.id,
|
||||
)
|
||||
|
||||
resp = _transition(client, task.id, "completed", auth_header(other), comment="I finished it")
|
||||
assert resp.status_code == 403
|
||||
|
||||
|
||||
# -----------------------------------------------------------------------
|
||||
# Close task (from various states)
|
||||
# -----------------------------------------------------------------------
|
||||
|
||||
class TestCloseTask:
|
||||
|
||||
@pytest.mark.parametrize("initial_status", [
|
||||
TaskStatus.PENDING,
|
||||
TaskStatus.OPEN,
|
||||
TaskStatus.UNDERGOING,
|
||||
])
|
||||
def test_close_from_valid_states(
|
||||
self, initial_status,
|
||||
client, db, make_user, make_project, make_milestone,
|
||||
make_task, seed_roles_and_permissions, make_member, auth_header,
|
||||
):
|
||||
"""Close is allowed from pending/open/undergoing with permission."""
|
||||
_, mgr_role, _ = seed_roles_and_permissions
|
||||
user = make_user()
|
||||
project = make_project(owner_id=user.id)
|
||||
make_member(project.id, user.id, mgr_role.id)
|
||||
ms = make_milestone(project.id, user.id, status=MilestoneStatus.UNDERGOING)
|
||||
task = make_task(project.id, ms.id, user.id, status=initial_status)
|
||||
|
||||
resp = _transition(client, task.id, "closed", auth_header(user))
|
||||
assert resp.status_code == 200, resp.text
|
||||
assert resp.json()["status"] == "closed"
|
||||
|
||||
@pytest.mark.parametrize("initial_status", [
|
||||
TaskStatus.COMPLETED,
|
||||
TaskStatus.CLOSED,
|
||||
])
|
||||
def test_close_from_terminal_states_fails(
|
||||
self, initial_status,
|
||||
client, db, make_user, make_project, make_milestone,
|
||||
make_task, seed_roles_and_permissions, make_member, auth_header,
|
||||
):
|
||||
"""Cannot close from completed or already closed."""
|
||||
_, mgr_role, _ = seed_roles_and_permissions
|
||||
user = make_user()
|
||||
project = make_project(owner_id=user.id)
|
||||
make_member(project.id, user.id, mgr_role.id)
|
||||
ms = make_milestone(project.id, user.id, status=MilestoneStatus.UNDERGOING)
|
||||
task = make_task(project.id, ms.id, user.id, status=initial_status)
|
||||
|
||||
resp = _transition(client, task.id, "closed", auth_header(user))
|
||||
assert resp.status_code == 400
|
||||
|
||||
def test_close_without_permission_fails(
|
||||
self, client, db, make_user, make_project, make_milestone,
|
||||
make_task, seed_roles_and_permissions, make_member, auth_header,
|
||||
):
|
||||
"""User without task.close permission cannot close."""
|
||||
from app.models.role_permission import Role
|
||||
_, _, dev_role = seed_roles_and_permissions
|
||||
|
||||
# Create a role with NO task.close permission
|
||||
no_close_role = Role(name="viewer", is_global=False)
|
||||
db.add(no_close_role)
|
||||
db.commit()
|
||||
|
||||
# Give viewer only basic perms (project.read, task.read)
|
||||
from app.models.role_permission import Permission, RolePermission
|
||||
for pname in ("project.read", "task.read"):
|
||||
p = db.query(Permission).filter(Permission.name == pname).first()
|
||||
if p:
|
||||
db.add(RolePermission(role_id=no_close_role.id, permission_id=p.id))
|
||||
db.commit()
|
||||
|
||||
user = make_user()
|
||||
project = make_project(owner_id=user.id)
|
||||
make_member(project.id, user.id, no_close_role.id)
|
||||
ms = make_milestone(project.id, user.id, status=MilestoneStatus.UNDERGOING)
|
||||
task = make_task(project.id, ms.id, user.id, status=TaskStatus.OPEN)
|
||||
|
||||
resp = _transition(client, task.id, "closed", auth_header(user))
|
||||
assert resp.status_code == 403
|
||||
|
||||
|
||||
# -----------------------------------------------------------------------
|
||||
# Reopen (completed → open, closed → open)
|
||||
# -----------------------------------------------------------------------
|
||||
|
||||
class TestReopen:
|
||||
|
||||
def test_reopen_completed(
|
||||
self, client, db, make_user, make_project, make_milestone,
|
||||
make_task, seed_roles_and_permissions, make_member, auth_header,
|
||||
):
|
||||
"""Reopen from completed → open with task.reopen_completed permission."""
|
||||
_, mgr_role, _ = seed_roles_and_permissions
|
||||
user = make_user()
|
||||
project = make_project(owner_id=user.id)
|
||||
make_member(project.id, user.id, mgr_role.id)
|
||||
ms = make_milestone(project.id, user.id, status=MilestoneStatus.UNDERGOING)
|
||||
task = make_task(project.id, ms.id, user.id, status=TaskStatus.COMPLETED)
|
||||
|
||||
resp = _transition(client, task.id, "open", auth_header(user))
|
||||
assert resp.status_code == 200
|
||||
assert resp.json()["status"] == "open"
|
||||
# finished_on should be cleared
|
||||
db.refresh(task)
|
||||
assert task.finished_on is None
|
||||
|
||||
def test_reopen_closed(
|
||||
self, client, db, make_user, make_project, make_milestone,
|
||||
make_task, seed_roles_and_permissions, make_member, auth_header,
|
||||
):
|
||||
"""Reopen from closed → open with task.reopen_closed permission."""
|
||||
_, mgr_role, _ = seed_roles_and_permissions
|
||||
user = make_user()
|
||||
project = make_project(owner_id=user.id)
|
||||
make_member(project.id, user.id, mgr_role.id)
|
||||
ms = make_milestone(project.id, user.id, status=MilestoneStatus.UNDERGOING)
|
||||
task = make_task(project.id, ms.id, user.id, status=TaskStatus.CLOSED)
|
||||
|
||||
resp = _transition(client, task.id, "open", auth_header(user))
|
||||
assert resp.status_code == 200
|
||||
assert resp.json()["status"] == "open"
|
||||
|
||||
def test_reopen_without_permission_fails(
|
||||
self, client, db, make_user, make_project, make_milestone,
|
||||
make_task, seed_roles_and_permissions, make_member, auth_header,
|
||||
):
|
||||
"""User without reopen permission cannot reopen."""
|
||||
from app.models.role_permission import Role, Permission, RolePermission
|
||||
|
||||
# Create a role with task.close but NO reopen permissions
|
||||
limited_role = Role(name="limited", is_global=False)
|
||||
db.add(limited_role)
|
||||
db.commit()
|
||||
for pname in ("project.read", "task.read", "task.write", "task.close"):
|
||||
p = db.query(Permission).filter(Permission.name == pname).first()
|
||||
if p:
|
||||
db.add(RolePermission(role_id=limited_role.id, permission_id=p.id))
|
||||
db.commit()
|
||||
|
||||
user = make_user()
|
||||
project = make_project(owner_id=user.id)
|
||||
make_member(project.id, user.id, limited_role.id)
|
||||
ms = make_milestone(project.id, user.id, status=MilestoneStatus.UNDERGOING)
|
||||
task = make_task(project.id, ms.id, user.id, status=TaskStatus.COMPLETED)
|
||||
|
||||
resp = _transition(client, task.id, "open", auth_header(user))
|
||||
assert resp.status_code == 403
|
||||
|
||||
|
||||
# -----------------------------------------------------------------------
|
||||
# Invalid transitions
|
||||
# -----------------------------------------------------------------------
|
||||
|
||||
class TestInvalidTransitions:
|
||||
|
||||
@pytest.mark.parametrize("from_status,to_status", [
|
||||
(TaskStatus.PENDING, "undergoing"),
|
||||
(TaskStatus.PENDING, "completed"),
|
||||
(TaskStatus.OPEN, "completed"),
|
||||
(TaskStatus.OPEN, "pending"),
|
||||
(TaskStatus.UNDERGOING, "open"),
|
||||
(TaskStatus.UNDERGOING, "pending"),
|
||||
(TaskStatus.COMPLETED, "undergoing"),
|
||||
(TaskStatus.COMPLETED, "closed"),
|
||||
(TaskStatus.CLOSED, "undergoing"),
|
||||
(TaskStatus.CLOSED, "completed"),
|
||||
])
|
||||
def test_disallowed_transition(
|
||||
self, from_status, to_status,
|
||||
client, db, make_user, make_project, make_milestone,
|
||||
make_task, seed_roles_and_permissions, make_member, auth_header,
|
||||
):
|
||||
"""State machine rejects transitions not in VALID_TRANSITIONS."""
|
||||
_, mgr_role, _ = seed_roles_and_permissions
|
||||
user = make_user()
|
||||
project = make_project(owner_id=user.id)
|
||||
make_member(project.id, user.id, mgr_role.id)
|
||||
ms = make_milestone(project.id, user.id, status=MilestoneStatus.UNDERGOING)
|
||||
task = make_task(
|
||||
project.id, ms.id, user.id,
|
||||
status=from_status,
|
||||
assignee_id=user.id,
|
||||
)
|
||||
|
||||
resp = _transition(client, task.id, to_status, auth_header(user))
|
||||
assert resp.status_code == 400
|
||||
assert "cannot transition" in resp.json()["detail"].lower()
|
||||
|
||||
|
||||
# -----------------------------------------------------------------------
|
||||
# Edit restrictions (PATCH)
|
||||
# -----------------------------------------------------------------------
|
||||
|
||||
class TestEditRestrictions:
|
||||
|
||||
def test_undergoing_body_edit_blocked(
|
||||
self, client, db, make_user, make_project, make_milestone,
|
||||
make_task, seed_roles_and_permissions, make_member, auth_header,
|
||||
):
|
||||
"""Cannot PATCH body fields on an undergoing task."""
|
||||
_, mgr_role, _ = seed_roles_and_permissions
|
||||
user = make_user()
|
||||
project = make_project(owner_id=user.id)
|
||||
make_member(project.id, user.id, mgr_role.id)
|
||||
ms = make_milestone(project.id, user.id, status=MilestoneStatus.UNDERGOING)
|
||||
task = make_task(project.id, ms.id, user.id, status=TaskStatus.UNDERGOING, assignee_id=user.id)
|
||||
|
||||
resp = client.patch(
|
||||
f"/tasks/{task.id}",
|
||||
json={"title": "New Title"},
|
||||
headers=auth_header(user),
|
||||
)
|
||||
assert resp.status_code == 400
|
||||
assert "undergoing" in resp.json()["detail"].lower()
|
||||
|
||||
def test_completed_body_edit_blocked(
|
||||
self, client, db, make_user, make_project, make_milestone,
|
||||
make_task, seed_roles_and_permissions, make_member, auth_header,
|
||||
):
|
||||
"""Cannot PATCH body fields on a completed task."""
|
||||
_, mgr_role, _ = seed_roles_and_permissions
|
||||
user = make_user()
|
||||
project = make_project(owner_id=user.id)
|
||||
make_member(project.id, user.id, mgr_role.id)
|
||||
ms = make_milestone(project.id, user.id, status=MilestoneStatus.UNDERGOING)
|
||||
task = make_task(project.id, ms.id, user.id, status=TaskStatus.COMPLETED)
|
||||
|
||||
resp = client.patch(
|
||||
f"/tasks/{task.id}",
|
||||
json={"title": "Changed"},
|
||||
headers=auth_header(user),
|
||||
)
|
||||
assert resp.status_code == 400
|
||||
|
||||
def test_open_assignee_only_edit(
|
||||
self, client, db, make_user, make_project, make_milestone,
|
||||
make_task, seed_roles_and_permissions, make_member, auth_header,
|
||||
):
|
||||
"""Open task with assignee: only assignee can edit body."""
|
||||
_, mgr_role, _ = seed_roles_and_permissions
|
||||
owner = make_user()
|
||||
other = make_user()
|
||||
project = make_project(owner_id=owner.id)
|
||||
make_member(project.id, owner.id, mgr_role.id)
|
||||
make_member(project.id, other.id, mgr_role.id)
|
||||
ms = make_milestone(project.id, owner.id, status=MilestoneStatus.UNDERGOING)
|
||||
task = make_task(
|
||||
project.id, ms.id, owner.id,
|
||||
status=TaskStatus.OPEN,
|
||||
assignee_id=owner.id,
|
||||
)
|
||||
|
||||
# Other user cannot edit
|
||||
resp = client.patch(
|
||||
f"/tasks/{task.id}",
|
||||
json={"title": "Hijack"},
|
||||
headers=auth_header(other),
|
||||
)
|
||||
assert resp.status_code == 403
|
||||
|
||||
# Assignee can edit
|
||||
resp = client.patch(
|
||||
f"/tasks/{task.id}",
|
||||
json={"title": "My Change"},
|
||||
headers=auth_header(owner),
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
assert resp.json()["title"] == "My Change"
|
||||
|
||||
def test_open_no_assignee_anyone_edits(
|
||||
self, client, db, make_user, make_project, make_milestone,
|
||||
make_task, seed_roles_and_permissions, make_member, auth_header,
|
||||
):
|
||||
"""Open task without assignee: any project member can edit."""
|
||||
_, mgr_role, _ = seed_roles_and_permissions
|
||||
user = make_user()
|
||||
project = make_project(owner_id=user.id)
|
||||
make_member(project.id, user.id, mgr_role.id)
|
||||
ms = make_milestone(project.id, user.id, status=MilestoneStatus.UNDERGOING)
|
||||
task = make_task(
|
||||
project.id, ms.id, user.id,
|
||||
status=TaskStatus.OPEN,
|
||||
assignee_id=None,
|
||||
)
|
||||
|
||||
resp = client.patch(
|
||||
f"/tasks/{task.id}",
|
||||
json={"title": "Anyone's Change"},
|
||||
headers=auth_header(user),
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
assert resp.json()["title"] == "Anyone's Change"
|
||||
211
tests/test_tasks.py
Normal file
211
tests/test_tasks.py
Normal file
@@ -0,0 +1,211 @@
|
||||
"""P14.1 — Tasks CRUD API tests.
|
||||
|
||||
Covers:
|
||||
- List tasks (project-scoped, milestone-scoped)
|
||||
- Get task by ID
|
||||
- Create task
|
||||
- Update task
|
||||
- Delete task
|
||||
- Task filtering by status, assignee, etc.
|
||||
"""
|
||||
import pytest
|
||||
from datetime import datetime, timedelta
|
||||
|
||||
|
||||
class TestTasksCRUD:
|
||||
"""Task CRUD endpoints."""
|
||||
|
||||
def test_list_tasks(self, client, db, make_user, make_project, auth_header, seed_roles_and_permissions, make_member):
|
||||
"""List tasks for a project."""
|
||||
admin_role, mgr_role, dev_role = seed_roles_and_permissions
|
||||
user = make_user()
|
||||
project = make_project()
|
||||
make_member(project.id, user.id, dev_role.id)
|
||||
|
||||
# Create milestone and tasks
|
||||
from app.models.milestone import Milestone, MilestoneStatus
|
||||
from app.models.task import Task, TaskStatus, TaskPriority
|
||||
milestone = Milestone(title="M1", project_id=project.id, status=MilestoneStatus.OPEN)
|
||||
db.add(milestone)
|
||||
db.commit()
|
||||
|
||||
task1 = Task(
|
||||
title="Task 1", project_id=project.id, milestone_id=milestone.id,
|
||||
reporter_id=user.id, status=TaskStatus.OPEN, priority=TaskPriority.MEDIUM
|
||||
)
|
||||
task2 = Task(
|
||||
title="Task 2", project_id=project.id, milestone_id=milestone.id,
|
||||
reporter_id=user.id, status=TaskStatus.OPEN, priority=TaskPriority.MEDIUM
|
||||
)
|
||||
db.add_all([task1, task2])
|
||||
db.commit()
|
||||
|
||||
resp = client.get(f"/projects/{project.id}/tasks", headers=auth_header(user))
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert len(data) >= 2
|
||||
|
||||
def test_get_task_by_id(self, client, db, make_user, make_project, auth_header, seed_roles_and_permissions, make_member):
|
||||
"""Get specific task."""
|
||||
admin_role, mgr_role, dev_role = seed_roles_and_permissions
|
||||
user = make_user()
|
||||
project = make_project()
|
||||
make_member(project.id, user.id, dev_role.id)
|
||||
|
||||
from app.models.milestone import Milestone, MilestoneStatus
|
||||
from app.models.task import Task, TaskStatus, TaskPriority
|
||||
milestone = Milestone(title="M1", project_id=project.id, status=MilestoneStatus.OPEN)
|
||||
db.add(milestone)
|
||||
db.commit()
|
||||
|
||||
task = Task(
|
||||
title="Test Task", project_id=project.id, milestone_id=milestone.id,
|
||||
reporter_id=user.id, status=TaskStatus.OPEN, priority=TaskPriority.HIGH,
|
||||
task_type="issue"
|
||||
)
|
||||
db.add(task)
|
||||
db.commit()
|
||||
|
||||
resp = client.get(f"/projects/{project.id}/tasks/{task.id}", headers=auth_header(user))
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert data["id"] == task.id
|
||||
assert data["title"] == "Test Task"
|
||||
assert data["task_type"] == "issue"
|
||||
|
||||
def test_create_task(self, client, db, make_user, make_project, auth_header, seed_roles_and_permissions, make_member):
|
||||
"""Create new task."""
|
||||
admin_role, mgr_role, dev_role = seed_roles_and_permissions
|
||||
user = make_user()
|
||||
project = make_project(project_code="PROJ")
|
||||
make_member(project.id, user.id, dev_role.id)
|
||||
|
||||
from app.models.milestone import Milestone, MilestoneStatus
|
||||
milestone = Milestone(title="M1", project_id=project.id, status=MilestoneStatus.OPEN)
|
||||
db.add(milestone)
|
||||
db.commit()
|
||||
|
||||
resp = client.post(
|
||||
f"/projects/{project.id}/tasks",
|
||||
json={
|
||||
"title": "New Task",
|
||||
"description": "Task description",
|
||||
"milestone_id": milestone.id,
|
||||
"task_type": "issue",
|
||||
"priority": "high"
|
||||
},
|
||||
headers=auth_header(user)
|
||||
)
|
||||
assert resp.status_code == 201
|
||||
data = resp.json()
|
||||
assert data["title"] == "New Task"
|
||||
assert data["status"] == "open"
|
||||
assert data["task_code"].startswith("PROJ:")
|
||||
|
||||
def test_update_task(self, client, db, make_user, make_project, auth_header, seed_roles_and_permissions, make_member):
|
||||
"""Update task."""
|
||||
admin_role, mgr_role, dev_role = seed_roles_and_permissions
|
||||
user = make_user()
|
||||
project = make_project()
|
||||
make_member(project.id, user.id, dev_role.id)
|
||||
|
||||
from app.models.milestone import Milestone, MilestoneStatus
|
||||
from app.models.task import Task, TaskStatus, TaskPriority
|
||||
milestone = Milestone(title="M1", project_id=project.id, status=MilestoneStatus.OPEN)
|
||||
db.add(milestone)
|
||||
db.commit()
|
||||
|
||||
task = Task(
|
||||
title="Old Title", project_id=project.id, milestone_id=milestone.id,
|
||||
reporter_id=user.id, status=TaskStatus.OPEN, priority=TaskPriority.MEDIUM
|
||||
)
|
||||
db.add(task)
|
||||
db.commit()
|
||||
|
||||
resp = client.patch(
|
||||
f"/projects/{project.id}/tasks/{task.id}",
|
||||
json={"title": "Updated Title"},
|
||||
headers=auth_header(user)
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert data["title"] == "Updated Title"
|
||||
|
||||
def test_delete_task(self, client, db, make_user, make_project, auth_header, seed_roles_and_permissions, make_member):
|
||||
"""Delete task."""
|
||||
admin_role, mgr_role, dev_role = seed_roles_and_permissions
|
||||
user = make_user()
|
||||
project = make_project()
|
||||
make_member(project.id, user.id, dev_role.id)
|
||||
|
||||
from app.models.milestone import Milestone, MilestoneStatus
|
||||
from app.models.task import Task, TaskStatus, TaskPriority
|
||||
milestone = Milestone(title="M1", project_id=project.id, status=MilestoneStatus.OPEN)
|
||||
db.add(milestone)
|
||||
db.commit()
|
||||
|
||||
task = Task(
|
||||
title="To Delete", project_id=project.id, milestone_id=milestone.id,
|
||||
reporter_id=user.id, status=TaskStatus.OPEN, priority=TaskPriority.MEDIUM
|
||||
)
|
||||
db.add(task)
|
||||
db.commit()
|
||||
|
||||
resp = client.delete(f"/projects/{project.id}/tasks/{task.id}", headers=auth_header(user))
|
||||
assert resp.status_code == 204
|
||||
|
||||
def test_task_filter_by_status(self, client, db, make_user, make_project, auth_header, seed_roles_and_permissions, make_member):
|
||||
"""Filter tasks by status."""
|
||||
admin_role, mgr_role, dev_role = seed_roles_and_permissions
|
||||
user = make_user()
|
||||
project = make_project()
|
||||
make_member(project.id, user.id, dev_role.id)
|
||||
|
||||
from app.models.milestone import Milestone, MilestoneStatus
|
||||
from app.models.task import Task, TaskStatus, TaskPriority
|
||||
milestone = Milestone(title="M1", project_id=project.id, status=MilestoneStatus.OPEN)
|
||||
db.add(milestone)
|
||||
db.commit()
|
||||
|
||||
open_task = Task(
|
||||
title="Open", project_id=project.id, milestone_id=milestone.id,
|
||||
reporter_id=user.id, status=TaskStatus.OPEN, priority=TaskPriority.MEDIUM
|
||||
)
|
||||
closed_task = Task(
|
||||
title="Closed", project_id=project.id, milestone_id=milestone.id,
|
||||
reporter_id=user.id, status=TaskStatus.CLOSED, priority=TaskPriority.MEDIUM
|
||||
)
|
||||
db.add_all([open_task, closed_task])
|
||||
db.commit()
|
||||
|
||||
resp = client.get(f"/projects/{project.id}/tasks?status=open", headers=auth_header(user))
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert all(t["status"] == "open" for t in data)
|
||||
|
||||
def test_task_filter_by_assignee(self, client, db, make_user, make_project, auth_header, seed_roles_and_permissions, make_member):
|
||||
"""Filter tasks by assignee."""
|
||||
admin_role, mgr_role, dev_role = seed_roles_and_permissions
|
||||
user = make_user()
|
||||
assignee = make_user(username="assignee")
|
||||
project = make_project()
|
||||
make_member(project.id, user.id, dev_role.id)
|
||||
|
||||
from app.models.milestone import Milestone, MilestoneStatus
|
||||
from app.models.task import Task, TaskStatus, TaskPriority
|
||||
milestone = Milestone(title="M1", project_id=project.id, status=MilestoneStatus.OPEN)
|
||||
db.add(milestone)
|
||||
db.commit()
|
||||
|
||||
assigned_task = Task(
|
||||
title="Assigned", project_id=project.id, milestone_id=milestone.id,
|
||||
reporter_id=user.id, assignee_id=assignee.id,
|
||||
status=TaskStatus.OPEN, priority=TaskPriority.MEDIUM
|
||||
)
|
||||
db.add(assigned_task)
|
||||
db.commit()
|
||||
|
||||
resp = client.get(f"/projects/{project.id}/tasks?assignee_id={assignee.id}", headers=auth_header(user))
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert all(t["assignee_id"] == assignee.id for t in data)
|
||||
100
tests/test_users.py
Normal file
100
tests/test_users.py
Normal file
@@ -0,0 +1,100 @@
|
||||
"""P14.1 — Users API tests.
|
||||
|
||||
Covers:
|
||||
- List users
|
||||
- Get user by ID
|
||||
- Create user
|
||||
- Update user
|
||||
- Delete user
|
||||
- User self-service restrictions
|
||||
"""
|
||||
import pytest
|
||||
|
||||
|
||||
class TestUsers:
|
||||
"""User management endpoints."""
|
||||
|
||||
def test_list_users(self, client, db, make_user, auth_header, seed_roles_and_permissions):
|
||||
"""Admin can list all users."""
|
||||
seed_roles_and_permissions
|
||||
admin = make_user(username="admin", is_admin=True)
|
||||
make_user(username="user1")
|
||||
make_user(username="user2")
|
||||
|
||||
resp = client.get("/users", headers=auth_header(admin))
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert len(data) >= 2
|
||||
|
||||
def test_get_user_by_id(self, client, db, make_user, auth_header, seed_roles_and_permissions):
|
||||
"""Get specific user details."""
|
||||
seed_roles_and_permissions
|
||||
admin = make_user(username="admin", is_admin=True)
|
||||
user = make_user(username="testuser")
|
||||
|
||||
resp = client.get(f"/users/{user.id}", headers=auth_header(admin))
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert data["id"] == user.id
|
||||
assert data["username"] == "testuser"
|
||||
|
||||
def test_create_user(self, client, db, make_user, auth_header, seed_roles_and_permissions):
|
||||
"""Admin can create new user."""
|
||||
seed_roles_and_permissions
|
||||
admin = make_user(username="admin", is_admin=True)
|
||||
|
||||
resp = client.post(
|
||||
"/users",
|
||||
json={
|
||||
"username": "newuser",
|
||||
"password": "newpass123",
|
||||
"is_admin": False
|
||||
},
|
||||
headers=auth_header(admin)
|
||||
)
|
||||
assert resp.status_code == 201
|
||||
data = resp.json()
|
||||
assert data["username"] == "newuser"
|
||||
assert "id" in data
|
||||
|
||||
def test_update_user(self, client, db, make_user, auth_header, seed_roles_and_permissions):
|
||||
"""Admin can update user."""
|
||||
seed_roles_and_permissions
|
||||
admin = make_user(username="admin", is_admin=True)
|
||||
user = make_user(username="testuser")
|
||||
|
||||
resp = client.patch(
|
||||
f"/users/{user.id}",
|
||||
json={"username": "updateduser"},
|
||||
headers=auth_header(admin)
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert data["username"] == "updateduser"
|
||||
|
||||
def test_delete_user(self, client, db, make_user, auth_header, seed_roles_and_permissions):
|
||||
"""Admin can delete user."""
|
||||
seed_roles_and_permissions
|
||||
admin = make_user(username="admin", is_admin=True)
|
||||
user = make_user(username="testuser")
|
||||
|
||||
resp = client.delete(f"/users/{user.id}", headers=auth_header(admin))
|
||||
assert resp.status_code == 204
|
||||
|
||||
def test_regular_user_cannot_list_all_users(self, client, db, make_user, auth_header, seed_roles_and_permissions):
|
||||
"""Non-admin cannot list all users."""
|
||||
seed_roles_and_permissions
|
||||
user = make_user(username="regular")
|
||||
|
||||
resp = client.get("/users", headers=auth_header(user))
|
||||
assert resp.status_code == 403
|
||||
|
||||
def test_user_can_view_self(self, client, db, make_user, auth_header, seed_roles_and_permissions):
|
||||
"""User can view their own profile."""
|
||||
seed_roles_and_permissions
|
||||
user = make_user(username="testuser")
|
||||
|
||||
resp = client.get(f"/users/{user.id}", headers=auth_header(user))
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert data["id"] == user.id
|
||||
Reference in New Issue
Block a user